17.3 用于线程同步的对象

在几乎所有的线程使用中,数据都是几个线程共享的.当有两个或以上的线程试图访问同一个数据的时候,无论这个数据是一个对象还是一个资源,这种访问都应该被同步,以避免数据在同一时刻被超过1个线程访问或者修改.因为应用程序中充满了所谓的不变量,比如,对于一个链表来说,我们总认为它的第一个元素是有效的,每个元素都指向它的下一个元素,最后的一个元素是空指针.但是在对链表进行插入新元素操作的时候,有一小段时间间隔,这个所谓不变量是被打破的.这时候假如有两个线程同时在使用这个链表,如果没有进行数据同步的动作,就会出现不可知的问题.因此你必须保证,在你插入元素这一小段时间间隔内,没有别的线程在访问同样的数据.

保证所有的共享数据被各个访问它的线程快速的并且是以一个合理的顺序访问是程序员自己的责任.因此这一小节我们来介绍一下wxWidgets提供了哪些类来帮助程序员达到这个目的.

wxMutex

这个名字来源于mutual exclusion(共有的互斥量),它是最简单的一种数据同步手段.它可以保证同一个时刻只有一个线程在访问某一部分数据.要获取数据的访问权,线程必须调用wxMutex::Lock函数,这将阻塞当前线程的执行直到它所请求的数据已经不再有任何别的线程使用.而在它开始使用这个数据以后,别的线程对 wxMutex::Lock的调用同样将被阻塞,直至当前使用的线程调用wxMutex::Unlock函数释放它所使用的资源.尽管你可以直接使用 wxMutex的Lock和Unlock函数,我们还是推荐你使用wxMutexLocker类来使用wxMutex,这将确保你不会忘记在Lock以后调用Unlock函数(译者注:你可以想像假如你忘了Unlock的后果,呵呵),因为这两个函数被隐藏在wxMutexLocker类的构造函数和析构函数中,因此,即使发生了异常,wxMutexLocker类仍然会在自己被释放的时候进行Unlock.

下面的代码中,我们确信MyApp有一个wxMutex类型的变量m_mutex:

void MyApp::DoSomething()
{
    wxMutexLocker lock(m_mutex);
    if (lock.IsOk())
    {
        ... do something
    }
    else
    {
        ... we have not been able to
        ... acquire the mutex, fatal error
    }
}

使用互斥量有三个重要的规则:

  1. 线程不可以锁定已经被锁定的互斥量(不允许互斥量递归). 尽管有些系统允许你这样做,但这是不可移植的.
  2. 线程不允许解锁别的线程锁定的互斥量. 如果你需要这个功能,参考我们马上会讲到的信号量机制.
  3. 如果你的线程即使无法锁定互斥量也还有别的事情可以做,你应该先使用wxMutex::TryLock函数判断是否可以锁定.这个函数是立即返回的,返回值为可以锁定(wxMUTEX_NO_ERROR)或者不可锁定(wxMUTEX_DEAD_LOCK或 wxMUTEX_BUSY). 这在主线程中尤其有用,因为主线程(GUI线程)是不可以被阻塞的,否则它将不能响应任何用户的输入.

死锁

如果两个线程在互相等待对方已经锁定的互斥量,我们称之为发生了死锁.举例来说,假如线程A已经锁定了互斥量1,线程B已经锁定了互斥量 2,线程A正等待锁定互斥量2,而线程B正等待锁定互斥量1,那么,他们两个人将无限期的等待下去,在某些系统上,如果出现这种情况,Lock或者 Unlock或者TryLock函数将返回错误码wxMUTEX_DEAD_LOCK,但是在另外一些系统上,除非你把整个程序杀死,否则他们将一直等下去.

