Qt 信号槽的工作原理（三）——队列连接与跨线程通信

原文链接 https://woboq.com/blog/how-qt-signals-slots-work-part3-queuedconnection.html Olivier Goffart 于 2016年02月04日

这篇博客是系列文章的一部分，用来解释 Qt 信号槽的内部实现机制。

在本文中，我们将深入探讨 Qt 队列连接（Queued Connection）背后的工作机制。

第一部分回顾

在第一部分中，我们看到：信号本质上只是普通的函数，其函数体由 moc 生成。信号函数所做的事情，就是调用 QMetaObject::activate，并传入一个位于栈上的参数指针数组。

下面是 moc 生成的一个信号函数示例（摘自第一部分）：

1
// SIGNAL 0
2
void Counter::valueChanged(int _t1)
3
{
4
    void *_a[] = { Q_NULLPTR, const_cast<void*>(reinterpret_cast<const void*>(&_t1)) };
5
    QMetaObject::activate(this, &staticMetaObject, 0, _a);
6
}

随后，QMetaObject::activate 会在内部数据结构中查找：哪些槽函数连接到了这个信号。

正如第一部分所展示的，对每一个槽，都会执行类似下面的代码：

xxxxxxxxxx
10
1
// 判断该连接是立即调用，还是放入事件队列
2
if ((c->connectionType == Qt::AutoConnection && !receiverInSameThread)
3
        || (c->connectionType == Qt::QueuedConnection)) {
4
    queued_activate(sender, signal_index, c, argv, locker);
5
    continue;
6
} else if (c->connectionType == Qt::BlockingQueuedConnection) {
7
    /* ... 省略 ... */
8
    continue;
9
}
10
/* ... DirectConnection：像第一部分那样直接调用槽函数 */

也就是说：

• 如果是 自动连接 且发送者和接收者不在同一线程

• 或者显式指定了 Qt::QueuedConnection

那么就会调用 queued_activate，把调用放入事件队列中。

如果是 Qt::BlockingQueuedConnection，则会走另一条分支（本文稍后会讲到，之前被跳过了）。

否则，就是 直接连接（DirectConnection），像第一部分中那样直接调用槽函数。

因此，在这篇博客中，我们将重点看看：

• queued_activate 里到底发生了什么

• 以及之前为 Qt::BlockingQueuedConnection 跳过的那些代码究竟是如何工作的

Qt 事件循环（Qt Event Loop）

一次 QueuedConnection（队列连接） 会向事件循环投递（post）一个事件，等待稍后被处理。

在投递事件时（QCoreApplication::postEvent），

事件会被压入一个按线程划分的队列中（QThreadData::postEventList）。

这个事件队列受到互斥锁的保护，因此当多个线程同时向另一个线程的事件队列投递事件时，不会产生竞态条件。

一旦事件被加入队列，如果接收者对象位于另一个线程中，

就会通过调用 QAbstractEventDispatcher::wakeUp 来通知该线程的事件分发器。

这会在事件分发器因等待新事件而休眠时将其唤醒。

如果接收者位于同一线程，事件则会在之后事件循环迭代时被正常处理。

事件会在处理它的线程中，在处理完成后立即被删除。

通过 QueuedConnection 投递的事件类型是 QMetaCallEvent。

当该事件被处理时，它会以与直接连接（DirectConnection）相同的方式调用槽函数。

所有必要的信息（要调用的槽、参数值等）都存储在这个事件对象内部。

参数拷贝（Copying the parameters）

从信号传过来的 argv 是一个指向参数的指针数组。

问题在于：这些指针指向的是信号函数栈上的参数，一旦信号函数返回，这些指针就会变得无效。

因此，我们必须把这些参数值拷贝到堆上。

为此，Qt 会直接向 QMetaType 求助。

正如 QMetaType 文章中所说，

QMetaType::create 能够在已知 QMetaType ID 和一个指向原始对象的指针的情况下，拷贝任意类型的对象。

那么，如何知道某个参数对应的 QMetaType ID 呢？

Qt 会从 QMetaObject 中获取参数类型的名称（moc 已经把所有类型名存进去了），

然后再用这个名字到 QMetaType 数据库中查找对应的类型 ID。

队列激活（queued_activate）

现在我们可以把前面的内容串起来，阅读 queued_activate 的代码了。

