Khi làm việc với multiprocessing trong Python, chúng ta thường sử dụng module này để quản lý đa tiến trình. Bài viết này sẽ phân tích một số nguyên tắc và cơ chế hoạt động của module multiprocessing trong việc quản lý các tiến trình con.

1. Câu hỏi đầu tiên: Có cần gọi rõ ràng phương thức close và join của pool không? Nếu không gọi, các tiến trình con có bị kẹt lại không?

Khi khởi tạo Pool, tham số processes xác định số lượng worker trong pool. Khi khởi tạo, các worker được khởi động dưới dạng tiến trình con với thuộc tính daemon=True.

    def _repopulate_pool(self):
        """Tăng số lượng tiến trình worker trong pool lên mức được chỉ định,
        sau khi thu hồi các worker đã thoát."""
        for i in range(self._processes - len(self._pool)):
            w = self.Process(target=worker_process,
                             args=(self._input_queue, self._output_queue,
                                   self._initializer,
                                   self._initargs, self._max_tasks_per_child)
                            )
            self._pool.append(w)
            w.name = w.name.replace('Process', 'PoolWorker')
            w.daemon = True
            w.start()
            debug('added worker')

Khuyến nghị rằng sau khi sử dụng xong pool, nên gọi phương thức close() và join() để giải phóng các worker và đợi các tác vụ con hoàn thành. Tuy nhiên, nếu không gọi rõ ràng các phương thức này, khi tiến trình chính thoát, các tiến trình con cũng sẽ tự động thoát do cờ daemon đã được thiết lập.

def _exit_function(info=info, debug=debug, _run_finalizers=_run_finalizers,
                   active_children=active_children,
                   current_process=current_process):
    # Lưu giữ tham chiếu đến các hàm trong danh sách tham số do tình huống
    # mô tả bên dưới, khi hàm này được gọi sau khi globals của module bị hủy.

    global _exiting

    info('process shutting down')
    debug('running all "atexit" finalizers with priority >= 0')
    _run_finalizers(0)

    if current_process() is not None:
        # Kiểm tra nếu process hiện tại là None tại đây vì nếu nó là None,
        # bất kỳ lời gọi nào đến active_children() sẽ ném ra AttributeError.

        for p in active_children():
            if p._daemonic:
                info('calling terminate() for daemon %s', p.name)
                p._popen.terminate()

        for p in active_children():
            info('calling process for process %s', p.name)
            p.join()

    debug('running the remaining "atexit" finalizers')
    _run_finalizers()

Khi tiến trình chính thoát, hàm _exit_function sẽ được gọi. Nếu phát hiện các tiến trình con đang hoạt động với thuộc tính _daemonic, nó sẽ gọi phương thức terminate để kết thúc các tiến trình con. Cơ chế này được đăng ký thông qua atexit.register(_exit_function), tận dụng hook thoát của hệ thống để kích hoạt hàm tương ứng khi thoát.

2. Câu hỏi thứ hai: Nếu khởi động xong rồi kill -9 tiến trình chính, các tiến trình con có bị kẹt lại không?

Dưới đây là logic chính của worker trong pool. Nếu kill -9 tiến trình chính và worker không đang xử lý tác vụ, do tiến trình chính đã thoát nên phương thức get() lấy task từ queue sẽ ném ra exception, khiến worker thoát. Nếu worker đang xử lý tác vụ, khi tác vụ kết thúc và cần đưa kết quả vào queue do tiến trình chính đã thoát cũng sẽ gây exception và worker sẽ thoát.

def worker_process(in_queue, out_queue, initializer=None, init_args=(), max_tasks=None):
    assert max_tasks is None or (type(max_tasks) == int and max_tasks > 0)
    put = out_queue.put
    get = in_queue.get
    if hasattr(in_queue, '_writer'):
        in_queue._writer.close()
        out_queue._reader.close()

    if initializer is not None:
        initializer(*init_args)
    completed = 0
    while max_tasks is None or (max_tasks and completed < max_tasks):
        try:
            task = get()
        except (EOFError, IOError):
            debug('worker got EOFError or IOError -- exiting')
            break

        if task is None:
            debug('worker got sentinel -- exiting')
            break

        job_id, idx, func, args, kwds = task
        try:
            result = (True, func(*args, **kwds))
        except Exception as e:
            result = (False, e)
        try:
            put((job_id, idx, result))
        except Exception as e:
            wrapped = MaybeEncodingError(e, result[1])
            debug("Possible encoding error while sending result: %s" % (
                wrapped))
            put((job_id, idx, (False, wrapped)))
        completed += 1
    debug('worker exiting after %d tasks' % completed)

Khi worker thoát, xem đoạn code sau:

## process.py
def _bootstrap(self):
        from . import util
        global _current_process

        try:
            self._children = set()
            self._counter = itertools.count(1)
            try:
                sys.stdin.close()
                sys.stdin = open(os.devnull)
            except (OSError, ValueError):
                pass
            _current_process = self
            util._finalizer_registry.clear()
            util._run_after_forkers()
            util.info('child process calling self.run()')
            try:
                self.run()
                exitcode = 0
            finally:
                util._exit_function()

Tiến trình con sẽ gọi hàm run() và kết thúc, sau đó gọi _exit_function() để dọn dẹp và gọi _run_finalizers() để kết thúc tiến trình.

Tuy nhiên, nếu worker trong pool đang chạy tác vụ dài hạn không thoát, tiến trình con sẽ bị kẹt lại và tiếp tục chạy. Nếu các tác vụ đều là công việc ngắn, ngay cả khi tiến trình chính bị kill -9, các tiến trình con cũng sẽ thoát sau khi hoàn thành công việc.