线程池

摘抄自:https://chenxiaowei.gitbooks.io/cppconcurrencyin_action/content/content/chapter9/chapter9-chinese.html

第9章 高级线程管理

本章主要内容

• 线程池
  
• 处理线程池中任务的依赖关系
  
• 池中线程如何获取任务
  
• 中断线程
  

之前的章节中，我们通过创建std::thread对象来对线程进行管理。在一些情况下，这种方式不可行了，因为需要在线程的整个生命周期中对其进行管理，并根据硬件来确定线程数量，等等。理想情况是将代码划分为最小块，再并发执行，之后交给处理器和标准库，进行性能优化。

另一种情况是，当使用多线程来解决某个问题时，在某个条件达成的时候，可以提前结束。可能是因为结果已经确定，或者因为产生错误，亦或是用户执行终止操作。无论是哪种原因，线程都需要发送“请停止”请求，放弃任务，清理，然后尽快停止。

本章，我们将了解一下管理线程和任务的机制，从自动管理线程数量和自动管理任务划分开始。

9.1 线程池

很多公司里，雇员通常会在办公室度过他们的办公时光(偶尔也会外出访问客户或供应商)，或是参加贸易展会。虽然外出可能很有必要，并且可能需要很多人一起去，不过对于一些特别的雇员来说，一趟可能就是几个月，甚至是几年。公司要给每个雇员都配一辆车，这基本上是不可能的，不过公司可以提供一些共用车辆；这样就会有一定数量车，来让所有雇员使用。当一个员工要去异地旅游时，那么他就可以从共用车辆中预定一辆，并在返回公司的时候将车交还。如果某天没有闲置的共用车辆，雇员就得不延后其旅程了。

线程池就是类似的一种方式，在大多数系统中，将每个任务指定给某个线程是不切实际的，不过可以利用现有的并发性，进行并发执行。线程池就提供了这样的功能，提交到线程池中的任务将并发执行，提交的任务将会挂在任务队列上。队列中的每一个任务都会被池中的工作线程所获取，当任务执行完成后，再回到线程池中获取下一个任务。

创建一个线程池时，会遇到几个关键性的设计问题，比如：可使用的线程数量，高效的任务分配方式，以及是否需要等待一个任务完成。

在本节，我们将看到线程池是如何解决这些问题的，从最简单的线程池开始吧！

9.1.1 最简单的线程池

作为最简单的线程池，其拥有固定数量的工作线程(通常工作线程数量与std::thread::hardware_concurrency()相同)。当工作需要完成时，可以调用函数将任务挂在任务队列中。每个工作线程都会从任务队列上获取任务，然后执行这个任务，执行完成后再回来获取新的任务。在最简单的线程池中，线程就不需要等待其他线程完成对应任务了。如果需要等待，就需要对同步进行管理。

下面清单中的代码就展示了一个最简单的线程池实现。

清单9.1 简单的线程池

class threadpool
{
  std::atomicbool done;
  threadsafequeue<std::function<void()> > workqueue;  // 1
  std::vector<std::thread> threads;  // 2
  jointhreads joiner;  // 3


void workerthread()
  {
    while(!done)  // 4
    {
      std::function<void()> task;
      if(workqueue.trypop(task))  // 5
      {
        task();  // 6
      }
      else
      {
        std::thisthread::yield();  // 7
      }
    }
  }

public:
  threadpool():
    done(false),joiner(threads)
  {
    unsigned const threadcount=std::thread::hardware_concurrency();  // 8
try
{
  for(unsigned i=0;i&lt;thread_count;++i)
  {
    threads.push_back( 
      std::thread(&amp;thread_pool::worker_thread,this));  // 9
  }
}
catch(...)
{
  done=true;  // 10
  throw;
}

}

~thread_pool()
  {
    done=true;  // 11
  }

template<typename FunctionType>
  void submit(FunctionType f)
  {
    workqueue.push(std::function<void()>(f));  // 12
  }
};

实现中有一组工作线程②，并且使用了一个线程安全队列(见第6章)①来管理任务队列。这种情况下，用户不用等待任务，并且任务不需要返回任何值，所以可以使用std::function<void()>对任务进行封装。submit()函数会将函数或可调用对象包装成一个std::function<void()>实例，并将其推入队列中⑫。

线程始于构造函数：使用std::thread::hardwareconcurrency()来获取硬件支持多少个并发线程⑧，这些线程会在workerthread()成员函数中执行⑨。

当有异常抛出时，线程启动就会失败，所以需要保证任何已启动的线程都能停止，并且能在这种情况下清理干净。当有异常抛出时，通过使用try-catch来设置done标志⑩，还有jointhreads类的实例(来自于第8章)③用来汇聚所有线程。当然也需要析构函数：仅设置done标志⑪，并且jointhreads确保所有线程在线程池销毁前全部执行完成。注意成员声明的顺序很重要：done标志和workerqueue必须在threads数组之前声明，而数据必须在joiner前声明。这就能确保成员能以正确的顺序销毁；比如，所有线程都停止运行时，队列就可以安全的销毁了。

workerthread函数很简单：从任务队列上获取任务⑤，以及同时执行这些任务⑥，执行一个循环直到done标志被设置④。如果任务队列上没有任务，函数会调用std::thisthread::yield()让线程休息⑦，并且给予其他线程向任务队列上推送任务的机会。

一些简单的情况，这样线程池就足以满足要求，特别是任务没有返回值，或需要执行一些阻塞操作的时候。不过，在很多情况下，这样简单的线程池完全不够用，其他情况使用这样简单的线程池可能会出现问题，比如：死锁。同样，在简单例子中，使用std::async能提供更好的功能(如第8章中的例子)。

