详解linux系统中断线程的那些事

系统管理员 2023-07-24 13:13 27阅读 0赞

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

可能会给每种情况制定一个独立的机制，但这样做的意义不大。不仅因为用统一的机制会更容易在之后的场景中实现，而且写出来的中断代码不用担心在哪里使用。C11标准没有提供这样的机制(草案上有积极的建议，说不定中断线程会在以后的C标准中添加\[1\])，不过实现这样的机制也并不困难。

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

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

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

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

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

interruptible\_thread的基本实现

class interrupt_flag
    { 
    public:
      void set();
      bool is_set() const;
    };
    thread_local interrupt_flag this_thread_interrupt_flag;  // 1
    
    class interruptible_thread
    { 
      std::thread internal_thread;
      interrupt_flag* flag;
    public:
      template<typename FunctionType>
      interruptible_thread(FunctionType f)
      { 
        std::promise<interrupt_flag*> p;  // 2
        internal_thread=std::thread([f,&p]{   // 3
          p.set_value(&this_thread_interrupt_flag);
          f();  // 4
        });
        flag=p.get_future().get();  // 5
      }
      void interrupt()
      { 
        if(flag)
        { 
          flag->set();  // 6
        }
      }
    };

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

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

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

void interruption_point()
    { 
      if(this_thread_interrupt_flag.is_set())
      { 
        throw thread_interrupted();
      }
    }

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

void foo()
    { 
      while(!done)
      { 
        interruption_point();
        process_next_item();
      }
    }

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

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

为std::condition\_variable实现的interruptible\_wait有问题版

void interruptible_wait(std::condition_variable& cv,
    std::unique_lock<std::mutex>& lk)
    { 
      interruption_point();
      this_thread_interrupt_flag.set_condition_variable(cv);  // 1
      cv.wait(lk);  // 2
      this_thread_interrupt_flag.clear_condition_variable();  // 3
      interruption_point();
    }

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

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

为std::condition\_variable在interruptible\_wait中使用超时

class interrupt_flag
    { 
      std::atomic<bool> flag;
      std::condition_variable* thread_cond;
      std::mutex set_clear_mutex;
    
    public:
      interrupt_flag():
        thread_cond(0)
      { }
    
      void set()
      { 
        flag.store(true,std::memory_order_relaxed);
        std::lock_guard<std::mutex> lk(set_clear_mutex);
        if(thread_cond)
        { 
          thread_cond->notify_all();
        }
      }
    
      bool is_set() const
      { 
        return flag.load(std::memory_order_relaxed);
      }
    
      void set_condition_variable(std::condition_variable& cv)
      { 
        std::lock_guard<std::mutex> lk(set_clear_mutex);
        thread_cond=&cv;
      }
    
      void clear_condition_variable()
      { 
        std::lock_guard<std::mutex> lk(set_clear_mutex);
        thread_cond=0;
      }
    
      struct clear_cv_on_destruct
      { 
        ~clear_cv_on_destruct()
        { 
          this_thread_interrupt_flag.clear_condition_variable();
        }
      };
    };
    
    void interruptible_wait(std::condition_variable& cv,
      std::unique_lock<std::mutex>& lk)
    { 
      interruption_point();
      this_thread_interrupt_flag.set_condition_variable(cv);
      interrupt_flag::clear_cv_on_destruct guard;
      interruption_point();
      cv.wait_for(lk,std::chrono::milliseconds(1));
      interruption_point();
    }

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

template<typename Predicate>
    void interruptible_wait(std::condition_variable& cv,
                            std::unique_lock<std::mutex>& lk,
                            Predicate pred)
    { 
      interruption_point();
      this_thread_interrupt_flag.set_condition_variable(cv);
      interrupt_flag::clear_cv_on_destruct guard;
      while(!this_thread_interrupt_flag.is_set() && !pred())
      { 
        cv.wait_for(lk,std::chrono::milliseconds(1));
      }
      interruption_point();
    }

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

使用std::condition\_variable\_any中断等待  
std::condition\_variable\_any与std::condition\_variable的不同在于，std::condition\_variable\_any可以使用任意类型的锁，而不仅有std::unique\_lockstd::mutex。可以让事情做起来更加简单，并且std::condition\_variable\_any可以比std::condition\_variable做的更好。因为能与任意类型的锁一起工作，就可以设计自己的锁，上锁/解锁interrupt\_flag的内部互斥量set\_clear\_mutex，并且锁也支持等待调用，就像下面的代码。

