多线程编程的有力武器——ThreadQueue


#1

开发专业级音频软件,最核心与困难的部分莫过于多线程并发编程,即使一个超过20年编程经验的职业C++程序员,在面对复杂的音频系统多线程问题时,也会感到头大如斗。近期,我在编写和重构一个简单的音频项目时,就遇到了这个问题。我的指导老师Vinnie先生给出了很多具体的意见和建议,非常宝贵。我将多线程方面的部分内容翻译并整理了一下,特发表于此,与来自华语地区的JUCE People分享。 :slight_smile:

SwingCoder (中文网名:小T)

以下大部分内容摘自Vinnie和我的私人通信记录。未经许可,谢绝转载。

//=====================================================================================

简介
ThreadQueue是Vinnie写的一个类,一种典型的FIFO队列数据结构,效率极高。该类的代码虽然极其简练,但结构有点复杂:队列类中嵌套了一个抽象基类,该抽象基类派生一个子类,子类是类模板,这两个类均是嵌套类。队列类有几个核心成员函数:process(), call(), put()等等。ThreadQueue用于回调函数的异步执行,任意返回类型、任意多个参数的函数均可拿下。特别适合于多线程编程中的数据共享与传递,是修改多线程共享数据的另一种方案,用来替代JUCE::CriticalSection类的传统式锁定技术,它可以直接修改数据(采用函数异步回调的方式),并且在发出回调指令的线程上运行。队列中的线程决定何时调用ThreadQueue类的process()函数。当调用process()时,线程队列中所有的functor(仿函数)开始执行。这将使程序员写出更加强壮的语句,确保并发系统中数据的正确性。

正文
考虑一个线程T,该线程拥有一块共享数据X。传统方式下,其它线程要改变X,必须这么做:

修改数据:

  1. 获得互斥(更准确的翻译应该是“争取到互斥”,Vinnie的原信中使用的是“Acquire mutex”)
  2. 分配X的值
  3. 释放互斥

这种方式下,其它线程每次读取数据时,均要使线程T首先获得互斥:

读取数据:

  1. 获得互斥
  2. 读取X的值
  3. 释放互斥

这种方式仅仅适用于练习或小型程序,如果使用线程队列,情况将迥异。线程T能够在不获得互斥的前提下读取X,当其它线程要改变X的值时,它执行以下操作:

修改数据 (线程队列):

  1. 队列中的函数在线程队列上通过call()函数直接改变X的值。

这种设计使每个线程均持有共享数据的副本。比如:音频回调线程T1,消息线程T2,这两个线程的共享数据为tempo变量(double型,代表音频实时变速的比率),此时,这两个线程均持有tempo的副本。当T2要改变tempo的值时,实际上改变的是它所持有的副本,此时,该线程的某个函数(该函数已经置入队列中)在T1的线程队列上被调用,通知T1该值已经改变。

线程从不关注不一致的数据,尽管它所关注的数据也许有点“旧”(如果未调用process())。对音频编程来说,调用process()的理想时间是audioDeviceIOCallback的最开始。

上述tempo仅仅是一个简单的double值,也许不足以说明问题。但ThreadQueue在不同的线程之间传递更复杂的对象时,将发挥出极大的价值。比如:操作主界面中新增一个包含音频数据的音轨(添加音轨或效果器)。如果从音频设备进出回调函数中加载音频数据,那么一定会导致当前的播放出现停顿和断续。正确的做法应该是:GUI将加载音频数据的接口方法进行入队处理,使用辅助线程调用该方法。辅助线程收到回调后,开始执行加载,构造并初始化一个完整的对象,而后再次通过线程队列,将该对象传递给音频设备进出回调。process()函数此时将发挥极大的作用(因为该函数始终在音频进出回调的最开始处被调用)。

通常,在线程队列中开始传递对象时,往往将它们作为引用计数(reference counted)来使用,这将有助于控制对象的生命期,并可以解决“谁来销毁”这种常见问题。事实上,职业C++程序员通常采用的对象管理技术并非直接delete,而是将它们置入一个线程队列中,让辅助线程来销毁,释放堆中内存。这种方式下,回调线程和消息线程将不会互相阻塞。

关于JUCE::TimeSliceThread(时间片线程)和ThreadQueue(线程队列)

JUCE类库的TimeSliceThread类和Vinnie的ThreadQueue类不是一个概念, 虽然它们可以互相配合,共同实现一个目标。JUCE时间片线程和时间片线程客户端这两个类的功能与用法可参阅其API文档或JUCE Demo,而ThreadQueue则用于异步回调函数,比如:

void Object::doSomething ()
{
//...
   this->performActivity();
//...
}

performActivity()运行于调用者所在的线程。现在考虑一下这个:

void Object::doSomething ()
{
  //...
  threadQueue.call (std::bind (&Object::performActivity, this));
  //...
}

performActivity()运行于调用threadQueue.process()函数的线程——假设这是不同的线程。

注:std::bind (&Object::performActivity, this)的返回值是一个functor,即汉语程序员常说的“仿函数”,也就是重载了“()”运算符的类所实例化的对象,该对象可像函数一样带参使用。关于std::bind()与functor,参见Vinnie先生的另一封来信。C++标准库和boost准标准库中大量使用了仿函数技术,可参阅其源代码。Vinnie曾自己写了一个bond()函数,用来替代boost库的同类方法,有时间另外撰文介绍。

ThreadQueue的执行效率更高,因为它们实际上没有执行任何实际工作,也不应该由它来执行实际的工作。这也是仿函数和线程队列类的最大特色。实际工作应该由拥有队列的线程来执行,如果某个线程需要执行由另一个线程所启动的某些不间断的任务,那么该线程可以将具体的执行函数put到ThreadQueue中,创建一个TimeSliceClient,将该对象添加到TimeSliceThread对象中并立即返回。这种情况下,TimeSliceClient稍后才会执行。

更简洁的说,ThreadQueue类的设计初衷不是用来执行实际工作的,它仅仅用来在并发编程中同步多个线程,当线程需要读写数据时,它可以用来替代线程类的传统式lock方式。或者说:有一个线程T关联了线程队列,该线程有若干需要执行的工作项目(比如功能性函数),这些项目位于一个list中,如果打算添加新的工作项目,则:

void T::addWork (Work* item) 
{
  ScopedLock lock (criticalSection);  // 先锁定,再添加
  list->add (item);
}

不幸的是,每次我们要看看list中下一个要执行什么,均需要再次lock。这意味着:当添加新的工作项目时,多个线程之间将发生阻塞的问题,或者需要遍历整个list。现在考虑ThreadQueue的执行:

void T::addWork (Work* item) 
{
   threadQueue.call (std::bind (&T::appendWorkListItem, this, item));
}

void T::appendWorkListItem (Work* item) 
{
  list->add (item);
}

可看出,使用ThreadQueue之后,无需再用CriticalSection来保护list,线程T可以不经lock直接访问list,任何线程都可以毫无顾忌的调用addWork(),无需考虑list是否已被锁定。这一点是非常强悍的。

总结
ThreadQueue用来替代传统的lock锁定方式。进行并发编程时,使用CriticalSection管理线程的共享数据并使原语保持同步是最困难的步骤,很难确保最终的行为正确。而使用ThreadQueue则使之容易许多,而且便于理解、诊断、调试和维护——尽管此时的多线程与并发编程依然是一个较复杂的系统。


#2

This should probably be moved to “Useful Tools and Components” since it’s not strictly Juce related.


#3

Oh, yes.

I’ll be careful next time.

Sorry.