在本章中，我们将了解一下更加复杂的线程池实现，通过添加特性满足用户需求，或减少问题的发生几率。

首先，从已经提交的任务开始说起。

9.1.2 等待提交到线程池中的任务

第8章中的例子中，线程间的任务划分完成后，代码会显式生成新线程，主线程通常就是等待新线程在返回调用之前结束，确保所有任务都完成。使用线程池，就需要等待任务提交到线程池中，而非直接提交给单个线程。这与基于std::async的方法(第8章等待future的例子)类似，使用清单9.1中的简单线程池，使用第4章中提到的工具：条件变量和future。虽然，会增加代码的复杂度，不过，要比直接对任务进行等待的方式好很多。

通过增加线程池的复杂度，可以直接等待任务完成。使用submit()函数返回一个对任务描述的句柄，用来等待任务的完成。任务句柄会用条件变量或future进行包装，这样能使用线程池来简化代码。

一种特殊的情况是，执行任务的线程需要返回一个结果到主线程上进行处理。你已经在本书中看到多个这样的例子，比如：parallelaccumulate()(第2章)。这种情况下，需要用future对最终的结果进行转移。清单9.2展示了对简单线程池的修改，通过修改就能等待任务完成，以及在工作线程完成后，返回一个结果到等待线程中去，不过std::packagedtask<>实例是不可拷贝的，仅是可移动的，所以不能再使用std::function<>来实现任务队列，因为std::function<>需要存储可复制构造的函数对象。包装一个自定义函数，用来处理只可移动的类型。这就是一个带有函数操作符的类型擦除类。只需要处理那些没有函数和无返回的函数，所以这是一个简单的虚函数调用。

清单9.2 可等待任务的线程池

class functionwrapper
{
  struct implbase {
    virtual void call()=0;
    virtual ~impl_base() {}
  };


std::uniqueptr<implbase> impl;
  template<typename F>
  struct impltype: implbase
  {
    F f;
    impltype(F&& f): f(std::move(f)) {}
    void call() { f(); }
  };
public:
  template<typename F>
  functionwrapper(F&& f):
    impl(new impl_type<F>(std::move(f)))
  {}

void operator()() { impl->call(); }

function_wrapper() = default;

functionwrapper(functionwrapper&& other):
    impl(std::move(other.impl))
  {}

functionwrapper& operator=(functionwrapper&& other)
  {
    impl=std::move(other.impl);
    return *this;
  }

functionwrapper(const functionwrapper&)=delete;
  functionwrapper(functionwrapper&)=delete;
  functionwrapper& operator=(const functionwrapper&)=delete;
};

class threadpool
{
  threadsafequeue<functionwrapper> workqueue;  // 使用functionwrapper，而非使用std::function

void workerthread()
  {
    while(!done)
    {
      functionwrapper task;
      if(workqueue.trypop(task))
      {
        task();
      }
      else
      {
        std::thisthread::yield();
      }
    }
  }
public:
  template<typename FunctionType>
  std::future<typename std::resultof<FunctionType()>::type>  // 1
    submit(FunctionType f)
  {
    typedef typename std::resultof<FunctionType()>::type
      resulttype;  // 2
std::packaged_task&lt;result_type()&gt; task(std::move(f));  // 3
std::future&lt;result_type&gt; res(task.get_future());  // 4
work_queue.push(std::move(task));  // 5
return res;  // 6

}
  // 休息一下
};

首先，修改的是submit()函数①返回一个std::future<>保存任务的返回值，并且允许调用者等待任务完全结束。因为需要知道提供函数f的返回类型，所以使用std::resultof<>：std::resultof<FunctionType()>::type是FunctionType类型的引用实例(如，f)，并且没有参数。同样，函数中可以对resulttype typedef②使用std::resultof<>。

然后，将f包装入std::packagedtask<resulttype()>③，因为f是一个无参数的函数或是可调用对象，能够返回resulttype类型的实例。向任务队列推送任务⑤和返回future⑥前，就可以从std::packagedtask<>中获取future④。注意，要将任务推送到任务队列中时，只能使用std::move()，因为std::packagedtask<>是不可拷贝的。为了对任务进行处理，队列里面存的就是functionwrapper对象，而非std::function<void()>对象。

现在线程池允许等待任务，并且返回任务后的结果。下面的清单就展示了，如何让parallelaccumuate函数使用线程池。

清单9.3 parallelaccumulate使用一个可等待任务的线程池

template<typename Iterator,typename T>
T parallel_accumulate(Iterator first,Iterator last,T init)
{
  unsigned long const length=std::distance(first,last);


if(!length)
    return init;

unsigned long const blocksize=25;
  unsigned long const numblocks=(length+blocksize-1)/blocksize;  // 1

std::vector<std::future<T> > futures(numblocks-1);
  threadpool pool;

Iterator blockstart=first;
  for(unsigned long i=0;i<(numblocks-1);++i)
  {
    Iterator blockend=blockstart;
    std::advance(blockend,blocksize);
    futures[i]=pool.submit(accumulateblock<Iterator,T>());  // 2
    blockstart=blockend;
  }
  T lastresult=accumulateblock<Iterator,T>()(blockstart,last);
  T result=init;
  for(unsigned long i=0;i<(numblocks-1);++i)
  {
    result+=futures[i].get();
  }
  result += lastresult;
  return result;
}

