From 3e3679bebf2ee1b3c5e481e4b2e57b3cc44f7154 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Adrien=20B=C3=A9raud?= <adrien.beraud@savoirfairelinux.com>
Date: Mon, 18 Jun 2018 15:52:20 -0400
Subject: [PATCH] dhtrunner: drop packet if processing took too long

---
 include/opendht/dhtrunner.h |  8 +++++++-
 src/dhtrunner.cpp           | 15 +++++++++------
 2 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h
index 2bf98df7..8a74aca4 100644
--- a/include/opendht/dhtrunner.h
+++ b/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};
diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp
index c0921f5c..05d8b0df 100644
--- a/src/dhtrunner.cpp
+++ b/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();
                     }