Asyncore update
- request downloads in bigger chunks - don't put whole objects into the receiveDataQueue
This commit is contained in:
parent
0f3a69adf4
commit
2d7d9c2f92
|
@ -11,8 +11,8 @@ from network.connectionpool import BMConnectionPool
|
||||||
import protocol
|
import protocol
|
||||||
|
|
||||||
class DownloadThread(threading.Thread, StoppableThread):
|
class DownloadThread(threading.Thread, StoppableThread):
|
||||||
maxPending = 50
|
maxPending = 200
|
||||||
requestChunk = 100
|
requestChunk = 1000
|
||||||
requestTimeout = 60
|
requestTimeout = 60
|
||||||
cleanInterval = 60
|
cleanInterval = 60
|
||||||
requestExpires = 600
|
requestExpires = 600
|
||||||
|
|
|
@ -24,17 +24,37 @@ class ReceiveQueueThread(threading.Thread, StoppableThread):
|
||||||
def run(self):
|
def run(self):
|
||||||
while not self._stopped and state.shutdown == 0:
|
while not self._stopped and state.shutdown == 0:
|
||||||
try:
|
try:
|
||||||
connection = receiveDataQueue.get(block=True, timeout=1)
|
dest = receiveDataQueue.get(block=True, timeout=1)
|
||||||
receiveDataQueue.task_done()
|
receiveDataQueue.task_done()
|
||||||
except Queue.Empty:
|
except Queue.Empty:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if self._stopped:
|
if self._stopped:
|
||||||
break
|
break
|
||||||
|
|
||||||
# cycle as long as there is data
|
# cycle as long as there is data
|
||||||
# methods should return False if there isn't enough data, or the connection is to be aborted
|
# methods should return False if there isn't enough data, or the connection is to be aborted
|
||||||
|
|
||||||
|
# state_* methods should return False if there isn't enough data,
|
||||||
|
# or the connection is to be aborted
|
||||||
|
|
||||||
try:
|
try:
|
||||||
connection.process()
|
BMConnectionPool().inboundConnections[dest].process()
|
||||||
|
except KeyError:
|
||||||
|
pass
|
||||||
|
except AttributeError:
|
||||||
|
logger.error("Unknown state %s, ignoring", connection.state)
|
||||||
|
|
||||||
|
try:
|
||||||
|
BMConnectionPool().outboundConnections[dest].process()
|
||||||
|
except KeyError:
|
||||||
|
pass
|
||||||
|
except AttributeError:
|
||||||
|
logger.error("Unknown state %s, ignoring", connection.state)
|
||||||
|
|
||||||
|
try:
|
||||||
|
BMConnectionPool().udpSockets[dest].process()
|
||||||
|
except KeyError:
|
||||||
|
pass
|
||||||
except AttributeError:
|
except AttributeError:
|
||||||
# missing command
|
|
||||||
logger.error("Unknown state %s, ignoring", connection.state)
|
logger.error("Unknown state %s, ignoring", connection.state)
|
||||||
|
|
|
@ -190,7 +190,7 @@ class TCPConnection(BMProto, TLSDispatcher):
|
||||||
self.append_write_buf(protocol.assembleVersionMessage(self.destination.host, self.destination.port, network.connectionpool.BMConnectionPool().streams, False))
|
self.append_write_buf(protocol.assembleVersionMessage(self.destination.host, self.destination.port, network.connectionpool.BMConnectionPool().streams, False))
|
||||||
#print "%s:%i: Sending version" % (self.destination.host, self.destination.port)
|
#print "%s:%i: Sending version" % (self.destination.host, self.destination.port)
|
||||||
self.connectedAt = time.time()
|
self.connectedAt = time.time()
|
||||||
receiveDataQueue.put(self)
|
receiveDataQueue.put(self.destination)
|
||||||
|
|
||||||
def handle_read(self):
|
def handle_read(self):
|
||||||
TLSDispatcher.handle_read(self)
|
TLSDispatcher.handle_read(self)
|
||||||
|
@ -201,7 +201,7 @@ class TCPConnection(BMProto, TLSDispatcher):
|
||||||
knownnodes.knownNodes[s][self.destination]["lastseen"] = time.time()
|
knownnodes.knownNodes[s][self.destination]["lastseen"] = time.time()
|
||||||
except KeyError:
|
except KeyError:
|
||||||
pass
|
pass
|
||||||
receiveDataQueue.put(self)
|
receiveDataQueue.put(self.destination)
|
||||||
|
|
||||||
def handle_write(self):
|
def handle_write(self):
|
||||||
TLSDispatcher.handle_write(self)
|
TLSDispatcher.handle_write(self)
|
||||||
|
|
|
@ -169,5 +169,5 @@ class TLSDispatcher(AdvancedDispatcher):
|
||||||
|
|
||||||
self.bm_proto_reset()
|
self.bm_proto_reset()
|
||||||
self.set_state("connection_fully_established")
|
self.set_state("connection_fully_established")
|
||||||
receiveDataQueue.put(self)
|
receiveDataQueue.put(self.destination)
|
||||||
return False
|
return False
|
||||||
|
|
|
@ -139,7 +139,7 @@ class UDPSocket(BMProto):
|
||||||
# overwrite the old buffer to avoid mixing data and so that self.local works correctly
|
# overwrite the old buffer to avoid mixing data and so that self.local works correctly
|
||||||
self.read_buf = recdata
|
self.read_buf = recdata
|
||||||
self.bm_proto_reset()
|
self.bm_proto_reset()
|
||||||
receiveDataQueue.put(self)
|
receiveDataQueue.put(self.destination)
|
||||||
|
|
||||||
def handle_write(self):
|
def handle_write(self):
|
||||||
try:
|
try:
|
||||||
|
|
Reference in New Issue
Block a user