与清单8.4相比，有几个点需要注意一下。首先，工作量是依据使用的块数(numblocks①)，而不是线程的数量。为了利用线程池的最大化可扩展性，需要将工作块划分为最小工作块。当线程池中线程不多时，每个线程将会处理多个工作块，不过随着硬件可用线程数量的增长，会有越来越多的工作块并发执行。

最简单的方法就是在threadpool中添加一个新函数，来执行任务队列上的任务，并对线程池进行管理。高级线程池的实现可能会在等待函数中添加逻辑，或等待其他函数来处理这个任务，优先的任务会让其他的任务进行等待。下面清单中的实现，就展示了一个新runpendingtask()函数，对于快速排序的修改将会在清单9.5中展示。

清单9.4 runpendingtask()函数实现

void threadpool::runpendingtask()
{
  functionwrapper task;
  if(workqueue.trypop(task))
  {
    task();
  }
  else
  {
    std::thisthread::yield();
  }
}

runpendingtask()的实现去掉了在workerthread()函数的主循环。函数任务队列中有任务的时候，执行任务；要是没有的话，就会让操作系统对线程进行重新分配。

下面快速排序算法的实现要比清单8.1中版本简单许多，因为所有线程管理逻辑都被移入到线程池。

清单9.5 基于线程池的快速排序实现

template<typename T>
struct sorter  // 1
{
  thread_pool pool;  // 2


std::list<T> dosort(std::list<T>& chunkdata)
  {
    if(chunkdata.empty())
    {
      return chunkdata;
    }
std::list&lt;T&gt; result;
result.splice(result.begin(),chunk_data,chunk_data.begin());
T const&amp; partition_val=*result.begin();

typename std::list&lt;T&gt;::iterator divide_point=
  std::partition(chunk_data.begin(),chunk_data.end(),
                 [&amp;](T const&amp; val){return val&lt;partition_val;});

std::list&lt;T&gt; new_lower_chunk;
new_lower_chunk.splice(new_lower_chunk.end(),
                       chunk_data,chunk_data.begin(),
                       divide_point);

std::future&lt;std::list&lt;T&gt; &gt; new_lower=  // 3
  pool.submit(std::bind(&amp;sorter::do_sort,this,
                        std::move(new_lower_chunk)));

std::list&lt;T&gt; new_higher(do_sort(chunk_data));

result.splice(result.end(),new_higher);
while(!new_lower.wait_for(std::chrono::seconds(0)) ==
  std::future_status::timeout)
{
  pool.run_pending_task();  // 4
}

result.splice(result.begin(),new_lower.get());
return result;

}
};

template<typename T>
std::list<T> parallelquicksort(std::list<T> input)
{
  if(input.empty())
  {
    return input;
  }
  sorter<T> s;

return s.dosort(input);
}

与清单8.1相比，这里将实际工作放在sorter类模板的dosort()成员函数中执行①，即使例子中仅对threadpool实例进行包装②。

线程和任务管理，在线程等待的时候，就会少向线程池中提交一个任务③，并且执行任务队列上未完成的任务④。需要显式的管理线程和栈上要排序的数据块。当有任务提交到线程池中，可以使用std::bind()绑定this指针到dosort()上，绑定是为了让数据块进行排序。这种情况下，需要对newlowerchunk使用std::move()将其传入函数，数据移动要比拷贝的方式开销少。

虽然，使用等待其他任务的方式，解决了死锁问题，这个线程池距离理想的线程池很远。

首先，每次对submit()的调用和对runpendingtask()的调用，访问的都是同一个队列。在第8章中，当多线程去修改一组数据，就会对性能有所影响，所以需要解决这个问题。

9.1.4 避免队列中的任务竞争

线程每次调用线程池的submit()函数，都会推送一个任务到工作队列中。就像工作线程为了执行任务，从任务队列中获取任务一样。这意味着随着处理器的增加，在任务队列上就会有很多的竞争，这会让性能下降。使用无锁队列会让任务没有明显的等待，但是乒乓缓存会消耗大量的时间。

为了避免乒乓缓存，每个线程建立独立的任务队列。这样，每个线程就会将新任务放在自己的任务队列上，并且当线程上的任务队列没有任务时，去全局的任务列表中取任务。下面列表中的实现，使用了一个threadlocal变量，来保证每个线程都拥有自己的任务列表(如全局列表那样)。

class threadpool
{
  threadsafequeue<functionwrapper> poolwork_queue;


typedef std::queue<functionwrapper> localqueuetype;  // 1
  static threadlocal std::uniqueptr<localqueuetype>
    localwork_queue;  // 2

void workerthread()
  {
    localworkqueue.reset(new localqueuetype);  // 3
    while(!done)
    {
      runpending_task();
    }
  }

public:
  template<typename FunctionType>
  std::future<typename std::resultof<FunctionType()>::type>
    submit(FunctionType f)
  {
    typedef typename std::resultof<FunctionType()>::type result_type;
std::packaged_task&lt;result_type()&gt; task(f);
std::future&lt;result_type&gt; res(task.get_future());
if(local_work_queue)  // 4
{
  local_work_queue-&gt;push(std::move(task));
}
else
{
  pool_work_queue.push(std::move(task));  // 5
}
return res;

}

void runpendingtask()
  {
    functionwrapper task;
    if(localworkqueue && !localworkqueue->empty())  // 6
    {
      task=std::move(localworkqueue->front());
      localworkqueue->pop();
      task();
    }
    else if(poolworkqueue.trypop(task))  // 7
    {
      task();
    }
    else
    {
      std::thisthread::yield();
    }
  }
// rest as before
};

