python线程和进程

发表于 | 分类于
it is so pythonic

唠叨话

进程和线程应该是编程语言老生常谈的话题了，但是python在这俩个方面，只能说能用，不怎么关心，起码在早期的python原生的库中，相关的支持很简陋，比如thread库，基本的同步原语都没，感觉早期python目标场景，比如运维领域，不需要多进程、多线程，后来，在需要高并发的场景，比如web,python又在不断迭代协程来实现

threading

threading模块封装了很多的同步原语，

  • Lock
    • 原始锁，实际上就是使用匿名信号量实现的，一般语言里面的原始锁可以由互斥量、临界区、信号量实现的，python这里选择了匿名信号量实现，也就是说，这个锁，谁(所有线程)都可以减去获取锁acquire、就是信号量减一，释放锁release、信号量加一
  • Rlock
    • 重入锁RLock,相当于java的重入锁，基于Lock实现的，就是在acquire的时候记录当前线程id，release的时候判定线程是不是acquire的线程
  • condition，条件，维护一个锁Lock/RLock，和一个waiting池，线程通过acquire获得condition对象，当wait的时候，线程会释放condition内部的锁进入block状态，同时waiting池中记录这个线程，当notify，condition对象从waiting池中挑选一个线程，通知其acquire
  • Semaphore，信号量，基于Condition实现
  • BoundedSemaphore，防止Semaphore被无限释放
  • event 基于Condition，控制线程间运行顺序
  • Barrier 基于Condition，让特定线程先运行
  • Timer 基于Thread和Event实现，定时任务

从锁上看，python原生库提供的支持很基础，而且没有特定的优化，相对于java在锁方面，大量运用CAS原语，提供各种字节码指令，优化各种并发情况下的同步操作，以达到高性能，python在这方面就很随意，为什么呢？因为python本身有着GIL，在多线程效率方面就天生有着限制，我觉得GIL就是对单核操作系统执行多进程的模拟，另外，如果说高并发的话，python可以用协程实现，比如tornado、asyncio，底层使用是IO多路复用、select/poll/epoll/kqueue等，不过老实说，在协程方面，python又不如go、scala、erlang等完善，python自己的协程框架asyncio的三方生态太少，单说高并发，也不及Go等

创建线程调用链： start -> _thread._start_new_thread -> thread_Pythread_start_new_thread -> PyThread_start_new_thread -> pthread_create

通过这个调用链看出，在linux，python的线程实现，就是通过pthread库函数创建的，是一种用户态的线程，和linux内核进程一对一

线程池

一个基本模型

不管是线程池还是进程池，都要维护俩个核心东西：进(线)程队列和任务队列，进(线)程队列中的进程从任务队列中取任务去执行，为了取任务的时候，各个进(线)程不冲突，需要一个锁，并将返回给一个对象，一般都叫future对象，后续通过这个future获取任务完成后的返回值。

关于以上这个模型，有一个很好的现实场景对应，假设一个银行有三个功能一模一样的窗口(进/线程队列)，一群人（任务队列）去这个银行办理业务，这些都是领号，然后当某个窗口空了，大厅的屏幕上，就分配一个号对应的人，去那个窗口办理业务，这个分配号到窗口的系统，保证一次就只会把一个号分配到一个空闲的窗口，这就是锁的作用。

1
2
3
4
5
6
7
8
9
10
窗口1    窗口2     窗口3
客户-1   客户0     VIP客户0
---     ----      ---

客户1
客户2
VIP客户1
VIP客户2
客户3

这里有个分配号到窗口的具体逻辑，如果是按照号的顺序去分配，就是典型的FIFO队列，java的ArrayBlockingQueue/LinkedBlockingQueue就是这样的，但是如果某些银行的vip客户，可以优先分配到空闲窗口，那么就是优先级队列，java的PriorityBlockingQueue就是这样实现的。当然，现实场景往往是俩者结合的，想想，如果优先级队列的优先级，按照任务入队列的时间顺序递减，那么不就是FIFO队列吗？还有一种队列，就是随机选客户，众生平等，这会让编程变得不可预期，所以实际应用应该很少，起码java好像没有这种队列

ThreadPoolExecutor

java中也有一个ThreadPoolExecutor实现线程池，感觉python像是直接参考过来的。

