clientget: Guard callback functors during execution

src/clientget.cpp

@@ -119,6 +119,9 @@ struct GPROp : public OperationBase
     // For GET, PUT a duplicate of RequestInfo::prototype
     Value arg;
     Result result;
+    bool builder_busy = false;
+    bool done_busy = false;
+    bool onInit_busy = false;
     bool getOput = false;
     bool autoExec = true;
 
@@ -156,6 +159,7 @@ struct GPROp : public OperationBase
     }
 
     void notify() {
+        done_busy = true;
         try {
             if(done)
                 done(std::move(result));
@@ -168,17 +172,23 @@ struct GPROp : public OperationBase
             if(!result.error())
                 result = Result(std::current_exception());
         }
+        done_busy = false;
     }
 
     virtual bool cancel() override final
     {
-        decltype (done) junk;
+        decltype (builder) junkB;
+        decltype (done) junkD;
         decltype (onInit) junkI;
         bool ret = false;
-        (void)loop.tryCall([this, &junk, &junkI, &ret](){
+        (void)loop.tryCall([&, this](){
             ret = _cancel(false);
-            junk = std::move(done);
-            junkI = std::move(onInit);
+            if(!builder_busy)
+                junkB = std::move(builder);
+            if(!done_busy)
+                junkD = std::move(done);
+            if(!onInit_busy)
+                junkI = std::move(onInit);
             // leave opByIOID for GC
         });
         return ret;
@@ -218,10 +228,16 @@ struct GPROp : public OperationBase
             if(self->state!=Idle)
                 return;
 
+            if(self->done_busy)
+                throw std::logic_error("done callback in progress");
+
             if(self->op==RPC) {
                 self->arg = std::move(a);
 
             } else if(put && self->op==Put) {
+                if(self->builder_busy)
+                    throw std::logic_error("builder callback in progress");
+
                 self->builder = [a](Value&&) noexcept -> Value {
                     // caller should be passing a Value of the correct prototype
                     // given through onInit().
@@ -275,6 +291,7 @@ struct GPROp : public OperationBase
         if(state==GPROp::BuildPut) {
             temp = arg.clone();
 
+            builder_busy = true;
             try {
                 temp = builder(std::move(temp));
                 state = GPROp::Exec;
@@ -283,6 +300,7 @@ struct GPROp : public OperationBase
                 result = Result(std::current_exception());
                 state = GPROp::Done;
             }
+            builder_busy = false;
         }
 
         // act on new operation state
@@ -517,10 +535,13 @@ void Connection::handle_GPR(pva_app_msg_t cmd)
         if(cmd==CMD_PUT || cmd==CMD_GET)
             gpr->arg = data; // save for later use in sendReply() when RequestInfo not available
 
+        gpr->onInit_busy = true;
         try {
             if(gpr->onInit)
                 gpr->onInit(data);
+            gpr->onInit_busy = false;
         } catch(std::exception& e) {
+            gpr->onInit_busy = false;
             log_err_printf(setup, "Server %s op%02x \"%s\" onInit() error: %s\n",
                            peerName.c_str(), cmd, gpr->chan->name.c_str(), e.what());
             gpr->result = Result(std::current_exception());