因为不希望非线程池中的线程也拥有一个任务队列，使用std::uniqueptr<>指向线程本地的工作队列②；这个指针在workerthread()中进行初始化③。std:uniqueptr<>的析构函数会保证在线程退出的时候，工作队列被销毁。

submit()会检查当前线程是否具有一个工作队列④。如果有，就是线程池中的线程，可以将任务放入线程的本地队列中；否者，就像之前一样将这个任务放在线程池中的全局队列中⑤。

runpendingtask()⑥中的检查和之前类似，只是要对是否存在本地任务队列进行检查。如果存在，就会从队列中的第一个任务开始处理；注意本地任务队列可以是一个普通的std::queue<>①，因为这个队列只能被一个线程所访问，就不存在竞争。如果本地线程上没有任务，就会从全局工作列表上获取任务⑦。

这样就能有效避免竞争，不过当任务分配不均时，造成的结果就是：某个线程本地队列中有很多任务的同时，其他线程无所事事。例如：举一个快速排序的例子，只有一开始的数据块能在线程池上被处理，因为剩余部分会放在工作线程的本地队列上进行处理，这样的使用方式也违背使用线程池的初衷。

幸好，这个问题是有解：本地工作队列和全局工作队列上没有任务时，可从别的线程队列中窃取任务。

9.1.5 窃取任务

为了让没有任务的线程能从其他线程的任务队列中获取任务，就需要本地任务列表可以进行访问，这样才能让runpendingtasks()窃取任务。需要每个线程在线程池队列上进行注册，或由线程池指定一个线程。同样，还需要保证数据队列中的任务适当的被同步和保护，这样队列的不变量就不会被破坏。

实现一个无锁队列，让其拥有线程在其他线程窃取任务的时候，能够推送和弹出一个任务是可能的；不过，这个队列的实现就超出了本书的讨论范围。为了证明这种方法的可行性，将使用一个互斥量来保护队列中的数据。我们希望任务窃取是一个不常见的现象，这样就会减少对互斥量的竞争，并且使得简单队列的开销最小。下面，实现了一个简单的基于锁的任务窃取队列。

清单9.7 基于锁的任务窃取队列

class workstealingqueue
{
private:
  typedef functionwrapper datatype;
  std::deque<datatype> thequeue;  // 1
  mutable std::mutex the_mutex;


public:
  workstealingqueue()
  {}

workstealingqueue(const workstealingqueue& other)=delete;
  workstealingqueue& operator=(
    const workstealingqueue& other)=delete;

void push(datatype data)  // 2
  {
    std::lockguard<std::mutex> lock(themutex);
    thequeue.push_front(std::move(data));
  }

bool empty() const
  {
    std::lockguard<std::mutex> lock(themutex);
    return the_queue.empty();
  }

bool trypop(datatype& res)  // 3
  {
    std::lockguard<std::mutex> lock(themutex);
    if(the_queue.empty())
    {
      return false;
    }
res=std::move(the_queue.front());
the_queue.pop_front();
return true;

}

bool trysteal(datatype& res)  // 4
  {
    std::lockguard<std::mutex> lock(themutex);
    if(the_queue.empty())
    {
      return false;
    }
res=std::move(the_queue.back());
the_queue.pop_back();
return true;

}
};

这个队列对std::deque<fuctionwrapper>进行了简单的包装①，就能通过一个互斥锁来对所有访问进行控制了。push()②和trypop()③对队列的前端进行操作，trysteal()④对队列的后端进行操作。

这就说明每个线程中的“队列”是一个后进先出的栈，最新推入的任务将会第一个执行。从缓存角度来看，这将对性能有所提升，因为任务相关的数据一直存于缓存中，要比提前将任务相关数据推送到栈上好。同样，这种方式很好的映射到某个算法上，例如：快速排序。之前的实现中，每次调用dosort()都会推送一个任务到栈上，并且等待这个任务执行完毕。通过对最新推入任务的处理，就可以保证在将当前所需数据块处理完成前，其他任务是否需要这些数据块，从而可以减少活动任务的数量和栈的使用次数。trysteal()从队列末尾获取任务，为了减少与trypop()之间的竞争；使用在第6、7章中的所讨论的技术来让trypop()和trysteal()并发执行。

OK，现在拥有了一个很不错的任务队列，并且支持窃取；那这个队列将如何在线程池中使用呢？这里简单的展示一下。

清单9.8 使用任务窃取的线程池

