dhtrunner: force execution of pending ops on shutdown

src/dht.cpp

@@ -873,6 +873,11 @@ Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filte
 size_t
 Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
 {
+    if (not id) {
+        if (logger_)
+            logger_->w(id, "Listen called with invalid key");
+        return 0;
+    }
     scheduler.syncTime();
 
     auto token = ++listener_token;
@@ -955,7 +960,9 @@ struct GetStatus : public OpStatus {
 void
 Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
 {
-    if (not val) {
+    if (not id or not val) {
+        if (logger_)
+            logger_->w(id, "Put called with invalid key or value");
         if (callback)
             callback(false, {});
         return;
@@ -1025,6 +1032,13 @@ bool callbackWrapper(Cb get_cb, DoneCallback done_cb, const std::vector<Sp<T>>&
 void
 Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where)
 {
+    if (not id) {
+        if (logger_)
+            logger_->w(id, "Get called with invalid key");
+        if (donecb)
+            donecb(false, {});
+        return;
+    }
     scheduler.syncTime();
 
     auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
@@ -1065,6 +1079,13 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 
 void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Query&& q)
 {
+    if (not id) {
+        if (logger_)
+            logger_->w(id, "Query called with invalid key");
+        if (done_cb)
+            done_cb(false, {});
+        return;
+    }
     scheduler.syncTime();
     auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
     auto f = q.where.getFilter();

src/dhtrunner.cpp

@@ -294,6 +294,11 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
         if (dht_)
             dht_->shutdown(onShutdown, stop);
     });
+    // Force execution of pending ops
+    if (getStatus() == NodeStatus::Connecting) {
+        status4 = NodeStatus::Disconnected;
+        status6 = NodeStatus::Disconnected;
+    }
     cv.notify_all();
 }
 
@@ -368,6 +373,7 @@ DhtRunner::join()
     {
         std::lock_guard<std::mutex> lck(dht_mtx);
         resetDht();
+        std::lock_guard<std::mutex> lck2(storage_mtx);
         status4 = NodeStatus::Disconnected;
         status6 = NodeStatus::Disconnected;
     }
@@ -703,8 +709,11 @@ DhtRunner::loop_()
     NodeStatus nstatus4 = dht->updateStatus(AF_INET);
     NodeStatus nstatus6 = dht->updateStatus(AF_INET6);
     if (nstatus4 != status4 || nstatus6 != status6) {
-        status4 = nstatus4;
-        status6 = nstatus6;
+        {
+            std::lock_guard<std::mutex> lck(storage_mtx);
+            status4 = nstatus4;
+            status6 = nstatus6;
+        }
         if (statusCb)
             statusCb(status4, status6);
     }