What is the method to forcefully close threads, coroutines and processes in Python?

Multi-threading

First of all, when exiting in a thread, we often use a method: the loop condition of the sub-thread execution sets a condition, and when we need to exit the sub-thread, set the condition , at this time the sub-thread will actively exit, but when the sub-thread is blocked, the conditions are not judged in the loop, and the blocking time is uncertain, it will be impossible for us to recycle the thread. At this time, the following methods are needed:

Daemon thread:

If you set a thread as a daemon thread, it means that you are saying that this thread is not important and will not be used in the process. When exiting, there is no need to wait for this thread to exit.
If your main thread does not need to wait for those child threads to complete when it exits, then set the daemon attributes of these threads. That is, before the thread starts (thread.start()), call the setDeamon() function to set the daemon flag of the thread. (thread.setDaemon(True)) means that this thread is "not important".

If you want to wait for the child thread to complete before exiting, then there is nothing to do. , or explicitly call thread.setDaemon(False) to set the daemon value to false. The new child thread will inherit the daemon flag of the parent thread. The entire Python will end after all non-daemon threads exit, that is, when there are no non-daemon threads in the process.

That is, the sub-thread is a non-deamon thread, and the main thread does not exit immediately

import threading
import time
import gc
import datetime

def circle():
    print("begin")
    try:
        while True:
            current_time = datetime.datetime.now()
            print(str(current_time) + ' circle.................')
            time.sleep(3)
    except Exception as e:
        print('error:',e)
    finally:
        print('end')


if __name__ == "__main__":
   t = threading.Thread(target=circle)
   t.setDaemon(True)
   t.start()
   time.sleep(1)
   # stop_thread(t)
   # print('stoped threading Thread') 
   current_time = datetime.datetime.now()
   print(str(current_time) + ' stoped after') 
   gc.collect()
   while True:
      time.sleep(1)
      current_time = datetime.datetime.now()
      print(str(current_time) + ' end circle')

Is it the main thread that controls it?

The daemon thread needs the main thread to exit to complete the sub-thread exit, the following is the code, and then encapsulate one layer to verify whether the main thread needs to exit

def Daemon_thread():
   circle_thread= threading.Thread(target=circle)
#    circle_thread.daemon = True
   circle_thread.setDaemon(True)
   circle_thread.start()   
   while running:
        print('running:',running) 
        time.sleep(1)
   print('end..........') 


if __name__ == "__main__":
    t = threading.Thread(target=Daemon_thread)
    t.start()   
    time.sleep(3)
    running = False
    print('stop running:',running) 
    print('stoped 3') 
    gc.collect()
    while True:
        time.sleep(3)
        print('stoped circle')

Replace the main function execution , and found that the circle thread continued to execute after printing the stopped 3 flag.

Conclusion: The main thread relies on processing signals. Only by ensuring that it is alive can the signal be processed correctly.

Throwing exceptions in Python threads

Although using PyThreadState_SetAsyncExc can satisfy our direct exit from the thread in most cases; the PyThreadState_SetAsyncExc method only executes the "plan" for the thread to exit. It does not kill the thread, especially when it is executing an external C library. Try sleep(100) your method to kill one. It will be "killed" after 100 seconds. while flag: It works the same as ->flag = False method.

So when the sub-thread has blocking functions such as sleep, during the sleep process, the sub-thread cannot respond and will be captured by the main thread, resulting in the inability to cancel the sub-thread. In fact, when the thread is sleeping, it is not possible to directly use the async_raise function to kill the thread, because if the thread is busy outside the Python interpreter, it will not catch the interrupt

Sample code:

import ctypes
import inspect
import threading
import time
import gc
import datetime

def async_raise(tid, exctype):
   """raises the exception, performs cleanup if needed"""
   tid = ctypes.c_long(tid)
   if not inspect.isclass(exctype):
      exctype = type(exctype)
   res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
   if res == 0:
      raise ValueError("invalid thread id")
   elif res != 1:
      # """if it returns a number greater than one, you're in trouble,  
      # and you should call it again with exc=NULL to revert the effect"""  
      ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
      raise SystemError("PyThreadState_SetAsyncExc failed")
      
def stop_thread(thread):
   async_raise(thread.ident, SystemExit)
   
def circle():
    print("begin")
    try:
        while True:
            current_time = datetime.datetime.now()
            print(str(current_time) + ' circle.................')
            time.sleep(3)
    except Exception as e:
        print('error:',e)
    finally:
        print('end')

if __name__ == "__main__":
   t = threading.Thread(target=circle)
   t.start()
   time.sleep(1)
   stop_thread(t)
   print('stoped threading Thread') 
   current_time = datetime.datetime.now()
   print(str(current_time) + ' stoped after') 
   gc.collect()
   while True:
      time.sleep(1)
      current_time = datetime.datetime.now()
      print(str(current_time) + ' end circle')