class threadpool
{
  typedef functionwrapper task_type;


std::atomicbool done;
  threadsafequeue<tasktype> poolworkqueue;
  std::vector<std::uniqueptr<workstealingqueue> > queues;  // 1
  std::vector<std::thread> threads;
  jointhreads joiner;

static threadlocal workstealingqueue* localworkqueue;  // 2
  static threadlocal unsigned my_index;

void workerthread(unsigned myindex)
  {
    myindex=myindex;
    localworkqueue=queues[myindex].get();  // 3
    while(!done)
    {
      runpending_task();
    }
  }

bool poptaskfromlocalqueue(tasktype& task)
  {
    return localworkqueue && localworkqueue->trypop(task);
  }

bool poptaskfrompoolqueue(tasktype& task)
  {
    return poolworkqueue.trypop(task);
  }

bool poptaskfromotherthreadqueue(tasktype& task)  // 4
  {
    for(unsigned i=0;i<queues.size();++i)
    {
      unsigned const index=(myindex+i+1)%queues.size();  // 5
      if(queues[index]->trysteal(task))
      {
        return true;
      }
    }
    return false;
  }

public:
  threadpool():
    done(false),joiner(threads)
  {
    unsigned const threadcount=std::thread::hardware_concurrency();
try
{
  for(unsigned i=0;i&lt;thread_count;++i)
  {
    queues.push_back(std::unique_ptr&lt;work_stealing_queue&gt;(  // 6
                     new work_stealing_queue));
    threads.push_back(
      std::thread(&amp;thread_pool::worker_thread,this,i));
  }
}
catch(...)
{
  done=true;
  throw;
}

}

~thread_pool()
  {
    done=true;
  }

template<typename FunctionType>
  std::future<typename std::resultof<FunctionType()>::type> submit(
    FunctionType f)
  { 
    typedef typename std::resultof<FunctionType()>::type resulttype;
    std::packagedtask<resulttype()> task(f);
    std::future<resulttype> res(task.getfuture());
    if(localworkqueue)
    {
      localworkqueue->push(std::move(task));
    }
    else
    {
      poolwork_queue.push(std::move(task));
    }
    return res;
  }

void runpendingtask()
  {
    tasktype task;
    if(poptaskfromlocalqueue(task) ||  // 7
       poptaskfrompoolqueue(task) ||  // 8
       poptaskfromotherthreadqueue(task))  // 9
    {
      task();
    }
    else
    {
      std::thisthread::yield();
    }
  }
};

这段代码与清单9.6很相似。第一个不同在于，每个线程都有一个workstealingqueue，而非只是普通的std::queue<>②。当每个线程被创建，就创建了一个属于自己的工作队列⑥，每个线程自己的工作队列将存储在线程池的全局工作队列中①。列表中队列的序号，会传递给线程函数，然后使用序号来索引对应队列③。这就意味着线程池可以访问任意线程中的队列，为了给闲置线程窃取任务。runpendingtask()将会从线程的任务队列中取出一个任务来执行⑦，或从线程池队列中获取一个任务⑧，亦或从其他线程的队列中获取一个任务⑨。

poptaskfromotherthreadqueue()④会遍历池中所有线程的任务队列，然后尝试窃取任务。为了避免每个线程都尝试从列表中的第一个线程上窃取任务，每一个线程都会从下一个线程开始遍历，通过自身的线程序号来确定开始遍历的线程序号。

使用线程池有很多好处，还有很多很多的方式能为某些特殊用法提升性能，不过这就留给读者作为作业吧。

特别是还没有探究动态变换大小的线程池，即使线程被阻塞的时候(例如：I/O或互斥锁)，程序都能保证CPU最优的使用率。

下面，我们将来了解一下线程管理的“高级”用法——中断线程。

9.2 中断线程

很多情况下，使用信号来终止一个长时间运行的线程是合理的。这种线程的存在，可能是因为工作线程所在的线程池被销毁，或是用户显式的取消了这个任务，亦或其他各种原因。不管是什么原因，原理都一样：需要使用信号来让未结束线程停止运行。这里需要一种合适的方式让线程主动的停下来，而非让线程戛然而止。

你可能会给每种情况制定一个独立的机制，这样做的意义不大。不仅因为用统一的机制会更容易在之后的场景中实现，而且写出来的中断代码不用担心在哪里使用。C++11标准没有提供这样的机制，不过实现这样的机制也并不困难。

在了解一下应该如何实现这种机制前，先来了解一下启动和中断线程的接口。

9.2.1 启动和中断线程

先看一下外部接口，需要从可中断线程上获取些什么？最起码需要和std::thread相同的接口，还要多加一个interrupt()函数：

class interruptiblethread
{
public:
  template<typename FunctionType>
  interruptiblethread(FunctionType f);
  void join();
  void detach();
  bool joinable() const;
  void interrupt();
};

类内部可以使用std::thread来管理线程，并且使用一些自定义数据结构来处理中断。现在，从线程的角度能看到什么呢？“能用这个类来中断线程”——需要一个断点(interruption point)。在不添加多余的数据的前提下，为了使断点能够正常使用，就需要使用一个没有参数的函数：interruptionpoint()。这意味着中断数据结构可以访问threadlocal变量，并在线程运行时，对变量进行设置，因此当线程调用interruptionpoint()函数时，就会去检查当前运行线程的数据结构。我们将在后面看到interruptionpoint()的具体实现。

threadlocal标志是不能使用普通的std::thread管理线程的主要原因；需要使用一种方法分配出一个可访问的interruptiblethread实例，就像新启动一个线程一样。在使用已提供函数来做这件事情前，需要将interruptiblethread实例传递给std::thread的构造函数，创建一个能够执行的线程，就像下面的代码清单所实现。

清单9.9 interruptiblethread的基本实现

class interruptflag
{
public:
  void set();
  bool isset() const;
};
threadlocal interruptflag thisthreadinterrupt_flag;  // 1


class interruptiblethread
{
  std::thread internalthread;
  interruptflag* flag;
public:
  template<typename FunctionType>
  interruptiblethread(FunctionType f)
  {
    std::promise<interruptflag*> p;  // 2
    internalthread=std::thread([f,&p]{  // 3
      p.setvalue(&thisthreadinterruptflag);
      f();  // 4
    });
    flag=p.getfuture().get();  // 5
  }
  void interrupt()
  {
    if(flag)
    {
      flag->set();  // 6
    }
  }
};

