Implementation of a Thread-Safe Signal and Slot Mechanism Based on C++17

Why do we need our own “Qt-style communication mechanism”?

In GUI programming, asynchronous systems, and event-driven architectures, signals and slots are a classic design pattern. They enable loose coupling communication between objects, avoiding complex callback nesting and direct dependencies.

The signal-slot mechanism in the Qt framework is well-known, but it relies on a large meta-object compiler (moc) and runtime type information (RTTI), which can be “too heavy” for lightweight projects or embedded systems.

This article will take you deep into the analysis of a thread-safe signal and slot mechanism implemented in pure C++17, supporting synchronous/asynchronous calls and customizable thread pool scheduling. We will comprehensively interpret this elegant communication model from principles, architecture, implementation details to performance optimization.

Overall Architecture Overview

Let’s first look at the module division of the entire system:

+------------------+
|   Object Base Class | <-- Provides thread pool binding capability
+------------------+
         |
         v
+------------------+
|   Signal<...>     | <-- Template class, manages connections and emits signals
+------------------+
         |
         v
+------------------+
|  Connection<...>  | <-- Manages the lifecycle of a single connection
+------------------+
         |
         v
+------------------+
|  ThreadPool       | <-- External dependency, provides asynchronous execution capability
+------------------+

Core objectives:

• Support member functions, free functions, and Lambdas as slot functions;
• Support synchronous calls (emit) and asynchronous calls (emitAsync);
• Support thread safety, safe connection, disconnection, and emission in a multithreaded environment;
• Support automatic memory management, avoiding dangling pointers; extensible to signals of any parameter type; and implement receiver thread pool awareness through the Object base class.

Core Component Details

Object Base Class: Thread Context Binding

class Object {
public:
    Object() : thread_pool_(&global_thread_pool) {}
    virtual ~Object() = default;

    void setThreadPool(ThreadPool* pool) { ... }
    ThreadPool* getThreadPool() const { return thread_pool_; }

private:
    ThreadPool* thread_pool_;
};

Design Intent

• All classes that can act as “signal receivers” must inherit from Object.
• Each Object instance can bind to a ThreadPool, indicating its “belonging thread environment”.
• When using connect(signal, obj, &Obj::slot), the system automatically retrieves obj->getThreadPool() to determine which thread pool the slot function should execute in.

⚠️ Notes

• Uses static_assert(std::is_base_of_v<Object, R>) to enforce constraints on the receiver type, ensuring type safety.
• Copy construction and assignment are prohibited (deleted) to prevent shallow copies leading to incorrect sharing of thread pool pointers.

Connection Class: Lifecycle Management of Connections

template<typename... SignalArgs>
class Connection {
    struct ConnectionData {
        bool connected = true;
        size_t id = 0;
        mutable std::shared_mutex signal_mutex;
        std::function<void(size_t)> disconnect_callback;
    };
    std::weak_ptr<ConnectionData> connection_data_;
};

Core Design: Shared State + RAII + Weak Reference

• ConnectionData is a shared state object for all connections, containing:
• connected: whether it has been disconnected;
• id: unique identifier;
• disconnect_callback: notifies Signal to clean itself when the connection is disconnected.
• connection_data_ is a weak_ptr to avoid circular references leading to memory leaks.
• The disconnect() method obtains a shared_ptr via lock(), and if it exists, marks it as disconnected and triggers the callback.
• isConnected() safely checks the validity of the connection.

🧩 Why use weak_ptr? To prevent Signal from holding a shared_ptr to ConnectionData while Connection holds the same object, causing it to be unreleased.

Signal Class: The Hub for Registering and Emitting Signals

This is the core of the entire mechanism, and we will break it down by functionality.

(1) Connecting Slot Functions: Two Overloads of connect()① Free Functions / Lambdas

template<typename T>
Connection<SignalArgs...> connect(T&& slot) {
    return connect_impl(nullptr, std::forward<T>(slot));
}

• The target thread pool is nullptr, meaning it executes directly during synchronous calls and uses the global thread pool for asynchronous calls.