解决死锁的方法有一下两种:

  • 修改顺序.一个一致的互斥量锁定顺序将减少死锁发生的概率.在前面的例子中,如果线程A和线程B都要求先锁定互斥量1再锁定互斥量2,则死锁将不会发生.
  • 使用TryLock. 在成功锁定第一个互斥量以后,在后续的互斥量锁定之前都使用TryLock函数判断,如果TryLock返回失败,解锁第一个然后重新开始锁定第一个. 这种方法系统开销较大,但是如果修改顺序的方法有明显的缺陷或者导致你的代码乱七八糟,你可以考虑使用这种方法.

wxCriticalSection

关键区域用来保证某一段代码在某一个时刻只被一个线程执行,而前面介绍的互斥量则用来保证互斥量在某一个时刻只被一个线程锁定.他们之间是非常相似的,除了在某些系统上,互斥量是系统范围内的变量而关键区域只在本应用程序范围内有效.在这样的系统上,使用关键区域的效率会比使用互斥量高一点点.也因为这些细微的差别,他们的一些术语也略有不同,互斥量称为锁定(或者装载)和解锁(或者卸载),而关键区域称为进入或者离开.

关键区域也有对应的wxCriticalSectionLocker对象,出于和wxMutexLocker同样的原因,你应该尽量使用它而不要直接使用wxCriticalSection的函数.

wxCondition

所谓条件变量wxCondition,是用来指示共享数据的某些条件已经满足.比如,你可以使用它来指示一个消息队列已经有数据到来.而共享数据本身(在这里指的这个消息队列)通常还需要另外使用一个互斥量来保护.

你可以通过锁定互斥量,检测队列有无数据,然后释放信号量这样的循环来进行消息队列数据的处理,不过如果队列里一直没有数据,这样的作法也太浪费了,时间全部浪费在锁定和解锁互斥量上面了.象这种情况,最好是使用条件变量,这样消息处理线程就可以被阻塞直到等到别的线程把事件放入事件队列以后发出通知事件.

多个线程可能都在等待同一个条件,这时你可以选择唤醒一个线程还是唤醒多个线程,唤醒一个线程的函数是Signal,唤醒所有正在等待的线程的函数是Broadcast.如果有多个条件都是由同一个wxCondition通知的,你必须使用Broadcast函数,否则可能某个线程被唤醒了但是却什么也做不了,因为它的条件还没有满足,而另外的可以满足条件的那个线程却无法唤醒了.

wxCondition使用举例

我们来假设一下我们有两个线程:

一个是生产线程,它负责产生10个元素并且将其放入队列,然后发送队列满信号并且在继续填充元素之前等待队列空信号.

一个是消费线程,它在收到队列满信号的时候移除队列中所有的元素.

我们需要一个互斥量m_mutex,用来保护整个队列和两个条件变量:m_isFull和m_isEmpty.这个互斥量被传递给两个条件变量的构造函数作为参数.另外你需要总是显示判断条件是否满足,然后再开始等待通知,因为可能在你还没有开始等待之前,已经有一个信号通知了,由于你还没有等待,那个信号就丢失了.

我们来看看生产线程的Entry函数的伪代码:

while ( notDone )
{
   wxMutexLocker lock(m_mutex) ;
   while( m_queue.GetCount() > 0 )
   {
      m_isEmpty.Wait() ;
   }
   for ( int i = 0 ; i < 10 ; ++i )
   {
      m_queue.Append( wxString::Format(wxT("Element %d"),i) ) ;
   }
   m_isFull.Signal();
}

消费线程:

while ( notDone )
{
   wxMutexLocker lock(m_mutex) ;
   while( m_queue.GetCount() == 0 )
   {
      m_isFull.Wait() ;
   }
   for ( int i = queue.GetCount() ; i > 0 ; i )
   {
      m_queue.RemoveAt( i ) ;
   }
   m_isEmpty.Signal();
}

Wait函数首先Unlock其内部的互斥量 ,然后等待条件被通知.当它被通知唤醒时,会首先再次锁定内部的信号量,因此数据同步时非常严格满足的.