它由 QMetaObject::activate 调用，用于为 Qt::QueuedConnection 的槽调用做准备。

下面展示的代码做了少量简化并加了注释：

​x
1
static void queued_activate(QObject *sender, int signal,
2
                            QObjectPrivate::Connection *c, void **argv,
3
                            QMutexLocker &locker)
4
{
5
  const int *argumentTypes = c->argumentTypes;
6
  // c->argumentTypes 是一个 int 数组，保存各个参数的类型 id。
7
  // 在使用新语法连接时，可能在 connect 语句里就初始化了；
8
  // 但通常情况下它在第一次进行 queued 激活之前都是 nullptr。
9

10
  // DIRECT_CONNECTION_ONLY 是一个哑元 int，表示获取参数类型 id 时发生了错误。
11

12
  if (!argumentTypes) {
13
    // 向 QMetaObject 查询参数类型名，然后通过 QMetaType 系统查到类型 id
14
    QMetaMethod m = QMetaObjectPrivate::signal(sender->metaObject(), signal);
15
    argumentTypes = queuedConnectionTypes(m.parameterTypes());
16
    if (!argumentTypes) // 参数无法用于队列连接
17
      argumentTypes = &DIRECT_CONNECTION_ONLY;
18
    c->argumentTypes = argumentTypes; /* ... 此处省略：原子更新 ... */
19
  }
20
  if (argumentTypes == &DIRECT_CONNECTION_ONLY) // 无法激活
21
      return;
22

23
  int nargs = 1; // 把返回值类型也算进去
24
  while (argumentTypes[nargs-1])
25
      ++nargs;
26

27
  // 复制 argumentTypes 数组，因为 event 将会接管它的所有权
28
  int *types = (int *) malloc(nargs*sizeof(int));
29
  void **args = (void **) malloc(nargs*sizeof(void *));
30

31
  // 忽略返回值：队列连接里返回值没有意义
32
  types[0] = 0; // 返回值类型
33
  args[0] = 0;  // 返回值存放地址
34

35
  if (nargs > 1) {
36
    for (int n = 1; n < nargs; ++n)
37
      types[n] = argumentTypes[n-1];
38

39
    // 在调用参数的拷贝构造函数时，必须先解锁信号互斥锁，
40
    // 因为拷贝构造过程中可能会发生重入，导致死锁
41
    locker.unlock();
42
    for (int n = 1; n < nargs; ++n)
43
      args[n] = QMetaType::create(types[n], argv[n]);
44
    locker.relock();
45

46
    if (!c->receiver) {
47
      // 我们在解锁期间可能已经被断开连接了
48
      /* ... 省略清理代码 ... */
49
      return;
50
    }
51
  }
52

53
  // 投递事件
54
  QMetaCallEvent *ev = c->isSlotObject ?
55
    new QMetaCallEvent(c->slotObj, sender, signal, nargs, types, args) :
56
    new QMetaCallEvent(c->method_offset, c->method_relative, c->callFunction,
57
                       sender, signal, nargs, types, args);
58
  QCoreApplication::postEvent(c->receiver, ev);
59
}
60

当这个事件到达接收方后，QObject::event 会设置 sender，并调用 QMetaCallEvent::placeMetaCall。

后者会像第一部分所示的直接连接那样分发调用——也就是以 QMetaObject::activate 类似的方式来调用槽函数。

xxxxxxxxxx
9
1
case QEvent::MetaCall:
2
{
3
  QMetaCallEvent *mce = static_cast<QMetaCallEvent*>(e);
4

5
  QConnectionSenderSwitcher sw(this, const_cast<QObject*>(mce->sender()), mce->signalId());
6

7
  mce->placeMetaCall(this);
8
  break;
9
}

阻塞队列连接（BlockingQueuedConnection）

BlockingQueuedConnection 是 DirectConnection 和 QueuedConnection 的混合体：

• 像 DirectConnection 一样 → 参数可以留在栈上（因为调用线程会被阻塞）

• 像 QueuedConnection 一样 → 事件会被投递到另一个线程的事件循环中

该事件中还包含一个 QSemaphore（信号量）：

• 发送信号的线程： → 在投递事件后调用 semaphore.acquire() 阻塞

• 接收线程： → 在槽函数执行完毕后释放信号量

示例代码：

