Peter Surda
ba4162d7fe
- get rid of per-connection writeQueue/receiveQueue, and instead use strings and locking - minor code cleanup - all state handlers now should set expectBytes - almost all data processing happens in ReceiveDataThread, and AsyncoreThread is almost only I/O (plus TLS). AsyncoreThread simply puts the connection object into the queue when it has some data for processing - allow poll, epoll and kqueue handlers. kqueue is untested and unoptimised, poll and epoll seem to work ok (linux) - stack depth threshold handler in decode_payload_content, this is recursive and I think was causing occasional RuntimeErrors. Fixes #964 - longer asyncore loops, as now data is handled in ReceiveDataThread - randomise node order when deciding what to download. Should prevent retries being stuck to the same node - socks cleanup (socks5 works ok, socks4a untested but should work too)
49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
import Queue
|
|
import threading
|
|
|
|
import addresses
|
|
from helper_threading import StoppableThread
|
|
from network.connectionpool import BMConnectionPool
|
|
from queues import invQueue
|
|
import protocol
|
|
import state
|
|
|
|
class InvThread(threading.Thread, StoppableThread):
|
|
def __init__(self):
|
|
threading.Thread.__init__(self, name="InvThread")
|
|
self.initStop()
|
|
self.name = "InvThread"
|
|
|
|
def run(self):
|
|
while not state.shutdown:
|
|
chunk = []
|
|
while True:
|
|
try:
|
|
data = invQueue.get(False)
|
|
if len(data) == 2:
|
|
BMConnectionPool().handleReceivedObject(data[0], data[1])
|
|
else:
|
|
BMConnectionPool().handleReceivedObject(data[0], data[1], data[2])
|
|
chunk.append((data[0], data[1]))
|
|
except Queue.Empty:
|
|
break
|
|
|
|
if chunk:
|
|
for connection in BMConnectionPool().inboundConnections.values() + \
|
|
BMConnectionPool().outboundConnections.values():
|
|
hashes = []
|
|
for inv in chunk:
|
|
if inv[0] not in connection.streams:
|
|
continue
|
|
try:
|
|
with connection.objectsNewToThemLock:
|
|
del connection.objectsNewToThem[inv[1]]
|
|
hashes.append(inv[1])
|
|
except KeyError:
|
|
continue
|
|
if hashes:
|
|
connection.append_write_buf(protocol.CreatePacket('inv', \
|
|
addresses.encodeVarint(len(hashes)) + "".join(hashes)))
|
|
invQueue.iterate()
|
|
self.stop.wait(1)
|