为std::condition\_variable\_any设计的interruptible\_wait

class interrupt_flag
    { 
      std::atomic<bool> flag;
      std::condition_variable* thread_cond;
      std::condition_variable_any* thread_cond_any;
      std::mutex set_clear_mutex;
    
    public:
      interrupt_flag(): 
        thread_cond(0),thread_cond_any(0)
      { }
    
      void set()
      { 
        flag.store(true,std::memory_order_relaxed);
        std::lock_guard<std::mutex> lk(set_clear_mutex);
        if(thread_cond)
        { 
          thread_cond->notify_all();
        }
        else if(thread_cond_any)
        { 
          thread_cond_any->notify_all();
        }
      }
    
      template<typename Lockable>
      void wait(std::condition_variable_any& cv,Lockable& lk)
      { 
        struct custom_lock
        { 
          interrupt_flag* self;
          Lockable& lk;
    
          custom_lock(interrupt_flag* self_,
                      std::condition_variable_any& cond,
                      Lockable& lk_):
            self(self_),lk(lk_)
          { 
            self->set_clear_mutex.lock();  // 1
            self->thread_cond_any=&cond;  // 2
          }
    
          void unlock()  // 3
          { 
            lk.unlock();
            self->set_clear_mutex.unlock();
          }
    
          void lock()
          { 
            std::lock(self->set_clear_mutex,lk);  // 4
          }
    
          ~custom_lock()
          { 
            self->thread_cond_any=0;  // 5
            self->set_clear_mutex.unlock();
          }
        };
        custom_lock cl(this,cv,lk);
        interruption_point();
        cv.wait(cl);
        interruption_point();
      }
      // rest as before
    };
    
    template<typename Lockable>
    void interruptible_wait(std::condition_variable_any& cv,
                            Lockable& lk)
    { 
      this_thread_interrupt_flag.wait(cv,lk);
    }

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

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

template<typename T>
    void interruptible_wait(std::future<T>& uf)
    { 
      while(!this_thread_interrupt_flag.is_set())
      { 
        if(uf.wait_for(lk,std::chrono::milliseconds(1))==
           std::future_status::ready)
          break;
      }
      interruption_point();
    }

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

OK，我们已经了解如何使用interruption\_point()和interruptible\_wait()函数检查中断。当中断被检查出来了，要如何处理它呢？

处理中断  
从中断线程的角度看，中断就是thread\_interrupted异常，因此能像处理其他异常那样进行处理。特别是使用标准catch块对其进行捕获：

try
    { 
      do_something();
    }
    catch(thread_interrupted&)
    { 
      handle_interruption();
    }

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

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

internal_thread=std::thread([f,&p]{ 
            p.set_value(&this_thread_interrupt_flag);
            
            try
            { 
              f();
            }
            catch(thread_interrupted const&)
            { }
          });

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

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

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

后台监视文件系统

std::mutex config_mutex;
    std::vector<interruptible_thread> background_threads;
    
    void background_thread(int disk_id)
    { 
      while(true)
      { 
        interruption_point();  // 1
        fs_change fsc=get_fs_changes(disk_id);  // 2
        if(fsc.has_changes())
        { 
          update_index(fsc);  // 3
        }
      }
    }
    
    void start_background_processing()
    { 
      background_threads.push_back(
        interruptible_thread(background_thread,disk_1));
      background_threads.push_back(
        interruptible_thread(background_thread,disk_2));
    }
    
    int main()
    { 
      start_background_processing();  // 4
      process_gui_until_exit();  // 5
      std::unique_lock<std::mutex> lk(config_mutex);
      for(unsigned i=0;i<background_threads.size();++i)
      { 
        background_threads[i].interrupt();  // 6
      }
      for(unsigned i=0;i<background_threads.size();++i)
      { 
        background_threads[i].join(); // 7
      }
    }

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

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

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

需要C/C++ Linux服务器架构师学习资料加群563998835获取 （资料包括C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg等）![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70]: /images/20230528/4c058cbece7d45239add6e2a034a9dc6.png