② Member Functions (Bound to Object Derived Classes)

template<typename T, typename R>
Connection<SignalArgs...> connect(R* receiver, T&& slot) {
    static_assert(...);
    ThreadPool* pool = receiver ? receiver->getThreadPool() : nullptr;
    auto wrapper = [receiver, slot](SignalArgs... args) {
        (receiver->*slot)(args...);
    };
    return connect_impl(pool, std::move(wrapper));
}

• Wraps the member function pointer &Class::func into a callable lambda;
• Simultaneously extracts the receiver’s thread pool;
• Finally, it is uniformly converted to std::function<void(SignalArgs…)> for storage.

💡 Tip: Use [receiver, slot] to capture this and the member function pointer, forming a closure.

(2) Internal Storage Structure: ConnectedSlot

struct ConnectedSlot {
    std::function<void(SignalArgs...)> slot;
    std::shared_ptr<ConnectionData> connection_data;
    size_t id;
    ThreadPool* target_thread_pool;
};

• All connection information is stored in a std::list;
• Using std::list instead of vector: efficient insertion/deletion, iterators do not become invalid (important!);
• target_thread_pool records the thread pool the receiver expects to run on.

Synchronous Emission: emit(…)

void emit(SignalArgs... args) {
    std::shared_lock<std::shared_mutex> lock(mutex_);
    auto local_slots = slots_; // Copy the list
    lock.unlock();

    for (const auto&& slot_info : local_slots) {
        if (slot_info.connection_data->connected) {
            try {
                slot_info.slot(args...);
            } catch (...) { ... }
        }
    }
}

Key Point Analysis

• Uses std::shared_mutex to implement read-write separation: multiple emits can occur concurrently (read-only), but connect/disconnect requires exclusive access.
• Copies slots_ to a local variable: prevents other threads from modifying the list during iteration (e.g., disconnect() deleting elements), avoiding iterator invalidation.
• Locking time is very short, only used for copying, improving concurrency performance.
• Exception handling protection mechanism prevents a crash in one slot function from affecting others.

Asynchronous Emission: emitAsync(…) — The Real Highlight!

void emitAsync(SignalArgs... args) {
    std::shared_lock<std::shared_mutex> lock(mutex_);
    auto local_slots = slots_;
    lock.unlock();

    for (const auto&& slot_info : local_slots) {
        if (slot_info.connection_data->connected) {
            auto slot_copy = std::move(slot_info.slot);
            auto args_tuple = std::make_tuple(std::forward<SignalArgs>(args)...);
            ThreadPool* target_pool = slot_info.target_thread_pool ? ... : &global_thread_pool;

            auto task = [slot_copy = std::move(slot_copy), args_tuple = std::move(args_tuple)]() mutable {
                try {
                    std::apply(slot_copy, std::move(args_tuple));
                } catch (...) { ... }
            };

            target_pool->push(std::move(task));
        }
    }
}

• std::make_tuple(std::forward<…>) perfectly forwards and packages variable-length parameters into a tuple
• auto task = … mutable {} constructs a no-argument callable object, adapting to the thread pool interface
• Move semantics std::move(slot_info.slot) reduces copy overhead, improving performance
• std::apply(func, tuple) expands the tuple at runtime to call the function
• Capturing mutable allows the lambda to modify captured copies (e.g., moving args_tuple)

Completely decouples signal emission from slot execution, with the emitting thread not blocking.

⚠️ Note: Parameters are copied or moved into the task, requiring all parameters to support copy or move semantics.

Global connect Function and Macro

#define SLOT(...) (&__VA_ARGS__)

template<typename... SigArgs, typename SlotFunc>
Connection<SigArgs...> connect(Signal<SigArgs...> &signal, SlotFunc&& slot) {
    return signal.connect(std::forward<SlotFunc>(slot));
}

// Overloaded version supports receiver
template<typename SignalType, typename ReceiverType, typename MemberFuncType>
auto connect(SignalType&& signal, ReceiverType* receiver, MemberFuncType slot)
-> decltype(signal.connect(receiver, slot))

