Revert "broadcast shares in serial", strongly suspected of causing a memory leak

[p2pool.git] / p2pool / util / p2protocol.py

diff --git a/p2pool/util/p2protocol.py b/p2pool/util/p2protocol.py
index 1886713..28ace01 100644 (file)
--- a/p2pool/util/p2protocol.py
+++ b/p2pool/util/p2protocol.py
@@ -9,16 +9,21 @@ from twisted.internet import protocol
 from twisted.python import log
 
 import p2pool
-from p2pool.util import datachunker
+from p2pool.util import datachunker, variable
 
 class TooLong(Exception):
     pass
 
 class Protocol(protocol.Protocol):
-    def __init__(self, message_prefix, max_payload_length):
+    def __init__(self, message_prefix, max_payload_length, traffic_happened=variable.Event()):
         self._message_prefix = message_prefix
         self._max_payload_length = max_payload_length
-        self.dataReceived = datachunker.DataChunker(self.dataReceiver())
+        self.dataReceived2 = datachunker.DataChunker(self.dataReceiver())
+        self.traffic_happened = traffic_happened
+    
+    def dataReceived(self, data):
+        self.traffic_happened.happened('p2p/in', len(data))
+        self.dataReceived2(data)
     
     def dataReceiver(self):
         while True:
@@ -36,6 +41,7 @@ class Protocol(protocol.Protocol):
             
             if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum:
                 print 'invalid hash for', self.transport.getPeer().host, repr(command), length, checksum.encode('hex'), hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4].encode('hex'), payload.encode('hex')
+                self.badPeerHappened()
                 continue
             
             type_ = getattr(self, 'message_' + command, None)
@@ -49,7 +55,7 @@ class Protocol(protocol.Protocol):
             except:
                 print 'RECV', command, payload[:100].encode('hex') + ('...' if len(payload) > 100 else '')
                 log.err(None, 'Error handling message: (see RECV line)')
-                self.badPeerHappened()
+                self.transport.loseConnection()
     
     def packetReceived(self, command, payload2):
         handler = getattr(self, 'handle_' + command, None)
@@ -58,7 +64,8 @@ class Protocol(protocol.Protocol):
                 print 'no handler for', repr(command)
             return
         
-        handler(**payload2)
+        if getattr(self, 'connected', True) and not getattr(self, 'disconnecting', False):
+            handler(**payload2)
     
     def badPeerHappened(self):
         self.transport.loseConnection()
@@ -73,7 +80,9 @@ class Protocol(protocol.Protocol):
         payload = type_.pack(payload2)
         if len(payload) > self._max_payload_length:
             raise TooLong('payload too long')
-        self.transport.write(self._message_prefix + struct.pack('<12sI', command, len(payload)) + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] + payload)
+        data = self._message_prefix + struct.pack('<12sI', command, len(payload)) + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] + payload
+        self.traffic_happened.happened('p2p/out', len(data))
+        self.transport.write(data)
     
     def __getattr__(self, attr):
         prefix = 'send_'