more fuxing around

include/llarp/dht/bucket.hpp

@@ -76,7 +76,7 @@ namespace llarp
       void
       PutNode(const Val_t& val)
       {
-        nodes[val.ID] = val;
+        nodes.insert(std::make_pair(val.ID, val));
       }
 
       void

include/llarp/dht/context.hpp

@@ -42,8 +42,8 @@ namespace llarp
 
       void
       LookupIntroSet(const service::Address& addr, const Key_t& whoasked,
-                     uint64_t whoaskedTX, const Key_t& askpeer,
-                     bool interative = false, std::set< Key_t > excludes = {});
+                     uint64_t whoaskedTX, const Key_t& askpeer, uint64_t R,
+                     std::set< Key_t > excludes = {});
 
       void
       LookupTag(const service::Tag& tag, const Key_t& whoasked,

include/llarp/dht/key.hpp

@@ -28,7 +28,13 @@ namespace llarp
       bool
       operator<(const Key_t& other) const
       {
-        return memcmp(data_l(), other.data_l(), 32) < 0;
+        return memcmp(data(), other.data(), 32) < 0;
+      }
+
+      bool
+      operator>(const Key_t& other) const
+      {
+        return memcmp(data(), other.data(), 32) > 0;
       }
     };
   }  // namespace dht

include/llarp/dht/messages/findintro.hpp

@@ -10,8 +10,7 @@ namespace llarp
   {
     struct FindIntroMessage : public IMessage
     {
-      uint64_t R     = 0;
-      bool iterative = false;
+      uint64_t R = 0;
       llarp::service::Address S;
       llarp::service::Tag N;
       uint64_t T   = 0;
@@ -22,14 +21,16 @@ namespace llarp
         relayed = relay;
       }
 
-      FindIntroMessage(const llarp::service::Tag& tag, uint64_t txid)
-          : IMessage({}), N(tag), T(txid)
+      FindIntroMessage(const Key_t& from, const llarp::service::Tag& tag,
+                       uint64_t txid)
+          : IMessage(from), N(tag), T(txid)
       {
         S.Zero();
       }
 
-      FindIntroMessage(const llarp::service::Address& addr, uint64_t txid)
-          : IMessage({}), S(addr), T(txid)
+      FindIntroMessage(const Key_t& from, const llarp::service::Address& addr,
+                       uint64_t txid)
+          : IMessage(from), S(addr), T(txid)
       {
         N.Zero();
       }

include/llarp/dht/node.hpp

@@ -40,7 +40,8 @@ namespace llarp
       ISNode(const llarp::service::IntroSet& other)
       {
         introset = other;
-        other.A.CalculateAddress(ID);
+        introset.A.CalculateAddress(ID);
+        llarp::LogInfo("make ISNode with topic ", introset.topic.ToString());
       }
     };
   }  // namespace dht

include/llarp/service/IntroSet.hpp

@@ -36,7 +36,8 @@ namespace llarp
         topic   = other.topic;
         if(W)
           delete W;
-        W = other.W;
+        if(other.W)
+          W = new llarp::PoW(*other.W);
         Z = other.Z;
         return *this;
       }

include/llarp/service/endpoint.hpp

@@ -74,6 +74,11 @@ namespace llarp
       bool
       HasPathToService(const Address& remote) const;
 
+      /// return true if we have a pending job to build to a hidden service but
+      /// it's not done yet
+      bool
+      HasPendingPathToService(const Address& remote) const;
+
       /// return false if we don't have a path to the service
       /// return true if we did and we removed it
       bool

include/llarp/service/tag.hpp

@@ -29,6 +29,13 @@ namespace llarp
 #endif
       }
 