结合上述模型，下面来分析源码，源码来自于github，我这里都是看的3.10-master上的代码，这里做了些省略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class ThreadPoolExecutor(_base.Executor):

    def __init__(self, max_workers=None, thread_name_prefix='',
                 initializer=None, initargs=()):
        """Initializes a new ThreadPoolExecutor instance.
        Args:
            max_workers: The maximum number of threads that can be used to
                execute the given calls.
            thread_name_prefix: An optional name prefix to give our threads.
            initializer: A callable used to initialize worker threads.
            initargs: A tuple of arguments to pass to the initializer.
        """
        if max_workers is None:
            # ThreadPoolExecutor is often used to:
            # * CPU bound task which releases GIL
            # * I/O bound task (which releases GIL, of course)
            #
            # We use cpu_count + 4 for both types of tasks.
            # But we limit it to 32 to avoid consuming surprisingly large resource
            # on many core machine.
            max_workers = min(32, (os.cpu_count() or 1) + 4)
        if max_workers <= 0:
            raise ValueError("max_workers must be greater than 0")

        if initializer is not None and not callable(initializer):
            raise TypeError("initializer must be a callable")

        self._max_workers = max_workers #最大线程数
        self._work_queue = queue.SimpleQueue() #任务队列，实际上就是个无界的FIFO队列
        self._idle_semaphore = threading.Semaphore(0)
        self._threads = set()# 线程队列
        self._broken = False
        self._shutdown = False
        self._shutdown_lock = threading.Lock()
        self._thread_name_prefix = (thread_name_prefix or
                                    ("ThreadPoolExecutor-%d" % self._counter()))
        self._initializer = initializer
        self._initargs = initargs

	def submit(self, fn, /, *args, **kwargs):
        with self._shutdown_lock, _global_shutdown_lock:
            if self._broken:
                raise BrokenThreadPool(self._broken)

            if self._shutdown:
                raise RuntimeError('cannot schedule new futures after shutdown')
            if _shutdown:
                raise RuntimeError('cannot schedule new futures after '
                                   'interpreter shutdown')

            f = _base.Future()
            w = _WorkItem(f, fn, args, kwargs)# 将函数转变成_WorkItem对象

            self._work_queue.put(w) # 将任务入队
            self._adjust_thread_count() #实际上是这个函数负责启动线程，也就是将任务放入线程去执行，对应的就是将一个客户分配给一个窗口，锁就是_idle_semaphore(信号量)
            return f # 返回一个future对象,之后就是通过这个对象获取线程执行的结果

	def shutdown(self, wait=True, *, cancel_futures=False):
		# 关闭线程池，对应的场景就是银行下班了，窗口都不在服务了，只不过现实中窗口不会服务一半就直接停了
        with self._shutdown_lock:
            self._shutdown = True
            if cancel_futures:
                # Drain all work items from the queue, and then cancel their
                # associated futures.
                while True:
                    try:
                        work_item = self._work_queue.get_nowait()
                    except queue.Empty:
                        break
                    if work_item is not None:
                        work_item.future.cancel()

            # Send a wake-up to prevent threads calling
            # _work_queue.get(block=True) from permanently blocking.
            self._work_queue.put(None)
        if wait:
            for t in self._threads:
                t.join()

通过分析代码，发现python的线程池十分基础，既没有java的ThreadPoolExecutor的核心线程与非核心线程，也不支持除了FIFO的其他任务队列，不过都可以自己参考来实现，也实现了基本的线程池模型

flag: 找找有没有功能更全的threadpoolexecutor或者自己实现

进程

什么是进程？一般回答是进程是操作系统分配资源的基本单位，相对应的，线程是cpu调度的最小单位，在不同的操作系统上，线程的实现也有所差异。

multiprocessing.Process

