Replaced state.streamsInWhichIAmParticipating with pool.streams #2226
|
@ -237,8 +237,7 @@ class Main(object):
|
||||||
upnpThread = upnp.uPnPThread()
|
upnpThread = upnp.uPnPThread()
|
||||||
upnpThread.start()
|
upnpThread.start()
|
||||||
else:
|
else:
|
||||||
# Populate with hardcoded value (same as connectToStream above)
|
network.connectionpool.pool.connectToStream(1)
|
||||||
state.streamsInWhichIAmParticipating.append(1)
|
|
||||||
|
|
||||||
if not daemon and state.enableGUI:
|
if not daemon and state.enableGUI:
|
||||||
|
|||||||
if state.curses:
|
if state.curses:
|
||||||
|
|
|
@ -20,7 +20,7 @@ import widgets
|
||||||
from bmconfigparser import config as config_obj
|
from bmconfigparser import config as config_obj
|
||||||
from helper_sql import sqlExecute, sqlStoredProcedure
|
from helper_sql import sqlExecute, sqlStoredProcedure
|
||||||
from helper_startup import start_proxyconfig
|
from helper_startup import start_proxyconfig
|
||||||
from network import knownnodes
|
from network import connectionpool, knownnodes
|
||||||
from network.announcethread import AnnounceThread
|
from network.announcethread import AnnounceThread
|
||||||
from network.asyncore_pollchoose import set_rates
|
from network.asyncore_pollchoose import set_rates
|
||||||
from tr import _translate
|
from tr import _translate
|
||||||
|
@ -165,7 +165,7 @@ class SettingsDialog(QtGui.QDialog):
|
||||||
if self._proxy_type:
|
if self._proxy_type:
|
||||||
for node, info in six.iteritems(
|
for node, info in six.iteritems(
|
||||||
knownnodes.knownNodes.get(
|
knownnodes.knownNodes.get(
|
||||||
min(state.streamsInWhichIAmParticipating), [])
|
min(connectionpool.pool.streams), [])
|
||||||
):
|
):
|
||||||
if (
|
if (
|
||||||
node.host.endswith('.onion') and len(node.host) > 22
|
node.host.endswith('.onion') and len(node.host) > 22
|
||||||
|
|
|
@ -108,7 +108,7 @@ class singleCleaner(StoppableThread):
|
||||||
# Cleanup knownnodes and handle possible severe exception
|
# Cleanup knownnodes and handle possible severe exception
|
||||||
# while writing it to disk
|
# while writing it to disk
|
||||||
if state.enableNetwork:
|
if state.enableNetwork:
|
||||||
knownnodes.cleanupKnownNodes()
|
knownnodes.cleanupKnownNodes(connectionpool.pool)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
if "Errno 28" in str(err):
|
if "Errno 28" in str(err):
|
||||||
self.logger.fatal(
|
self.logger.fatal(
|
||||||
|
|
|
@ -4,7 +4,6 @@ Announce myself (node address)
|
||||||
import time
|
import time
|
||||||
|
|
||||||
# magic imports!
|
# magic imports!
|
||||||
import state
|
|
||||||
import connectionpool
|
import connectionpool
|
||||||
from bmconfigparser import config
|
from bmconfigparser import config
|
||||||
from protocol import assembleAddrMessage
|
from protocol import assembleAddrMessage
|
||||||
|
@ -34,7 +33,7 @@ class AnnounceThread(StoppableThread):
|
||||||
for connection in connectionpool.pool.udpSockets.values():
|
for connection in connectionpool.pool.udpSockets.values():
|
||||||
if not connection.announcing:
|
if not connection.announcing:
|
||||||
continue
|
continue
|
||||||
for stream in state.streamsInWhichIAmParticipating:
|
for stream in connectionpool.pool.streams:
|
||||||
addr = (
|
addr = (
|
||||||
stream,
|
stream,
|
||||||
Peer(
|
Peer(
|
||||||
|
|
|
@ -6,6 +6,7 @@ import time
|
||||||
|
|
||||||
import protocol
|
import protocol
|
||||||
import state
|
import state
|
||||||
|
import connectionpool
|
||||||
from highlevelcrypto import calculateInventoryHash
|
from highlevelcrypto import calculateInventoryHash
|
||||||
|
|
||||||
logger = logging.getLogger('default')
|
logger = logging.getLogger('default')
|
||||||
|
@ -98,7 +99,7 @@ class BMObject(object): # pylint: disable=too-many-instance-attributes
|
||||||
logger.warning(
|
logger.warning(
|
||||||
'The object has invalid stream: %s', self.streamNumber)
|
'The object has invalid stream: %s', self.streamNumber)
|
||||||
raise BMObjectInvalidError()
|
raise BMObjectInvalidError()
|
||||||
if self.streamNumber not in state.streamsInWhichIAmParticipating:
|
if self.streamNumber not in connectionpool.pool.streams:
|
||||||
logger.debug(
|
logger.debug(
|
||||||
'The streamNumber %i isn\'t one we are interested in.',
|
'The streamNumber %i isn\'t one we are interested in.',
|
||||||
self.streamNumber)
|
self.streamNumber)
|
||||||
|
|
|
@ -445,7 +445,7 @@ class BMProto(AdvancedDispatcher, ObjectTracker):
|
||||||
for seenTime, stream, _, ip, port in self._decode_addr():
|
for seenTime, stream, _, ip, port in self._decode_addr():
|
||||||
ip = str(ip)
|
ip = str(ip)
|
||||||
if (
|
if (
|
||||||
stream not in state.streamsInWhichIAmParticipating
|
stream not in connectionpool.pool.streams
|
||||||
# FIXME: should check against complete list
|
# FIXME: should check against complete list
|
||||||
or ip.startswith('bootstrap')
|
or ip.startswith('bootstrap')
|
||||||
):
|
):
|
||||||
|
|
|
@ -88,7 +88,6 @@ class BMConnectionPool(object):
|
||||||
def connectToStream(self, streamNumber):
|
def connectToStream(self, streamNumber):
|
||||||
"""Connect to a bitmessage stream"""
|
"""Connect to a bitmessage stream"""
|
||||||
self.streams.append(streamNumber)
|
self.streams.append(streamNumber)
|
||||||
state.streamsInWhichIAmParticipating.append(streamNumber)
|
|
||||||
|
|
||||||
def getConnectionByAddr(self, addr):
|
def getConnectionByAddr(self, addr):
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -226,7 +226,7 @@ def dns():
|
||||||
1, Peer('bootstrap%s.bitmessage.org' % port, port))
|
1, Peer('bootstrap%s.bitmessage.org' % port, port))
|
||||||
|
|
||||||
|
|
||||||
def cleanupKnownNodes():
|
def cleanupKnownNodes(pool):
|
||||||
"""
|
"""
|
||||||
Cleanup knownnodes: remove old nodes and nodes with low rating
|
Cleanup knownnodes: remove old nodes and nodes with low rating
|
||||||
"""
|
"""
|
||||||
|
@ -236,7 +236,7 @@ def cleanupKnownNodes():
|
||||||
|
|
||||||
with knownNodesLock:
|
with knownNodesLock:
|
||||||
for stream in knownNodes:
|
for stream in knownNodes:
|
||||||
if stream not in state.streamsInWhichIAmParticipating:
|
if stream not in pool.streams:
|
||||||
continue
|
continue
|
||||||
keys = knownNodes[stream].keys()
|
keys = knownNodes[stream].keys()
|
||||||
for node in keys:
|
for node in keys:
|
||||||
|
|
|
@ -8,6 +8,7 @@ import time
|
||||||
# magic imports!
|
# magic imports!
|
||||||
import protocol
|
import protocol
|
||||||
import state
|
import state
|
||||||
|
import connectionpool
|
||||||
from queues import receiveDataQueue
|
from queues import receiveDataQueue
|
||||||
|
|
||||||
from bmproto import BMProto
|
from bmproto import BMProto
|
||||||
|
@ -81,7 +82,7 @@ class UDPSocket(BMProto): # pylint: disable=too-many-instance-attributes
|
||||||
remoteport = False
|
remoteport = False
|
||||||
for seenTime, stream, _, ip, port in addresses:
|
for seenTime, stream, _, ip, port in addresses:
|
||||||
decodedIP = protocol.checkIPAddress(str(ip))
|
decodedIP = protocol.checkIPAddress(str(ip))
|
||||||
if stream not in state.streamsInWhichIAmParticipating:
|
if stream not in connectionpool.pool.streams:
|
||||||
continue
|
continue
|
||||||
if (seenTime < time.time() - protocol.MAX_TIME_OFFSET
|
if (seenTime < time.time() - protocol.MAX_TIME_OFFSET
|
||||||
or seenTime > time.time() + protocol.MAX_TIME_OFFSET):
|
or seenTime > time.time() + protocol.MAX_TIME_OFFSET):
|
||||||
|
|
|
@ -3,7 +3,6 @@ Global runtime variables.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
neededPubkeys = {}
|
neededPubkeys = {}
|
||||||
streamsInWhichIAmParticipating = []
|
|
||||||
|
|
||||||
extPort = None
|
extPort = None
|
||||||
"""For UPnP"""
|
"""For UPnP"""
|
||||||
|
|
|
@ -165,7 +165,7 @@ class TestCore(unittest.TestCase):
|
||||||
"""test knownnodes starvation leading to IndexError in Asyncore"""
|
"""test knownnodes starvation leading to IndexError in Asyncore"""
|
||||||
self._outdate_knownnodes()
|
self._outdate_knownnodes()
|
||||||
# time.sleep(303) # singleCleaner wakes up every 5 min
|
# time.sleep(303) # singleCleaner wakes up every 5 min
|
||||||
knownnodes.cleanupKnownNodes()
|
knownnodes.cleanupKnownNodes(connectionpool.pool)
|
||||||
self.assertTrue(knownnodes.knownNodes[1])
|
self.assertTrue(knownnodes.knownNodes[1])
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
|
@ -179,7 +179,7 @@ class TestCore(unittest.TestCase):
|
||||||
config.set('bitmessagesettings', 'dontconnect', 'true')
|
config.set('bitmessagesettings', 'dontconnect', 'true')
|
||||||
self._wipe_knownnodes()
|
self._wipe_knownnodes()
|
||||||
knownnodes.addKnownNode(1, Peer('127.0.0.1', 8444), is_self=True)
|
knownnodes.addKnownNode(1, Peer('127.0.0.1', 8444), is_self=True)
|
||||||
knownnodes.cleanupKnownNodes()
|
knownnodes.cleanupKnownNodes(connectionpool.pool)
|
||||||
time.sleep(5)
|
time.sleep(5)
|
||||||
|
|
||||||
def _check_connection(self, full=False):
|
def _check_connection(self, full=False):
|
||||||
|
|
Reference in New Issue
Block a user
This makes difference in streamsInWhichIAmParticipating & pool streams