Changeset 0a73303

Timestamp:

May 10, 2019 5:02:30 PM (11 months ago)

Author:

zzz <zzz@…>

Branches:

master

Children:

c1adcfc

Parents:

d8980d1

Message:

NetDB: Store Meta LS2 to floodfills (proposal #123)

Location:

router/java/src/net/i2p/router

Files:

• ClientManagerFacade.java (modified) (2 diffs)
• client/ClientManager.java (modified) (6 diffs)
• client/ClientManagerFacadeImpl.java (modified) (1 diff)
• networkdb/kademlia/StoreJob.java (modified) (4 diffs)

router/java/src/net/i2p/router/ClientManagerFacade.java

@@ -14 +14 @@
 import java.util.Set;
 
+import net.i2p.client.I2PSessionException;
 import net.i2p.crypto.SessionKeyManager;
 import net.i2p.data.Destination;
@@ -105 +106 @@
     /** @since 0.8.8 */
     public abstract void shutdown(String msg);
+
+    /**
+     *  Declare that we're going to publish a meta LS for this destination.
+     *  Must be called before publishing the leaseset.
+     *
+     *  @throws I2PSessionException on duplicate dest
+     *  @since 0.9.41
+     */
+    public void registerMetaDest(Destination dest) throws I2PSessionException {}
+
+    /**
+     *  Declare that we're no longer going to publish a meta LS for this destination.
+     *
+     *  @since 0.9.41
+     */
+    public void unregisterMetaDest(Destination dest) {}
 }

router/java/src/net/i2p/router/client/ClientManager.java

@@ -44 +44 @@
 import net.i2p.router.JobImpl;
 import net.i2p.router.RouterContext;
+import net.i2p.util.ConcurrentHashSet;
 import net.i2p.util.I2PThread;
 import net.i2p.util.Log;
@@ -68 +69 @@
     private final Set<ClientConnectionRunner> _pendingRunners;
     private final Set<SessionId> _runnerSessionIds;
+    private final Set<Destination> _metaDests;
+    private final Set<Hash> _metaHashes;
     protected final RouterContext _ctx;
     protected final int _port;
@@ -101 +104 @@
         //                                      "ClientMessages", 
         //                                      new long[] { 60*1000l, 60*60*1000l, 24*60*60*1000l });
-        _listeners = new ArrayList<ClientListenerRunner>();
-        _runners = new ConcurrentHashMap<Destination, ClientConnectionRunner>();
-        _runnersByHash = new ConcurrentHashMap<Hash, ClientConnectionRunner>();
-        _pendingRunners = new HashSet<ClientConnectionRunner>();
-        _runnerSessionIds = new HashSet<SessionId>();
+        _listeners = new ArrayList<ClientListenerRunner>(4);
+        _runners = new ConcurrentHashMap<Destination, ClientConnectionRunner>(4);
+        _runnersByHash = new ConcurrentHashMap<Hash, ClientConnectionRunner>(4);
+        _pendingRunners = new HashSet<ClientConnectionRunner>(4);
+        _runnerSessionIds = new HashSet<SessionId>(4);
+        _metaDests = new ConcurrentHashSet<Destination>(4);
+        _metaHashes = new ConcurrentHashSet<Hash>(4);
         _port = port;
         _clientTimestamper = new ClientTimestamper();
@@ -369 +374 @@
 
     /**
+     *  Declare that we're going to publish a meta LS for this destination.
+     *  Must be called before publishing the leaseset.
+     *
+     *  @throws I2PSessionException on duplicate dest
+     *  @since 0.9.41
+     */
+    public void registerMetaDest(Destination dest) throws I2PSessionException {
+        synchronized (_runners) {
+            if (_runners.containsKey(dest) || _metaDests.contains(dest)) {
+                String msg = "Client attempted to register duplicate destination " + dest.toBase32();
+                _log.error(msg);
+                throw new I2PSessionException(msg);
+            }
+            _metaDests.add(dest);
+            _metaHashes.add(dest.calculateHash());
+        }
+    }
+
+    /**
+     *  Declare that we're no longer going to publish a meta LS for this destination.
+     *
+     *  @since 0.9.41
+     */
+    public void unregisterMetaDest(Destination dest) {
+        synchronized (_runners) {
+            _metaDests.remove(dest);
+            _metaHashes.remove(dest.calculateHash());
+        }
+    }
+
+    /**
      *  Generate a new random, unused sessionId. Caller must synch on _runners.
      *  @return null on failure
@@ -516 +552 @@
     
     /**
-     *  Unsynchronized
+     *  Unsynchronized.
+     *  DOES contain meta destinations.
      */
     public boolean isLocal(Destination dest) { 
-        return _runners.containsKey(dest);
-    }
-
-    /**
-     *  Unsynchronized
+        return _runners.containsKey(dest) || _metaDests.contains(dest);
+    }
+
+    /**
+     *  Unsynchronized.
+     *  DOES contain meta destinations.
      */
     public boolean isLocal(Hash destHash) { 
         if (destHash == null) return false;
-        return _runnersByHash.containsKey(destHash);
+        return _runnersByHash.containsKey(destHash) || _metaHashes.contains(destHash);
     }
     
@@ -543 +581 @@
 
     /**
-     *  Unsynchronized
+     *  Unsynchronized.
+     *  Does NOT contain meta destinations.
      */
     public Set<Destination> listClients() {

router/java/src/net/i2p/router/client/ClientManagerFacadeImpl.java

@@ -267 +267 @@
         throw new I2PSessionException("No manager yet");
     }