Design Advantages

• Provides a global function syntax similar to Qt’s connect(…);
• Supports function overload resolution, automatically matching the correct version;
• SLOT(…) macro is used to obtain member function addresses, with clear semantics.

Thread Safety and Performance Analysis

Thread Safety Strategy

Operation	Lock Mechanism	Description
connect / disconnect	unique_lock<shared_mutex>	Write operation, mutually exclusive access
emit / emitAsync	shared_lock<shared_mutex>	Multiple emissions can occur concurrently
Iterating slots_	Copying a copy	Avoid holding locks during iteration

Advantages: Excellent performance in read-heavy write-light scenarios.Potential Issues: If the number of connections is extremely large, the copying cost of slots_ is high.

Memory Management and Exception Safety

• Uses std::shared_ptr and std::weak_ptr for automatic lifecycle management;
• disconnect_callback automatically cleans up entries in Signal when ConnectionData is destroyed;
• All exceptions are caught to prevent crash propagation;
• Move semantics reduce temporary object overhead.

Complete Source Code

#pragma once

#include <iostream>
#include <functional>
#include <thread>
#include <mutex>
#include <shared_mutex>
#include <condition_variable>
#include <queue>
#include <list>
#include <memory>
#include <atomic>
#include <sstream>
#include <tuple>

#include "ThreadPool.hpp"

// --- Global Thread Pool Management (Simplified) ---
static ThreadPool global_thread_pool(4);

// --- Object Base Class ---
class Object
{
public:
 Object() : thread_pool_(&global_thread_pool)
 { }
 virtual ~Object() = default;

 void setThreadPool(ThreadPool* pool)
 {
  if(pool)
  {
   thread_pool_ = pool;
  }
 }

 ThreadPool *getThreadPool() const
 {
  return thread_pool_;
 }

 Object(const Object &) = delete;
 Object &operator=(const Object &) = delete;

private:
 ThreadPool * thread_pool_;
};

// --- Signal and Slot Mechanism ---

// Connection Class
template<typename... SignalArgs>
class Signal;

template<typename... SignalArgs>
class Connection
{
 friend class Signal<SignalArgs...>;
public:
 Connection() = default;
 ~Connection() = default;

 void disconnect()
 {
  if(auto shared_con_data = connection_data_.lock())
  {
   std::unique_lock<std::shared_mutex> lock(shared_con_data->signal_mutex);
   if(shared_con_data->connected)
   {
    shared_con_data->connected = false;
    if(shared_con_data->disconnect_callback)
    {
     shared_con_data->disconnect_callback(shared_con_data->id);
    }
   }
  }
 }

 bool isConnected() const
 {
  auto shared_con_data = connection_data_.lock();
  return shared_con_data && shared_con_data->connected;
 }

private:
 struct ConnectionData
 {
  bool connected = true;
  size_t id = 0;
  mutable std::shared_mutex signal_mutex;
  std::function<void(size_t)> disconnect_callback;
 };
 std::weak_ptr<typename Connection::ConnectionData> connection_data_;
};

// Signal Class
template<typename... SignalArgs>
class Signal
{
private:
 // Storage for connected slot information
 struct ConnectedSlot
 {
  std::function<void(SignalArgs...)> slot; // Unified as std::function
  std::shared_ptr<typename Connection<SignalArgs...>::ConnectionData> connection_data;
  size_t id;
  ThreadPool *target_thread_pool; // Thread pool associated with the receiver
 };

 mutable std::shared_mutex mutex_;
 std::list<ConnectedSlot> slots_;
 std::atomic<size_t> current_id_{0};

