prefetch introset via tag (non working)

.vscode/settings.json

@@ -61,6 +61,8 @@
     "*.ipp": "cpp",
     "csignal": "cpp",
     "future": "cpp",
-    "map": "cpp"
+    "map": "cpp",
+    "vector": "cpp",
+    "new": "cpp"
   }
 }

Makefile

@@ -88,8 +88,6 @@ test: debug-configure
 format:
 	clang-format -i $$(find daemon llarp include | grep -E '\.[h,c](pp)?$$')
 
-
-
 fuzz-configure: clean
 	cmake -GNinja -DCMAKE_BUILD_TYPE=Fuzz -DCMAKE_C_COMPILER=afl-gcc -DCMAKE_CXX_COMPILER=afl-g++
 

contrib/testnet/genconf.py

@@ -88,7 +88,8 @@ def main():
             config.write(f)
         config = CP()
         config['test-service'] = {
-
+            'tag': 'test',
+            'prefetch-tag': "test"
         }
         with open(hiddenservice, 'w') as f:
             config.write(f)

include/llarp/aligned.hpp

@@ -151,7 +151,10 @@ namespace llarp
       if(!bencode_read_string(buf, &strbuf))
         return false;
       if(strbuf.sz != sz)
+      {
+        llarp::LogError("bdecode buffer size missmatch ", strbuf.sz, "!=", sz);
         return false;
+      }
       memcpy(b, strbuf.base, sz);
       return true;
     }

include/llarp/dht/context.hpp

@@ -8,6 +8,7 @@
 #include <llarp/dht/message.hpp>
 #include <llarp/dht/node.hpp>
 #include <llarp/dht/search_job.hpp>
+#include <llarp/service/IntroSet.hpp>
 
 #include <set>
 
@@ -52,6 +53,13 @@ namespace llarp
       void
       LookupRouterViaJob(llarp_router_lookup_job* job);
 
+      void
+      LookupTagForPath(const service::Tag& tag, uint64_t txid,
+                       const llarp::PathID_t& path, const Key_t& askpeer);
+
+      std::set< service::IntroSet >
+      FindIntroSetsWithTag(const service::Tag& tag);
+
       void
       LookupRouterRelayed(const Key_t& requester, uint64_t txid,
                           const Key_t& target, bool recursive,

include/llarp/dht/messages/findintro.hpp

@@ -14,23 +14,27 @@ namespace llarp
       bool iterative = false;
       llarp::service::Address S;
       llarp::service::Tag N;
-      uint64_t T = 0;
+      uint64_t T   = 0;
+      bool relayed = false;
 
-      FindIntroMessage(const Key_t& from) : IMessage(from)
+      FindIntroMessage(const Key_t& from, bool relay) : IMessage(from)
       {
+        relayed = relay;
       }
 
       FindIntroMessage(const llarp::service::Tag& tag, uint64_t txid)
           : IMessage({}), N(tag), T(txid)
       {
+        S.Zero();
       }
 
       FindIntroMessage(const llarp::service::Address& addr, uint64_t txid)
           : IMessage({}), S(addr), T(txid)
       {
+        N.Zero();
       }
 
-      virtual ~FindIntroMessage();
+      ~FindIntroMessage();
 
       bool
       BEncode(llarp_buffer_t* buf) const;
@@ -38,19 +42,6 @@ namespace llarp
       bool
       DecodeKey(llarp_buffer_t key, llarp_buffer_t* val);
 
-      virtual bool
-      HandleMessage(llarp_dht_context* ctx,
-                    std::vector< IMessage* >& replies) const;
-    };
-
-    struct RelayedFindIntroMessage : public FindIntroMessage
-    {
-      RelayedFindIntroMessage() : FindIntroMessage({})
-      {
-      }
-
-      ~RelayedFindIntroMessage();
-
       bool
       HandleMessage(llarp_dht_context* ctx,
                     std::vector< IMessage* >& replies) const;

include/llarp/dht/search_job.hpp

@@ -28,6 +28,9 @@ namespace llarp
       SearchJob(const Key_t& requester, uint64_t requesterTX,
                 const Key_t& target, const std::set< Key_t >& excludes,
                 IntroSetHookFunc found);
+      // for introsets via tag
+      SearchJob(const Key_t& requester, uint64_t requseterTX,
+                IntroSetHookFunc found);
 
       void
       FoundRouter(const llarp_rc* router) const;

include/llarp/logger.hpp

@@ -4,6 +4,7 @@
 #include <ctime>
 #include <iomanip>
 #include <iostream>
+#include <mutex>
 #include <sstream>
 #include <string>
 
@@ -22,6 +23,7 @@ namespace llarp
   {
     LogLevel minlevel = eLogInfo;
     std::ostream& out = std::cout;
+    std::mutex access;
   };
 
   extern Logger _glog;
@@ -86,10 +88,13 @@ namespace llarp
     ss << "\t";
     LogAppend(ss, std::forward< TArgs >(args)...);
     ss << (char)27 << "[0;0m";
-    _glog.out << ss.str() << std::endl;
+    {
+      std::unique_lock< std::mutex > lock(_glog.access);
+      _glog.out << ss.str() << std::endl;
 #ifdef SHADOW_TESTNET
-    _glog.out << "\n" << std::flush;
+      _glog.out << "\n" << std::flush;
 #endif
+    }
   }
 }  // namespace llarp
 

