diff --git a/other/docker/modules/check b/other/docker/modules/check index 8cbaa87a3..eba660357 100755 --- a/other/docker/modules/check +++ b/other/docker/modules/check @@ -20,6 +20,7 @@ STD_MODULE = """module std [system] { textual header "/usr/include/c++/14.2.0/chrono" textual header "/usr/include/c++/14.2.0/climits" textual header "/usr/include/c++/14.2.0/compare" + textual header "/usr/include/c++/14.2.0/condition_variable" textual header "/usr/include/c++/14.2.0/cstddef" textual header "/usr/include/c++/14.2.0/cstdint" textual header "/usr/include/c++/14.2.0/cstdio" @@ -27,6 +28,7 @@ STD_MODULE = """module std [system] { textual header "/usr/include/c++/14.2.0/cstring" textual header "/usr/include/c++/14.2.0/deque" textual header "/usr/include/c++/14.2.0/functional" + textual header "/usr/include/c++/14.2.0/future" textual header "/usr/include/c++/14.2.0/iomanip" textual header "/usr/include/c++/14.2.0/iosfwd" textual header "/usr/include/c++/14.2.0/iostream" diff --git a/testing/support/BUILD.bazel b/testing/support/BUILD.bazel index bc32df56c..d49cfbd06 100644 --- a/testing/support/BUILD.bazel +++ b/testing/support/BUILD.bazel @@ -18,6 +18,7 @@ cc_library( "src/simulated_environment.cc", "src/simulation.cc", "src/tox_network.cc", + "src/tox_runner.cc", ], hdrs = [ "doubles/fake_clock.hh", @@ -31,11 +32,13 @@ cc_library( "public/fuzz_data.hh", "public/fuzz_helpers.hh", "public/memory.hh", + "public/mpsc_queue.hh", "public/network.hh", "public/random.hh", "public/simulated_environment.hh", "public/simulation.hh", "public/tox_network.hh", + "public/tox_runner.hh", ], copts = select({ "//tools/config:windows": ["/wd4200"], # Zero-sized array in struct/union @@ -46,6 +49,7 @@ cc_library( "//c-toxcore/toxcore:mem", "//c-toxcore/toxcore:network", "//c-toxcore/toxcore:tox", + "//c-toxcore/toxcore:tox_events", "//c-toxcore/toxcore:tox_memory", "//c-toxcore/toxcore:tox_options", "//c-toxcore/toxcore:tox_random", diff --git a/testing/support/CMakeLists.txt b/testing/support/CMakeLists.txt index 00da142e1..6d1d471b4 100644 --- a/testing/support/CMakeLists.txt +++ b/testing/support/CMakeLists.txt @@ -18,6 +18,7 @@ set(support_SOURCES src/simulated_environment.cc src/simulation.cc src/tox_network.cc + src/tox_runner.cc doubles/fake_clock.hh doubles/fake_memory.hh doubles/fake_network_stack.hh @@ -29,11 +30,13 @@ set(support_SOURCES public/fuzz_data.hh public/fuzz_helpers.hh public/memory.hh + public/mpsc_queue.hh public/network.hh public/random.hh public/simulated_environment.hh public/simulation.hh public/tox_network.hh + public/tox_runner.hh ) add_library(support STATIC ${support_SOURCES}) diff --git a/testing/support/doubles/fake_clock.hh b/testing/support/doubles/fake_clock.hh index 0bfadeb5b..555d25852 100644 --- a/testing/support/doubles/fake_clock.hh +++ b/testing/support/doubles/fake_clock.hh @@ -1,6 +1,8 @@ #ifndef C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_CLOCK_H #define C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_CLOCK_H +#include + #include "../public/clock.hh" namespace tox::test { @@ -15,7 +17,7 @@ public: void advance(uint64_t ms); private: - uint64_t now_ms_; + std::atomic now_ms_; }; } // namespace tox::test diff --git a/testing/support/doubles/fake_memory.hh b/testing/support/doubles/fake_memory.hh index 4754efb25..b25b16b1f 100644 --- a/testing/support/doubles/fake_memory.hh +++ b/testing/support/doubles/fake_memory.hh @@ -1,6 +1,7 @@ #ifndef C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_MEMORY_H #define C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_MEMORY_H +#include #include #include "../public/memory.hh" @@ -31,7 +32,13 @@ public: // Get the C-compatible struct struct Tox_Memory get_c_memory(); + size_t current_allocation() const; + size_t max_allocation() const; + private: + void on_allocation(size_t size); + void on_deallocation(size_t size); + struct Header { size_t size; size_t magic; @@ -39,8 +46,8 @@ private: static constexpr size_t kMagic = 0xDEADC0DE; static constexpr size_t kFreeMagic = 0xBAADF00D; - size_t current_allocation_ = 0; - size_t max_allocation_ = 0; + std::atomic current_allocation_{0}; + std::atomic max_allocation_{0}; FailureInjector failure_injector_; Observer observer_; diff --git a/testing/support/public/mpsc_queue.hh b/testing/support/public/mpsc_queue.hh new file mode 100644 index 000000000..fb0130af6 --- /dev/null +++ b/testing/support/public/mpsc_queue.hh @@ -0,0 +1,83 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later + * Copyright © 2026 The TokTok team. + */ +#ifndef C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H +#define C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H + +#include +#include +#include + +namespace tox::test { + +/** + * @brief Multiple Producer, Single Consumer Queue. + * + * This queue implementation provides thread-safe access for multiple producers + * pushing items and a single consumer popping items. It uses a `std::mutex` + * and `std::condition_variable` for synchronization. + * + * @tparam T The type of elements stored in the queue. + */ +template +class MpscQueue { +public: + MpscQueue() = default; + ~MpscQueue() = default; + + // Disable copy/move to prevent accidental sharing/slicing issues + MpscQueue(const MpscQueue &) = delete; + MpscQueue &operator=(const MpscQueue &) = delete; + + /** + * @brief Pushes a value onto the queue. + * Thread-safe (Multiple Producers). + */ + void push(T value) + { + { + std::lock_guard lock(mutex_); + queue_.push_back(std::move(value)); + } + cv_.notify_one(); + } + + /** + * @brief Pops a value from the queue, blocking if empty. + * Thread-safe (Single Consumer). + */ + T pop() + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return !queue_.empty(); }); + T value = std::move(queue_.front()); + queue_.pop_front(); + return value; + } + + /** + * @brief Tries to pop a value from the queue without blocking. + * Thread-safe (Single Consumer). + * + * @param out Reference to store the popped value. + * @return true if a value was popped, false if the queue was empty. + */ + bool try_pop(T &out) + { + std::lock_guard lock(mutex_); + if (queue_.empty()) + return false; + out = std::move(queue_.front()); + queue_.pop_front(); + return true; + } + +private: + std::deque queue_; + std::mutex mutex_; + std::condition_variable cv_; +}; + +} // namespace tox::test + +#endif // C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H diff --git a/testing/support/public/simulation.hh b/testing/support/public/simulation.hh index 63cbdab4c..abbfd9c3a 100644 --- a/testing/support/public/simulation.hh +++ b/testing/support/public/simulation.hh @@ -1,6 +1,7 @@ #ifndef C_TOXCORE_TESTING_SUPPORT_SIMULATION_H #define C_TOXCORE_TESTING_SUPPORT_SIMULATION_H +#include #include #include #include @@ -42,6 +43,57 @@ public: void advance_time(uint64_t ms); void run_until(std::function condition, uint64_t timeout_ms = 5000); + // Synchronization Barrier + // These methods coordinate the lock-step execution of multiple Tox runners. + + /** + * @brief Registers a new runner with the simulation barrier. + * @return The current generation ID of the simulation. + */ + uint64_t register_runner(); + + /** + * @brief Unregisters a runner from the simulation barrier. + * + * This ensures the simulation does not block waiting for a terminated runner. + */ + void unregister_runner(); + + using TickListenerId = int; + + /** + * @brief Registers a callback to be invoked when a new simulation tick starts. + * + * @param listener The function to call with the new generation ID. + * @return An ID handle for unregistering the listener. + */ + TickListenerId register_tick_listener(std::function listener); + + /** + * @brief Unregisters a tick listener. + */ + void unregister_tick_listener(TickListenerId id); + + /** + * @brief Blocks until the simulation advances to the next tick. + * + * Called by runner threads to wait for the global clock to advance. + * + * @param last_gen The generation ID of the last processed tick. + * @param stop_token Atomic flag to signal termination while waiting. + * @param timeout_ms Maximum time to wait for the tick. + * @return The new generation ID, or `last_gen` on timeout/stop. + */ + uint64_t wait_for_tick( + uint64_t last_gen, const std::atomic &stop_token, uint64_t timeout_ms = 10); + + /** + * @brief Signals that a runner has completed its work for the current tick. + * + * @param next_delay_ms The requested delay until the next tick (from `tox_iteration_interval`). + */ + void tick_complete(uint32_t next_delay_ms = 50); + // Global Access FakeClock &clock() { return *clock_; } NetworkUniverse &net() { return *net_; } @@ -53,6 +105,21 @@ private: std::unique_ptr clock_; std::unique_ptr net_; uint32_t node_count_ = 0; + + // Barrier State + std::mutex barrier_mutex_; + std::condition_variable barrier_cv_; + uint64_t current_generation_ = 0; + int registered_runners_ = 0; + std::atomic active_runners_{0}; + std::atomic next_step_min_{50}; + + struct TickListener { + TickListenerId id; + std::function callback; + }; + std::vector tick_listeners_; + TickListenerId next_listener_id_ = 0; }; /** @@ -90,6 +157,8 @@ public: struct Tox_Random get_c_random() { return random_->get_c_random(); } struct Tox_Memory get_c_memory() { return memory_->get_c_memory(); } + Simulation &simulation() { return sim_; } + // For fuzzing compatibility (exposes first bound UDP socket as "endpoint") FakeUdpSocket *get_primary_socket(); diff --git a/testing/support/public/tox_network.hh b/testing/support/public/tox_network.hh index f799c09e3..4e60260ee 100644 --- a/testing/support/public/tox_network.hh +++ b/testing/support/public/tox_network.hh @@ -8,18 +8,19 @@ #include #include "simulation.hh" +#include "tox_runner.hh" namespace tox::test { struct ConnectedFriend { std::unique_ptr node; - SimulatedNode::ToxPtr tox; + std::unique_ptr runner; uint32_t friend_number; - ConnectedFriend(std::unique_ptr node_in, SimulatedNode::ToxPtr tox_in, + ConnectedFriend(std::unique_ptr node_in, std::unique_ptr runner_in, uint32_t friend_number_in) : node(std::move(node_in)) - , tox(std::move(tox_in)) + , runner(std::move(runner_in)) , friend_number(friend_number_in) { } @@ -44,7 +45,8 @@ struct ConnectedFriend { * @return A vector of ConnectedFriend structures, each representing a friend. */ std::vector setup_connected_friends(Simulation &sim, Tox *main_tox, - SimulatedNode &main_node, int num_friends, const Tox_Options *options = nullptr); + SimulatedNode &main_node, int num_friends, const Tox_Options *options = nullptr, + bool verbose = false); /** * @brief Connects two existing Tox instances as friends. diff --git a/testing/support/public/tox_runner.hh b/testing/support/public/tox_runner.hh new file mode 100644 index 000000000..0f99a9c8d --- /dev/null +++ b/testing/support/public/tox_runner.hh @@ -0,0 +1,116 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later + * Copyright © 2026 The TokTok team. + */ + +#ifndef C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H +#define C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H + +#include +#include +#include +#include +#include +#include + +#include "../../../toxcore/tox_events.h" +#include "mpsc_queue.hh" +#include "simulation.hh" + +namespace tox::test { + +class ToxRunner { +public: + explicit ToxRunner(SimulatedNode &node, const Tox_Options *options = nullptr); + ~ToxRunner(); + + ToxRunner(const ToxRunner &) = delete; + ToxRunner &operator=(const ToxRunner &) = delete; + + struct ToxEventsDeleter { + void operator()(Tox_Events *e) const { tox_events_free(e); } + }; + using ToxEventsPtr = std::unique_ptr; + + /** + * @brief Schedules a task for execution on the runner's thread. + * + * This method is thread-safe and non-blocking. The task is queued and will + * be executed during the runner's event loop cycle. + * + * @param task The function to execute, taking a raw Tox pointer. + */ + void execute(std::function task); + + /** + * @brief Executes a task on the runner's thread and waits for the result. + * + * This method blocks the calling thread until the task has been executed + * by the runner. It automatically handles return value propagation and + * exception safety (though exceptions are not currently propagated). + * + * @tparam Func The type of the callable object. + * @param func The callable to execute, taking a raw Tox pointer. + * @return The result of the callable execution. + */ + template + auto invoke(Func &&func) -> std::invoke_result_t + { + using R = std::invoke_result_t; + auto promise = std::make_shared>(); + auto future = promise->get_future(); + + execute([p = promise, f = std::forward(func)](Tox *tox) { + if constexpr (std::is_void_v) { + f(tox); + p->set_value(); + } else { + p->set_value(f(tox)); + } + }); + + return future.get(); + } + + /** + * @brief Retrieves all accumulated Tox event batches. + * + * Returns a vector of unique pointers to Tox_Events structures that have + * been collected by the runner since the last call. Ownership is transferred + * to the caller. This method is thread-safe. + * + * @return A vector of Tox_Events pointers. + */ + std::vector poll_events(); + + /** + * @brief Accesses the underlying Tox instance directly. + * + * @warning Thread-Safety Violation: This method provides unsafe access to the + * Tox instance. It should ONLY be used when the runner thread is known to be + * idle (e.g., before the loop starts) or for accessing constant/read-only properties. + * For all other operations, use `execute` or `invoke`. + */ + Tox *unsafe_tox() { return tox_.get(); } + +private: + void loop(); + + SimulatedNode::ToxPtr tox_; + std::thread thread_; + + struct Message { + enum Type { Task, Tick, Stop } type; + std::function task; + uint64_t generation = 0; + }; + + MpscQueue queue_; + MpscQueue events_queue_; + + Simulation::TickListenerId tick_listener_id_ = -1; + SimulatedNode &node_; +}; + +} // namespace tox::test + +#endif // C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H diff --git a/testing/support/src/fake_memory.cc b/testing/support/src/fake_memory.cc index 4d67cf4c4..bb0d97036 100644 --- a/testing/support/src/fake_memory.cc +++ b/testing/support/src/fake_memory.cc @@ -45,15 +45,9 @@ void *FakeMemory::malloc(size_t size) header->size = size; header->magic = kMagic; - current_allocation_ += size; - if (current_allocation_ > max_allocation_) { - max_allocation_ = current_allocation_; - } + on_allocation(size); - void *res = header + 1; - // std::cerr << "[FakeMemory] malloc(" << size << ") -> " << res << " (header=" << header << ")" - // << std::endl; - return res; + return header + 1; } void *FakeMemory::realloc(void *ptr, size_t size) @@ -82,7 +76,6 @@ void *FakeMemory::realloc(void *ptr, size_t size) } if (fail) { - // If realloc fails, original block is left untouched. return nullptr; } @@ -92,19 +85,14 @@ void *FakeMemory::realloc(void *ptr, size_t size) return nullptr; } - Header *header = static_cast
(new_ptr); - current_allocation_ -= old_size; - current_allocation_ += size; - if (current_allocation_ > max_allocation_) { - max_allocation_ = current_allocation_; - } + on_deallocation(old_size); + on_allocation(size); + Header *header = static_cast
(new_ptr); header->size = size; header->magic = kMagic; - void *res = header + 1; - // std::cerr << "[FakeMemory] realloc(" << ptr << ", " << size << ") -> " << res << " (header=" - // << header << ")" << std::endl; - return res; + + return header + 1; } void FakeMemory::free(void *ptr) @@ -127,7 +115,7 @@ void FakeMemory::free(void *ptr) } size_t size = header->size; - current_allocation_ -= size; + on_deallocation(size); header->magic = kFreeMagic; // Mark as free std::free(header); } @@ -141,4 +129,17 @@ void FakeMemory::set_observer(Observer observer) { observer_ = std::move(observe struct Tox_Memory FakeMemory::get_c_memory() { return Tox_Memory{&kFakeMemoryVtable, this}; } +size_t FakeMemory::current_allocation() const { return current_allocation_.load(); } + +size_t FakeMemory::max_allocation() const { return max_allocation_.load(); } + +void FakeMemory::on_allocation(size_t size) +{ + size_t current = current_allocation_.fetch_add(size) + size; + size_t max = max_allocation_.load(std::memory_order_relaxed); + while (current > max && !max_allocation_.compare_exchange_weak(max, current)) { } +} + +void FakeMemory::on_deallocation(size_t size) { current_allocation_.fetch_sub(size); } + } // namespace tox::test diff --git a/testing/support/src/simulation.cc b/testing/support/src/simulation.cc index 3a2d3364a..42835a05a 100644 --- a/testing/support/src/simulation.cc +++ b/testing/support/src/simulation.cc @@ -1,7 +1,9 @@ #include "../public/simulation.hh" #include +#include #include +#include namespace tox::test { @@ -24,11 +26,123 @@ void Simulation::advance_time(uint64_t ms) void Simulation::run_until(std::function condition, uint64_t timeout_ms) { uint64_t start_time = clock_->current_time_ms(); - while (!condition()) { + + // Initial check + if (condition()) + return; + + while (true) { if (clock_->current_time_ms() - start_time > timeout_ms) { break; } - advance_time(10); // 10ms ticks + + // 1. Advance Global Time + // Determine the time step based on the minimum requested delay from all runners + // during the previous tick. We default to 50ms if no specific request was made. + // The `exchange` operation resets the minimum accumulator for the current tick. + uint32_t step = next_step_min_.exchange(50); + advance_time(step); + + // 2. Start Barrier (Signal Runners) + // Notify all registered runners that time has advanced and they should proceed + // with their next iteration. + { + std::lock_guard lock(barrier_mutex_); + current_generation_++; + // Initialize the countdown of active runners for this tick. + active_runners_.store(registered_runners_); + + for (const auto &l : tick_listeners_) { + l.callback(current_generation_); + } + } + barrier_cv_.notify_all(); + + // 3. End Barrier (Wait for Completion) + // Block until all active runners have reported completion via `tick_complete()`. + { + std::unique_lock lock(barrier_mutex_); + // We use a lambda predicate to handle spurious wakeups. + // The wait finishes when `active_runners_` reaches zero. + barrier_cv_.wait(lock, [this] { return active_runners_.load() == 0; }); + } + + // 4. Check condition + if (condition()) + return; + } +} + +Simulation::TickListenerId Simulation::register_tick_listener( + std::function listener) +{ + std::lock_guard lock(barrier_mutex_); + TickListenerId id = next_listener_id_++; + tick_listeners_.push_back({id, std::move(listener)}); + return id; +} + +void Simulation::unregister_tick_listener(TickListenerId id) +{ + std::lock_guard lock(barrier_mutex_); + for (auto it = tick_listeners_.begin(); it != tick_listeners_.end(); ++it) { + if (it->id == id) { + tick_listeners_.erase(it); + break; + } + } +} + +uint64_t Simulation::register_runner() +{ + std::lock_guard lock(barrier_mutex_); + registered_runners_++; + return current_generation_; +} + +void Simulation::unregister_runner() +{ + std::lock_guard lock(barrier_mutex_); + registered_runners_--; + + // If we are currently running a tick (active_runners > 0), we need to decrement it + // because this runner will not be calling tick_complete() + if (active_runners_.load() > 0) { + if (active_runners_.fetch_sub(1) == 1) { + barrier_cv_.notify_all(); + } + } +} + +uint64_t Simulation::wait_for_tick( + uint64_t last_gen, const std::atomic &stop_token, uint64_t timeout_ms) +{ + std::unique_lock lock(barrier_mutex_); + // Wait until generation increases (new tick started) OR we are stopped OR timeout + bool result = barrier_cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms), + [&] { return current_generation_ > last_gen || stop_token; }); + + if (stop_token) + return last_gen; + if (!result) + return last_gen; // Timeout + return current_generation_; +} + +void Simulation::tick_complete(uint32_t next_delay_ms) +{ + // Atomic min reduction + uint32_t current = next_step_min_.load(std::memory_order_relaxed); + while ( + next_delay_ms < current && !next_step_min_.compare_exchange_weak(current, next_delay_ms)) { + // If exchange failed, current was updated to actual value, so loop checks again + } + + // We don't need the mutex to decrement the atomic + if (active_runners_.fetch_sub(1) == 1) { + // Last runner to finish: notify main thread + std::lock_guard lock(barrier_mutex_); + barrier_cv_.notify_all(); } } diff --git a/testing/support/src/tox_network.cc b/testing/support/src/tox_network.cc index d306e6256..f30545711 100644 --- a/testing/support/src/tox_network.cc +++ b/testing/support/src/tox_network.cc @@ -5,17 +5,21 @@ #include "../public/tox_network.hh" #include +#include #include +#include #include "../../../toxcore/network.h" #include "../../../toxcore/tox.h" +#include "../../../toxcore/tox_events.h" +#include "../public/tox_runner.hh" namespace tox::test { ConnectedFriend::~ConnectedFriend() = default; std::vector setup_connected_friends(Simulation &sim, Tox *main_tox, - SimulatedNode &main_node, int num_friends, const Tox_Options *options) + SimulatedNode &main_node, int num_friends, const Tox_Options *options, bool verbose) { std::vector friends; friends.reserve(num_friends); @@ -42,32 +46,30 @@ std::vector setup_connected_friends(Simulation &sim, Tox *main_ for (int i = 0; i < num_friends; ++i) { auto node = sim.create_node(); - auto tox = node->create_tox(options); - if (!tox) { - return {}; - } + auto runner = std::make_unique(*node, options); uint8_t friend_pk[TOX_PUBLIC_KEY_SIZE]; - tox_self_get_public_key(tox.get(), friend_pk); + runner->invoke([&](Tox *tox) { tox_self_get_public_key(tox, friend_pk); }); Tox_Err_Friend_Add err; uint32_t fn = tox_friend_add_norequest(main_tox, friend_pk, &err); if (fn == UINT32_MAX || err != TOX_ERR_FRIEND_ADD_OK) { return {}; } - if (tox_friend_add_norequest(tox.get(), main_pk, &err) == UINT32_MAX - || err != TOX_ERR_FRIEND_ADD_OK) { - return {}; - } - // Bootstrap to the main node AND the PREVIOUS node in the chain - tox_bootstrap(tox.get(), main_ip_str, main_port, main_dht_id, nullptr); - if (i > 0) { - tox_bootstrap(tox.get(), prev_ip_str, prev_port, prev_dht_id, nullptr); - } + // Execute add friend and bootstrap on runner + runner->execute([=](Tox *tox) { + tox_friend_add_norequest(tox, main_pk, nullptr); + tox_bootstrap(tox, main_ip_str, main_port, main_dht_id, nullptr); + if (i > 0) { + tox_bootstrap(tox, prev_ip_str, prev_port, prev_dht_id, nullptr); + } + }); + + // Retrieve previous node's DHT ID and update IP for the next iteration. + // We use invoke to safely fetch data from the runner thread. + runner->invoke([&](Tox *tox) { tox_self_get_dht_id(tox, prev_dht_id); }); - // Update prev for next node - tox_self_get_dht_id(tox.get(), prev_dht_id); ip_parse_addr(&node->ip, prev_ip_str, sizeof(prev_ip_str)); FakeUdpSocket *node_socket = node->get_primary_socket(); @@ -76,67 +78,72 @@ std::vector setup_connected_friends(Simulation &sim, Tox *main_ } prev_port = node_socket->local_port(); - friends.push_back({std::move(node), std::move(tox), fn}); + friends.push_back({std::move(node), std::move(runner), fn}); - // Run simulation to let DHT stabilize - sim.run_until( - [&]() { - tox_iterate(main_tox, nullptr); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - } - return false; - }, - 200); + // Run the simulation periodically to allow the DHT to stabilize incrementally + // as we add nodes, rather than waiting until the end. + if (friends.size() % 10 == 0) { + sim.run_until([&]() { return false; }, 20); + } } - // Optional: Bootstrap main_tox to the last node to complete the circle if (!friends.empty()) { tox_bootstrap(main_tox, prev_ip_str, prev_port, prev_dht_id, nullptr); } - // Run simulation until all are connected + std::vector friends_connected(friends.size(), false); + sim.run_until( [&]() { - bool all_connected = true; - int connected_count = 0; tox_iterate(main_tox, nullptr); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - if (tox_friend_get_connection_status(main_tox, f.friend_number, nullptr) - != TOX_CONNECTION_NONE - && tox_friend_get_connection_status(f.tox.get(), 0, nullptr) - != TOX_CONNECTION_NONE) { - connected_count++; - } else { - all_connected = false; - } - } - static uint64_t last_print = 0; - if (sim.clock().current_time_ms() - last_print > 1000) { - std::cerr << "[setup_connected_friends] Friends connected: " << connected_count - << "/" << friends.size() << " (time: " << sim.clock().current_time_ms() - << "ms)" << std::endl; - if (connected_count < static_cast(friends.size()) - && sim.clock().current_time_ms() > 10000) { - for (size_t i = 0; i < friends.size(); ++i) { - auto s1 = tox_friend_get_connection_status( - main_tox, friends[i].friend_number, nullptr); - auto s2 - = tox_friend_get_connection_status(friends[i].tox.get(), 0, nullptr); - if (s1 == TOX_CONNECTION_NONE || s2 == TOX_CONNECTION_NONE) { - std::cerr << " Friend " << i << " not connected (Main->F: " << s1 - << ", F->Main: " << s2 << ")" << std::endl; + // Check connection status + int connected_count = 0; + + for (size_t i = 0; i < friends.size(); ++i) { + // Check if main sees friend + bool main_sees_friend + = tox_friend_get_connection_status(main_tox, friends[i].friend_number, nullptr) + != TOX_CONNECTION_NONE; + + // Check if friend sees main by polling events from the runner + auto batches = friends[i].runner->poll_events(); + for (const auto &batch : batches) { + size_t size = tox_events_get_size(batch.get()); + for (size_t k = 0; k < size; ++k) { + const Tox_Event *e = tox_events_get(batch.get(), k); + if (tox_event_get_type(e) == TOX_EVENT_FRIEND_CONNECTION_STATUS) { + auto *ev = tox_event_get_friend_connection_status(e); + if (tox_event_friend_connection_status_get_connection_status(ev) + != TOX_CONNECTION_NONE) { + friends_connected[i] = true; + } else { + friends_connected[i] = false; + } } } } + if (main_sees_friend && friends_connected[i]) { + connected_count++; + } + } + + if (connected_count == static_cast(friends.size())) { + return true; + } + + static uint64_t last_print = 0; + if (verbose && sim.clock().current_time_ms() - last_print > 1000) { + std::cerr << "[setup_connected_friends] Friends connected: " << connected_count + << "/" << friends.size() << " (time: " << sim.clock().current_time_ms() + << "ms)" << std::endl; last_print = sim.clock().current_time_ms(); } - return all_connected; + + return false; }, - 300000); // 5 minutes simulation time for 100 nodes to converge + 300000); return friends; } @@ -144,6 +151,10 @@ std::vector setup_connected_friends(Simulation &sim, Tox *main_ bool connect_friends( Simulation &sim, SimulatedNode &node1, Tox *tox1, SimulatedNode &node2, Tox *tox2) { + // This helper function assumes the Tox instances are running in the current thread + // (e.g., standard unit test) or that the caller is handling thread safety if they + // are part of a runner. It uses direct tox_iterate calls. + uint8_t pk1[TOX_PUBLIC_KEY_SIZE]; uint8_t pk2[TOX_PUBLIC_KEY_SIZE]; tox_self_get_public_key(tox1, pk1); @@ -217,52 +228,26 @@ uint32_t setup_connected_group( &err_new); if (main_state.group_number == UINT32_MAX || err_new != TOX_ERR_GROUP_NEW_OK) { - std::cerr << "tox_group_new failed with error: " << err_new << std::endl; return UINT32_MAX; } - std::vector> friend_states; - friend_states.reserve(friends.size()); + // Friend states tracked via events + std::vector friend_states(friends.size()); - for (size_t i = 0; i < friends.size(); ++i) { - auto state = std::make_unique(); - tox_callback_group_peer_join( - friends[i].tox.get(), [](Tox *, uint32_t, uint32_t, void *user_data) { - static_cast(user_data)->peer_count++; - }); + // Main tox sends invites; friends accept via events polled from their runners. - tox_callback_group_invite(friends[i].tox.get(), - [](Tox *tox, uint32_t friend_number, const uint8_t *invite_data, - size_t invite_data_length, const uint8_t *, size_t, void *user_data) { - NodeGroupState *ng_state = static_cast(user_data); - Tox_Err_Group_Invite_Accept err_accept; - ng_state->group_number - = tox_group_invite_accept(tox, friend_number, invite_data, invite_data_length, - reinterpret_cast("peer"), 4, nullptr, 0, &err_accept); - if (ng_state->group_number == UINT32_MAX - || err_accept != TOX_ERR_GROUP_INVITE_ACCEPT_OK) { - ng_state->group_number = UINT32_MAX; - } - }); - - friend_states.push_back(std::move(state)); - } - - // Run until all have joined and see everyone bool success = false; - uint64_t last_print = 0; size_t invites_sent = 0; sim.run_until( [&]() { tox_iterate(main_tox, &main_state); - // Throttle invites: keep max 5 pending + // Throttle invites size_t accepted_count = 0; - for (size_t k = 0; k < invites_sent; ++k) { - if (friend_states[k]->group_number != UINT32_MAX) { + for (const auto &fs : friend_states) { + if (fs.group_number != UINT32_MAX) accepted_count++; - } } while (invites_sent < friends.size() && (invites_sent - accepted_count) < 5) { @@ -271,58 +256,58 @@ uint32_t setup_connected_group( friends[invites_sent].friend_number, &err_invite)) { invites_sent++; } else { - if (err_invite != TOX_ERR_GROUP_INVITE_FRIEND_FAIL_SEND) { - std::cerr << "Invite failed for friend " << invites_sent << ": " - << err_invite << std::endl; - } - break; // Stop trying to send for this tick if we failed + break; } } - bool all_see_all = true; - - if (main_state.peer_count < friends.size()) { - all_see_all = false; - } - + // Process friend events for (size_t i = 0; i < friends.size(); ++i) { - tox_iterate(friends[i].tox.get(), friend_states[i].get()); - if (friend_states[i]->group_number == UINT32_MAX - || friend_states[i]->peer_count < friends.size()) { - all_see_all = false; - } - } + auto batches = friends[i].runner->poll_events(); + for (const auto &batch : batches) { + size_t size = tox_events_get_size(batch.get()); + for (size_t k = 0; k < size; ++k) { + const Tox_Event *e = tox_events_get(batch.get(), k); + Tox_Event_Type type = tox_event_get_type(e); - if ((sim.clock().current_time_ms() - last_print) % 5000 == 0) { - int joined = 0; - int fully_connected = 0; - if (main_state.group_number != UINT32_MAX) - joined++; - if (main_state.peer_count >= friends.size()) - fully_connected++; + if (type == TOX_EVENT_GROUP_INVITE) { + auto *ev = tox_event_get_group_invite(e); + uint32_t friend_number = tox_event_group_invite_get_friend_number(ev); + const uint8_t *data = tox_event_group_invite_get_invite_data(ev); + size_t len = tox_event_group_invite_get_invite_data_length(ev); - for (const auto &fs : friend_states) { - if (fs->group_number != UINT32_MAX) { - joined++; - if (fs->peer_count >= friends.size()) - fully_connected++; + // Accept invite on runner thread. + // We must copy data because the event structure will be freed. + std::vector invite_data(data, data + len); + + friends[i].runner->execute([=](Tox *tox) { + Tox_Err_Group_Invite_Accept err; + tox_group_invite_accept(tox, friend_number, invite_data.data(), + invite_data.size(), reinterpret_cast("peer"), + 4, nullptr, 0, &err); + }); + } else if (type == TOX_EVENT_GROUP_PEER_JOIN) { + friend_states[i].peer_count++; + } else if (type == TOX_EVENT_GROUP_SELF_JOIN) { + auto *ev = tox_event_get_group_self_join(e); + friend_states[i].group_number + = tox_event_group_self_join_get_group_number(ev); + } } } - std::cerr << "[setup_connected_group] Main peer count: " << main_state.peer_count - << "/" << friends.size() << ", Nodes joined: " << joined << "/" - << (friends.size() + 1) << ", fully connected: " << fully_connected << "/" - << (friends.size() + 1) << " (time: " << sim.clock().current_time_ms() - << "ms)" << std::endl; - last_print = sim.clock().current_time_ms(); } - if (all_see_all) { - success = true; - return true; + if (main_state.peer_count < friends.size()) + return false; + + for (const auto &fs : friend_states) { + if (fs.group_number == UINT32_MAX || fs.peer_count < friends.size()) + return false; } - return false; + + success = true; + return true; }, - 300000); // 5 minutes + 300000); return success ? main_state.group_number : UINT32_MAX; } diff --git a/testing/support/src/tox_runner.cc b/testing/support/src/tox_runner.cc new file mode 100644 index 000000000..055c6d5dd --- /dev/null +++ b/testing/support/src/tox_runner.cc @@ -0,0 +1,92 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later + * Copyright © 2026 The TokTok team. + */ + +#include "../public/tox_runner.hh" + +namespace tox::test { + +ToxRunner::ToxRunner(SimulatedNode &node, const Tox_Options *options) + : tox_(node.create_tox(options)) + , node_(node) +{ + tox_events_init(tox_.get()); + node_.simulation().register_runner(); + + tick_listener_id_ = node_.simulation().register_tick_listener([this](uint64_t gen) { + Message msg; + msg.type = Message::Tick; + msg.generation = gen; + queue_.push(std::move(msg)); + }); + + thread_ = std::thread([this] { loop(); }); +} + +ToxRunner::~ToxRunner() +{ + // Unregister first to prevent new ticks and update simulation counters + node_.simulation().unregister_tick_listener(tick_listener_id_); + node_.simulation().unregister_runner(); + + Message msg; + msg.type = Message::Stop; + queue_.push(std::move(msg)); + + if (thread_.joinable()) { + thread_.join(); + } +} + +void ToxRunner::execute(std::function task) +{ + Message msg; + msg.type = Message::Task; + msg.task = std::move(task); + queue_.push(std::move(msg)); +} + +std::vector ToxRunner::poll_events() +{ + std::vector ret; + ToxEventsPtr ptr; + while (events_queue_.try_pop(ptr)) { + ret.push_back(std::move(ptr)); + ptr = nullptr; // Reset ptr to avoid use-after-move warning, although try_pop overwrites + // it. + } + return ret; +} + +void ToxRunner::loop() +{ + while (true) { + Message msg = queue_.pop(); // Blocking wait + + switch (msg.type) { + case Message::Stop: + return; + + case Message::Task: + if (msg.task) { + msg.task(tox_.get()); + } + break; + + case Message::Tick: { + // Run Tox Events + Tox_Err_Events_Iterate err; + Tox_Events *events = tox_events_iterate(tox_.get(), false, &err); + if (events) { + events_queue_.push(ToxEventsPtr(events)); + } + + uint32_t interval = tox_iteration_interval(tox_.get()); + node_.simulation().tick_complete(interval); + break; + } + } + } +} + +} // namespace tox::test diff --git a/testing/support/tox_network_test.cc b/testing/support/tox_network_test.cc index 1293fd5ad..3b16223cc 100644 --- a/testing/support/tox_network_test.cc +++ b/testing/support/tox_network_test.cc @@ -6,6 +6,8 @@ #include +#include + namespace tox::test { namespace { @@ -22,6 +24,8 @@ namespace { ASSERT_EQ(friends.size(), num_friends); + // Verification of connection status is done inside setup_connected_friends now, + // but we can double check main_tox's view. for (const auto &f : friends) { EXPECT_NE(tox_friend_get_connection_status(main_tox.get(), f.friend_number, nullptr), TOX_CONNECTION_NONE); @@ -29,7 +33,7 @@ namespace { // Verify they can actually communicate struct Context { - int count = 0; + std::atomic count{0}; } ctx; tox_callback_friend_message(main_tox.get(), @@ -38,16 +42,14 @@ namespace { }); for (const auto &f : friends) { - const uint8_t msg[] = "hello"; - tox_friend_send_message( - f.tox.get(), 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr); + f.runner->execute([](Tox *tox) { + const uint8_t msg[] = "hello"; + tox_friend_send_message(tox, 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr); + }); } sim.run_until([&]() { tox_iterate(main_tox.get(), &ctx); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - } return ctx.count == num_friends; }); @@ -68,7 +70,7 @@ namespace { ASSERT_EQ(friends.size(), num_friends); struct Context { - int count = 0; + std::atomic count{0}; } ctx; tox_callback_friend_message(main_tox.get(), @@ -77,17 +79,15 @@ namespace { }); for (const auto &f : friends) { - const uint8_t msg[] = "hello"; - tox_friend_send_message( - f.tox.get(), 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr); + f.runner->execute([](Tox *tox) { + const uint8_t msg[] = "hello"; + tox_friend_send_message(tox, 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr); + }); } sim.run_until( [&]() { tox_iterate(main_tox.get(), &ctx); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - } return ctx.count == num_friends; }, 60000); @@ -113,10 +113,10 @@ namespace { EXPECT_NE(tox_friend_get_connection_status(tox2.get(), 0, nullptr), TOX_CONNECTION_NONE); // Verify communication - bool received = false; + std::atomic received{false}; tox_callback_friend_message(tox2.get(), [](Tox *, uint32_t, Tox_Message_Type, const uint8_t *, size_t, void *user_data) { - *static_cast(user_data) = true; + *static_cast *>(user_data) = true; }); const uint8_t msg[] = "hello"; @@ -125,7 +125,7 @@ namespace { sim.run_until([&]() { tox_iterate(tox1.get(), nullptr); tox_iterate(tox2.get(), &received); - return received; + return received.load(); }); EXPECT_TRUE(received); @@ -148,7 +148,7 @@ namespace { // Verify we can send a group message struct Context { - int count = 0; + std::atomic count{0}; } ctx; tox_callback_group_message(main_tox.get(), @@ -156,20 +156,18 @@ namespace { void *user_data) { static_cast(user_data)->count++; }); for (const auto &f : friends) { - const uint8_t msg[] = "hello"; - uint32_t f_gn = 0; // It should be 0 since it's the first group. - Tox_Err_Group_Send_Message err_send; - tox_group_send_message( - f.tox.get(), f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send); - EXPECT_EQ(err_send, TOX_ERR_GROUP_SEND_MESSAGE_OK); + f.runner->execute([](Tox *tox) { + const uint8_t msg[] = "hello"; + uint32_t f_gn = 0; // First group + Tox_Err_Group_Send_Message err_send; + tox_group_send_message( + tox, f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send); + }); } sim.run_until( [&]() { tox_iterate(main_tox.get(), &ctx); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - } return ctx.count == num_friends; }, 10000); @@ -193,7 +191,7 @@ namespace { EXPECT_NE(group_number, UINT32_MAX); struct Context { - int count = 0; + std::atomic count{0}; } ctx; tox_callback_group_message(main_tox.get(), @@ -201,20 +199,18 @@ namespace { void *user_data) { static_cast(user_data)->count++; }); for (const auto &f : friends) { - const uint8_t msg[] = "hello"; - uint32_t f_gn = 0; - Tox_Err_Group_Send_Message err_send; - tox_group_send_message( - f.tox.get(), f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send); - EXPECT_EQ(err_send, TOX_ERR_GROUP_SEND_MESSAGE_OK); + f.runner->execute([](Tox *tox) { + const uint8_t msg[] = "hello"; + uint32_t f_gn = 0; + Tox_Err_Group_Send_Message err_send; + tox_group_send_message( + tox, f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send); + }); } sim.run_until( [&]() { tox_iterate(main_tox.get(), &ctx); - for (auto &f : friends) { - tox_iterate(f.tox.get(), nullptr); - } return ctx.count == num_friends; }, 120000);