Python进程卡住排查

import multiprocessing, threading, logging, sys, traceback
import time
from logging import StreamHandler

SLEEP_TIME = 0.000001

logger = logging.getLogger()

ch = logging.StreamHandler()
ch.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(funcName)s(): %(message)s'))
ch.setLevel(logging.DEBUG)

logger.setLevel(logging.DEBUG)
logger.addHandler(ch)


class LoggingThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while True:
            logger.debug('LoggingThread: {}'.format(self))
            time.sleep(SLEEP_TIME)


def action(i):
    logger.debug('action: {}'.format(i))


def do_parallel_job():

    processes = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes=processes)
    for i in range(20):
        pool.apply_async(action, args=(i,))
    pool.close()
    pool.join()


if __name__ == '__main__':

    logger.debug('START')

    #
    # multithread part
    #
    for _ in range(10):
        lt = LoggingThread()
        lt.setDaemon(True)
        lt.start()

    #
    # multiprocess part
    #
    do_parallel_job()

    logger.debug('FINISH')

def os_forkpty_wrapper():
    """Wraps os.forkpty() to run atfork handlers."""
    pid = None
    prepare_to_fork_acquire()
    try:
        pid, fd = _orig_os_forkpty()
    finally:
        if pid == 0:
            child_after_fork_release()
        else:
            parent_after_fork_release()
    return pid, fd


def atfork(prepare=None, parent=None, child=None):
    """A Python work-a-like of pthread_atfork.

    Any time a fork() is called from Python, all 'prepare' callables will
    be called in the order they were registered using this function.

    After the fork (successful or not), all 'parent' callables will be called in
    the parent process.  If the fork succeeded, all 'child' callables will be
    called in the child process.

    No exceptions may be raised from any of the registered callables.  If so
    they will be printed to sys.stderr after the fork call once it is safe
    to do so.
    """
    assert not prepare or callable(prepare)
    assert not parent or callable(parent)
    assert not child or callable(child)
    _fork_lock.acquire()
    try:
        if prepare:
            _prepare_call_list.append(prepare)
        if parent:
            _parent_call_list.append(parent)
        if child:
            _child_call_list.append(child)
    finally:
        _fork_lock.release()


# 以下截取对logging模块的patch部分

def fork_safe_createLock(self):
    self._orig_createLock()
    atfork(self.lock.acquire, self.lock.release, self.lock.release)

# Fix the logging.Handler lock (a major source of deadlocks).
logging.Handler._orig_createLock = logging.Handler.createLock
logging.Handler.createLock = fork_safe_createLock

# Fix the module level lock.
atfork(logging._acquireLock,
                logging._releaseLock, logging._releaseLock)

monkeypatch_os_fork_functions()
fix_logging_module()

os.register_at_fork(*, before=None, after_in_parent=None, after_in_child=None)

Register callables to be executed when a new child process is forked using os.fork() or similar process cloning APIs. The parameters are optional and keyword-only. Each specifies a different call point.

 - before is a function called before forking a child process.
 - after_in_parent is a function called from the parent process after forking a child process.
 - after_in_child is a function called from the child process

ch = logging.StreamHandler()
ch.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(funcName)s(): %(message)s'))
ch.setLevel(logging.DEBUG)
ch.createLock()