 // Connection implementation helper function
template<typename Callable>
 Connection<SignalArgs...> connect_impl(ThreadPool* pool, Callable&& callable)
 {
  auto con_data = std::make_shared<typename Connection<SignalArgs...>::ConnectionData>();
  con_data->disconnect_callback = [this](size_t slot_id)
  {
   std::unique_lock<std::shared_mutex> lock(this->mutex_);
   this->slots_.remove_if([slot_id](const ConnectedSlot && info)
   {
    return info.id == slot_id;
   });
  };

  std::unique_lock<std::shared_mutex> lock(mutex_);
  size_t id = ++current_id_;
  con_data->id = id;

  slots_.emplace_back(ConnectedSlot
  {
   std::function<void(SignalArgs...)>(std::forward<Callable>(callable)),
   con_data,
   id,
   pool
  });

  Connection<SignalArgs...> conn;
  conn.connection_data_ = con_data;
  return conn;
 }

public:
 Signal() = default;
 ~Signal() = default;

 // Connect free functions or Lambdas
 template<typename T>
 Connection<SignalArgs...> connect(T&& slot)
 {
  return connect_impl(nullptr, std::forward<T>(slot));
 }

 // Connect member functions to Object derived class instances
 template<typename T, typename R>
 Connection<SignalArgs...> connect(R* receiver, T&& slot)
 {
  static_assert(std::is_base_of_v<Object, R>, "Receiver must inherit from Object");
  ThreadPool* pool = (receiver) ? receiver->getThreadPool() : nullptr;
  // Wrap member function call into std::function
  auto wrapper = [receiver, slot](SignalArgs... args)
  {
   (receiver->*slot)(args...);
  };
  return connect_impl(pool, std::move(wrapper));
 }

 // Synchronously emit signal
 void emit(SignalArgs... args)
 {
  std::shared_lock<std::shared_mutex> lock(mutex_);
  // Copy connection list to avoid modification during iteration
  auto local_slots = slots_;
  lock.unlock();

  for(const auto&& slot_info : local_slots)
  {
   if(slot_info.connection_data->connected)
   {
    try
    {
     // Directly call the wrapped std::function
     slot_info.slot(args...);
    }
    catch(...)
    {
     std::cerr << "[Signal] Exception caught in slot (sync)." << std::endl;
    }
   }
  }
 }

 // Asynchronously emit signal
 void emitAsync(SignalArgs... args)
 {
  std::shared_lock<std::shared_mutex> lock(mutex_);
  // Copy connection list to avoid modification during iteration
  auto local_slots = slots_;
  lock.unlock();

  for(const auto&& slot_info : local_slots)
  {
   if(slot_info.connection_data->connected)
   {
    // 1. Move std::function copy
    auto slot_copy = std::move(slot_info.slot); // Move the slot function
    // 2. Move parameters packaged into tuple
    auto args_tuple = std::make_tuple(std::forward<SignalArgs>(args)...); // Forward args to tuple
    // 3. Determine target thread pool
    ThreadPool* target_pool = slot_info.target_thread_pool ? slot_info.target_thread_pool : &global_thread_pool;

    // 4. Create the final no-argument task lambda
    //    Capture all necessary data (by moving)
    auto task = [slot_copy = std::move(slot_copy), args_tuple = std::move(args_tuple)]() mutable
    {
     try
     {
      // 5. Use std::apply to expand tuple and call slot function
      std::apply(slot_copy, std::move(args_tuple)); // Apply with moved tuple
     }
     catch(...)
     {
      std::cerr << "[Signal] Exception caught in slot (async)." << std::endl;
     }
    };

    // 6. Submit this no-argument task to the correct thread pool
    try
    {
     target_pool->push(std::move(task)); // Enqueue the moved task
    }
    catch(const std::exception&& e)
    {
     std::cerr << "[Signal] Failed to enqueue task: " << e.what() << std::endl;
    }
   }
  }
 }

 size_t getConnectionCount() const
 {
  std::shared_lock<std::shared_mutex> lock(mutex_);
  return slots_.size();
 }
};

// --- Macro Definitions ---
#define SLOT(...) (&__VA_ARGS__)

// --- Global connect function template ---
// Provides convenience for connecting free functions/Lambdas
template<typename... SigArgs, typename SlotFunc>
Connection<SigArgs...> connect(Signal<SigArgs...> &signal, SlotFunc&& slot)
{
 return signal.connect(std::forward<SlotFunc>(slot));
}

