I2P tweaks

minode/connection.py

@@ -185,15 +185,15 @@ class Connection(threading.Thread):
             if len(shared.objects) > 0:
                 to_send = {vector for vector in shared.objects.keys() if shared.objects[vector].expires_time > time.time()}
                 while len(to_send) > 0:
-                    if len(to_send) > 50000:
-                        pack = random.sample(to_send, 50000)
+                    if len(to_send) > 10000:
+                        # We limit size of inv messaged to 10000 entries because they might time out in very slow networks (I2P)
+                        pack = random.sample(to_send, 10000)
                         self.send_queue.put(message.Inv(pack))
                         to_send.difference_update(pack)
                     else:
                         self.send_queue.put(message.Inv(to_send))
                         to_send.clear()
         addr = {structure.NetAddr(c.remote_version.services, c.host, c.port) for c in shared.connections.copy() if c.network != 'i2p' and not c.server and c.status == 'fully_established'}
-        addr = set()
         if len(shared.node_pool) > 10:
             addr.update({structure.NetAddr(1, a[0], a[1]) for a in random.sample(shared.node_pool, 10) if a[1] != 'i2p'})
         if len(shared.unchecked_node_pool) > 10:
@@ -307,8 +307,8 @@ class Connection(threading.Thread):
         if self.vectors_to_get:
             self.vectors_to_get.difference_update(shared.objects.keys())
             if self.vectors_to_get:
-                if len(self.vectors_to_get) > 16:
-                    pack = random.sample(self.vectors_to_get, 16)
+                if len(self.vectors_to_get) > 64:
+                    pack = random.sample(self.vectors_to_get, 64)
                     self.send_queue.put(message.GetData(pack))
                     self.vectors_to_get.difference_update(pack)
                 else:
@@ -317,8 +317,8 @@ class Connection(threading.Thread):
 
     def _send_objects(self):
         if self.vectors_to_send:
-            if len(self.vectors_to_send) > 16:
-                to_send = random.sample(self.vectors_to_send, 16)
+            if len(self.vectors_to_send) > 32:
+                to_send = random.sample(self.vectors_to_send, 32)
                 self.vectors_to_send.difference_update(to_send)
             else:
                 to_send = self.vectors_to_send.copy()

minode/i2p/controller.py

@@ -4,6 +4,7 @@ import logging
 import os
 import socket
 import threading
+import time
 
 from i2p.util import receive_line, pub_from_priv
 import shared
@@ -63,9 +64,16 @@ class I2PController(threading.Thread):
         assert self.dest_priv
 
     def create_session(self):
-        self._send(b'SESSION CREATE STYLE=STREAM ID=' + self.nick + b' DESTINATION=' + self.dest_priv + b'\n')
+        self._send(b'SESSION CREATE STYLE=STREAM ID=' + self.nick +
+                   b' inbound.length=' + str(shared.i2p_tunnel_length).encode() +
+                   b' outbound.length=' + str(shared.i2p_tunnel_length).encode() +
+                   b' DESTINATION=' + self.dest_priv + b'\n')
         reply = self._receive_line().split()
-        assert b'RESULT=OK' in reply
+        if b'RESULT=OK' not in reply:
+            logging.warning(reply)
+            logging.warning('We could not create I2P session, retrying in 5 seconds.')
+            time.sleep(5)
+            self.create_session()
 
     def run(self):
         self.s.settimeout(1)

minode/i2p/listener.py

@@ -47,8 +47,7 @@ class I2PListener(threading.Thread):
         while not shared.shutting_down:
             try:
                 destination = self._receive_line().split()[0]
-                print(destination)
-                logging.info('Incoming I2P connection from: {}'.format(destination))
+                logging.info('Incoming I2P connection from: {}'.format(destination.decode()))
                 c = Connection(destination, 'i2p', self.s, 'i2p', True, destination)
                 c.start()
                 shared.connections.add(c)

minode/main.py

@@ -30,6 +30,9 @@ def parse_arguments():
     parser.add_argument('--trusted-peer', help='Specify a trusted peer we should connect to')
     parser.add_argument('--connection-limit', help='Maximum number of connections', type=int)
     parser.add_argument('--i2p', help='Enable I2P support (uses SAMv3)', action='store_true')
+    parser.add_argument('--i2p-tunnel-length', help='Length of I2P tunnels', type=int)
+    parser.add_argument('--i2p-sam-host', help='Host of I2P SAMv3 bridge')
+    parser.add_argument('--i2p-sam-port', help='Port of I2P SAMv3 bridge', type=int)
 
     args = parser.parse_args()
     if args.port:
@@ -65,6 +68,12 @@ def parse_arguments():
         shared.connection_limit = args.connection_limit
     if args.i2p:
         shared.i2p_enabled = True
+    if args.i2p_tunnel_length:
+        shared.i2p_tunnel_length = args.i2p_tunnel_length
+    if args.i2p_sam_host:
+        shared.i2p_sam_host = args.i2p_sam_host
+    if args.i2p_sam_port:
+        shared.i2p_sam_port = args.i2p_sam_port
 
 
 def main():

minode/manager.py

@@ -80,8 +80,11 @@ class Manager(threading.Thread):
                 continue
             if addr[1] == 'i2p' and shared.i2p_enabled:
                 if shared.i2p_session_nick:
-                    c = I2PDialer(addr[0], shared.i2p_session_nick, shared.i2p_sam_host, shared.i2p_sam_port).get_connection()
-                    c.start()
+                    try:
+                        c = I2PDialer(addr[0], shared.i2p_session_nick, shared.i2p_sam_host, shared.i2p_sam_port).get_connection()
+                        c.start()
+                    except:
+                        pass
                 else:
                     logging.debug('We were going to connect to an I2P peer but our tunnels are not ready')
                     continue

minode/shared.py

@@ -26,6 +26,7 @@ header_length = 24
 i2p_enabled = False
 i2p_sam_host = '127.0.0.1'
 i2p_sam_port = 7656
+i2p_tunnel_length = 3
 i2p_session_nick = b''
 i2p_dest_pub = b''