+
+    /**
+     *  Declare that we're going to publish a meta LS for this destination.
+     *  Must be called before publishing the leaseset.
+     *
+     *  @throws I2PSessionException on duplicate dest
+     *  @since 0.9.41
+     */
+    @Override
+    public void registerMetaDest(Destination dest) throws I2PSessionException {
+        if (_manager != null)
+            _manager.registerMetaDest(dest);
+    }
+
+    /**
+     *  Declare that we're no longer going to publish a meta LS for this destination.
+     *
+     *  @since 0.9.41
+     */
+    @Override
+    public void unregisterMetaDest(Destination dest) {
+        if (_manager != null)
+            _manager.unregisterMetaDest(dest);
+    }
 }

router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java

@@ -347 +347 @@
             // if it is an encrypted leaseset...
             if (getContext().keyRing().get(msg.getKey()) != null)
-                sendStoreThroughGarlic(msg, peer, expiration);
+                sendStoreThroughExploratory(msg, peer, expiration);
+            else if (msg.getEntry().getType() == DatabaseEntry.KEY_TYPE_META_LS2)
+                sendWrappedStoreThroughExploratory(msg, peer, expiration);
             else
                 sendStoreThroughClient(msg, peer, expiration);
@@ -356 +358 @@
                 sendDirect(msg, peer, expiration);
             else
-                sendStoreThroughGarlic(msg, peer, expiration);
+                sendStoreThroughExploratory(msg, peer, expiration);
         }
     }
@@ -388 +390 @@
     
     /**
-     * This is misnamed, it means sending it out through an exploratory tunnel,
+     * Send it out through an exploratory tunnel,
      * with the reply to come back through an exploratory tunnel.
      * There is no garlic encryption added.
      *
-     */
-    private void sendStoreThroughGarlic(DatabaseStoreMessage msg, RouterInfo peer, long expiration) {
+     * @since 0.9.41 renamed from sendStoreThroughGarlic()
+     */
+    private void sendStoreThroughExploratory(DatabaseStoreMessage msg, RouterInfo peer, long expiration) {
         long token = 1 + getContext().random().nextLong(I2NPMessage.MAX_ID_VALUE);
         
@@ -516 +519 @@
     
     /**
+     * Send a leaseset store message out an exploratory tunnel,
+     * with the reply to come back through a exploratory tunnel.
+     * Stores are garlic encrypted to hide the identity from the OBEP.
+     *
+     * Only for Meta LS2, for now.
+     *
+     * @param msg must contain a leaseset
+     * @since 0.9.41
+     */
+    private void sendWrappedStoreThroughExploratory(DatabaseStoreMessage msg, RouterInfo peer, long expiration) {
+        long token = 1 + getContext().random().nextLong(I2NPMessage.MAX_ID_VALUE);
+        Hash to = peer.getIdentity().getHash();
+        TunnelInfo replyTunnel = getContext().tunnelManager().selectInboundExploratoryTunnel(to);
+        if (replyTunnel == null) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("No inbound expl. tunnels for reply - delaying...");
+            // continueSending() above did an addPending() so remove it here.
+            // This means we will skip the peer next time, can't be helped for now
+            // without modding StoreState
+            _state.replyTimeout(to);
+            Job waiter = new WaitJob(getContext());
+            waiter.getTiming().setStartAfter(getContext().clock().now() + 3*1000);
+            getContext().jobQueue().addJob(waiter);
+            return;
+        }
+        TunnelId replyTunnelId = replyTunnel.getReceiveTunnelId(0);
+        msg.setReplyToken(token);
+        msg.setReplyTunnel(replyTunnelId);
+        msg.setReplyGateway(replyTunnel.getPeer(0));
+
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug(getJobId() + ": send(dbStore) w/ token expected " + token);
+
+        TunnelInfo outTunnel = getContext().tunnelManager().selectOutboundExploratoryTunnel(to);
+        if (outTunnel != null) {
+            I2NPMessage sent;
+            // garlic encrypt using router SKM
+            MessageWrapper.WrappedMessage wm = MessageWrapper.wrap(getContext(), msg, null, peer);
+            if (wm == null) {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Fail garlic encrypting");
+                fail();
+                return;
+            }
+            sent = wm.getMessage();
+            _state.addPending(to, wm);
+
+            SendSuccessJob onReply = new SendSuccessJob(getContext(), peer, outTunnel, sent.getMessageSize());
+            FailedJob onFail = new FailedJob(getContext(), peer, getContext().clock().now());
+            StoreMessageSelector selector = new StoreMessageSelector(getContext(), getJobId(), peer, token, expiration);
+    
+            if (_log.shouldLog(Log.DEBUG)) {
+                _log.debug(getJobId() + ": sending encrypted store to " + peer.getIdentity().getHash() + " through " + outTunnel + ": " + sent);
+            }
+            getContext().messageRegistry().registerPending(selector, onReply, onFail);
+            getContext().tunnelDispatcher().dispatchOutbound(sent, outTunnel.getSendTunnelId(0), null, to);
+        } else {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("No outbound expl. tunnels to send a dbStore out - delaying...");
+            // continueSending() above did an addPending() so remove it here.
+            // This means we will skip the peer next time, can't be helped for now
+            // without modding StoreState
+            _state.replyTimeout(to);
+            Job waiter = new WaitJob(getContext());
+            waiter.getTiming().setStartAfter(getContext().clock().now() + 3*1000);
+            getContext().jobQueue().addJob(waiter);
+        }
+    }
+    
+    /**
      * Called to wait a little while
      * @since 0.7.10