FM / multiprocessing

/snap/core18/2979/usr/lib/python3.6/multiprocessing/

/snap/core18/2979/usr/lib/python3.6/multiprocessing

Name	Size	Mode	Actions
dummy/	-	0755	rm
__pycache__/	-	0755	rm
connection.py	30891	0644	edit dl rm
context.py	10917	0644	edit dl rm
forkserver.py	8694	0644	edit dl rm
heap.py	8319	0644	edit dl rm
managers.py	38151	0644	edit dl rm
pool.py	26059	0644	edit dl rm
popen_fork.py	2307	0644	edit dl rm
popen_forkserver.py	1956	0644	edit dl rm
popen_spawn_posix.py	1904	0644	edit dl rm
popen_spawn_win32.py	2999	0644	edit dl rm
process.py	9211	0644	edit dl rm
queues.py	10763	0644	edit dl rm
reduction.py	9226	0644	edit dl rm
resource_sharer.py	5325	0644	edit dl rm
semaphore_tracker.py	5394	0644	edit dl rm
sharedctypes.py	6245	0644	edit dl rm
spawn.py	8863	0644	edit dl rm
synchronize.py	12050	0644	edit dl rm
util.py	11886	0644	edit dl rm
__init__.py	923	0644	edit dl rm

Edit: /snap/core18/2979/usr/lib/python3.6/multiprocessing/semaphore_tracker.py (5394B)

#
# On Unix we run a server process which keeps track of unlinked
# semaphores. The server ignores SIGINT and SIGTERM and reads from a
# pipe.  Every other process of the program has a copy of the writable
# end of the pipe, so we get EOF when all other processes have exited.
# Then the server process unlinks any remaining semaphore names.
#
# This is important because the system only supports a limited number
# of named semaphores, and they will not be automatically removed till
# the next reboot.  Without this semaphore tracker process, "killall
# python" would probably leave unlinked semaphores.
#

import os
import signal
import sys
import threading
import warnings
import _multiprocessing

from . import spawn
from . import util

__all__ = ['ensure_running', 'register', 'unregister']

class SemaphoreTracker(object):

def __init__(self):
        self._lock = threading.Lock()
        self._fd = None
        self._pid = None

def getfd(self):
        self.ensure_running()
        return self._fd

def ensure_running(self):
        '''Make sure that semaphore tracker process is running.

This can be run from any process.  Usually a child process will use
        the semaphore created by its parent.'''
        with self._lock:
            if self._pid is not None:
                # semaphore tracker was launched before, is it still running?
                pid, status = os.waitpid(self._pid, os.WNOHANG)
                if not pid:
                    # => still alive
                    return
                # => dead, launch it again
                os.close(self._fd)
                self._fd = None
                self._pid = None

warnings.warn('semaphore_tracker: process died unexpectedly, '
                              'relaunching.  Some semaphores might leak.')

fds_to_pass = []
            try:
                fds_to_pass.append(sys.stderr.fileno())
            except Exception:
                pass
            cmd = 'from multiprocessing.semaphore_tracker import main;main(%d)'
            r, w = os.pipe()
            try:
                fds_to_pass.append(r)
                # process will out live us, so no need to wait on pid
                exe = spawn.get_executable()
                args = [exe] + util._args_from_interpreter_flags()
                args += ['-c', cmd % r]
                pid = util.spawnv_passfds(exe, args, fds_to_pass)
            except:
                os.close(w)
                raise
            else:
                self._fd = w
                self._pid = pid
            finally:
                os.close(r)

def register(self, name):
        '''Register name of semaphore with semaphore tracker.'''
        self._send('REGISTER', name)

def unregister(self, name):
        '''Unregister name of semaphore with semaphore tracker.'''
        self._send('UNREGISTER', name)

def _send(self, cmd, name):
        self.ensure_running()
        msg = '{0}:{1}\n'.format(cmd, name).encode('ascii')
        if len(name) > 512:
            # posix guarantees that writes to a pipe of less than PIPE_BUF
            # bytes are atomic, and that PIPE_BUF >= 512
            raise ValueError('name too long')
        nbytes = os.write(self._fd, msg)
        assert nbytes == len(msg)

_semaphore_tracker = SemaphoreTracker()
ensure_running = _semaphore_tracker.ensure_running
register = _semaphore_tracker.register
unregister = _semaphore_tracker.unregister
getfd = _semaphore_tracker.getfd

def main(fd):
    '''Run semaphore tracker.'''
    # protect the process from ^C and "killall python" etc
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)

for f in (sys.stdin, sys.stdout):
        try:
            f.close()
        except Exception:
            pass

cache = set()
    try:
        # keep track of registered/unregistered semaphores
        with open(fd, 'rb') as f:
            for line in f:
                try:
                    cmd, name = line.strip().split(b':')
                    if cmd == b'REGISTER':
                        cache.add(name)
                    elif cmd == b'UNREGISTER':
                        cache.remove(name)
                    else:
                        raise RuntimeError('unrecognized command %r' % cmd)
                except Exception:
                    try:
                        sys.excepthook(*sys.exc_info())
                    except:
                        pass
    finally:
        # all processes have terminated; cleanup any remaining semaphores
        if cache:
            try:
                warnings.warn('semaphore_tracker: There appear to be %d '
                              'leaked semaphores to clean up at shutdown' %
                              len(cache))
            except Exception:
                pass
        for name in cache:
            # For some reason the process which created and registered this
            # semaphore has failed to unregister it. Presumably it has died.
            # We therefore unlink it.
            try:
                name = name.decode('ascii')
                try:
                    _multiprocessing.sem_unlink(name)
                except Exception as e:
                    warnings.warn('semaphore_tracker: %r: %s' % (name, e))
            finally:
                pass

cmd: