PyBitmessage/src/network/invthread.py
Peter Surda ba4162d7fe
Asyncore update
- 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)
2017-07-06 19:45:36 +02:00

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)