提供函数f是包装了一个lambda函数③，线程将会持有f副本和本地promise变量(p)的引用②。在新线程中，lambda函数设置promise变量的值到thisthreadinterruptflag(在threadlocal①中声明)的地址中，为的是让线程能够调用提供函数的副本④。调用线程会等待与其future相关的promise就绪，并且将结果存入到flag成员变量中⑤。注意，即使lambda函数在新线程上执行，对本地变量p进行悬空引用，都没有问题，因为在新线程返回之前，interruptiblethread构造函数会等待变量p，直到变量p不被引用。实现没有考虑处理汇入线程，或分离线程。所以，需要flag变量在线程退出或分离前已经声明，这样就能避免悬空问题。

interrupt()函数相对简单：需要一个线程去做中断时，需要一个合法指针作为一个中断标志，所以可以仅对标志进行设置⑥。

9.2.2 检查线程是否中断

现在就可以设置中断标志了，不过不检查线程是否被中断，这样的意义就不大了。使用interruptionpoint()函数最简单的情况；可以在一个安全的地方调用这个函数，如果标志已经设置，就可以抛出一个threadinterrupted异常：

void interruptionpoint()
{
  if(thisthreadinterruptflag.isset())
  {
    throw threadinterrupted();
  }
}

代码中可以在适当的地方使用这个函数：

void foo()
{
  while(!done)
  {
    interruptionpoint();
    processnextitem();
  }
}

虽然也能工作，但不理想。最好实在线程等待或阻塞的时候中断线程，因为这时的线程不能运行，也就不能调用interruptionpoint()函数！在线程等待的时候，什么方式才能去中断线程呢？

9.2.3 中断等待——条件变量

OK，需要仔细选择中断的位置，并通过显式调用interruptionpoint()进行中断，不过在线程阻塞等待的时候，这种办法就显得苍白无力了，例如：等待条件变量的通知。就需要一个新函数——interruptiblewait()——就可以运行各种需要等待的任务，并且可以知道如何中断等待。之前提到，可能会等待一个条件变量，所以就从它开始：如何做才能中断一个等待的条件变量呢？最简单的方式是，当设置中断标志时，需要提醒条件变量，并在等待后立即设置断点。为了让其工作，需要提醒所有等待对应条件变量的线程，就能确保感谢兴趣的线程能够苏醒。伪苏醒是无论如何都要处理的，所以其他线程(非感兴趣线程)将会被当作伪苏醒处理——两者之间没什么区别。interruptflag结构需要存储一个指针指向一个条件变量，所以用set()函数对其进行提醒。为条件变量实现的interruptiblewait()可能会看起来像下面清单中所示。

清单9.10 为std::conditionvariable实现的interruptiblewait有问题版

void interruptiblewait(std::conditionvariable& cv,
std::uniquelock<std::mutex>& lk)
{
  interruptionpoint();
  thisthreadinterruptflag.setconditionvariable(cv);  // 1
  cv.wait(lk);  // 2
  thisthreadinterruptflag.clearconditionvariable();  // 3
  interruptionpoint();
}

假设函数能够设置和清除相关条件变量上的中断标志，代码会检查中断，通过interruptflag为当前线程关联条件变量①，等待条件变量②，清理相关条件变量③，并且再次检查中断。如果线程在等待期间被条件变量所中断，中断线程将广播条件变量，并唤醒等待该条件变量的线程，所以这里就可以检查中断。不幸的是，代码有两个问题。第一个问题比较明显，如果想要线程安全：std::conditionvariable::wait()可以抛出异常，所以这里会直接退出，而没有通过条件变量删除相关的中断标志。这个问题很容易修复，就是在析构函数中添加相关删除操作即可。

第二个问题就不大明显了，这段代码存在条件竞争。虽然，线程可以通过调用interruptionpoint()被中断，不过在调用wait()后，条件变量和相关中断标志就没有什么系了，因为线程不是等待状态，所以不能通过条件变量的方式唤醒。就需要确保线程不会在最后一次中断检查和调用wait()间被唤醒。这里，不对std::conditionvariable的内部结构进行研究；不过，可通过一种方法来解决这个问题：使用lk上的互斥量对线程进行保护，这就需要将lk传递到setconditionvariable()函数中去。不幸的是，这将产生两个新问题：需要传递一个互斥量的引用到一个不知道生命周期的线程中去(这个线程做中断操作)为该线程上锁(调用interrupt()的时候)。这里可能会死锁，并且可能访问到一个已经销毁的互斥量，所以这种方法不可取。当不能完全确定能中断条件变量等待——没有interruptiblewait()情况下也可以时(可能有些严格)，那有没有其他选择呢？一个选择就是放置超时等待，使用waitfor()并带有一个简单的超时量(比如，1ms)。在线程被中断前，算是给了线程一个等待的上限(以时钟刻度为基准)。如果这样做了，等待线程将会看到更多因为超时而“伪”苏醒的线程，不过超时也不轻易的就帮助到我们。与interruptflag相关的实现的一个实现放在下面的清单中展示。

清单9.11 为std::conditionvariable在interruptiblewait中使用超时

class interruptflag
{
  std::atomic<bool> flag;
  std::conditionvariable* threadcond;
  std::mutex setclear_mutex;


public:
  interruptflag():
    threadcond(0)
  {}

void set()
  {
    flag.store(true,std::memoryorderrelaxed);
    std::lockguard<std::mutex> lk(setclearmutex);
    if(threadcond)
    {
      threadcond->notifyall();
    }
  }

bool isset() const
  {
    return flag.load(std::memoryorder_relaxed);
  }

void setconditionvariable(std::conditionvariable& cv)
  {
    std::lockguard<std::mutex> lk(setclearmutex);
    thread_cond=&cv;
  }

void clearconditionvariable()
  {
    std::lockguard<std::mutex> lk(setclearmutex);
    threadcond=0;
  }

struct clearcvondestruct
  {
    ~clearcvondestruct()
    {
      thisthreadinterruptflag.clearcondition_variable();
    }
  };
};

