added sharereq/sharereply p2p messages with id field to allow more robust share reque...

diff --git a/p2pool/main.py b/p2pool/main.py
index 2514766..838912f 100644 (file)
--- a/p2pool/main.py
+++ b/p2pool/main.py
@@ -293,7 +293,7 @@ def main(args, net, datadir_path, merged_urls, worker_endpoint):
                             break
                         shares.append(share)
                 print 'Sending %i shares to %s:%i' % (len(shares), peer.addr[0], peer.addr[1])
-                peer.sendShares(shares)
+                return shares
         
         @deferral.retry('Error submitting block: (will retry)', 10, 10)
         @defer.inlineCallbacks

diff --git a/p2pool/p2p.py b/p2pool/p2p.py
index 1dddcde..31a6d8c 100644 (file)
--- a/p2pool/p2p.py
+++ b/p2pool/p2p.py
@@ -196,7 +196,7 @@ class Protocol(p2protocol.Protocol):
         ('stops', pack.ListType(pack.IntType(256))),
     ])
     def handle_getshares(self, hashes, parents, stops):
-        self.node.handle_get_shares(hashes, parents, stops, self)
+        self.sendShares(self.node.handle_get_shares(hashes, parents, stops, self))
     
     message_shares = pack.ComposedType([
         ('shares', pack.ListType(p2pool_data.share_type)),
@@ -214,6 +214,29 @@ class Protocol(p2protocol.Protocol):
         if shares:
             att(self.send_shares, shares=[share.as_share() for share in shares])
     
+    
+    message_sharereq = pack.ComposedType([
+        ('id', pack.IntType(256)),
+        ('hashes', pack.ListType(pack.IntType(256))),
+        ('parents', pack.VarIntType()),
+        ('stops', pack.ListType(pack.IntType(256))),
+    ])
+    def handle_sharereq(self, id, hashes, parents, stops):
+        shares = self.node.handle_get_shares(hashes, parents, stops, self)
+        try:
+            self.send_sharereply(id=id, result='good', shares=[share.as_share() for share in shares])
+        except p2protocol.TooLong:
+            self.send_sharereply(id=id, result='too long', shares=[])
+    
+    message_sharereply = pack.ComposedType([
+        ('id', pack.IntType(256)),
+        ('result', pack.EnumType(pack.VarIntType(), {'good': 0, 'too long': 1, 'unk2': 2, 'unk2': 3, 'unk2': 4, 'unk2': 5, 'unk2': 6})),
+        ('shares', pack.ListType(p2pool_data.share_type)),
+    ])
+    def handle_sharereply(self, id, result, shares):
+        self.node.handle_share_reply(id, result, shares, self)
+    
+    
     def connectionLost(self, reason):
         if self.connected2:
             self.factory.proto_disconnected(self, reason)
@@ -448,6 +471,9 @@ class Node(object):
     def handle_get_shares(self, hashes, parents, stops, peer):
         print 'handle_get_shares', (hashes, parents, stops, peer)
     
+    def handle_share_reply(self, id, result, shares, peer):
+        raise PeerMisbehavingError('sent share reply without being sent a request')
+    
     def get_good_peers(self, max_count):
         t = time.time()
         return [x[0] for x in sorted(self.addr_store.iteritems(), key=lambda (k, (services, first_seen, last_seen)): -max(3600, last_seen - first_seen)/max(3600, t - last_seen)*random.expovariate(1))][:max_count]

diff --git a/p2pool/test/test_p2p.py b/p2pool/test/test_p2p.py
new file mode 100644 (file)
index 0000000..36803cb
--- /dev/null
+++ b/p2pool/test/test_p2p.py
@@ -0,0 +1,25 @@
+import random
+
+from twisted.internet import defer, reactor
+from twisted.trial import unittest
+
+from p2pool import data, networks, p2p
+from p2pool.util import deferral
+
+class MyNode(p2p.Node):
+    def handle_share_hashes(self, hashes, peer):
+        peer.send_sharereq(id=random.randrange(2**256), hashes=[hashes[0]], parents=5, stops=[])
+        print 'handle_share_hashes', (hashes, peer)
+    
+    def handle_share_reply(self, id, result, shares, peer):
+        print (id, result, shares)
+
+class Test(unittest.TestCase):
+    @defer.inlineCallbacks
+    def test_sharereq(self):
+        n = MyNode(lambda: None, 29333, networks.nets['bitcoin'], {}, set([('127.0.0.1', 9333)]), 0, 0, 0, 0)
+        n.start()
+        try:
+            yield deferral.sleep(10)
+        finally:
+            n.stop()