另外,在Wait函数被唤醒之后再次检测条件是否满足也是必要的,因为在信号被发送和线程被唤醒之间可能发生某些事情,导致条件又一次不满足了;另外,系统有时候也会产生一些假的信号导致Wait函数返回.

Signal可能在Wait之前发生,正象pthread中的那样,这时这个信号会丢失.因此如果你想要确定你没有错过任何信号,你必须保证和条件变量绑定的互斥量在最开始就处于锁定状态,并且在你调用Signal函数之前再次尝试锁定它,这意味着对Signal的调用将被阻塞直到另外一个线程调用了Wait函数.

OK,上面的这段话读起来比较费劲,我们来看一个例子,在这个例子中,主线程创建了一个工作线程,工作线程的Signal函数直到主线程调用了Wait以后才能被调用:

class MySignallingThread : public wxThread
{
public:
    MySignallingThread(wxMutex *mutex, wxCondition *condition)
    {
        m_mutex = mutex;
        m_condition = condition;
        Create();
    }
    virtual ExitCode Entry()
    {
        ... do our job ...
        // 告诉其它线程我们马上就要退出了.
        // 我们必须先锁定信号量,这个动作会阻塞自己
        // 直到主线程调用了Wait
        wxMutexLocker lock(m_mutex);
        m_condition.Broadcast(); // 我们只有一个线程在等待,所以等同于Signal()
        return 0;
    }
private:
    wxCondition *m_condition;
    wxMutex *m_mutex;
};
void TestThread()
{
    wxMutex mutex;
    wxCondition condition(mutex);
    // 互斥量应该先出于锁定状态
    mutex.Lock();
    // 先创建和运行工作线程,注意这个线程不能退出
    // 除非我们解锁了互斥量
    MySignallingThread *thread =
        new MySignallingThread(&mutex, &condition);
    thread->Run();
    // Wait工作线程退出,Wait函数将自动解锁和它绑定的互斥量
    // 因此工作线程可以继续直至发出Signal并且终至自己.
    condition.Wait();
    // 我们收到了Signal就可以退出了.
}

当然上面的这个例子指示出于演示如何实现条件变量中第一个Singal在第一个Wait之后执行,如果单就代码例子实现的功能来说,我们应该直接使用一个联合线程,然后在主线程调用wxThread::Join函数就可以了.

wxSemaphore

信号量(wxSemaphore)可以通俗的看成一个互斥量和一个记数器的结合,它和记数器最大的不同在于信号量的值可以被任何线程更改,而不仅仅是拥有它的那个线程.所以你也可以把信号量看作是一个没有主人的记数器.

如果一个线程调用信号量的Wait函数,这个调用将阻塞,除非记数器当前为一个正数,然后Wait函数将记数器减一,然后返回.而对Post函数的调用则将增加记数器的值然后返回.

wxWidgets实现的信号量还有一个额外的特性,你可以在其构造函数中指定一个记数器的最大值,默认为0表明最大值没有限制,如果你给定了一个最大值,而Post函数的调用使得当前的记数器超过了这个最大值,你将会得到一个wxSEMA_OVERFLOW错误.让我们再回到前面说的用信号量实现特殊互斥的描述:

  • 一个可以被不同的线程锁定和解锁的互斥量可以通过一个记数器最大值为1的信号量实现,互斥量的Lock函数等同于信号量的Wait函数而互斥量的Unlock函数等同于信号量的Post函数.
  • 前一个线程调用Lock(Wait)发现是一个整数值,于是减一,然后立即继续.
  • 第二个线程调用Lock发现是零,将必须等待某个线程(不一定是前一个线程)调用Unlock(Post).

你可以在wxWidgets自带的samples/thread中找到一个用来演示多线程编程的例子.如下图所示.在这个例子中,你可以启动,停止,暂停,恢复线程的运行.它演示了一个工作线程周期性的通过wxPostEvent往主程序发送事件,一个进度条对话框用来指示当前进度并在进度到达最后的时候取消工作线程的运行.