void interruptiblewait(std::conditionvariable& cv,
  std::uniquelock<std::mutex>& lk)
{
  interruptionpoint();
  thisthreadinterruptflag.setconditionvariable(cv);
  interruptflag::clearcvondestruct guard;
  interruptionpoint();
  cv.waitfor(lk,std::chrono::milliseconds(1));
  interruptionpoint();
}

如果有谓词(相关函数)进行等待，1ms的超时将会完全在谓词循环中完全隐藏：

template<typename Predicate>
void interruptiblewait(std::conditionvariable& cv,
                        std::uniquelock<std::mutex>& lk,
                        Predicate pred)
{
  interruptionpoint();
  thisthreadinterruptflag.setconditionvariable(cv);
  interruptflag::clearcvondestruct guard;
  while(!thisthreadinterruptflag.isset() && !pred())
  {
    cv.waitfor(lk,std::chrono::milliseconds(1));
  }
  interruptionpoint();
}

这会让谓词被检查的次数增加许多，不过对于简单调用wait()这套实现还是很好用的。超时变量很容易实现：通过制定时间，比如：1ms或更短。OK，对于std::conditionvariable的等待，就需要小心应对了；std::conditionvariableany呢？还是能做的更好吗？

9.2.4 使用std::conditionvariableany中断等待

std::conditionvariableany与std::conditionvariable的不同在于，std::conditionvariableany可以使用任意类型的锁，而不仅有std::uniquelock<std::mutex>。可以让事情做起来更加简单，并且std::conditionvariableany可以比std::conditionvariable做的更好。因为能与任意类型的锁一起工作，就可以设计自己的锁，上锁/解锁interruptflag的内部互斥量setclearmutex，并且锁也支持等待调用，就像下面的代码。

清单9.12 为std::conditionvariableany设计的interruptiblewait

class interruptflag
{
  std::atomic<bool> flag;
  std::conditionvariable* threadcond;
  std::conditionvariableany* threadcondany;
  std::mutex setclearmutex;


public:
  interruptflag(): 
    threadcond(0),threadcondany(0)
  {}

void set()
  {
    flag.store(true,std::memoryorderrelaxed);
    std::lockguard<std::mutex> lk(setclearmutex);
    if(threadcond)
    {
      threadcond->notifyall();
    }
    else if(threadcondany)
    {
      threadcondany->notify_all();
    }
  }

template<typename Lockable>
  void wait(std::conditionvariableany& cv,Lockable& lk)
  {
    struct customlock
    {
      interruptflag* self;
      Lockable& lk;
  custom_lock(interrupt_flag* self_,
              std::condition_variable_any&amp; cond,
              Lockable&amp; lk_):
    self(self_),lk(lk_)
  {
    self-&gt;set_clear_mutex.lock();  // 1
    self-&gt;thread_cond_any=&amp;cond;  // 2
  }

  void unlock()  // 3
  {
    lk.unlock();
    self-&gt;set_clear_mutex.unlock();
  }

  void lock()
  {
    std::lock(self-&gt;set_clear_mutex,lk);  // 4
  }

  ~custom_lock()
  {
    self-&gt;thread_cond_any=0;  // 5
    self-&gt;set_clear_mutex.unlock();
  }
};
custom_lock cl(this,cv,lk);
interruption_point();
cv.wait(cl);
interruption_point();

}
  // rest as before
};

template<typename Lockable>
void interruptiblewait(std::conditionvariableany& cv,
                        Lockable& lk)
{
  thisthreadinterruptflag.wait(cv,lk);
}

自定义的锁类型在构造的时候，需要所锁住内部setclearmutex①，对threadcondany指针进行设置，并引用std::conditionvariableany传入锁的构造函数中②。Lockable引用将会在之后进行存储，其变量必须被锁住。现在可以安心的检查中断，不用担心竞争了。如果这时中断标志已经设置，那么标志一定是在锁住setclearmutex时设置的。当条件变量调用自定义锁的unlock()函数中的wait()时，就会对Lockable对象和setclearmutex进行解锁③。这就允许线程可以尝试中断其他线程获取setclearmutex锁；以及在内部wait()调用之后，检查threadcondany指针。这就是在替换std::conditionvariable后，所拥有的功能(不包括管理)。当wait()结束等待(因为等待，或因为伪苏醒)，因为线程将会调用lock()函数，这里依旧要求锁住内部setclearmutex，并且锁住Lockable对象④。现在，在wait()调用时，customlock的析构函数中⑤清理threadcondany指针(同样会解锁setclearmutex)之前，可以再次对中断进行检查。

9.2.5 中断其他阻塞调用

这次轮到中断条件变量的等待了，不过其他阻塞情况，比如：互斥锁，等待future等等，该怎么办呢？通常情况下，可以使用std::conditionvariable的超时选项，因为在实际运行中不可能很快的将条件变量的等待终止(不访问内部互斥量或future的话)。不过，在某些情况下，你知道知道你在等待什么，这样就可以让循环在interruptiblewait()函数中运行。作为一个例子，这里为std::future<>重载了interruptiblewait()的实现：