// Enable this overload only when signal.connect(...) is valid
template<typename SignalType, typename ReceiverType, typename MemberFuncType>
auto connect(SignalType&& signal, ReceiverType* receiver, MemberFuncType slot)
-> decltype(signal.connect(receiver, slot))
{
 // SFINAE: Only enable this overload when signal.connect(...) is valid
 static_assert(std::is_base_of_v<Object, ReceiverType>, "Receiver must inherit from Object");
 return signal.connect(receiver, slot);
}

// ThreadPool.hpp
#pragma once
#include <vector>
#include <thread>
#include <mutex>
#include <queue>
#include <functional>
#include <condition_variable>
#include <future>
#include <memory>
#include <stdexcept>
#include <iostream>

class ThreadPool
{
public:
 /**
  * @brief Thread pool constructor
  *
  * @param[in] threads Number of threads, defaults to hardware concurrency
  */
 explicit ThreadPool(size_t threads = std::thread::hardware_concurrency())
  : stop(false)
 {
  if(threads == 0)
  {
   threads = 1;
  }

  for (size_t i = 0; i < threads; ++i)
  {
   workers.emplace_back([this]
   {
    while (true)
    {
     std::function<void()> task;
     {
      std::unique_lock<std::mutex> lock(this->queue_mutex);
      this->condition.wait(lock, [this]
      {
       return this->stop || !this->tasks.empty();
      });

      if (this->stop && this->tasks.empty())
      {
       return;
      }

      task = std::move(this->tasks.front());
      this->tasks.pop();
     }
     try
     {
      task();
     }
     catch (...)
     {
      std::cerr << "Thread Pool: something wrong.\n";
     }
    }
   });
  }
 }

 /**
  * @brief Add a task to the thread pool queue for execution
  *
  * This function takes a callable object and its parameters, wraps it into a packaged_task and adds it to the task queue,
  * then notifies worker threads that a new task has arrived. Returns a future object to get the task execution result.
  *
  * @tparam[in] F Type of the callable object
  * @tparam[in] Args Types of the callable object's parameters
  * @param[in] f Callable object
  * @param[in] args Parameters for the callable object
  *
  * @return std::future<typename std::result_of<F(Args...)>::type>
  *         Returns a future object that can be used to get the task execution result or wait for the task to complete
  *
  * @throws std::runtime_error Throws an exception when the thread pool is stopped
  */
 template<class F, class... Args>
 auto enqueue(F&& f, Args&&... args) -> std::future<typename std::invoke_result<F(Args...)>::type>
 {
  // Get the return type of the callable object
  using return_type = typename std::invoke_result<F(Args...)>::type;

  // Bind the callable object and parameters, create a packaged_task object
  auto task = std::make_shared< std::packaged_task<return_type()> >
                  (std::bind(std::forward<F>(f), std::forward<Args>(args)...)
              );

  std::future<return_type> res = task->get_future();
  {
   std::unique_lock<std::mutex> lock(queue_mutex);
   if(stop)
   {
    throw std::runtime_error("enqueue on stopped ThreadPool");
   }

   tasks.emplace([task]()
   {
    (*task)();
   });
  }
  condition.notify_one();
  return res;
 }

 template<class F>
 void push(F&& f)
 {
  {
   std::unique_lock<std::mutex> lock(queue_mutex);
   if (stop)
   {
    throw std::runtime_error("enqueue on stopped ThreadPool");
   }
   tasks.emplace(std::forward<F>(f));
  }
  condition.notify_one();
 }

 /**
  * @brief Thread pool destructor
  *
  * Safely stops all worker threads and cleans up resources
  */
 ~ThreadPool()
 {
  {
   std::unique_lock<std::mutex> lock(queue_mutex);
   stop = true;
  }
  condition.notify_all();
  for(std::thread &worker : workers)
   if(worker.joinable())
   {
    worker.join();
   }
 }

