Process and Thread resource recycling issue

[Han-Jiangtao]

Summary

For a child process with daemon=False creates the multiprocessing.Manager object, if it creates a thread for polling SyncManager-created Lock/Event objects, the default recycling mechanism will first release the grandchild process corresponding to the created multiprocessing.Manager object, which can interrupt the polling thread and raise Error.

Description

The test code is as follows:

import time
import queue
import threading
import multiprocessing

class GlobalMgr(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.mgr = multiprocessing.Manager()
        self.task_q = self.mgr.Queue()
        self.stop_event = threading.Event()

    def run(self):
        while not self.stop_event.is_set():
            try:
                self.task_q.get(timeout=0.001)
            except queue.Empty as e:
                continue

def subprocess():
    mgr = GlobalMgr(daemon=True)
    mgr.start()
    time.sleep(1)

if __name__ == "__main__":
    process = multiprocessing.Process(target=subprocess, daemon=False)
    process.start()
    process.join()

The above test code will throw the following exception:

Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1009, in _bootstrap_inner
multiprocessing/process.py 318 source exit
    self.run()
  File "/home/hanjiangtao/workspace/program_learning/python_workspace/mp_resource_manager.py", line 18, in run
    self.task_q.get(timeout=0.001)
  File "<string>", line 2, in get
  File "/usr/lib/python3.10/multiprocessing/managers.py", line 833, in _callmethod
    raise convert_to_error(kind, result)
multiprocessing.managers.RemoteError: 
---------------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/lib/python3.10/multiprocessing/managers.py", line 260, in serve_client
    self.id_to_local_proxy_obj[ident]
KeyError: '7fcff61b4d00'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.10/multiprocessing/managers.py", line 262, in serve_client
    raise ke
  File "/usr/lib/python3.10/multiprocessing/managers.py", line 256, in serve_client
    obj, exposed, gettypeid = id_to_obj[ident]
KeyError: '7fcff61b4d00'
---------------------------------------------------------------------------

After debugging, it was found that the main cause of the above exception was the invocation of BaseProcess._bootstrap after the subprocess was created and started.While releasing the program resources, that is, when the process object try to exit, the finally brach in BaseProcess._bootstrap will be taken. Which leds to calling util._exit_function in advance. The util._exit_function method will futher call the BaseManager._finalize_manager registered in the BaseManger.start method in advance, and this will trigger the exception during the runtime of thread for the accidental release of SyncManager object.

However, the multiprocessing.util module registers util._exit_function in the atexit module automatically when it is imported, which means that util._exit_function will be called naturally while subprocess exits. Consequently, can we consider the active invocation of BaseProcess._bootstrap's finally branch as unnecessary?

Of course, after learning the multiprocessing.util module, we can avoid the issue by the following:

import time
import queue
import threading
import multiprocessing


class GlobalMgr(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.mgr = multiprocessing.Manager()
        self.task_q = self.mgr.Queue()
        self.stop_event = threading.Event()

    def run(self):
        while not self.stop_event.is_set():
            try:
                self.task_q.get(timeout=0.001)
            except queue.Empty as e:
                continue

def release(mgr):
    mgr.stop_event.set()
    mgr.join()

def subprocess():
    mgr = GlobalMgr(daemon=True)
    mgr.start()
    # Actively registers destructors for high-priority _exit_function calls
    multiprocessing.util.Finalize(None, release, args=(mgr,), exitpriority=100)
    time.sleep(1)

if __name__ == "__main__":
    process = multiprocessing.Process(target=subprocess, daemon=False)
    process.start()
    process.join()

The multiprocessing.util.Finalize module is not an externally visible method, so it's only a hedge.'

Environment

Python 3.10.4

Ubuntu x86_64

[gaogaotiantian]

I believe atexit module won't be triggered in forked process, as os._exit(retcode) is used to exit the program.

However, this does not seem like a resource release order issue. I think the problem is that you never appropriately end your thread.

What is the expected behavior of the code? Your thread is trying to get data from a queue unconditionally and the only way to stop it is for the process to finish. However, the process has to clean up the SyncManager - the data source of your thread - before it exits. That creates an impossible loop for your code to work properly.

The solution of your issue does not have to be as complicated as you provided. You can simply do release(mgr) at the end of your subprocess function, which is a reasonable thing to do - to stop and finish the thread before your process ends.

[Han-Jiangtao]

tks