+      Tag&
+      operator=(const Tag& other)
+      {
+        memcpy(data(), other.data(), 16);
+        return *this;
+      }
+
       llarp::dht::Key_t
       Key() const
       {

llarp/dht/context.cpp

@@ -134,7 +134,7 @@ namespace llarp
       pendingTX[ownerKey] = job;
 
       auto msg    = new llarp::DHTImmeidateMessage(askpeer);
-      auto dhtmsg = new FindIntroMessage(tag, id);
+      auto dhtmsg = new FindIntroMessage({}, tag, id);
       dhtmsg->R   = 5;
       msg->msgs.push_back(dhtmsg);
       llarp::LogInfo("asking ", askpeer, " for tag ", tag.ToString(), " with ",
@@ -159,7 +159,10 @@ namespace llarp
 
       auto msg    = new llarp::DHTImmeidateMessage(askpeer);
       auto dhtmsg = new FindIntroMessage(addr, id);
+      dhtmsg->R   = 5;
       msg->msgs.push_back(dhtmsg);
+      llarp::LogInfo("askng ", askpeer, " for ", addr.ToString(),
+                     " with txid=", id);
       router->SendToOrQueue(askpeer, msg);
     }
 
@@ -169,8 +172,9 @@ namespace llarp
       std::set< service::IntroSet > found;
       auto &nodes = services->nodes;
       if(nodes.size() == 0)
+      {
         return found;
-
+      }
       auto itr = nodes.begin();
       // start at random middle point
       auto start = llarp_randint() % nodes.size();
@@ -336,8 +340,12 @@ namespace llarp
       llarp::routing::DHTMessage reply;
       if(!msg->HandleMessage(router->dht, reply.M))
         return false;
-      auto path = router->paths.GetByUpstream(router->pubkey(), id);
-      return path && path->SendRoutingMessage(&reply, router);
+      if(reply.M.size())
+      {
+        auto path = router->paths.GetByUpstream(router->pubkey(), id);
+        return path && path->SendRoutingMessage(&reply, router);
+      }
+      return true;
     }
 
     /// handles replying with a GIM for a lookup
@@ -393,7 +401,7 @@ namespace llarp
       pendingTX[ownerKey] = job;
 
       auto msg    = new llarp::DHTImmeidateMessage(askpeer);
-      auto dhtmsg = new FindIntroMessage(tag, id);
+      auto dhtmsg = new FindIntroMessage({}, tag, id);
       dhtmsg->R   = R;
       msg->msgs.push_back(dhtmsg);
       router->SendToOrQueue(askpeer, msg);
@@ -401,7 +409,7 @@ namespace llarp
 
     void
     Context::LookupIntroSet(const service::Address &addr, const Key_t &whoasked,
-                            uint64_t txid, const Key_t &askpeer, bool iterative,
+                            uint64_t txid, const Key_t &askpeer, uint64_t R,
                             std::set< Key_t > excludes)
     {
       auto id = ++ids;
@@ -418,7 +426,8 @@ namespace llarp
       pendingTX[ownerKey] = job;
 
       auto msg    = new llarp::DHTImmeidateMessage(askpeer);
-      auto dhtmsg = new FindIntroMessage(addr, id);
+      auto dhtmsg = new FindIntroMessage({}, addr, id);
+      dhtmsg->R   = R;
       msg->msgs.push_back(dhtmsg);
       router->SendToOrQueue(askpeer, msg);
     }

llarp/dht/decode.cpp

@@ -35,7 +35,8 @@ namespace llarp
           // bad msg size?
           if(strbuf.sz != 1)
             return false;