xxxxxxxxxx
18
1
} else if (c->connectionType == Qt::BlockingQueuedConnection) {
2
  locker.unlock(); // 解锁 QObject 的信号互斥锁。
3
  if (receiverInSameThread) {
4
    qWarning("Qt: Dead lock detected while activating a BlockingQueuedConnection: "
5
             "Sender is %s(%p), receiver is %s(%p)",
6
             sender->metaObject()->className(), sender,
7
             receiver->metaObject()->className(), receiver);
8
  }
9
  QSemaphore semaphore;
10
  QMetaCallEvent *ev = c->isSlotObject ?
11
    new QMetaCallEvent(c->slotObj, sender, signal_index, 0, 0, argv, &semaphore) :
12
    new QMetaCallEvent(c->method_offset, c->method_relative, c->callFunction,
13
                       sender, signal_index, 0, 0, argv , &semaphore);
14
  QCoreApplication::postEvent(receiver, ev);
15
  semaphore.acquire();
16
  locker.relock();
17
  continue;
18
}

这里是由 QMetaCallEvent 的析构函数 来释放（release）信号量。这样设计很合理，因为事件会在以下两种情况下被删除：

1. 事件成功投递并处理后（也就是槽函数已经被调用后）

2. 事件没能被投递/处理（例如接收对象被删除了）

因此无论哪种情况，等待中的线程都不会永远卡住。

BlockingQueuedConnection 在做线程间通信时很有用：

当你需要在另一个线程中调用一个函数，并且希望 等它执行完再继续 时，可以用它。

不过，这种方式使用起来必须非常小心。

BlockingQueuedConnection 的风险

使用 BlockingQueuedConnection 时必须非常小心，否则很容易造成死锁。

显而易见，如果你在 同一个线程 中的两个对象之间使用 BlockingQueuedConnection 进行连接，会立刻发生死锁。

原因是：你向当前线程自己的事件循环发送了一个事件，然后又阻塞这个线程，等待该事件被处理。

但由于线程已经被阻塞，事件永远不会被处理，线程也就永远卡住了。

Qt 会在运行时检测到这种情况并打印一条警告，但 不会尝试帮你修复这个问题。

有人曾建议：如果发现发送者和接收者在同一个线程里，Qt 是否可以自动退化为普通的 DirectConnection。

但 Qt 并没有这么做，因为 BlockingQueuedConnection 本身就是一种 只有在你非常清楚自己在做什么时才应该使用的机制：

你必须明确知道事件是从哪个线程发送到哪个线程的。

真正的危险在于：一旦你的应用中存在从线程 A 到线程 B 的 BlockingQueuedConnection，

那么线程 B 绝对不能再等待线程 A，否则就会再次产生死锁。

此外，在发射信号或调用 QMetaObject::invokeMethod() 时，

你不能持有任何线程 B 也可能尝试获取的互斥锁，否则同样可能导致死锁。

一个典型问题出现在使用 BlockingQueuedConnection 来终止线程时，例如下面这段伪代码：

xxxxxxxxxx
12
1
void MyOperation::stop()
2
{
3
    m_thread->quit();
4
    m_thread->wait(); // 等待被调用线程，可能发生死锁
5
    cleanup();
6
}
7

8
// 通过 BlockingQueuedConnection 连接
9
Stuff MyOperation::slotGetSomeData(const Key &k)
10
{
11
    return m_data->get(k);
12
}

你不能在这里直接调用 wait()，因为子线程可能已经发射了，或者正准备发射一个信号，而这个信号会等待父线程；

但父线程此时不会再回到事件循环中，于是就产生了死锁。

所有线程清理和信息传递都必须通过在线程之间投递事件来完成，而不能使用 wait()。

一种更好的写法是：

xxxxxxxxxx
6
1
void MyOperation::stop()
2
{
3
    connect(m_thread, &QThread::finished, this, &MyOperation::cleanup);
4
    m_thread->quit();
5
    /*（注意：在调用 quit 之前先建立连接，以避免竞态条件）*/
6
}

这样做的缺点是：MyOperation::cleanup() 现在是 异步调用 的，这可能会让整体设计变得更复杂。

总结（Conclusion）

这篇文章也为整个系列画上了句号。希望这些文章能够揭开 Qt 信号与槽的“神秘面纱”，

并且对其底层工作原理有一定了解之后，能帮助你在实际应用中更好地使用它们。