template<typename T>
void interruptiblewait(std::future<T>& uf)
{
  while(!thisthreadinterruptflag.isset())
  {
    if(uf.waitfor(lk,std::chrono::milliseconds(1)==
       std::futurestatus::ready)
      break;
  }
  interruptionpoint();
}

等待会在中断标志设置好的时候，或future准备就绪的时候停止，不过实现中每次等待future的时间只有1ms。这就意味着，中断请求被确定前，平均等待的时间为0.5ms(这里假设存在一个高精度的时钟)。通常waitfor至少会等待一个时钟周期，所以如果时钟周期为15ms，那么结束等待的时间将会是15ms，而不是1ms。接受与不接受这种情况，都得视情况而定。如果这必要，且时钟支持的话，可以持续削减超时时间。这种方式将会让线程苏醒很多次，来检查标志，并且增加线程切换的开销。

OK，我们已经了解如何使用interruptionpoint()和interruptiblewait()函数检查中断。

当中断被检查出来了，要如何处理它呢？

9.2.6 处理中断

从中断线程的角度看，中断就是threadinterrupted异常，因此能像处理其他异常那样进行处理。

try
{
  dosomething();
}
catch(threadinterrupted&)
{
  handleinterruption();
}

捕获中断，进行处理。其他线程再次调用interrupt()时，线程将会再次被中断，这就被称为断点(interruption point)。如果线程执行的是一系列独立的任务，就会需要断点；中断一个任务，就意味着这个任务被丢弃，并且该线程就会执行任务列表中的其他任务。

因为threadinterrupted是一个异常，在能够被中断的代码中，之前线程安全的注意事项都是适用的，就是为了确保资源不会泄露，并在数据结构中留下对应的退出状态。通常，让线程中断是可行的，所以只需要让异常传播即可。不过，当异常传入std::thread的析构函数时，std::terminate()将会调用，并且整个程序将会终止。为了避免这种情况，需要在每个将interruptiblethread变量作为参数传入的函数中放置catch(threadinterrupted)处理块，可以将catch块包装进interruptflag的初始化过程中。因为异常将会终止独立进程，就能保证未处理的中断是异常安全的。interruptiblethread构造函数中对线程的初始化，实现如下：

internalthread=std::thread([f,&p]{
        p.setvalue(&thisthreadinterruptflag);

    try
    {
      f();
    }
    catch(thread_interrupted const&)
    {}
  });

下面，我们来看个更加复杂的例子。

9.2.7 应用退出时中断后台任务

试想，在桌面上查找一个应用。这就需要与用户互动，应用的状态需要能在显示器上显示，就能看出应用有什么改变。为了避免影响GUI的响应时间，通常会将处理线程放在后台运行。后台进程需要一直执行，直到应用退出；后台线程会作为应用启动的一部分被启动，并且在应用终止的时候停止运行。通常这样的应用只有在机器关闭时，才会退出，因为应用需要更新应用最新的状态，就需要全时间运行。在某些情况下，当应用被关闭，需要使用有序的方式将后台线程关闭，其中一种方式就是中断。

下面清单中为一个系统实现了简单的线程管理部分。

清单9.13 在后台监视文件系统

std::mutex configmutex;
std::vector<interruptiblethread> background_threads;


void backgroundthread(int diskid)
{
  while(true)
  {
    interruptionpoint();  // 1
    fschange fsc=getfschanges(diskid);  // 2
    if(fsc.haschanges())
    {
      update_index(fsc);  // 3
    }
  }
}

void startbackgroundprocessing()
{
  backgroundthreads.pushback(
    interruptiblethread(backgroundthread,disk1));
  backgroundthreads.pushback(
    interruptiblethread(backgroundthread,disk2));
}

int main()
{
  startbackgroundprocessing();  // 4
  processguiuntilexit();  // 5
  std::uniquelock<std::mutex> lk(configmutex);
  for(unsigned i=0;i<backgroundthreads.size();++i)
  {
    backgroundthreads[i].interrupt();  // 6
  }
  for(unsigned i=0;i<backgroundthreads.size();++i)
  {
    backgroundthreads[i].join(); // 7
  }
}

启动时，后台线程就已经启动④。之后，对应线程将会处理GUI⑤。当用户要求进程退出时，后台进程将会被中断⑥，并且主线程会等待每一个后台线程结束后才退出⑦。后台线程运行在一个循环中，并时刻检查磁盘的变化②，对其序号进行更新③。调用interruptionpoint()函数，可以在循环中对中断进行检查。

为什么中断线程前，对线程进行等待？为什么不中断每个线程，让它们执行下一个任务？答案就是“并发”。线程被中断后，不会马上结束，因为需要对下一个断点进行处理，并且在退出前执行析构函数和代码异常处理部分。因为需要汇聚每个线程，所以就会让中断线程等待，即使线程还在做着有用的工作——中断其他线程。只有当没有工作时(所有线程都被中断)，不需要等待。这就允许中断线程并行的处理自己的中断，并更快的完成中断。

中断机制很容易扩展到更深层次的中断调用，或在特定的代码块中禁用中断，这就当做留给读者的作业吧。

9.3 总结

在本章中，我们了解各种“高级”线程管理技术：线程池和中断线程。也了解了如何使用本地任务队列，使用任务窃取的方式减小同步开销，提高线程池的吞吐量；等待子任务完成的同时执行队列中其他任务，从而来避免死锁。

也了解了使用线程去中断另一个处理线程的各种方式，比如：使用特定的断点和函数执行中断，要不就是使用某种方法，对阻塞等待进行中断。