-          llarp::LogInfo("Handle DHT message ", *strbuf.base);
+          llarp::LogInfo("Handle DHT message ", *strbuf.base,
+                         " relayed=", dec->relayed);
           switch(*strbuf.base)
           {
             case 'F':

llarp/dht/find_intro.cpp

@@ -27,15 +27,7 @@ namespace llarp
     bool
     FindIntroMessage::DecodeKey(llarp_buffer_t k, llarp_buffer_t* val)
     {
-      uint64_t i = 0;
-      bool read  = false;
-      if(!BEncodeMaybeReadDictInt("I", i, read, k, val))
-        return false;
-      if(read)
-      {
-        iterative = i != 0;
-        return true;
-      }
+      bool read = false;
 
       if(!BEncodeMaybeReadDictEntry("N", N, read, k, val))
         return false;
@@ -65,9 +57,6 @@ namespace llarp
       // message id
       if(!BEncodeWriteDictMsgType(buf, "A", "F"))
         return false;
-      // iterative
-      if(!BEncodeWriteDictInt("I", iterative ? 1 : 0, buf))
-        return false;
       if(N.IsZero())
       {
         return false;
@@ -107,19 +96,25 @@ namespace llarp
         return false;
       }
       auto& dht = ctx->impl;
+      if((!relayed) && dht.FindPendingTX(From, T))
+      {
+        llarp::LogWarn("duplicate FIM from ", From, " txid=", T);
+        return false;
+      }
       Key_t peer;
       std::set< Key_t > exclude = {dht.OurKey(), From};
-      if(N.IsZero())
+      if(N.ToString().empty())
       {
         const auto introset = dht.GetIntroSetByServiceAddress(S);
         if(introset)
         {
+          llarp::LogInfo("introset found locally");
           service::IntroSet i = *introset;
           replies.push_back(new GotIntroMessage({i}, T));
         }
         else
         {
-          if(iterative)
+          if(R == 0 && !relayed)
           {
             // we are iterative and don't have it, reply with a direct reply
             replies.push_back(new GotIntroMessage({}, T));
@@ -131,12 +126,10 @@ namespace llarp
             {
               if(relayed)
                 dht.LookupIntroSetForPath(S, T, pathID, peer);
-              else if((peer ^ dht.OurKey())
-                      > (peer
-                         ^ From))  // peer is closer than us, recursive search
-                dht.LookupIntroSet(S, From, T, peer);
-              else  // we are closer than peer so do iterative search
-                dht.LookupIntroSet(S, From, T, peer, true);
+              else if(R >= 1)
+                dht.LookupIntroSet(S, From, T, peer, R - 1);
+              else
+                dht.LookupIntroSet(S, From, T, peer, 0);
             }
             else
             {
@@ -163,7 +156,7 @@ namespace llarp
         else
         {
           auto introsets = dht.FindRandomIntroSetsWithTag(N);
-          if(iterative || R == 0)
+          if(R == 0)
           {
             std::vector< service::IntroSet > reply;
             for(const auto& introset : introsets)

llarp/dht/publish_intro.cpp

@@ -67,7 +67,6 @@ namespace llarp
         return false;
       }
       dht.services->PutNode(I);
-      llarp::LogInfo("Put Introset for ", I.A.Addr().ToString());
       replies.push_back(new GotIntroMessage({I}, txID));
       Key_t peer;
       std::set< Key_t > exclude;

llarp/ev.cpp

@@ -2,6 +2,8 @@
 #include <llarp/logic.h>
 #include "mem.hpp"
 
+#define EV_TICK_INTERVAL 100
+
 #ifdef __linux__
 #include "ev_epoll.hpp"
 #endif
@@ -47,8 +49,8 @@ llarp_ev_loop_run(struct llarp_ev_loop *ev, struct llarp_logic *logic)
 {
   while(true)
   {
-    if(ev->tick(100) == -1)
-      break;
+    if(ev->tick(EV_TICK_INTERVAL) == -1)
+      return -1;
     llarp_logic_tick(logic);
   }
   return 0;
@@ -59,10 +61,9 @@ llarp_ev_loop_run_single_process(struct llarp_ev_loop *ev,
                                  struct llarp_threadpool *tp,
                                  struct llarp_logic *logic)
 {
-  while(true)
+  while(ev->running())
   {
-    if(ev->tick(10) == -1)
-      return;
+    ev->tick(EV_TICK_INTERVAL);
     llarp_logic_tick(logic);
     llarp_threadpool_tick(tp);
   }

llarp/ev.hpp

@@ -55,6 +55,9 @@ struct llarp_ev_loop
   virtual bool
   close_ev(llarp::ev_io* ev) = 0;
 
+  virtual bool
+  running() const = 0;
+
   virtual ~llarp_ev_loop(){};
 
   std::list< llarp_udp_io* > udp_listeners;

llarp/ev_epoll.hpp

@@ -82,6 +82,12 @@ struct llarp_epoll_loop : public llarp_ev_loop
       close(epollfd);
   }
 
+  bool
+  running() const
+  {
+    return epollfd != -1;
+  }
+
   bool
   init()
   {
@@ -105,7 +111,6 @@ struct llarp_epoll_loop : public llarp_ev_loop
   {
     epoll_event events[1024];
     int result;
-    byte_t readbuf[2048];
 
     result = epoll_wait(epollfd, events, 1024, ms);
     if(result > 0)
@@ -122,11 +127,7 @@ struct llarp_epoll_loop : public llarp_ev_loop
         llarp::ev_io* ev = static_cast< llarp::ev_io* >(events[idx].data.ptr);
         if(events[idx].events & EPOLLIN)
         {
-          if(ev->read(readbuf, sizeof(readbuf)) == -1)
-          {
-            llarp::LogDebug("close ev");
-            close_ev(ev);
-          }
+          ev->read(readbuf, sizeof(readbuf));
         }
         ++idx;
       }
@@ -142,10 +143,9 @@ struct llarp_epoll_loop : public llarp_ev_loop
   {
     epoll_event events[1024];
     int result;
-    byte_t readbuf[2048];
     do
     {
-      result = epoll_wait(epollfd, events, 1024, 10);
+      result = epoll_wait(epollfd, events, 1024, EV_TICK_INTERVAL);
       if(result > 0)
       {
         int idx = 0;
@@ -160,11 +160,7 @@ struct llarp_epoll_loop : public llarp_ev_loop
           llarp::ev_io* ev = static_cast< llarp::ev_io* >(events[idx].data.ptr);
           if(events[idx].events & EPOLLIN)
           {
-            if(ev->read(readbuf, sizeof(readbuf)) == -1)
-            {
-              llarp::LogDebug("close ev");
-              close_ev(ev);
-            }
+            ev->read(readbuf, sizeof(readbuf));
           }
           ++idx;
         }
@@ -271,6 +267,8 @@ struct llarp_epoll_loop : public llarp_ev_loop
     auto val = write(pipefds[1], &i, sizeof(i));
     (void)val;
   }
+
+  byte_t readbuf[2048];
 };
 
 #endif

llarp/ev_kqueue.hpp

@@ -98,6 +98,12 @@ struct llarp_kqueue_loop : public llarp_ev_loop
     return kqueuefd != -1;
   }
 
+  bool
+  running() const
+  {
+    return kqueuefd != -1;
+  }
+
   int
   tick(int ms)
   {
@@ -115,14 +121,7 @@ struct llarp_kqueue_loop : public llarp_ev_loop
       while(idx < result)
       {
         llarp::ev_io* ev = static_cast< llarp::ev_io* >(events[idx].udata);
-        if(ev->read(readbuf, sizeof(readbuf)) == -1)
-        {
-          llarp::LogInfo("tick close ev");
-          close_ev(ev);
-          // ev->fd = 0;
-          delete ev;
-          // break;
-        }
+        ev->read(readbuf, sizeof(readbuf));
         ++idx;
       }
     }
@@ -137,7 +136,7 @@ struct llarp_kqueue_loop : public llarp_ev_loop
   {
     timespec t;
     t.tv_sec  = 0;
-    t.tv_nsec = 10000UL;
+    t.tv_nsec = 1000UL * EV_TICK_INTERVAL;
     struct kevent events[1024];
     int result;
     byte_t readbuf[2048];
@@ -154,14 +153,7 @@ struct llarp_kqueue_loop : public llarp_ev_loop
           if(ev && ev->fd)
           {
             // printf("reading_ev [%x] fd[%d]\n", ev, ev->fd);
-            if(ev->read(readbuf, sizeof(readbuf)) == -1)
-            {
-              llarp::LogInfo("run error reading, should close ev");
-              close_ev(ev);
-              // ev->fd = 0;
-              // delete ev;
-              // break;
-            }
+            ev->read(readbuf, sizeof(readbuf));
           }
           else
           {

llarp/ev_win32.hpp

@@ -97,6 +97,8 @@ struct llarp_win32_loop : public llarp_ev_loop
 {
   HANDLE iocpfd;
 
+  bool _running = false;
+
   llarp_win32_loop() : iocpfd(INVALID_HANDLE_VALUE)
   {
     WSADATA wsockd;
@@ -123,10 +125,10 @@ struct llarp_win32_loop : public llarp_ev_loop
     if(iocpfd == INVALID_HANDLE_VALUE)
       iocpfd = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0);
 
-    if(iocpfd != INVALID_HANDLE_VALUE)
-      return true;  // we don't have a socket to attach to this IOCP yet
-
-    return false;
+    if(iocpfd == INVALID_HANDLE_VALUE)
+      return false;
+    _running = true;
+    return true;
   }
 
   // it works! -despair86, 3-Aug-18 @0420
@@ -153,11 +155,7 @@ struct llarp_win32_loop : public llarp_ev_loop
             reinterpret_cast< llarp::udp_listener* >(ev_id);
         if(ev && ev->fd)
         {
-          if(ev->getData(readbuf, sizeof(readbuf), iolen) == -1)
-          {
-            llarp::LogInfo("tick close ev");
-            close_ev(ev);
-          }
+          ev->getData(readbuf, sizeof(readbuf), iolen);
         }
       }
       ++idx;
@@ -203,11 +201,7 @@ struct llarp_win32_loop : public llarp_ev_loop
             reinterpret_cast< llarp::udp_listener* >(ev_id);
         if(ev && ev->fd)
         {
-          if(ev->getData(readbuf, sizeof(readbuf), iolen) == -1)
-          {
-            llarp::LogInfo("close ev");
-            close_ev(ev);
-          }
+          ev->getData(readbuf, sizeof(readbuf), iolen);
         }
       }
       ++idx;
@@ -321,10 +315,16 @@ struct llarp_win32_loop : public llarp_ev_loop
     return ret;
   }
 
+  bool
+  running() const
+  {
+    return _running;
+  }
+
   void
   stop()
   {
-    // do nothing, cancel io in close_ev()
+    _running = false;
   }
 };
 

llarp/mem.hpp

@@ -35,10 +35,6 @@ namespace llarp
       {
         printf("%c[1;31m", 27);
       }
-      else
-      {
-        printf("%c[0;0m", 27);
-      }
       if(std::isprint(buff.base[idx]))
       {
         printf("%c", buff.base[idx]);
@@ -47,6 +43,10 @@ namespace llarp
       {
         printf("X");
       }
+      if(buff.base + idx == buff.cur)
+      {
+        printf("%c[0;0m", 27);
+      }
       ++idx;
       if(idx % align == 0)
         printf("\n");

llarp/service.cpp

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

llarp/service/endpoint.cpp

@@ -88,6 +88,13 @@ namespace llarp
       return static_cast< Endpoint* >(user)->DoNetworkIsolation();
     }
 
+    bool
+    Endpoint::HasPendingPathToService(const Address& addr) const
+    {
+      return m_PendingServiceLookups.find(addr)
+          != m_PendingServiceLookups.end();
+    }
+
     void
     Endpoint::Tick(llarp_time_t now)
     {
@@ -166,6 +173,8 @@ namespace llarp
         }
         for(const auto& introset : itr->second.result)
         {
+          if(HasPendingPathToService(introset.A.Addr()))
+            continue;
           PathAlignJob* j = new PathAlignJob(introset.A.Addr());
           if(!EnsurePathToService(j->remote,
                                   std::bind(&PathAlignJob::HandleResult, j,
@@ -173,7 +182,7 @@ namespace llarp
                                   10000))
           {
             llarp::LogWarn("failed to ensure path to ", introset.A.Addr(),
-                           " for tag");
+                           " for tag ", tag.ToString());
             delete j;
           }
         }
@@ -432,7 +441,7 @@ namespace llarp
     Endpoint::CachedTagResult::BuildRequestMessage()
     {
       llarp::routing::DHTMessage* msg = new llarp::routing::DHTMessage();
-      msg->M.push_back(new llarp::dht::FindIntroMessage(tag, txid));
+      msg->M.push_back(new llarp::dht::FindIntroMessage({}, tag, txid));
       lastRequest = llarp_time_now_ms();
       parent->PutLookup(this, txid);
       return msg;
@@ -498,6 +507,7 @@ namespace llarp
       bool
       HandleResponse(const std::set< IntroSet >& results)
       {
+        llarp::LogInfo("found ", results.size(), " for ", remote.ToString());
         if(results.size() == 1)
         {
           llarp::LogInfo("hidden service lookup for ", remote.ToString(),