 /**
  * @brief Get the number of pending tasks
  *
  * This function accesses the task queue with a lock and returns the number of tasks waiting to be executed.
  *
  * @return size_t Returns the number of tasks in the task queue
  */
 size_t pending_tasks()
 {
  std::unique_lock<std::mutex> lock(queue_mutex);
  return tasks.size();
 }

 /**
  * @brief Block until all tasks are completed
  *
  * This function waits for all tasks in the task queue to finish executing, but does not stop the thread pool from receiving new tasks.
  * It determines whether all tasks are complete by checking if the task queue is empty.
  *
  * @note This function does not block other threads from adding new tasks to the queue
  */
 void wait_until_empty()
 {
  std::unique_lock<std::mutex> lock(queue_mutex);
  condition.wait(lock, [this] { return tasks.empty(); });
 }

private:
 std::vector<std::thread> workers; // Worker threads
 std::queue<std::function<void()>> tasks; // Task queue

 std::mutex queue_mutex; // Mutex, stop, tasks
 std::condition_variable condition; // Condition variable, queue_mutex
 bool stop; // Whether the thread pool is stopped
};

Usage Example

// --- Example Usage ---
class Sender : public Object
{
public:
    Signal<int, std::string> dataReady;
    Signal<> finished;

    void doWork()
    {
        for (int i = 0; i < 3; ++i)
        {
            std::cout << "[Sender] Emitting dataReady(" << i << ", 'Item " << i << "')\n";
            dataReady.emit(i, "Item " + std::to_string(i));
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        std::cout << "[Sender] Emitting finished()\n";
        finished.emit();
    }

    void doWorkAsync()
    {
        for (int i = 10; i < 13; ++i)
        {
            std::cout << "[Sender] Async Emitting dataReady(" << i << ", 'AsyncItem " << i << "')\n";
            dataReady.emitAsync(i, "AsyncItem " + std::to_string(i));
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }
        std::cout << "[Sender] Async Emitting finished()\n";
        finished.emitAsync();
    }
};

class Receiver : public Object
{
public:
    Receiver(const std::string&& name) : name(name)
    { }

    void onData(int value, const std::string&& msg)
    {
        std::ostringstream oss;
        oss << "[Receiver: " << name << "] Received data: " << value << ", " << msg
            << " (Thread ID: " << std::this_thread::get_id() << ")\n";
        std::cout << oss.str();
        // Simulate work
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }

    void onFinish()
    {
        std::ostringstream oss;
        oss << "[Receiver: " << name << "] Received finish signal."
            << " (Thread ID: " << std::this_thread::get_id() << ")\n";
        std::cout << oss.str();
    }

    const std::string&& getName() const
    {
        return name;
    }

private:
    std::string name;
};

int main()
{
    std::cout << "=== Qt-Style Signal/Slot with Inheritance (Fixed) ===" << std::endl;

    Sender sender;
    Receiver receiver1("R1");
    Receiver receiver2("R2");

    // Create and associate a dedicated thread pool for receiver1
    ThreadPool receiver1_pool(2);
    receiver1.setThreadPool(&receiver1_pool);

    // Connect signals and slots
    auto conn1 = connect(sender.dataReady, &receiver1, &Receiver::onData);
    auto conn2 = connect(sender.dataReady, &receiver2, &Receiver::onData);
    auto conn3 = connect(sender.finished, &receiver1, &Receiver::onFinish);
    auto conn4 = connect(sender.finished, []()
    {
        std::cout << "[Lambda Slot] Work is done!\n";
    });

    std::cout << "Connections established.\n\n";

    std::cout << "--- Synchronous Execution ---\n";
    sender.doWork();

    std::this_thread::sleep_for(std::chrono::milliseconds(200)); // Wait for synchronous slots to finish

    conn2.disconnect();
    std::cout << "\nDisconnected receiver2. Sending data again...\n";
    sender.dataReady.emit(-1, "After disconnection");

    std::cout << "\n--- Asynchronous Execution ---\n";
    sender.doWorkAsync();

    // Wait for asynchronous tasks to complete
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    std::cout << "\nMain function finished.\n";

    return 0; // ThreadPool destructor will wait for threads to finish
}

Running Result: