From c969383cf5c8411331a8b1d3d1fc8c50fa18083e Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Wed, 8 Jul 2026 16:46:27 -0700 Subject: [PATCH 1/6] clientmon: Guard event callback functor during execution It is not safe to allow the event callback function to be freed while it is executing. This is UB. Currently possible by cancellation, or the onEvent expert method. 1. onEvent() from a callback becomes logic_error 2. Explicit Subscription::cancel() during callback is made safe. Defer free to a later repeat cancel() or dtor. 3. Apply the same logic to onInit --- src/clientmon.cpp | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/src/clientmon.cpp b/src/clientmon.cpp index af5c4226d..6ce3b9e09 100644 --- a/src/clientmon.cpp +++ b/src/clientmon.cpp @@ -46,6 +46,8 @@ struct SubscriptionImpl final : public OperationBase, public Subscription std::function onInit; std::function 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,6 +280,8 @@ struct SubscriptionImpl final : public OperationBase, public Subscription virtual void _onEvent(std::function&& 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); }); @@ -283,10 +289,14 @@ struct SubscriptionImpl final : public OperationBase, public Subscription 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() : "", chan->name.c_str()); + + } else { + log_info_printf(io, "Server %s channel %s monitor cancel\n", + chan->conn ? chan->conn->peerName.c_str() : "", + chan->name.c_str()); } - log_info_printf(io, "Server %s channel %s monitor cancel\n", - chan->conn ? chan->conn->peerName.c_str() : "", - 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(); From c17812a4298ba67bea5f9de2d009cb67d238bcf7 Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Wed, 8 Jul 2026 17:47:57 -0700 Subject: [PATCH 2/6] clientget: Guard callback functors during execution --- src/clientget.cpp | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/src/clientget.cpp b/src/clientget.cpp index 59320b48b..092be6629 100644 --- a/src/clientget.cpp +++ b/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()); From eab32759df76e41748cab16320ccc9e715de771f Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Wed, 8 Jul 2026 17:51:04 -0700 Subject: [PATCH 3/6] clientdiscover: Guard callback functors during execution Also privatize struct Discovery with added notify_busy flag. --- src/clientdiscover.cpp | 26 +++++++++++++++++++++++++- src/clientimpl.h | 21 +-------------------- 2 files changed, 26 insertions(+), 21 deletions(-) diff --git a/src/clientdiscover.cpp b/src/clientdiscover.cpp index f9dc438c4..1fc14c9ae 100644 --- a/src/clientdiscover.cpp +++ b/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 context; + std::function notify; + bool notify_busy = false; + bool running = false; + + Discovery(const std::shared_ptr& 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 &&resultcb) override final; + virtual void _reExecPut(const Value &arg, std::function &&resultcb) override final; + virtual void createOp() override final; + virtual void disconnected(const std::shared_ptr &self) override final; +}; + Discovery::Discovery(const std::shared_ptr &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; @@ -106,11 +128,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; } } } diff --git a/src/clientimpl.h b/src/clientimpl.h index 0047a639c..c437330f2 100644 --- a/src/clientimpl.h +++ b/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 context; - std::function notify; - bool running = false; - - Discovery(const std::shared_ptr& 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 &&resultcb) override final; - virtual void _reExecPut(const Value &arg, std::function &&resultcb) override final; - virtual void createOp() override final; - virtual void disconnected(const std::shared_ptr &self) override final; -}; - struct ContextImpl : public std::enable_shared_from_this { SockAttach attach; From d5ecc88f3d34eca84cd869b6f7cb80ecf8df17ca Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Wed, 8 Jul 2026 17:58:32 -0700 Subject: [PATCH 4/6] clientintrospect: Guard callback functors during execution --- src/clientintrospect.cpp | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/clientintrospect.cpp b/src/clientintrospect.cpp index 1a011ee66..e93d15714 100644 --- a/src/clientintrospect.cpp +++ b/src/clientintrospect.cpp @@ -21,6 +21,7 @@ struct InfoOp : public OperationBase { std::function 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; From a79d65ba15be2866bb381d119f31b13cbb62b964 Mon Sep 17 00:00:00 2001 From: George McIntyre Date: Wed, 8 Jul 2026 21:19:18 -0700 Subject: [PATCH 5/6] testmon: cancelDuringEvent() --- test/testmon.cpp | 36 +++++++++++++++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/test/testmon.cpp b/test/testmon.cpp index e7f7f2daf..20875f9a7 100644 --- a/test/testmon.cpp +++ b/test/testmon.cpp @@ -128,6 +128,39 @@ struct BasicTest { testOk1(!done->wait(1.1)); } + void cancelDuringEvent() + { + testShow()<<__func__; + + mbox.open(initial); + serv.start(); + + // A heap-allocated captured value the callback reads AFTER cancelling. + // If the functor (and this capture) were freed by the in-call cancel, + // touching `canary` would be a use-after-free (caught under ASan). + auto canary(std::make_shared("alive")); + auto fired(std::make_shared()); + auto ok(std::make_shared>(false)); + + std::shared_ptr self; + self = cli.monitor("mailbox") + .maskConnected(true) + .maskDisconnected(true) + .event([canary, fired, ok](client::Subscription& s) { + // Re-entrant cancel on this same loop: move-destroys `event`. + s.cancel(); + // Touch the capture after the cancel; must still be valid. + ok->store(*canary == "alive"); + fired->signal(); + }) + .exec(); + + post(1); + + testOk1(fired->wait(5.0)); + testOk1(ok->load()); + } + void badRequest() { testShow()<<__func__; @@ -381,13 +414,14 @@ struct TestReconn : public BasicTest MAIN(testmon) { - testPlan(43); + testPlan(45); testSetup(); try{ logger_config_env(); BasicTest().orphan(); BasicTest().cancel(); BasicTest().asyncCancel(); + BasicTest().cancelDuringEvent(); BasicTest().badRequest(); BasicTest().testNoMark(); TestLifeCycle().testBasic(true); From cc7bc72dd7676c72871889c8586014947567ed1d Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Thu, 9 Jul 2026 11:46:11 -0700 Subject: [PATCH 6/6] client: log error when syncCancel op dtor on worker Log an error when an operation created with syncCancel=true is allowed to be free'd asynchronously on the client worker. This may indicate that the Operation/Subscription is directly owned by its own callback. Creating a self-reference loop. Or it might indicate a more complicated, but safe, situation like pvalink where a new operation is initiated as the previous one completes. Use the syncCancel=false flag, opting of blocking cancellation, as an indication that this risk is accepted by user code. --- ioc/pvalink_channel.cpp | 1 + src/clientdiscover.cpp | 6 ++++++ src/clientget.cpp | 6 ++++++ src/clientintrospect.cpp | 6 ++++++ src/clientmon.cpp | 6 ++++++ src/evhelper.cpp | 5 +++++ src/evhelper.h | 1 + 7 files changed, 31 insertions(+) diff --git a/ioc/pvalink_channel.cpp b/ioc/pvalink_channel.cpp index 9785e00e6..b119f81aa 100644 --- a/ioc/pvalink_channel.cpp +++ b/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 diff --git a/src/clientdiscover.cpp b/src/clientdiscover.cpp index 1fc14c9ae..d40f7921b 100644 --- a/src/clientdiscover.cpp +++ b/src/clientdiscover.cpp @@ -93,6 +93,12 @@ std::shared_ptr 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& op){ // on worker op->context->discoverers.erase(op.get()); diff --git a/src/clientget.cpp b/src/clientget.cpp index 092be6629..d167acc56 100644 --- a/src/clientget.cpp +++ b/src/clientget.cpp @@ -613,6 +613,12 @@ std::shared_ptr gpr_setup(const std::shared_ptr& 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& op) { diff --git a/src/clientintrospect.cpp b/src/clientintrospect.cpp index e93d15714..a0ccb1141 100644 --- a/src/clientintrospect.cpp +++ b/src/clientintrospect.cpp @@ -207,6 +207,12 @@ std::shared_ptr 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& op) { diff --git a/src/clientmon.cpp b/src/clientmon.cpp index 6ce3b9e09..c8d302288 100644 --- a/src/clientmon.cpp +++ b/src/clientmon.cpp @@ -830,6 +830,12 @@ std::shared_ptr 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& op) { diff --git a/src/evhelper.cpp b/src/evhelper.cpp index 41e8f5282..e4016e1c8 100644 --- a/src/evhelper.cpp +++ b/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()) { diff --git a/src/evhelper.h b/src/evhelper.h index 67d1a90f0..c8c155022 100644 --- a/src/evhelper.h +++ b/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.