Fix client self cancel

ioc/pvalink_channel.cpp

@@ -267,6 +267,7 @@ void pvaLinkChannel::put(bool force)
         // start net Put, cancels in-progress put
         op_put = linkGlobal->provider_remote.put(key.first)
                 .rawRequest(pvReq)
+                .syncCancel(false)
                 .build([this](Value&& prototype) -> Value
         {
                 return linkBuildPut(this, std::move(prototype)); // TODO

src/clientdiscover.cpp

@@ -15,6 +15,27 @@ DEFINE_LOGGER(io, "pvxs.client.io");
 namespace pvxs {
 namespace client {
 
+struct Discovery final : public OperationBase
+{
+    const std::shared_ptr<ContextImpl> context;
+    std::function<void(const Discovered &)> notify;
+    bool notify_busy = false;
+    bool running = false;
+
+    Discovery(const std::shared_ptr<ContextImpl>& context, const std::string& name);
+    ~Discovery();
+
+    virtual bool cancel() override final;
+private:
+    bool _cancel(bool implicit);
+
+    // unused for this special case
+    virtual void _reExecGet(std::function<void (Result &&)> &&resultcb) override final;
+    virtual void _reExecPut(const Value &arg, std::function<void (Result &&)> &&resultcb) override final;
+    virtual void createOp() override final;
+    virtual void disconnected(const std::shared_ptr<OperationBase> &self) override final;
+};
+
 Discovery::Discovery(const std::shared_ptr<ContextImpl> &context, const std::string& name)
     :OperationBase (Operation::Discover, context->tcp_loop, name)
     ,context(context)
@@ -31,7 +52,8 @@ bool Discovery::cancel()
     bool ret;
     loop.call([this, &junk, &ret](){
         ret = _cancel(false);
-        junk = std::move(notify);
+        if(!notify_busy)
+            junk = std::move(notify);
         // leave opByIOID for GC
     });
     return ret;
@@ -71,6 +93,12 @@ std::shared_ptr<Operation> DiscoverBuilder::exec()
         // (maybe) user thread
         auto loop(op->context->tcp_loop);
         auto temp(std::move(op));
+        if(syncCancel && loop.inLoop())
+            log_err_printf(io,
+                           "syncCancel discover '%s' being destroyed on worker thread.\n"
+                           "Possible self-reference loop.  Setup with .syncCancel(false) if intended.\n",
+                           temp->channelName.c_str());
+
         loop.tryInvoke(syncCancel, std::bind([](std::shared_ptr<Discovery>& op){
                            // on worker
                            op->context->discoverers.erase(op.get());
@@ -106,11 +134,13 @@ void ContextImpl::serverEvent(const Discovered &evt)
 {
     for(auto& pair : discoverers) {
         if(auto dis = pair.second.lock()) {
+            dis->notify_busy = true;
             try {
                 dis->notify(evt);
             } catch(std::exception& e) {
                 log_exc_printf(io, "Unhandled exception during Discovery callback : %s\n", e.what());
             }
+            dis->notify_busy = false;
         }
     }
 }

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());
@@ -592,6 +613,12 @@ std::shared_ptr<Operation> gpr_setup(const std::shared_ptr<ContextImpl>& context
         // (maybe) user thread
         auto temp(std::move(internal));
         auto loop(temp->loop);
+        if(syncCancel && loop.inLoop())
+            log_err_printf(io,
+                           "syncCancel op%u '%s' being destroyed on worker thread.\n"
+                           "Possible self-reference loop.  Setup with .syncCancel(false) if intended.\n",
+                           temp->op, temp->channelName.c_str());
+
         // std::bind for lack of c++14 generalized capture
         // to move internal ref to worker for dtor
         loop.tryInvoke(syncCancel, std::bind([](std::shared_ptr<GPROp>& op) {

src/clientimpl.h

@@ -25,6 +25,7 @@ namespace client {
 
 struct Channel;
 struct ContextImpl;
+struct Discovery;
 
 struct ResultWaiter {
     epicsMutex lock;
@@ -228,26 +229,6 @@ struct Channel {
                                    const std::string& server);
 };
 
-struct Discovery final : public OperationBase
-{
-    const std::shared_ptr<ContextImpl> context;
-    std::function<void(const Discovered &)> notify;
-    bool running = false;
-
-    Discovery(const std::shared_ptr<ContextImpl>& context, const std::string& name);
-    ~Discovery();
-
-    virtual bool cancel() override final;
-private:
-    bool _cancel(bool implicit);
-
-    // unused for this special case
-    virtual void _reExecGet(std::function<void (Result &&)> &&resultcb) override final;
-    virtual void _reExecPut(const Value &arg, std::function<void (Result &&)> &&resultcb) override final;
-    virtual void createOp() override final;
-    virtual void disconnected(const std::shared_ptr<OperationBase> &self) override final;
-};
-
 struct ContextImpl : public std::enable_shared_from_this<ContextImpl>
 {
     SockAttach attach;

src/clientintrospect.cpp

@@ -21,6 +21,7 @@ struct InfoOp : public OperationBase
 {
     std::function<void(Result&&)> done;
     Value result;
+    bool done_busy = false;
 
     enum state_t {
         Connecting, // waiting for an active Channel
@@ -45,7 +46,8 @@ struct InfoOp : public OperationBase
         bool ret = false;
         (void)loop.tryCall([this, &junk, &ret](){
             ret = _cancel(false);
-            junk = std::move(done);
+            if(!done_busy)
+                junk = std::move(done);
             // leave opByIOID for GC
         });
         return ret;
@@ -160,18 +162,19 @@ void Connection::handle_GET_FIELD()
     info->state = InfoOp::Done;
 
     if(info->done) {
-        auto done = std::move(info->done);
         Result res;
         if(sts.isSuccess()) {
             res = Result(std::move(prototype), peerName);
         } else {
             res = Result(std::make_exception_ptr(RemoteError(sts.msg)));
         }
+        info->done_busy = true;
         try {
-            done(std::move(res));
+            info->done(std::move(res));
         }catch(std::exception& e){
             log_exc_printf(setup, "Unhandled exception %s in Info result() callback: %s\n", typeid (e).name(), e.what());
         }
+        info->done_busy = false;
 
     } else {
         info->result = prototype;
@@ -204,6 +207,12 @@ std::shared_ptr<Operation> GetBuilder::_exec_info()
         // from user thread
         auto temp(std::move(op));
         auto loop(temp->loop);
+        if(syncCancel && loop.inLoop())
+            log_err_printf(io,
+                           "syncCancel info '%s' being destroyed on worker thread.\n"
+                           "Possible self-reference loop.  Setup with .syncCancel(false) if intended.\n",
+                           temp->channelName.c_str());
+
         // std::bind for lack of c++14 generalized capture
         // to move internal ref to worker for dtor
         loop.tryInvoke(syncCancel, std::bind([](std::shared_ptr<InfoOp>& op) {

src/clientmon.cpp

@@ -46,6 +46,8 @@ struct SubscriptionImpl final : public OperationBase, public Subscription
     std::function<void (Subscription&, const Value&)> onInit;
     std::function<void(Subscription&)> event;
     Value pvRequest;
+    bool event_busy = false;
+    bool onInit_busy = false;
     bool pipeline = false;
     bool autostart = true;
     bool maskConn = false, maskDiscon = true;
@@ -109,12 +111,14 @@ struct SubscriptionImpl final : public OperationBase, public Subscription
     void doNotify()
     {
         if(event) {
+            event_busy = true;
             try {
                 event(*this);
             }catch(std::exception& e){
                 log_exc_printf(io, "Unhandled user exception in Monitor %s %s : %s\n",
                                 __func__, typeid (e).name(), e.what());
             }
+            event_busy = false;
         }
     }
 
@@ -276,17 +280,23 @@ struct SubscriptionImpl final : public OperationBase, public Subscription
     virtual void _onEvent(std::function<void(Subscription&)>&& fn) override final {
         decltype (event) junk;
         loop.call([this, &junk, &fn]() {
+            if(event_busy)
+                throw std::logic_error("Must not replace Subscription::onEvent() while callback in progress");
             junk = std::move(event);
             this->event = std::move(fn);
         });
     }
 
     virtual bool cancel() override final {
         decltype (event) junk;
+        decltype (onInit) junkI;
         bool ret = false;
-        (void)loop.tryCall([this, &junk, &ret](){
+        (void)loop.tryCall([this, &junk, &junkI, &ret](){
             ret = _cancel(false);
-            junk = std::move(event);
+            if(!event_busy)
+                junk = std::move(event); // trash when cancelled from app. worker
+            if(!onInit_busy)
+                junkI = std::move(onInit);
             // leave opByIOID for GC
         });
         return ret;
@@ -297,10 +307,12 @@ struct SubscriptionImpl final : public OperationBase, public Subscription
             log_info_printf(io, "Server %s channel %s monitor implied cancel\n",
                             chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
                             chan->name.c_str());
+
+        } else {
+            log_info_printf(io, "Server %s channel %s monitor cancel\n",
+                            chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
+                            chan->name.c_str());
         }
-        log_info_printf(io, "Server %s channel %s monitor cancel\n",
-                        chan->conn ? chan->conn->peerName.c_str() : "<disconnected>",
-                        chan->name.c_str());
 
         if(state==Idle || state==Running) {
             chan->conn->sendDestroyRequest(chan->sid, ioid);
@@ -624,6 +636,7 @@ void Connection::handle_MONITOR()
 
         mon->state = SubscriptionImpl::Idle;
 
+        mon->onInit_busy = true;
         try {
             if(mon->onInit)
                 mon->onInit(*mon, info->prototype);
@@ -634,6 +647,7 @@ void Connection::handle_MONITOR()
                             peerName.c_str(),
                             mon->chan->name.c_str(), e.what());
         }
+        mon->onInit_busy = false;
 
         if(mon->autostart && mon->state == SubscriptionImpl::Idle)
             mon->resume();
@@ -816,6 +830,12 @@ std::shared_ptr<Subscription> MonitorBuilder::exec()
         // from user thread
         auto temp(std::move(op));
         auto loop(temp->loop);
+        if(syncCancel && loop.inLoop())
+            log_err_printf(io,
+                           "syncCancel monitor '%s' being destroyed on worker thread.\n"
+                           "Possible self-reference loop.  Setup with .syncCancel(false) if intended.\n",
+                           temp->channelName.c_str());
+
         // std::bind for lack of c++14 generalized capture
         // to move internal ref to worker for dtor
         loop.tryInvoke(syncCancel, std::bind([](std::shared_ptr<SubscriptionImpl>& op) {

src/evhelper.cpp

@@ -348,6 +348,11 @@ bool evbase::_call(mfunction&& fn, bool dothrow) const
     return true;
 }
 
+bool evbase::inLoop() const
+{
+    return pvt->worker.isCurrentThread();
+}
+
 void evbase::assertInLoop() const
 {
     if(!pvt->worker.isCurrentThread()) {

src/evhelper.h

@@ -198,6 +198,7 @@ struct PVXS_API evbase {
             return tryDispatch(std::move(fn));
     }
 
+    bool inLoop() const;
     void assertInLoop() const;
     //! Caller must be on the worker, or the worker must be stopped.
     //! @returns true if working is running.