源码对应cpython/Lib/multiprocessing/context.py，可以看到Process的实现基本是继承BaseProcess，所以来分析BaseProcess如何创建进程的吧,下面的中文注释就是分析过程同样，做了省略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class BaseProcess(object):
    '''
    Process objects represent activity that is run in a separate process

    The class is analogous to `threading.Thread`
    '''
    # 这个需要子类去实现的，Process这里根据当前的操作系统，做了选择不同创建进程的方法，
	# fork unix 使用os.fork，子进程完全克隆父进程的资源，unix默认
	# spawn unix，windows，darwin，父进程启动一个新的Python解释器， 子进程将只继承运行run()方法所需的资源，是win32/darwin默认的模式，比较消耗资源
	# forkserver unix，启动一个单线程的服务器进程，专门用于创建进程，这个服务器进程会帮助克隆资源，这种创建进程方式需要用户主动选择
	# 这么看这种直接raise Error的做法好像在模拟java的抽象方法，只不过，这里只能在运行时才会抛出Error
    def _Popen(self):
        raise NotImplementedError

    def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
                 *, daemon=None):
        assert group is None, 'group argument must be None for now'
        count = next(_process_counter)
        self._identity = _current_process._identity + (count,)
        self._config = _current_process._config.copy()
        self._parent_pid = os.getpid()
        self._parent_name = _current_process.name
        self._popen = None
        self._closed = False
        self._target = target #实际进程运行的内容，就是传入的函数
        self._args = tuple(args) # 进程的参数，
        self._kwargs = dict(kwargs)# 进程的参数
        self._name = name or type(self).__name__ + '-' + \
                     ':'.join(str(i) for i in self._identity)
        if daemon is not None: #是否为设置守护进程
            self.daemon = daemon
        _dangling.add(self)

    def run(self):
        '''
        Method to be run in sub-process; can be overridden in sub-class
        '''
		# 这里的run是不是很像java里面Runnable接口里面的run函数
        if self._target:
            self._target(*self._args, **self._kwargs)

    def start(self):
        '''
        Start child process
        '''
        self._check_closed()
        assert self._popen is None, 'cannot start a process twice'
        assert self._parent_pid == os.getpid(), \
               'can only start a process object created by current process'
        assert not _current_process._config.get('daemon'), \
               'daemonic processes are not allowed to have children'
        _cleanup()
		# 追溯这个_Popen，就会发现最终是调用popen_fork.py中的_launch函数，_launch先是创建了俩个匿名管道，因为匿名管道是半双工的，所以创建俩个，方便父子进程同时通信，然后调用os.fork创建子进程，然后在调用BaseProcess._bootstrap
        self._popen = self._Popen(self)
        self._sentinel = self._popen.sentinel
        # Avoid a refcycle if the target function holds an indirect
        # reference to the process object (see bpo-30775)
        del self._target, self._args, self._kwargs
        _children.add(self)

    def terminate(self):
        '''
        Terminate process; sends SIGTERM signal or uses TerminateProcess()
        '''
        self._check_closed()
        self._popen.terminate()

    def join(self, timeout=None):
        '''
        Wait until child process terminates
        '''
        self._check_closed()
        assert self._parent_pid == os.getpid(), 'can only join a child process'
        assert self._popen is not None, 'can only join a started process'
        res = self._popen.wait(timeout)
        if res is not None:
            _children.discard(self)

    def _bootstrap(self, parent_sentinel=None):
        from . import util, context
        global _current_process, _parent_process, _process_counter, _children

        try:
            if self._start_method is not None:
                context._force_start_method(self._start_method)
            _process_counter = itertools.count(1)
            _children = set()
            util._close_stdin()
            old_process = _current_process
            _current_process = self
            _parent_process = _ParentProcess(
                self._parent_name, self._parent_pid, parent_sentinel)
            if threading._HAVE_THREAD_NATIVE_ID:
                threading.main_thread()._set_native_id()
            try:
                util._finalizer_registry.clear() #清空_finalizer_registry
                util._run_after_forkers()#执行fork之后的任务，用于可以定义弱引用对象去实现这个阶段的hook
            finally:
                # delay finalization of the old process object until after
                # _run_after_forkers() is executed
                del old_process
            util.info('child process calling self.run()')
            try:
                self.run() #执行进程内容
                exitcode = 0
            finally:
                util._exit_function() #执行完毕，清理子进程，以及执行_run_finalizers(),这个函数默认为空，需要用户自己定义multiprocessing.util.Finalize，做进程退出时的hook函数
        except SystemExit as e:
            if e.code is None:
                exitcode = 0
            elif isinstance(e.code, int):
                exitcode = e.code
            else:
                sys.stderr.write(str(e.code) + '\n')
                exitcode = 1
        except:
            exitcode = 1
            import traceback
            sys.stderr.write('Process %s:\n' % self.name)
            traceback.print_exc()
        finally:
            threading._shutdown()
            util.info('process exiting with exitcode %d' % exitcode)
            util._flush_std_streams()

        return exitcode

进程间通信

线程共享堆上的数据，本身就可以通信。

回顾一下进程间通信的方法：共享内存、管道、信号量、套接字、文件、消息队列、信号等，下面就看看python这块的具体实现吧

  1. 首先，在分析进程创建的时候，就使用了匿名管道，这个时os模块实现的
  2. multiprocessing.Pipe，管道，在linux，使用域间套接字实现的
  3. multiprocessing.Queue,队列，实际上基于上面的Pipe实现的
  4. sharedMemory，共享内存，基于mmap实现的，将进程自己的虚拟内存映射到同一片物理内存
  5. shareableList，基于sharedMemory实现的

有了这些，基本的进程间通信时没什么问题的

这里没有说锁的实现，因为就是用的线程threading模块的锁

进程池mumltiproessing.Pool

结合在说明线程池的时候，描述的模型，和去银行窗口办业务，这里不详细分析源码了，直接解释关键的实现

  • _taskqueue存放任务的队列，对这个队列取任务时，会使用锁锁定
  • _pool worker进程队列
  • _inqueue 实际执行任务时，各个进程都是从这个队列取任务
  • _outqueue 发送结果的队列
  • _work_handler线程，保证进程池在worker进程有退出的时候创建新的进程，添加到pool
  • _task_handler线程，从任务队列中，取出任务，放入_inqueue
  • _handle_results线程，将处理完的任务的结果，从_outqueue取出