From cdc2fdf57be7c5e7c7c64c8571a503d4f8d6f92f Mon Sep 17 00:00:00 2001 From: Mykola Date: Fri, 23 Oct 2026 18:01:14 +0300 Subject: [PATCH 1/2] feat: add input port update subscriptions --- docs/INPUT_PORT_CALLBACKS.md | 89 +++++ include/behaviortree_cpp/blackboard.h | 34 +- .../behaviortree_cpp/scripting/operators.hpp | 12 +- include/behaviortree_cpp/tree_node.h | 13 + .../utils/thread_safe_signal.h | 86 +++++ src/blackboard.cpp | 72 +++- src/tree_node.cpp | 49 +++ tests/CMakeLists.txt | 1 + tests/gtest_input_port_callbacks.cpp | 316 ++++++++++++++++++ 9 files changed, 666 insertions(+), 6 deletions(-) create mode 100644 docs/INPUT_PORT_CALLBACKS.md create mode 100644 include/behaviortree_cpp/utils/thread_safe_signal.h create mode 100644 tests/gtest_input_port_callbacks.cpp diff --git a/docs/INPUT_PORT_CALLBACKS.md b/docs/INPUT_PORT_CALLBACKS.md new file mode 100644 index 000000000..42b72c2bb --- /dev/null +++ b/docs/INPUT_PORT_CALLBACKS.md @@ -0,0 +1,89 @@ +# Input port callbacks + +`TreeNode::subscribeToInputPort()` observes writes to the blackboard key backing +an input port. It is an instance method, so a callback can use node-specific +state. Port declarations in `providedPorts()` remain unchanged. + +For example, a node can react to a new topic name before its next tick: + +```cpp +class TopicListener : public BT::SyncActionNode +{ +public: + TopicListener(const std::string& name, const BT::NodeConfig& config) + : BT::SyncActionNode(name, config) + { + topic_subscription_ = subscribeToInputPort("topic", [this] { + if(auto topic = getInput("topic")) + { + configureSubscription(topic.value()); + } + }); + } + + static BT::PortsList providedPorts() + { + return { BT::InputPort("topic") }; + } + + BT::NodeStatus tick() override; + +private: + void configureSubscription(const std::string& topic); + BT::Blackboard::EntryUpdatedSubscriber topic_subscription_; +}; +``` + +The application must stop writers and wait for callbacks to finish before +destroying the node. Resetting the handle alone is not a synchronization barrier. +The example assumes the application serializes writes and access to the node's +subscription state. + +## Notification contract + +- Retain the returned handle to keep the callback registered. Resetting or + destroying it unsubscribes; a callback already selected for invocation may + still complete. +- Registration does not invoke the callback. Initialize from `getInput()` + separately when an initial value is needed. With concurrent writers, coordinate + initialization with the application's synchronization. +- Successful writes notify even if the new value equals the previous value. + Generic port values do not necessarily have an equality operation. +- Callbacks run synchronously on the writer's thread, outside blackboard locks. + They may call `getInput()`, write other keys, and change subscriptions. + Writing the observed key can recursively invoke callbacks. +- Concurrent writes may invoke the same callback concurrently. A callback reads + the latest value, not a snapshot of the triggering write; delivery order is + not guaranteed across writers. +- Callback exceptions propagate after the write has committed. As with other + synchronous signals, a throwing callback prevents the remaining callbacks in + that notification from running. + +## Supported inputs and writes + +Explicit subtree remapping, automatic remapping, `{=}`, `{@root_key}`, and +blackboard-backed port defaults are supported. A literal or constant default +returns an empty handle because it cannot change through the blackboard. +Missing inputs, output-only ports, invalid blackboards, and empty callbacks are +reported with `LogicError`. + +Subscriptions bind to the resolved key at registration time. Reconfiguring port +or subtree remapping requires a new subscription. Subscriptions survive removing +and recreating that key. Removal itself does not notify. + +Writes through `Blackboard::set()`, `TreeNode::setOutput()`, scripting assignments, +JSON import, and `Blackboard::cloneInto()` notify. Clone notifications run after +the entire snapshot has been copied, allowing a callback to read other restored +entries. Raw writes through `Entry`, `AnyPtrLocked`, or shared objects are not +observable automatically. Call `Blackboard::notifyEntryUpdated()` after releasing +the entry lock to explicitly publish such a write. + +The lower-level `Blackboard::subscribeToEntry()` supports subscriptions before +the first assignment without creating or changing the type of an entry. + +## Compatibility + +This adds an opt-in API; existing port declarations and unsubscribed nodes keep +their behavior. The additional subscription storage changes `Blackboard`'s +binary layout. Rebuild the library and clients together; this API is proposed +for the next major version, as requested by issue #872's BT.CPP-5 label. diff --git a/include/behaviortree_cpp/blackboard.h b/include/behaviortree_cpp/blackboard.h index b4e7f90c8..df359305a 100644 --- a/include/behaviortree_cpp/blackboard.h +++ b/include/behaviortree_cpp/blackboard.h @@ -10,6 +10,7 @@ #include "behaviortree_cpp/utils/locked_reference.hpp" #include "behaviortree_cpp/utils/polymorphic_cast_registry.hpp" #include "behaviortree_cpp/utils/safe_any.hpp" +#include "behaviortree_cpp/utils/thread_safe_signal.h" #include #include @@ -39,6 +40,9 @@ class Blackboard { public: using Ptr = std::shared_ptr; + using EntryUpdatedSignal = ThreadSafeSignal<>; + using EntryUpdatedSubscriber = EntryUpdatedSignal::Subscriber; + using EntryUpdatedCallback = EntryUpdatedSignal::CallableFunction; protected: // This is intentionally protected. Use Blackboard::create instead @@ -119,6 +123,27 @@ class Blackboard template void set(const std::string& key, const T& value); + /** Subscribe to successful writes to a key, including its first assignment. + * Subtree remapping and root keys are resolved at subscription time. + * The subscription survives unset()/set() of the same key. No initial + * notification is sent, and assigning the same value still notifies. + * + * Callbacks run synchronously on the writer's thread, outside blackboard + * locks, and may read the latest value. Concurrent writers may invoke a + * callback concurrently; intermediate values are not retained. Exceptions + * propagate to the writer after the value has been stored. + * Resetting the handle prevents future calls but does not wait for a call + * already in progress. Stop writers before destroying callback targets. + */ + [[nodiscard]] EntryUpdatedSubscriber subscribeToEntry(const std::string& key, + EntryUpdatedCallback callback); + + /** Notify subscribers after directly modifying an entry. Release its lock + * before calling this function. Raw Entry/AnyPtrLocked mutations do not + * notify automatically; set(), scripts, JSON import and cloneInto() do. + */ + void notifyEntryUpdated(const std::string& key); + void unset(const std::string& key); [[nodiscard]] const TypeInfo* entryInfo(const std::string& key); @@ -190,11 +215,15 @@ class Blackboard private: mutable std::shared_mutex storage_mutex_; std::unordered_map> storage_; + std::unordered_map> update_signals_; std::weak_ptr parent_bb_; std::unordered_map internal_to_external_; std::shared_ptr createEntryImpl(const std::string& key, const TypeInfo& info); + std::shared_ptr getUpdateSignal(const std::string& key, + bool create); + bool autoremapping_ = false; // Optional registry for polymorphic shared_ptr conversions @@ -326,7 +355,7 @@ inline void Blackboard::set(const std::string& key, const T& value) // if the type is the same or not. Entry& entry = *entry_ptr; - std::scoped_lock scoped_lock(entry.entry_mutex); + std::unique_lock scoped_lock(entry.entry_mutex); Any& previous_any = entry.value; Any new_value(value); @@ -339,6 +368,8 @@ inline void Blackboard::set(const std::string& key, const T& value) entry.sequence_id++; entry.stamp = std::chrono::steady_clock::now().time_since_epoch(); previous_any = std::move(new_value); + scoped_lock.unlock(); + notifyEntryUpdated(key); return; } @@ -394,6 +425,7 @@ inline void Blackboard::set(const std::string& key, const T& value) entry.sequence_id++; entry.stamp = std::chrono::steady_clock::now().time_since_epoch(); } + notifyEntryUpdated(key); } template diff --git a/include/behaviortree_cpp/scripting/operators.hpp b/include/behaviortree_cpp/scripting/operators.hpp index 0f7d74c78..0e8d4358f 100644 --- a/include/behaviortree_cpp/scripting/operators.hpp +++ b/include/behaviortree_cpp/scripting/operators.hpp @@ -562,7 +562,7 @@ struct ExprAssignment : ExprBase } auto value = rhs->evaluate(env); - std::scoped_lock lock(entry->entry_mutex); + std::unique_lock lock(entry->entry_mutex); auto* dst_ptr = &entry->value; auto errorPrefix = [dst_ptr, &key]() { @@ -624,7 +624,10 @@ struct ExprAssignment : ExprBase } entry->sequence_id++; entry->stamp = std::chrono::steady_clock::now().time_since_epoch(); - return *dst_ptr; + const Any result = *dst_ptr; + lock.unlock(); + env.vars->notifyEntryUpdated(key); + return result; } if(dst_ptr->empty()) @@ -680,7 +683,10 @@ struct ExprAssignment : ExprBase temp_variable.copyInto(*dst_ptr); entry->sequence_id++; entry->stamp = std::chrono::steady_clock::now().time_since_epoch(); - return *dst_ptr; + const Any result = *dst_ptr; + lock.unlock(); + env.vars->notifyEntryUpdated(key); + return result; } }; } // namespace BT::Ast diff --git a/include/behaviortree_cpp/tree_node.h b/include/behaviortree_cpp/tree_node.h index 9fe48568d..20588fb2d 100644 --- a/include/behaviortree_cpp/tree_node.h +++ b/include/behaviortree_cpp/tree_node.h @@ -214,6 +214,19 @@ class TreeNode [[nodiscard]] StatusChangeSubscriber subscribeToStatusChange(StatusChangeCallback callback); + /** Subscribe to writes to the blackboard key backing an input port. + * Unlike providedPorts(), this is an instance method: a callback may use + * node-specific state and getInput(). Retain the returned handle for as long + * as the callback should remain active. See Blackboard::subscribeToEntry() + * for threading and lifetime requirements. + * + * Constant inputs return an empty handle. Missing inputs, output-only ports + * and empty callbacks throw LogicError. Defaults and {=} remapping are + * supported. Changing the port mapping later does not rebind subscriptions. + */ + [[nodiscard]] Blackboard::EntryUpdatedSubscriber + subscribeToInputPort(const std::string& key, Blackboard::EntryUpdatedCallback callback); + /** This method attaches to the TreeNode a callback with signature: * * NodeStatus callback(TreeNode& node) diff --git a/include/behaviortree_cpp/utils/thread_safe_signal.h b/include/behaviortree_cpp/utils/thread_safe_signal.h new file mode 100644 index 000000000..394a9ad30 --- /dev/null +++ b/include/behaviortree_cpp/utils/thread_safe_signal.h @@ -0,0 +1,86 @@ +#pragma once + +#include +#include +#include +#include + +namespace BT +{ +/** A signal supporting concurrent notification and subscription. + * Callbacks run outside the subscription lock and may be invoked concurrently. + * Dropping a subscriber prevents future calls, but does not wait for callbacks + * already selected for invocation. The owner must synchronize their lifetime + * with callback targets. + */ +template +class ThreadSafeSignal +{ +public: + using CallableFunction = std::function; + + class Subscription + { + friend class ThreadSafeSignal; + + public: + explicit Subscription(CallableFunction func) + : callback_(std::make_shared(std::move(func))) + {} + + private: + std::shared_ptr callback_; + }; + + using Subscriber = std::shared_ptr; + + [[nodiscard]] Subscriber subscribe(CallableFunction func) + { + auto subscriber = std::make_shared(std::move(func)); + const std::scoped_lock lock(mutex_); + removeExpired(); + subscribers_.emplace_back(subscriber); + return subscriber; + } + + void notify(CallableArgs... args) + { + std::vector> subscribers; + { + const std::scoped_lock lock(mutex_); + removeExpired(); + subscribers = subscribers_; + } + for(const auto& weak_subscriber : subscribers) + { + if(auto subscriber = weak_subscriber.lock()) + { + // Keep the callable alive without keeping its subscription alive. + // A callback can reset its own handle and recursively notify safely. + auto callback = subscriber->callback_; + subscriber.reset(); + (*callback)(args...); + } + } + } + +private: + void removeExpired() + { + for(auto it = subscribers_.begin(); it != subscribers_.end();) + { + if(it->expired()) + { + it = subscribers_.erase(it); + } + else + { + ++it; + } + } + } + + std::mutex mutex_; + std::vector> subscribers_; +}; +} // namespace BT diff --git a/src/blackboard.cpp b/src/blackboard.cpp index 5f313a6f0..31ae94f1f 100644 --- a/src/blackboard.cpp +++ b/src/blackboard.cpp @@ -113,6 +113,65 @@ void Blackboard::enableAutoRemapping(bool remapping) autoremapping_ = remapping; } +std::shared_ptr +Blackboard::getUpdateSignal(const std::string& key, bool create) +{ + if(StartWith(key, '@')) + { + return rootBlackboard()->getUpdateSignal(key.substr(1), create); + } + + std::unique_lock lock(storage_mutex_); + // Match getEntry(): an existing local entry takes precedence over remapping. + if(storage_.find(key) == storage_.end()) + { + if(auto parent = parent_bb_.lock()) + { + auto it = internal_to_external_.find(key); + if(it != internal_to_external_.end()) + { + const auto remapped_key = it->second; + lock.unlock(); + return parent->getUpdateSignal(remapped_key, create); + } + if(autoremapping_ && !IsPrivateKey(key)) + { + lock.unlock(); + return parent->getUpdateSignal(key, create); + } + } + } + if(auto it = update_signals_.find(key); it != update_signals_.end()) + { + return it->second; + } + if(!create) + { + return {}; + } + auto signal = std::make_shared(); + update_signals_.emplace(key, signal); + return signal; +} + +Blackboard::EntryUpdatedSubscriber +Blackboard::subscribeToEntry(const std::string& key, EntryUpdatedCallback callback) +{ + if(!callback) + { + throw LogicError("subscribeToEntry(): empty callback"); + } + return getUpdateSignal(key, true)->subscribe(std::move(callback)); +} + +void Blackboard::notifyEntryUpdated(const std::string& key) +{ + if(auto signal = getUpdateSignal(key, false)) + { + signal->notify(); + } +} + AnyPtrLocked Blackboard::getAnyLocked(const std::string& key) { if(auto entry = getEntry(key)) @@ -380,6 +439,12 @@ void Blackboard::cloneInto(Blackboard& dst) const } } } + // Publish only after the complete snapshot has been restored and all locks + // released, so callbacks can read other restored entries as well. + for(const auto& task : tasks) + { + dst.notifyEntryUpdated(task.key); + } } Blackboard::Ptr Blackboard::parent() @@ -490,8 +555,11 @@ void ImportBlackboardFromJSON(const nlohmann::json& json, Blackboard& blackboard entry = blackboard.getEntry(it.key()); } // Lock entry_mutex before writing to prevent data races (BUG-4 fix). - std::scoped_lock lk(entry->entry_mutex); - entry->value = res->first; + { + std::scoped_lock lk(entry->entry_mutex); + entry->value = res->first; + } + blackboard.notifyEntryUpdated(it.key()); } } } diff --git a/src/tree_node.cpp b/src/tree_node.cpp index e43f19706..711f95de8 100644 --- a/src/tree_node.cpp +++ b/src/tree_node.cpp @@ -330,6 +330,55 @@ TreeNode::subscribeToStatusChange(TreeNode::StatusChangeCallback callback) return _p->state_change_signal.subscribe(std::move(callback)); } +Blackboard::EntryUpdatedSubscriber TreeNode::subscribeToInputPort( + const std::string& key, Blackboard::EntryUpdatedCallback callback) +{ + if(!callback) + { + throw LogicError("subscribeToInputPort(): empty callback"); + } + const PortInfo* info = nullptr; + if(config().manifest != nullptr) + { + auto it = config().manifest->ports.find(key); + if(it == config().manifest->ports.end() || + it->second.direction() == PortDirection::OUTPUT) + { + throw LogicError("subscribeToInputPort(): not an input port [", key, "]"); + } + info = &it->second; + } + + std::string mapping; + if(auto it = config().input_ports.find(key); it != config().input_ports.end()) + { + mapping = it->second; + } + else if(info != nullptr && !info->defaultValue().empty()) + { + if(!info->defaultValue().isString()) + { + return {}; + } + mapping = info->defaultValue().cast(); + } + else + { + throw LogicError("subscribeToInputPort(): missing input [", key, "]"); + } + + if(auto remapped_key = getRemappedKey(key, mapping)) + { + if(!config().blackboard) + { + throw LogicError("subscribeToInputPort(): invalid blackboard"); + } + return config().blackboard->subscribeToEntry(std::string(*remapped_key), + std::move(callback)); + } + return {}; +} + void TreeNode::setPreTickFunction(PreTickCallback callback) { const std::unique_lock lk(_p->callback_injection_mutex); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 49fcce397..dbab27cb3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -31,6 +31,7 @@ set(BT_TESTS gtest_parallel.cpp gtest_preconditions.cpp gtest_ports.cpp + gtest_input_port_callbacks.cpp gtest_port_type_rules.cpp gtest_postconditions.cpp gtest_match.cpp diff --git a/tests/gtest_input_port_callbacks.cpp b/tests/gtest_input_port_callbacks.cpp new file mode 100644 index 000000000..472843fa2 --- /dev/null +++ b/tests/gtest_input_port_callbacks.cpp @@ -0,0 +1,316 @@ +#include "behaviortree_cpp/bt_factory.h" +#include "behaviortree_cpp/scripting/operators.hpp" + +#include +#include + +#include + +using namespace BT; + +namespace +{ +class InputObserver : public SyncActionNode +{ +public: + InputObserver(const std::string& name, const NodeConfig& config) + : SyncActionNode(name, config) + {} + + static PortsList providedPorts() + { + return { InputPort("value"), InputPort("constant", 42, "constant"), + InputPort("default_key", "{source}", "remapped default"), + BidirectionalPort("both"), OutputPort("out") }; + } + + NodeStatus tick() override + { + return NodeStatus::SUCCESS; + } +}; +} // namespace + +TEST(InputPortCallbacks, NotifiesBeforeTickAndCanReadAndWrite) +{ + BehaviorTreeFactory factory; + factory.registerNodeType("Observer"); + auto tree = factory.createTreeFromText(R"( + + + )"); + auto* node = tree.rootNode(); + auto bb = tree.rootBlackboard(); + std::vector values; + auto subscription = node->subscribeToInputPort("value", [&] { + values.push_back(node->getInput("value").value()); + bb->set("other", values.back()); + }); + + EXPECT_TRUE(values.empty()); + bb->set("source", 3); + bb->set("source", 3); + bb->set("source", 7); + EXPECT_EQ(values, (std::vector{ 3, 3, 7 })); + EXPECT_EQ(bb->get("other"), 7); + EXPECT_EQ(node->status(), NodeStatus::IDLE); + subscription.reset(); + bb->set("source", 9); + EXPECT_EQ(values.size(), 3); +} + +TEST(InputPortCallbacks, RemappingAndPrivateKeys) +{ + auto root = Blackboard::create(); + auto child = Blackboard::create(root); + child->addSubtreeRemapping("local", "remote"); + child->addSubtreeRemapping("global", "@root_value"); + child->enableAutoRemapping(true); + int manual = 0; + int automatic = 0; + int global = 0; + int private_key = 0; + auto a = child->subscribeToEntry("local", [&] { ++manual; }); + auto b = child->subscribeToEntry("auto_key", [&] { ++automatic; }); + auto c = child->subscribeToEntry("global", [&] { ++global; }); + auto d = child->subscribeToEntry("_private", [&] { ++private_key; }); + root->set("remote", 1); + child->set("local", 2); + root->set("auto_key", 1); + child->set("auto_key", 2); + root->set("root_value", 1); + child->set("global", 2); + root->set("_private", 1); + EXPECT_EQ(private_key, 0); + child->set("_private", 2); + EXPECT_EQ(manual, 2); + EXPECT_EQ(automatic, 2); + EXPECT_EQ(global, 2); + EXPECT_EQ(private_key, 1); +} + +TEST(InputPortCallbacks, SubtreePort) +{ + BehaviorTreeFactory factory; + factory.registerNodeType("Observer"); + auto tree = factory.createTreeFromText(R"( + + + + )"); + auto* node = tree.subtrees.at(1)->nodes.front().get(); + int observed = 0; + auto subscription = node->subscribeToInputPort( + "value", [&] { observed = node->getInput("value").value(); }); + tree.rootBlackboard()->set("source", 17); + EXPECT_EQ(observed, 17); +} + +TEST(InputPortCallbacks, DefaultsConstantsIdentityAndValidation) +{ + NodeConfig config; + config.blackboard = Blackboard::create(); + TreeNodeManifest manifest = { + NodeType::ACTION, "Observer", InputObserver::providedPorts(), {} + }; + config.manifest = &manifest; + config.input_ports["value"] = "{=}"; + config.input_ports["both"] = "{@root_key}"; + InputObserver node("observer", config); + int count = 0; + auto callback = [&] { ++count; }; + auto a = node.subscribeToInputPort("value", callback); + auto b = node.subscribeToInputPort("default_key", callback); + auto c = node.subscribeToInputPort("both", callback); + EXPECT_FALSE(node.subscribeToInputPort("constant", callback)); + config.blackboard->set("value", 1); + config.blackboard->set("source", 2); + config.blackboard->set("root_key", 3); + EXPECT_EQ(count, 3); + EXPECT_THROW((void)node.subscribeToInputPort("missing", callback), LogicError); + EXPECT_THROW((void)node.subscribeToInputPort("out", callback), LogicError); + EXPECT_THROW((void)node.subscribeToInputPort("value", {}), LogicError); + + config.input_ports["value"] = "12"; + InputObserver literal("literal", config); + EXPECT_FALSE(literal.subscribeToInputPort("value", callback)); + config.input_ports.erase("value"); + InputObserver missing("missing", config); + EXPECT_THROW((void)missing.subscribeToInputPort("value", callback), LogicError); + config.input_ports["value"] = "{source}"; + config.blackboard.reset(); + InputObserver no_blackboard("no_blackboard", config); + EXPECT_THROW((void)no_blackboard.subscribeToInputPort("value", callback), LogicError); +} + +TEST(InputPortCallbacks, FirstWriteUnsetAndFailedAssignment) +{ + auto bb = Blackboard::create(); + int count = 0; + auto subscription = bb->subscribeToEntry("key", [&] { + EXPECT_TRUE(bb->getEntry("key")); + ++count; + }); + EXPECT_FALSE(bb->getEntry("key")); + bb->set("key", 1); + EXPECT_THROW(bb->set("key", std::string("not a number")), RuntimeError); + EXPECT_EQ(count, 1); + bb->unset("key"); + bb->set("key", 2); + EXPECT_EQ(count, 2); + EXPECT_THROW((void)bb->subscribeToEntry("key", {}), LogicError); +} + +TEST(InputPortCallbacks, ScriptsAndOutputPorts) +{ + auto bb = Blackboard::create(); + NodeConfig config; + config.blackboard = bb; + config.output_ports["out"] = "{key}"; + InputObserver writer("writer", config); + std::vector values; + auto subscription = + bb->subscribeToEntry("key", [&] { values.push_back(bb->get("key")); }); + Ast::Environment environment = { bb, {} }; + EXPECT_TRUE(ParseScriptAndExecute(environment, "key := 1")); + EXPECT_TRUE(ParseScriptAndExecute(environment, "key = 2; key += 3; key *= 2")); + EXPECT_TRUE(writer.setOutput("out", 12)); + EXPECT_EQ(values, (std::vector{ 1, 2, 5, 10, 12 })); +} + +TEST(InputPortCallbacks, ImportAndRestore) +{ + auto bb = Blackboard::create(); + auto backup = Blackboard::create(); + backup->set("key", 4); + backup->set("other", 8); + int observed = 0; + auto subscription = + bb->subscribeToEntry("key", [&] { observed = bb->get("key"); }); + ImportBlackboardFromJSON(nlohmann::json{ { "key", 2 } }, *bb); + EXPECT_EQ(observed, 2); + auto restore_subscription = + bb->subscribeToEntry("key", [&] { EXPECT_EQ(bb->get("other"), 8); }); + backup->cloneInto(*bb); + EXPECT_EQ(observed, 4); + restore_subscription.reset(); + bb->unset("key"); + backup->set("key", 6); + backup->cloneInto(*bb); + EXPECT_EQ(observed, 6); +} + +TEST(InputPortCallbacks, CallbackExceptionLeavesValueCommitted) +{ + auto bb = Blackboard::create(); + auto subscription = bb->subscribeToEntry("key", [] { throw RuntimeError("callback"); }); + EXPECT_THROW(bb->set("key", 42), RuntimeError); + EXPECT_EQ(bb->get("key"), 42); + subscription.reset(); + EXPECT_NO_THROW(bb->set("key", 43)); +} + +TEST(InputPortCallbacks, ReentrantSubscriptions) +{ + auto bb = Blackboard::create(); + int first_count = 0; + int second_count = 0; + Blackboard::EntryUpdatedSubscriber first; + Blackboard::EntryUpdatedSubscriber second; + first = bb->subscribeToEntry("key", [&] { + ++first_count; + first.reset(); + second = bb->subscribeToEntry("key", [&] { ++second_count; }); + bb->set("key", 2); + }); + bb->set("key", 1); + EXPECT_EQ(first_count, 1); + EXPECT_EQ(second_count, 1); + bb->set("key", 3); + EXPECT_EQ(second_count, 2); +} + +TEST(InputPortCallbacks, ConcurrentWritersAndSubscriptions) +{ + auto bb = Blackboard::create(); + bb->set("key", 0); + std::atomic count = 0; + auto subscription = bb->subscribeToEntry("key", [&] { + (void)bb->get("key"); + ++count; + }); + std::vector writers; + for(int i = 0; i < 4; ++i) + { + writers.emplace_back([&] { + for(int j = 0; j < 200; ++j) + { + auto temporary = bb->subscribeToEntry("key", [] {}); + bb->set("key", j); + } + }); + } + for(auto& writer : writers) + { + writer.join(); + } + EXPECT_EQ(count, 800); +} + +TEST(InputPortCallbacks, BlackboardMayBeDestroyedBeforeHandle) +{ + auto bb = Blackboard::create(); + auto subscription = bb->subscribeToEntry("key", [] {}); + bb.reset(); + EXPECT_NO_THROW(subscription.reset()); +} + +TEST(InputPortCallbacks, LocalEntryTakesPrecedence) +{ + auto root = Blackboard::create(); + auto child = Blackboard::create(root); + child->set("key", 1); + child->enableAutoRemapping(true); + int count = 0; + auto subscription = child->subscribeToEntry("key", [&] { ++count; }); + root->set("key", 2); + EXPECT_EQ(count, 0); + child->set("key", 3); + EXPECT_EQ(count, 1); +} + +TEST(InputPortCallbacks, DirectMutationRequiresExplicitNotification) +{ + auto bb = Blackboard::create(); + bb->set("key", 1); + int observed = 0; + auto subscription = + bb->subscribeToEntry("key", [&] { observed = bb->get("key"); }); + { + auto locked = bb->getAnyLocked("key"); + locked.assign(2); + } + EXPECT_EQ(observed, 0); + bb->notifyEntryUpdated("key"); + EXPECT_EQ(observed, 2); +} + +TEST(InputPortCallbacks, CallbackCanDisconnectLaterSubscriber) +{ + auto bb = Blackboard::create(); + Blackboard::EntryUpdatedSubscriber second; + auto first = bb->subscribeToEntry("key", [&] { second.reset(); }); + second = bb->subscribeToEntry("key", [] { FAIL() << "Disconnected callback"; }); + bb->set("key", 1); +} + +TEST(InputPortCallbacks, StatefulCallbackIsNotCopiedOnEachWrite) +{ + auto bb = Blackboard::create(); + int observed = 0; + auto subscription = + bb->subscribeToEntry("key", [&, count = 0]() mutable { observed = ++count; }); + bb->set("key", 1); + bb->set("key", 2); + EXPECT_EQ(observed, 2); +} From 1c374454ff39ef4eb17574dc0f2b986d6b735fe7 Mon Sep 17 00:00:00 2001 From: Davide Faconti Date: Sat, 24 Oct 2026 16:04:03 +0200 Subject: [PATCH 2/2] Store input port subscriptions in the blackboard Entry to keep the 4.x ABI Blackboard::create() is inline, so client binaries bake in sizeof(Blackboard): the per-key update_signals_ map broke the ABI of the 4.x series. It also made every write take storage_mutex_ exclusively and look the key up again, even without subscribers, and set() could notify the signal of a different key resolution than the one it wrote to. The subscribers are now kept in a shared_ptr appended as the last member of Blackboard::Entry. Entries are allocated only inside the library and inline code only touches the earlier members, whose offsets do not change; Blackboard is byte-identical to before. The pointer is guarded by entry_mutex and created lazily on the first subscription. Writers copy it under the entry lock they already hold and notify after releasing it, so a write without subscribers costs a null check. subscribeToEntry() follows remapping and '@' via getEntry() and, when the entry is missing, creates a generic one (subscribeToInputPort() uses the port's type, like tree creation does). As a consequence a subscription is bound to the entry: unset() drops it and a re-created entry is not observed. Writes from code compiled against older headers do not notify. Both are documented and covered by tests. Co-Authored-By: Claude Opus 5.5 --- docs/INPUT_PORT_CALLBACKS.md | 31 +++++--- include/behaviortree_cpp/blackboard.h | 36 ++++++--- .../behaviortree_cpp/scripting/operators.hpp | 12 ++- include/behaviortree_cpp/tree_node.h | 2 + src/blackboard.cpp | 79 ++++++++----------- src/tree_node.cpp | 9 ++- tests/gtest_input_port_callbacks.cpp | 67 ++++++++++++++-- 7 files changed, 159 insertions(+), 77 deletions(-) diff --git a/docs/INPUT_PORT_CALLBACKS.md b/docs/INPUT_PORT_CALLBACKS.md index 42b72c2bb..d059b01ba 100644 --- a/docs/INPUT_PORT_CALLBACKS.md +++ b/docs/INPUT_PORT_CALLBACKS.md @@ -67,9 +67,20 @@ returns an empty handle because it cannot change through the blackboard. Missing inputs, output-only ports, invalid blackboards, and empty callbacks are reported with `LogicError`. -Subscriptions bind to the resolved key at registration time. Reconfiguring port -or subtree remapping requires a new subscription. Subscriptions survive removing -and recreating that key. Removal itself does not notify. +Subscriptions bind to the blackboard entry the key resolves to at registration +time. Reconfiguring port or subtree remapping requires a new subscription. +A subscription lives in the entry: `unset()` (or `cloneInto()` removing the key) +drops it, without notifying, and an entry re-created afterwards under the same +name is not observed. Subscribe again after re-creating the key. + +If the entry doesn't exist yet, subscribing creates it, so the key is listed by +`getKeyNames()` before its first assignment, with an empty value. +`subscribeToInputPort()` creates it with the port's type, as tree creation does +for remapped ports. `Blackboard::subscribeToEntry()` creates a generic +(`AnyTypeAllowed`) entry, like `set()` does: the first strongly typed +write sets its type. Note that, as for any existing generic entry, a first write +of a `std::string` locks such an entry to `std::string`, whereas without the +subscription `set()` would have created a generic entry. Writes through `Blackboard::set()`, `TreeNode::setOutput()`, scripting assignments, JSON import, and `Blackboard::cloneInto()` notify. Clone notifications run after @@ -78,12 +89,14 @@ entries. Raw writes through `Entry`, `AnyPtrLocked`, or shared objects are not observable automatically. Call `Blackboard::notifyEntryUpdated()` after releasing the entry lock to explicitly publish such a write. -The lower-level `Blackboard::subscribeToEntry()` supports subscriptions before -the first assignment without creating or changing the type of an entry. - ## Compatibility This adds an opt-in API; existing port declarations and unsubscribed nodes keep -their behavior. The additional subscription storage changes `Blackboard`'s -binary layout. Rebuild the library and clients together; this API is proposed -for the next major version, as requested by issue #872's BT.CPP-5 label. +their behavior. It is ABI compatible with 4.x: the subscribers are stored in a +member appended at the end of `Blackboard::Entry`, which is allocated only inside +the library, and the layout of `Blackboard` is unchanged. Without subscribers, a +write only checks that member for null under the entry lock it already holds. + +Every write path (`set()`, `setOutput()`, scripting assignments) is inline code +in the headers. Writes performed by binaries compiled against older headers +(4.10 and earlier) store the value but do not notify, until they are rebuilt. diff --git a/include/behaviortree_cpp/blackboard.h b/include/behaviortree_cpp/blackboard.h index df359305a..50624e1a5 100644 --- a/include/behaviortree_cpp/blackboard.h +++ b/include/behaviortree_cpp/blackboard.h @@ -66,6 +66,12 @@ class Blackboard // timestamp since epoch std::chrono::nanoseconds stamp = std::chrono::nanoseconds{ 0 }; + // Subscribers of subscribeToEntry(), created on the first subscription. + // Protected by entry_mutex. It must remain the last member: Entry is + // allocated only inside the library, so appending it keeps the offsets of + // the members above (accessed by inline code) unchanged. + std::shared_ptr update_signal; + Entry(const TypeInfo& _info) : info(_info) {} @@ -123,10 +129,13 @@ class Blackboard template void set(const std::string& key, const T& value); - /** Subscribe to successful writes to a key, including its first assignment. - * Subtree remapping and root keys are resolved at subscription time. - * The subscription survives unset()/set() of the same key. No initial - * notification is sent, and assigning the same value still notifies. + /** Subscribe to successful writes to the entry of a key. + * Subtree remapping and root keys are resolved at subscription time. If the + * entry doesn't exist, a generic (AnyTypeAllowed) one is created: the first + * strongly typed write sets its type, as usual. The subscription is bound to + * that entry: unset() drops it and a re-created entry is not observed. + * No initial notification is sent, and assigning the same value still notifies. + * Writes compiled against headers older than this API do not notify. * * Callbacks run synchronously on the writer's thread, outside blackboard * locks, and may read the latest value. Concurrent writers may invoke a @@ -215,15 +224,11 @@ class Blackboard private: mutable std::shared_mutex storage_mutex_; std::unordered_map> storage_; - std::unordered_map> update_signals_; std::weak_ptr parent_bb_; std::unordered_map internal_to_external_; std::shared_ptr createEntryImpl(const std::string& key, const TypeInfo& info); - std::shared_ptr getUpdateSignal(const std::string& key, - bool create); - bool autoremapping_ = false; // Optional registry for polymorphic shared_ptr conversions @@ -325,6 +330,8 @@ inline void Blackboard::set(const std::string& key, const T& value) // follow the same remapping and return the pre-existing entry, and writing // to it blindly bypasses the type lock and the string conversion. auto entry_ptr = getEntry(key); + // copied under entry_mutex, notified after releasing it + std::shared_ptr update_signal; if(!entry_ptr) { // create a new entry @@ -348,6 +355,7 @@ inline void Blackboard::set(const std::string& key, const T& value) entry->value = new_value; entry->sequence_id++; entry->stamp = std::chrono::steady_clock::now().time_since_epoch(); + update_signal = entry->update_signal; } else { @@ -368,8 +376,12 @@ inline void Blackboard::set(const std::string& key, const T& value) entry.sequence_id++; entry.stamp = std::chrono::steady_clock::now().time_since_epoch(); previous_any = std::move(new_value); + update_signal = entry.update_signal; scoped_lock.unlock(); - notifyEntryUpdated(key); + if(update_signal) + { + update_signal->notify(); + } return; } @@ -424,8 +436,12 @@ inline void Blackboard::set(const std::string& key, const T& value) } entry.sequence_id++; entry.stamp = std::chrono::steady_clock::now().time_since_epoch(); + update_signal = entry.update_signal; + } + if(update_signal) + { + update_signal->notify(); } - notifyEntryUpdated(key); } template diff --git a/include/behaviortree_cpp/scripting/operators.hpp b/include/behaviortree_cpp/scripting/operators.hpp index 0e8d4358f..f52b550ae 100644 --- a/include/behaviortree_cpp/scripting/operators.hpp +++ b/include/behaviortree_cpp/scripting/operators.hpp @@ -625,8 +625,12 @@ struct ExprAssignment : ExprBase entry->sequence_id++; entry->stamp = std::chrono::steady_clock::now().time_since_epoch(); const Any result = *dst_ptr; + auto update_signal = entry->update_signal; lock.unlock(); - env.vars->notifyEntryUpdated(key); + if(update_signal) + { + update_signal->notify(); + } return result; } @@ -684,8 +688,12 @@ struct ExprAssignment : ExprBase entry->sequence_id++; entry->stamp = std::chrono::steady_clock::now().time_since_epoch(); const Any result = *dst_ptr; + auto update_signal = entry->update_signal; lock.unlock(); - env.vars->notifyEntryUpdated(key); + if(update_signal) + { + update_signal->notify(); + } return result; } }; diff --git a/include/behaviortree_cpp/tree_node.h b/include/behaviortree_cpp/tree_node.h index 20588fb2d..b4ae9506f 100644 --- a/include/behaviortree_cpp/tree_node.h +++ b/include/behaviortree_cpp/tree_node.h @@ -223,6 +223,8 @@ class TreeNode * Constant inputs return an empty handle. Missing inputs, output-only ports * and empty callbacks throw LogicError. Defaults and {=} remapping are * supported. Changing the port mapping later does not rebind subscriptions. + * The subscription is bound to the blackboard entry, which is created with + * the port's type if it doesn't exist yet; unset() of the key drops it. */ [[nodiscard]] Blackboard::EntryUpdatedSubscriber subscribeToInputPort(const std::string& key, Blackboard::EntryUpdatedCallback callback); diff --git a/src/blackboard.cpp b/src/blackboard.cpp index 31ae94f1f..13e906e9f 100644 --- a/src/blackboard.cpp +++ b/src/blackboard.cpp @@ -113,62 +113,39 @@ void Blackboard::enableAutoRemapping(bool remapping) autoremapping_ = remapping; } -std::shared_ptr -Blackboard::getUpdateSignal(const std::string& key, bool create) +Blackboard::EntryUpdatedSubscriber +Blackboard::subscribeToEntry(const std::string& key, EntryUpdatedCallback callback) { - if(StartWith(key, '@')) + if(!callback) { - return rootBlackboard()->getUpdateSignal(key.substr(1), create); + throw LogicError("subscribeToEntry(): empty callback"); } - - std::unique_lock lock(storage_mutex_); - // Match getEntry(): an existing local entry takes precedence over remapping. - if(storage_.find(key) == storage_.end()) + auto entry = getEntry(key); + if(!entry) { - if(auto parent = parent_bb_.lock()) - { - auto it = internal_to_external_.find(key); - if(it != internal_to_external_.end()) - { - const auto remapped_key = it->second; - lock.unlock(); - return parent->getUpdateSignal(remapped_key, create); - } - if(autoremapping_ && !IsPrivateKey(key)) - { - lock.unlock(); - return parent->getUpdateSignal(key, create); - } - } + // Generic, like the entry created by set(): the first + // strongly typed write sets its type. + entry = createEntryImpl(key, PortInfo(PortDirection::INOUT)); } - if(auto it = update_signals_.find(key); it != update_signals_.end()) + const std::scoped_lock lock(entry->entry_mutex); + if(!entry->update_signal) { - return it->second; + entry->update_signal = std::make_shared(); } - if(!create) - { - return {}; - } - auto signal = std::make_shared(); - update_signals_.emplace(key, signal); - return signal; + return entry->update_signal->subscribe(std::move(callback)); } -Blackboard::EntryUpdatedSubscriber -Blackboard::subscribeToEntry(const std::string& key, EntryUpdatedCallback callback) +void Blackboard::notifyEntryUpdated(const std::string& key) { - if(!callback) + std::shared_ptr update_signal; + if(auto entry = getEntry(key)) { - throw LogicError("subscribeToEntry(): empty callback"); + const std::scoped_lock lock(entry->entry_mutex); + update_signal = entry->update_signal; } - return getUpdateSignal(key, true)->subscribe(std::move(callback)); -} - -void Blackboard::notifyEntryUpdated(const std::string& key) -{ - if(auto signal = getUpdateSignal(key, false)) + if(update_signal) { - signal->notify(); + update_signal->notify(); } } @@ -396,6 +373,7 @@ void Blackboard::cloneInto(Blackboard& dst) const // Step 2: copy entry data under entry_mutex only (BUG-3 fix). std::vector>> new_entries; + std::vector> update_signals; for(auto& task : tasks) { @@ -408,6 +386,10 @@ void Blackboard::cloneInto(Blackboard& dst) const task.dst->info = task.src->info; task.dst->sequence_id++; task.dst->stamp = std::chrono::steady_clock::now().time_since_epoch(); + if(task.dst->update_signal) + { + update_signals.push_back(task.dst->update_signal); + } } else { @@ -441,9 +423,9 @@ void Blackboard::cloneInto(Blackboard& dst) const } // Publish only after the complete snapshot has been restored and all locks // released, so callbacks can read other restored entries as well. - for(const auto& task : tasks) + for(const auto& update_signal : update_signals) { - dst.notifyEntryUpdated(task.key); + update_signal->notify(); } } @@ -555,11 +537,16 @@ void ImportBlackboardFromJSON(const nlohmann::json& json, Blackboard& blackboard entry = blackboard.getEntry(it.key()); } // Lock entry_mutex before writing to prevent data races (BUG-4 fix). + std::shared_ptr update_signal; { std::scoped_lock lk(entry->entry_mutex); entry->value = res->first; + update_signal = entry->update_signal; + } + if(update_signal) + { + update_signal->notify(); } - blackboard.notifyEntryUpdated(it.key()); } } } diff --git a/src/tree_node.cpp b/src/tree_node.cpp index 711f95de8..b0c612f7d 100644 --- a/src/tree_node.cpp +++ b/src/tree_node.cpp @@ -373,8 +373,13 @@ Blackboard::EntryUpdatedSubscriber TreeNode::subscribeToInputPort( { throw LogicError("subscribeToInputPort(): invalid blackboard"); } - return config().blackboard->subscribeToEntry(std::string(*remapped_key), - std::move(callback)); + const auto bb_key = std::string(*remapped_key); + if(info != nullptr && !config().blackboard->getEntry(bb_key)) + { + // Like getLockedPortContent(): create the entry with the port's type. + config().blackboard->createEntry(bb_key, *info); + } + return config().blackboard->subscribeToEntry(bb_key, std::move(callback)); } return {}; } diff --git a/tests/gtest_input_port_callbacks.cpp b/tests/gtest_input_port_callbacks.cpp index 472843fa2..b945aa756 100644 --- a/tests/gtest_input_port_callbacks.cpp +++ b/tests/gtest_input_port_callbacks.cpp @@ -104,6 +104,18 @@ TEST(InputPortCallbacks, SubtreePort) "value", [&] { observed = node->getInput("value").value(); }); tree.rootBlackboard()->set("source", 17); EXPECT_EQ(observed, 17); + // Writes through the remapped key of the subtree reach the same entry + tree.subtrees.at(1)->blackboard->set("input", 23); + EXPECT_EQ(observed, 23); + + // A missing entry is created in the parent blackboard, with the port's type + tree.rootBlackboard()->unset("source"); + subscription = node->subscribeToInputPort( + "value", [&] { observed = node->getInput("value").value(); }); + ASSERT_NE(tree.rootBlackboard()->entryInfo("source"), nullptr); + EXPECT_EQ(tree.rootBlackboard()->entryInfo("source")->type(), typeid(int)); + tree.rootBlackboard()->set("source", std::string("5")); + EXPECT_EQ(observed, 5); } TEST(InputPortCallbacks, DefaultsConstantsIdentityAndValidation) @@ -147,20 +159,58 @@ TEST(InputPortCallbacks, FirstWriteUnsetAndFailedAssignment) { auto bb = Blackboard::create(); int count = 0; - auto subscription = bb->subscribeToEntry("key", [&] { - EXPECT_TRUE(bb->getEntry("key")); - ++count; - }); - EXPECT_FALSE(bb->getEntry("key")); + auto subscription = bb->subscribeToEntry("key", [&] { ++count; }); + // Subscribing creates a generic entry: the first strongly typed write sets its type + ASSERT_NE(bb->entryInfo("key"), nullptr); + EXPECT_FALSE(bb->entryInfo("key")->isStronglyTyped()); + EXPECT_EQ(count, 0); bb->set("key", 1); + EXPECT_EQ(bb->entryInfo("key")->type(), typeid(int)); EXPECT_THROW(bb->set("key", std::string("not a number")), RuntimeError); EXPECT_EQ(count, 1); - bb->unset("key"); - bb->set("key", 2); + bb->set("key", std::string("2")); EXPECT_EQ(count, 2); EXPECT_THROW((void)bb->subscribeToEntry("key", {}), LogicError); } +TEST(InputPortCallbacks, SubscriptionIsBoundToTheEntry) +{ + auto bb = Blackboard::create(); + int count = 0; + auto subscription = bb->subscribeToEntry("key", [&] { ++count; }); + bb->set("key", 1); + EXPECT_EQ(count, 1); + // unset() drops the subscription: the re-created entry is not observed + bb->unset("key"); + bb->set("key", 2); + EXPECT_EQ(count, 1); + bb->notifyEntryUpdated("key"); + EXPECT_EQ(count, 1); +} + +TEST(InputPortCallbacks, WritesWithoutSubscribersDoNotAllocateSignal) +{ + auto bb = Blackboard::create(); + bb->set("key", 1); + bb->set("key", 2); + bb->set("text", std::string("a")); + ImportBlackboardFromJSON(nlohmann::json{ { "key", 3 } }, *bb); + Ast::Environment environment = { bb, {} }; + EXPECT_TRUE(ParseScriptAndExecute(environment, "key = 4; created := 5")); + bb->notifyEntryUpdated("key"); + auto copy = Blackboard::create(); + bb->cloneInto(*copy); + for(const auto& blackboard : { bb, copy }) + { + for(const auto& name : blackboard->getKeyNames()) + { + EXPECT_FALSE(blackboard->getEntry(name)->update_signal) << name; + } + } + auto subscription = bb->subscribeToEntry("key", [] {}); + EXPECT_TRUE(bb->getEntry("key")->update_signal); +} + TEST(InputPortCallbacks, ScriptsAndOutputPorts) { auto bb = Blackboard::create(); @@ -194,10 +244,11 @@ TEST(InputPortCallbacks, ImportAndRestore) backup->cloneInto(*bb); EXPECT_EQ(observed, 4); restore_subscription.reset(); + // The entry re-created by cloneInto() is not observed bb->unset("key"); backup->set("key", 6); backup->cloneInto(*bb); - EXPECT_EQ(observed, 6); + EXPECT_EQ(observed, 4); } TEST(InputPortCallbacks, CallbackExceptionLeavesValueCommitted)