include/llarp/pow.hpp

@@ -40,6 +40,14 @@ namespace llarp
     {
       return !(*this == other);
     }
+
+    friend std::ostream&
+    operator<<(std::ostream& out, const PoW& p)
+    {
+      return out << "[pow timestamp=" << p.timestamp
+                 << " lifetime=" << p.extendedLifetime << " nonce=" << p.nonce
+                 << "]";
+    }
   };
 }  // namespace llarp
 

include/llarp/routing/message.hpp

@@ -22,11 +22,10 @@ namespace llarp
 
     struct InboundMessageParser
     {
-      llarp::PathID_t from;
       InboundMessageParser();
       bool
       ParseMessageBuffer(llarp_buffer_t buf, IMessageHandler* handler,
-                         llarp_router* r);
+                         const PathID_t& from, llarp_router* r);
 
      private:
       static bool

include/llarp/service/IntroSet.hpp

@@ -33,6 +33,7 @@ namespace llarp
         A       = other.A;
         I       = other.I;
         version = other.version;
+        topic   = other.topic;
         if(W)
           delete W;
         W = other.W;
@@ -60,6 +61,14 @@ namespace llarp
         {
           out << " topic=" << topic;
         }
+        else
+        {
+          out << " topic=" << i.topic;
+        }
+        if(i.W)
+        {
+          out << " W=" << *i.W;
+        }
         return out << " V=" << i.version << " Z=" << i.Z;
       }
 

include/llarp/service/tag.hpp

@@ -25,7 +25,8 @@ namespace llarp
         memcpy(data(), str.c_str(), std::min(16UL, str.size()));
       }
 
-      operator llarp::dht::Key_t() const
+      llarp::dht::Key_t
+      Key() const
       {
         llarp::dht::Key_t k;
         crypto_generichash(k, 32, data(), 16, nullptr, 0);

llarp/dht/context.cpp

@@ -45,11 +45,67 @@ namespace llarp
       ctx->ScheduleCleanupTimer();
     }
 
+    struct PathTagLookupJob
+    {
+      uint64_t txid;
+      PathID_t pathID;
+      llarp_router *m_router;
+      PathTagLookupJob(llarp_router *r, const PathID_t &localpath, uint64_t tx)
+          : txid(tx), pathID(localpath), m_router(r)
+      {
+      }
+
+      void
+      OnResult(const std::set< service::IntroSet > &results)
+      {
+        auto path =
+            m_router->paths.GetByUpstream(m_router->dht->impl.OurKey(), pathID);
+        if(path)
+        {
+          llarp::routing::DHTMessage msg;
+          msg.M.push_back(new llarp::dht::GotIntroMessage(results, txid));
+          path->SendRoutingMessage(&msg, m_router);
+        }
+        else
+        {
+          llarp::LogWarn("no local path for reply on PathTagLookupJob pathid=",
+                         pathID);
+        }
+        delete this;
+      }
+    };
+
     void