signal.pthread_kill operation:

This is the closest to the pthread kill operation in Unix. I saw some uses on the Internet, but I did not find the use in this library when I verified it.

This is From the description in python's official signal explanation document, I saw that it is a new function in version 3.3. I am python3.10 myself and there is no pthread_kill. It may be removed in subsequent versions.

This is some sample code I saw online, but it cannot be executed. If anyone knows how to use it, you can communicate.

from signal import pthread_kill, SIGTSTP
from threading import Thread
from itertools import count
from time import sleep

def target():
    for num in count():
        print(num)
        sleep(1)

thread = Thread(target=target)
thread.start()
sleep(5)
signal.pthread_kill(thread.ident, SIGTSTP)

Multiprocessing

Multiprocessing is a package that supports spawning processes using an API similar to the threading module. The multiprocessing package provides both local and remote concurrent operations, effectively bypassing the global interpreter lock by using child processes instead of threads. Therefore, the multiprocessing module allows programmers to take full advantage of multiple processors on a given machine.

Multiprocess libraries are used, and we can call its internal function terminate to help us release them. For example, t.terminate() can force the child process to exit.

However, the interaction method using multi-process data is more cumbersome. Shared memory, pipe or message queue must be used for data interaction between the child process and the parent process.

The sample code is as follows:

import time
import gc
import datetime
import multiprocessing

def circle():
    print("begin")
    try:
        while True:
            current_time = datetime.datetime.now()
            print(str(current_time) + ' circle.................')
            time.sleep(3)
    except Exception as e:
        print('error:',e)
    finally:
        print('end')


if __name__ == "__main__":
    t = multiprocessing.Process(target=circle, args=())
    t.start()
    # Terminate the process
    current_time = datetime.datetime.now()
    print(str(current_time) + ' stoped before') 
    time.sleep(1)
    t.terminate()  # sends a SIGTERM
    current_time = datetime.datetime.now()
    print(str(current_time) + ' stoped after') 
    gc.collect()
    while True:
        time.sleep(3)
        current_time = datetime.datetime.now()
        print(str(current_time) + ' end circle')

Multiple coroutines

Coroutines (coroutines) are also called micro-threads. They are another way to achieve multi-tasking and are more efficient than threads. A small execution unit, generally running on a single process and single thread. Because it has its own CPU context, it can switch tasks through a simple event loop, which is more efficient than switching between processes and threads, because the switching between processes and threads is performed by the operating system.

Python mainly relies on two libraries to implement coroutines: asyncio (asyncio is a standard library introduced from Python 3.4, which directly has built-in support for coroutine asynchronous IO. The essence of asyncio's programming model is a message Loop, we usually define a coroutine function (or task) first, obtain the event loop loop from the asyncio module, and then throw the coroutine task (or task list) that needs to be executed into the loop for execution, thus achieving asynchronous IO) and gevent (Gevent is a third-party library that can easily implement concurrent synchronous or asynchronous programming through gevent. The main mode used in gevent is Greenlet, which is a lightweight coroutine connected to Python in the form of a C extension module.).

由于asyncio已经成为python的标准库了无需pip安装即可使用，这意味着asyncio作为Python原生的协程实现方式会更加流行。本文仅会介绍asyncio模块的退出使用。

使用协程取消，有两个重要部分：第一，替换旧的休眠函数为多协程的休眠函数；第二取消使用cancel()函数。

其中cancel() 返回值为 True 表示 cancel 成功。

示例代码如下：创建一个coroutine，然后调用run_until_complete()来初始化并启动服务器来调用main函数，判断协程是否执行完成，因为设置的num协程是一个死循环，所以一直没有执行完，如果没有执行完直接使用 cancel()取消掉该协程，最后执行成功。

import asyncio
import time


async def num(n):
    try:
        i = 0
        while True:
            print(f'i={i} Hello')
            i=i+1
            # time.sleep(10)
            await asyncio.sleep(n*0.1)
        return n
    except asyncio.CancelledError:
        print(f"数字{n}被取消")
        raise


async def main():
    # tasks = [num(i) for i in range(10)]
    tasks = [num(10)]
    complete, pending = await asyncio.wait(tasks, timeout=0.5)
    for i in complete:
        print("当前数字",i.result())
    if pending:
        print("取消未完成的任务")
        for p in pending:
            p.cancel()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

The above is the detailed content of What is the method to forcefully close threads, coroutines and processes in Python?. For more information, please follow other related articles on the PHP Chinese website!