dhtrunner: drop packet if processing took too long

include/opendht/dhtrunner.h

@@ -457,7 +457,13 @@ private:
 
     std::thread rcv_thread {};
     std::mutex sock_mtx {};
-    std::vector<std::pair<Blob, SockAddr>> rcv {};
+
+    struct ReceivedPacket {
+        Blob data;
+        SockAddr from;
+        time_point received;
+    };
+    std::queue<ReceivedPacket> rcv {};
 
     /** true if currently actively boostraping */
     std::atomic_bool bootstraping {false};

src/dhtrunner.cpp

@@ -409,12 +409,15 @@ DhtRunner::loop_()
         received = std::move(rcv);
     }
     if (not received.empty()) {
-        for (const auto& pck : received) {
-            auto& buf = pck.first;
-            auto& from = pck.second;
-            wakeup = dht->periodic(buf.data(), buf.size()-1, from);
+        while (not received.empty()) {
+            auto& pck = received.front();
+            auto delay = clock::now() - pck.received;
+            if (delay > std::chrono::seconds(1))
+                std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
+            else
+                wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
+            received.pop();
         }
-        received.clear();
     } else {
         wakeup = dht->periodic(nullptr, 0, nullptr, 0);
     }
@@ -520,7 +523,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                     if (rc > 0) {
                         {
                             std::lock_guard<std::mutex> lck(sock_mtx);
-                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len));
+                            rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
                         }
                         cv.notify_all();
                     }