-    Context::LookupTag(const llarp::service::Tag &tag, const Key_t &whoasked,
-                       uint64_t txid, const Key_t &askpeer, bool iterative)
+    Context::LookupTagForPath(const service::Tag &tag, uint64_t txid,
+                              const llarp::PathID_t &path, const Key_t &askpeer)
+    {
+      auto id = ++ids;
+      TXOwner ownerKey;
+      ownerKey.node       = askpeer;
+      ownerKey.txid       = id;
+      PathTagLookupJob *j = new PathTagLookupJob(router, path, txid);
+      SearchJob job(
+          OurKey(), txid,
+          std::bind(&PathTagLookupJob::OnResult, j, std::placeholders::_1));
+      pendingTX[ownerKey] = job;
+
+      auto msg          = new llarp::DHTImmeidateMessage(askpeer);
+      auto dhtmsg       = new FindIntroMessage(tag, id);
+      dhtmsg->iterative = true;
+      msg->msgs.push_back(dhtmsg);
+      router->SendToOrQueue(askpeer, msg);
+    }
+
+    std::set< service::IntroSet >
+    Context::FindIntroSetsWithTag(const service::Tag &tag)
     {
-      // TODO: implement me
+      std::set< service::IntroSet > found;
+      for(const auto &itr : services->nodes)
+      {
+        if(itr.second.introset.topic == tag)
+          found.insert(itr.second.introset);
+      }
+      return found;
     }
 
     void
@@ -215,6 +271,29 @@ namespace llarp
       }
     };
 
+    void
+    Context::LookupTag(const llarp::service::Tag &tag, const Key_t &whoasked,
+                       uint64_t txid, const Key_t &askpeer, bool iterative)
+    {
+      auto id = ++ids;
+      if(txid == 0)
+        txid = id;
+
+      TXOwner ownerKey;
+      ownerKey.node        = askpeer;
+      ownerKey.txid        = id;
+      IntroSetInformJob *j = new IntroSetInformJob(router, askpeer, id);
+      SearchJob job(
+          whoasked, txid,
+          std::bind(&IntroSetInformJob::OnResult, j, std::placeholders::_1));
+      pendingTX[ownerKey] = job;
+
+      auto msg    = new llarp::DHTImmeidateMessage(askpeer);
+      auto dhtmsg = new FindIntroMessage(tag, id);
+      msg->msgs.push_back(dhtmsg);
+      router->SendToOrQueue(askpeer, msg);
+    }
+
     void
     Context::LookupIntroSet(const service::Address &addr, const Key_t &whoasked,
                             uint64_t txid, const Key_t &askpeer, bool iterative,

llarp/dht/decode.cpp

@@ -38,10 +38,8 @@ namespace llarp
           switch(*strbuf.base)
           {
             case 'F':
-              if(dec->relayed)
-                dec->msg = new RelayedFindIntroMessage();
-              else
-                dec->msg = new FindIntroMessage(dec->From);
+              dec->msg = new FindIntroMessage(dec->From, dec->relayed);
+              break;
             case 'R':
               if(dec->relayed)
                 dec->msg = new RelayedFindRouterMessage(dec->From);

llarp/dht/find_intro.cpp

@@ -7,6 +7,7 @@ namespace llarp
 {
   namespace dht
   {
+    /*
     struct IntroSetLookupInformer
     {
       llarp_router* router;
@@ -17,6 +18,7 @@ namespace llarp
       {
       }
     };
+    */
 
     FindIntroMessage::~FindIntroMessage()
     {
@@ -27,15 +29,13 @@ namespace llarp
     {
       uint64_t i = 0;
       bool read  = false;
-      if(BEncodeMaybeReadDictInt("I", i, read, k, val))
+      if(!BEncodeMaybeReadDictInt("I", i, read, k, val))
+        return false;
+      if(read)
       {
-        if(read)
-        {
-          iterative = i != 0;
-        }
+        iterative = i != 0;
+        return true;
       }
-      else
-        return false;
 
       if(!BEncodeMaybeReadDictEntry("N", N, read, k, val))
         return false;
@@ -101,32 +101,27 @@ namespace llarp
         llarp_dht_context* ctx,
         std::vector< llarp::dht::IMessage* >& replies) const
     {
-      auto& dht           = ctx->impl;
-      const auto introset = dht.GetIntroSetByServiceAddress(S);
-      if(introset)
-      {
-        replies.push_back(new GotIntroMessage({*introset}, T));
-        return true;
-      }
-      else if(iterative)
-      {
-        // we are iterative and don't have it, reply with a direct reply
-        replies.push_back(new GotIntroMessage({}, T));
-        return true;
-      }
-      else
+      auto& dht = ctx->impl;
+      Key_t peer;
+      std::set< Key_t > exclude = {dht.OurKey(), From};
+      if(N.IsZero())
       {
-        // we are recursive
-        Key_t peer;
-        std::set< Key_t > exclude = {dht.OurKey(), From};
-
-        if(N.IsZero())
+        const auto introset = dht.GetIntroSetByServiceAddress(S);
+        if(introset)
+        {
+          replies.push_back(new GotIntroMessage({*introset}, T));
+        }
+        else if(iterative)
+        {
+          // we are iterative and don't have it, reply with a direct reply
+          replies.push_back(new GotIntroMessage({}, T));
+        }
+        else
         {
-          // service address lookup
+          // we are recursive
           if(dht.nodes->FindCloseExcluding(S, peer, exclude))
           {
             dht.LookupIntroSet(S, From, T, peer);
-            return true;
           }
           else
           {
@@ -134,52 +129,51 @@ namespace llarp
                             S);
           }
         }
-        else
-        {
-          // tag lookup
-          if(dht.nodes->FindCloseExcluding(N, peer, exclude))
-          {
-            dht.LookupTag(N, From, T, peer);
-            return true;
-          }
-        }
-      }
-      return false;
-    }  // namespace dht
-
-    RelayedFindIntroMessage::~RelayedFindIntroMessage()
-    {
-    }
-
-    bool
-    RelayedFindIntroMessage::HandleMessage(
-        llarp_dht_context* ctx,
-        std::vector< llarp::dht::IMessage* >& replies) const
-    {
-      auto& dht           = ctx->impl;
-      const auto introset = dht.GetIntroSetByServiceAddress(S);
-      if(introset)
-      {
-        replies.push_back(new GotIntroMessage({*introset}, T));
-        return true;
       }
       else
       {
-        Key_t peer;
-        std::set< Key_t > exclude = {dht.OurKey()};
-        if(dht.nodes->FindCloseExcluding(S, peer, exclude))
+        if(relayed)
         {
-          dht.LookupIntroSet(S, From, 0, peer);
-          return true;
+          // tag lookup
+          if(dht.nodes->FindCloseExcluding(N.Key(), peer, exclude))
+          {
+            dht.LookupTagForPath(N, T, pathID, peer);
+          }
+          else
+          {
+            llarp::LogWarn("no closer peers for tag ", N.ToString());
+          }
         }
         else
         {
-          llarp::LogError("cannot find closer peers for introset lookup for ",
-                          S);
+          auto introsets = dht.FindIntroSetsWithTag(N);
+          if(introsets.size())
+          {
+            replies.push_back(new GotIntroMessage(introsets, T));
+          }
+          else
+          {
+            if(iterative)
+            {
+              // we are iterative and don't have it, reply with a direct reply
+              replies.push_back(new GotIntroMessage({}, T));
+            }
+            else
+            {
+              // tag lookup
+              if(dht.nodes->FindCloseExcluding(N.Key(), peer, exclude))
+              {
+                dht.LookupTag(N, From, T, peer);
+              }
+              else
+              {
+                llarp::LogWarn("no closer peers for tag ", N.ToString());
+              }
+            }
+          }
         }
       }
-      return false;
+      return true;
     }
-
   }  // namespace dht
 }  // namespace llarp

llarp/dht/got_intro.cpp

@@ -28,13 +28,15 @@ namespace llarp
     {
       auto &dht   = ctx->impl;
       auto crypto = &dht.router->crypto;
-      if(I.size() == 1)
+      std::set< service::IntroSet > introsets;
+
+      for(const auto &introset : I)
       {
-        const auto &introset = I.front();
         if(!introset.VerifySignature(crypto))
         {
           llarp::LogWarn(
-              "Invalid introset signature while handling direct GotIntro from ",
+              "Invalid introset signature while handling direct GotIntro "
+              "from ",
               From);
           return false;
         }
@@ -42,15 +44,25 @@ namespace llarp
         if(!introset.A.CalculateAddress(addr))
         {
           llarp::LogWarn(
-              "failed to calculate hidden service address for direct GotIntro "
+              "failed to calculate hidden service address for direct "
+              "GotIntro "
               "message from ",
               From);
           return false;
         }
-        // TODO: inform any pending tx
+        introsets.insert(introset);
+      }
+      auto pending = dht.FindPendingTX(From, T);
+      if(pending)
+      {
+        pending->FoundIntros(introsets);
+        dht.RemovePendingLookup(From, T);
         return true;
       }
-      return false;
+      else
+      {
+        return false;
+      }
     }
 
     bool

llarp/dht/search_job.cpp

@@ -34,6 +34,16 @@ namespace llarp
     {
     }
 
+    SearchJob::SearchJob(const Key_t &asker, uint64_t tx,
+                         IntroSetHookFunc found)
+        : foundIntroHook(found)
+        , started(llarp_time_now_ms())
+        , requester(asker)
+        , requesterTX(tx)
+    {
+      target.Zero();
+    }
+
     void
     SearchJob::FoundIntros(
         const std::set< llarp::service::IntroSet > &introsets) const

llarp/iwp/frame_state.cpp

@@ -324,9 +324,9 @@ bool
 frame_state::next_frame(llarp_buffer_t *buf)
 {
   auto left = sendqueue.size();
-  llarp::LogDebug("next frame, ", left, " frames left in send queue");
   if(left)
   {
+    llarp::LogDebug("next frame, ", left, " frames left in send queue");
     auto &send = sendqueue.front();
     buf->base  = send->data();
     buf->cur   = send->data();

llarp/mem.hpp

@@ -22,16 +22,32 @@ namespace llarp
     }
   }
 
-  template < typename T >
+  template < typename T, size_t align = 8 >
   void
-  dumphex_buffer(T buff)
+  DumpBuffer(const T &buff)
   {
     size_t idx = 0;
     printf("buffer of size %ld\n", buff.sz);
     while(idx < buff.sz)
     {
-      printf("%.2x ", buff.base[idx++]);
-      if(idx % 8 == 0)
+      if(buff.base + idx == buff.cur)
+      {
+        printf("%c[1;31m", 27);
+      }
+      else
+      {
+        printf("%c[0m", 27);
+      }
+      if(buff.base[idx])
+      {
+        printf("%c", buff.base[idx]);
+      }
+      else
+      {
+        printf("X");
+      }
+      ++idx;
+      if(idx % align == 0)
         printf("\n");
     }
   }

llarp/path.cpp

@@ -296,8 +296,6 @@ namespace llarp
       intro.router = hops[h->numHops - 1].router.pubkey;
       // TODO: or is it rxid ?
       intro.pathID = hops[h->numHops - 1].txID;
-
-      m_InboundMessageParser.from = RXID();
     }
 
     void
@@ -384,7 +382,7 @@ namespace llarp
     bool
     Path::HandleRoutingMessage(llarp_buffer_t buf, llarp_router* r)
     {
-      if(!m_InboundMessageParser.ParseMessageBuffer(buf, this, r))
+      if(!m_InboundMessageParser.ParseMessageBuffer(buf, this, RXID(), r))
       {
         llarp::LogWarn("Failed to parse inbound routing message");
         return false;

llarp/routing/message_parser.cpp

@@ -3,6 +3,7 @@
 #include <llarp/messages/path_latency.hpp>
 #include <llarp/messages/path_transfer.hpp>
 #include <llarp/routing/message.hpp>
+#include "mem.hpp"
 
 namespace llarp
 {
@@ -41,16 +42,16 @@ namespace llarp
         switch(self->key)
         {
           case 'L':
-            self->msg = new PathLatencyMessage;
+            self->msg = new PathLatencyMessage();
             break;
           case 'M':
-            self->msg = new DHTMessage;
+            self->msg = new DHTMessage();
             break;
           case 'P':
-            self->msg = new PathConfirmMessage;
+            self->msg = new PathConfirmMessage();
             break;
           case 'T':
-            self->msg = new PathTransferMessage;
+            self->msg = new PathTransferMessage();
             break;
           default:
             llarp::LogError("invalid routing message id: ", *strbuf.cur);
@@ -67,6 +68,7 @@ namespace llarp
     bool
     InboundMessageParser::ParseMessageBuffer(llarp_buffer_t buf,
                                              IMessageHandler* h,
+                                             const PathID_t& from,
                                              llarp_router* r)
     {
       bool result = false;
@@ -81,7 +83,10 @@ namespace llarp
         delete msg;
       }
       else
-        llarp::LogError("read dict failed");
+      {
+        llarp::LogError("read dict failed in routing layer");
+        llarp::DumpBuffer< llarp_buffer_t, 128 >(buf);
+      }
       return result;
     }
   }  // namespace routing

llarp/service.cpp

@@ -60,7 +60,7 @@ namespace llarp
       // end introduction list
 
       // topic tag
-      if(topic.ToString().size())
+      if(!topic.IsZero())
       {
         if(!BEncodeWriteDictEntry("n", topic, buf))
           return false;

llarp/service/endpoint.cpp

@@ -22,6 +22,7 @@ namespace llarp
       if(k == "tag")
       {
         m_Tag = v;
+        llarp::LogInfo("Setting tag to ", v);
       }
       if(k == "prefetch-tag")
       {
@@ -73,7 +74,11 @@ namespace llarp
           auto path = PickRandomEstablishedPath();
           if(path)
           {
-            itr->second.SendRequestViaPath(path, m_Router);
+            itr->second.pendingTX = GenTXID();
+            if(itr->second.SendRequestViaPath(path, m_Router))
+            {
+              m_PendingLookups[itr->second.pendingTX] = &itr->second;
+            }
           }
         }
       }
@@ -101,16 +106,17 @@ namespace llarp
       std::set< IntroSet > remote;
       for(const auto& introset : msg->I)
       {
-        if(!introset.VerifySignature(crypto))
-        {
-          llarp::LogWarn(
-              "invalid signature in got intro message for service endpoint ",
-              Name());
-          IntroSetPublishFail();
-          return false;
-        }
         if(m_Identity.pub == introset.A)
         {
+          if(!introset.VerifySignature(crypto))
+          {
+            llarp::LogWarn(
+                "invalid signature in got intro message for service endpoint ",
+                Name());
+
+            IntroSetPublishFail();
+            return false;
+          }
           llarp::LogInfo(
               "got introset publish confirmation for hidden service endpoint ",
               Name());
@@ -163,6 +169,8 @@ namespace llarp
     Endpoint::CachedTagResult::HandleResponse(
         const std::set< IntroSet >& results)
     {
+      llarp::LogInfo("Tag result for ", tag.ToString(), " got ", results.size(),
+                     " results");
       return true;
     }
 
@@ -190,10 +198,9 @@ namespace llarp
           llarp::LogInfo(Name(), " publishing introset");
           return true;
         }
-        return false;
       }
-      else
-        return false;
+      llarp::LogWarn(Name(), " publish introset failed");
+      return false;
     }
 
     void

llarp/transit_hop.cpp

@@ -84,7 +84,7 @@ namespace llarp
       r->crypto.xchacha20(buf, pathKey, Y);
       if(info.upstream == RouterID(r->pubkey()))
       {
-        return m_MessageParser.ParseMessageBuffer(buf, this, r);
+        return m_MessageParser.ParseMessageBuffer(buf, this, info.rxID, r);
       }
       else
       {