# `multiprocessing` — Process-based parallelism

Вихідний код: Lib/multiprocessing/

Availability: not Emscripten, not WASI.

This module does not work or is not available on WebAssembly platforms `wasm32-emscripten` and `wasm32-wasi`. See WebAssembly platforms for more information.

## Вступ

`multiprocessing` is a package that supports spawning processes using an API similar to the `threading` module. The `multiprocessing` package offers both local and remote concurrency, effectively side-stepping the Global Interpreter Lock by using subprocesses instead of threads. Due to this, the `multiprocessing` module allows the programmer to fully leverage multiple processors on a given machine. It runs on both Unix and Windows.

Модуль `multiprocessing` також представляє API, які не мають аналогів у модулі `threading`. Яскравим прикладом цього є об’єкт `Pool`, який пропонує зручний засіб розпаралелювання виконання функції для кількох вхідних значень, розподіляючи вхідні дані між процесами (паралелізм даних). Наступний приклад демонструє звичайну практику визначення таких функцій у модулі, щоб дочірні процеси могли успішно імпортувати цей модуль. Цей базовий приклад паралелізму даних з використанням `Pool`,

    from multiprocessing import Pool

    def f(x):
        return x*x

    if __name__ == '__main__':
        with Pool(5) as p:
            print(p.map(f, [1, 2, 3]))

друкуватиме стандартний вихід

    [1, 4, 9]

Дивись також

`concurrent.futures.ProcessPoolExecutor` offers a higher level interface to push tasks to a background process without blocking execution of the calling process. Compared to using the `Pool` interface directly, the `concurrent.futures` API more readily allows the submission of work to the underlying process pool to be separated from waiting for the results.

### Клас `Process`

У `multiprocessing` процеси породжуються шляхом створення об’єкта `Process` і виклику його методу `start()`. `Process` відповідає API `threading.Thread`. Тривіальним прикладом багатопроцесорної програми є:

    from multiprocessing import Process

    def f(name):
        print('hello', name)

    if __name__ == '__main__':
        p = Process(target=f, args=('bob',))
        p.start()
        p.join()

Щоб показати ідентифікатори окремих процесів, ось розширений приклад:

    from multiprocessing import Process
    import os

    def info(title):
        print(title)
        print('module name:', __name__)
        print('parent process:', os.getppid())
        print('process id:', os.getpid())

    def f(name):
        info('function f')
        print('hello', name)

    if __name__ == '__main__':
        info('main line')
        p = Process(target=f, args=('bob',))
        p.start()
        p.join()

Щоб отримати пояснення, чому потрібна частина `if __name__ == '__main__''`, див. Інструкції з програмування.

### Контексти та методи запуску

Залежно від платформи `multiprocessing` підтримує три способи запуску процесу. Це методи запуску

> спаун

>   The parent process starts a fresh Python interpreter process. The child process will only inherit those resources necessary to run the process object’s `run()` method. In particular, unnecessary file descriptors and handles from the parent process will not be inherited. Starting a process using this method is rather slow compared to using fork or forkserver.

>   Available on Unix and Windows. The default on Windows and macOS.

> вилка

>   Батьківський процес використовує `os.fork()` для розгалуження інтерпретатора Python. Дочірній процес, коли він починається, фактично ідентичний батьківському процесу. Усі ресурси батьківського процесу успадковуються дочірнім процесом. Зверніть увагу, що безпечне розгалуження багатопотокового процесу є проблематичним.

>   Available on Unix only. The default on Unix.

> форксервер

>   When the program starts and selects the forkserver start method, a server process is started. From then on, whenever a new process is needed, the parent process connects to the server and requests that it fork a new process. The fork server process is single threaded so it is safe for it to use `os.fork()`. No unnecessary resources are inherited.

>   Available on Unix platforms which support passing file descriptors over Unix pipes.

Змінено в версії 3.4: spawn added on all Unix platforms, and forkserver added for some Unix platforms. Child processes no longer inherit all of the parents inheritable handles on Windows.

Змінено в версії 3.8: On macOS, the spawn start method is now the default. The fork start method should be considered unsafe as it can lead to crashes of the subprocess as macOS system libraries may start threads. See bpo-33725.

On POSIX using the spawn or forkserver start methods will also start a resource tracker process which tracks the unlinked named system resources (such as named semaphores or `SharedMemory` objects) created by processes of the program. When all processes have exited the resource tracker unlinks any remaining tracked object. Usually there should be none, but if a process was killed by a signal there may be some «leaked» resources. (Neither leaked semaphores nor shared memory segments will be automatically unlinked until the next reboot. This is problematic for both objects because the system allows only a limited number of named semaphores, and shared memory segments occupy some space in the main memory.)

Щоб вибрати метод запуску, ви використовуєте `set_start_method()` в пункті `if __name__ == '__main__'` головного модуля. Наприклад:

    import multiprocessing as mp

    def foo(q):
        q.put('hello')

    if __name__ == '__main__':
        mp.set_start_method('spawn')
        q = mp.Queue()
        p = mp.Process(target=foo, args=(q,))
        p.start()
        print(q.get())
        p.join()

`set_start_method()` не слід використовувати більше одного разу в програмі.

Крім того, ви можете використовувати `get_context()`, щоб отримати об’єкт контексту. Контекстні об’єкти мають той самий API, що й багатопроцесорний модуль, і дозволяють використовувати кілька методів запуску в одній програмі.

    import multiprocessing as mp

    def foo(q):
        q.put('hello')

    if __name__ == '__main__':
        ctx = mp.get_context('spawn')
        q = ctx.Queue()
        p = ctx.Process(target=foo, args=(q,))
        p.start()
        print(q.get())
        p.join()

Зауважте, що об’єкти, пов’язані з одним контекстом, можуть бути несумісними з процесами для іншого контексту. Зокрема, блокування, створені за допомогою контексту fork, не можна передати процесам, запущеним за допомогою методів запуску spawn або forkserver.

Бібліотека, яка хоче використовувати певний метод запуску, ймовірно, повинна використовувати `get_context()`, щоб уникнути втручання у вибір користувача бібліотеки.

Попередження

The `'spawn'` and `'forkserver'` start methods cannot currently be used with «frozen» executables (i.e., binaries produced by packages like PyInstaller and cx_Freeze) on Unix. The `'fork'` start method does work.

### Обмін об’єктами між процесами

`multiprocessing` підтримує два типи каналів зв’язку між процесами:

Черги

> Клас `Queue` є майже клоном `queue.Queue`. Наприклад:

>     from multiprocessing import Process, Queue

>     def f(q):
>         q.put([42, None, 'hello'])

>     if __name__ == '__main__':
>         q = Queue()
>         p = Process(target=f, args=(q,))
>         p.start()
>         print(q.get())    # prints "[42, None, 'hello']"
>         p.join()

> Queues are thread and process safe.

Труби

> Функція `Pipe()` повертає пару об’єктів з’єднання, з’єднаних трубою, яка за умовчанням є дуплексною (двосторонньою). Наприклад:

>     from multiprocessing import Process, Pipe

>     def f(conn):
>         conn.send([42, None, 'hello'])
>         conn.close()

>     if __name__ == '__main__':
>         parent_conn, child_conn = Pipe()
>         p = Process(target=f, args=(child_conn,))
>         p.start()
>         print(parent_conn.recv())   # prints "[42, None, 'hello']"
>         p.join()

> Два об’єкти з’єднання, які повертає `Pipe()`, представляють два кінці труби. Кожен об’єкт підключення має методи `send()` і `recv()` (серед інших). Зауважте, що дані в каналі можуть бути пошкоджені, якщо два процеси (або потоки) намагаються зчитувати або писати в той самий кінець каналу одночасно. Звичайно, немає ризику пошкодження через процеси, що використовують різні кінці труби одночасно.

### Синхронізація між процесами

`multiprocessing` містить еквіваленти всіх примітивів синхронізації з `threading`. Наприклад, можна використовувати блокування, щоб гарантувати, що лише один процес друкує на стандартний вивід одночасно:

    from multiprocessing import Process, Lock

    def f(l, i):
        l.acquire()
        try:
            print('hello world', i)
        finally:
            l.release()

    if __name__ == '__main__':
        lock = Lock()

        for num in range(10):
            Process(target=f, args=(lock, num)).start()

Без використання блокування вихідні дані різних процесів можуть переплутатися.

### Спільне використання стану між процесами

Як згадувалося вище, під час паралельного програмування зазвичай краще уникати використання спільного стану, наскільки це можливо. Це особливо актуально при використанні кількох процесів.

Однак, якщо вам справді потрібно використовувати спільні дані, тоді `multiprocessing` надає кілька способів зробити це.

Спільна пам’ять

> Дані можна зберігати в спільній карті пам’яті за допомогою `Value` або `Array`. Наприклад, такий код:

>     from multiprocessing import Process, Value, Array

>     def f(n, a):
>         n.value = 3.1415927
>         for i in range(len(a)):
>             a[i] = -a[i]

>     if __name__ == '__main__':
>         num = Value('d', 0.0)
>         arr = Array('i', range(10))

>         p = Process(target=f, args=(num, arr))
>         p.start()
>         p.join()

>         print(num.value)
>         print(arr[:])

> надрукую

>     3.1415927
>     [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

> Аргументи `'d'` і `'i'`, що використовуються під час створення `num` і `arr`, є кодами типу, які використовуються модулем `array`: `' d''` вказує на число з плаваючою точкою подвійної точності, а `'i''` вказує на ціле число зі знаком. Ці спільні об’єкти будуть безпечними для процесу та потоків.

> Для більшої гнучкості використання спільної пам’яті можна використовувати модуль `multiprocessing.sharedctypes`, який підтримує створення довільних об’єктів ctypes, виділених із спільної пам’яті.

Серверний процес

> Об’єкт менеджера, який повертає `Manager()`, контролює серверний процес, який містить об’єкти Python і дозволяє іншим процесам маніпулювати ними за допомогою проксі-серверів.

> Менеджер, який повертає `Manager()`, підтримуватиме типи `list`, `dict`, `Namespace`, `Lock`, `RLock` і `Semaphore`, `BoundedSemaphore`, `Condition`, `Event`, `Barrier`, `Queue`, `Value` and `Array`. Наприклад,

>     from multiprocessing import Process, Manager

>     def f(d, l):
>         d[1] = '1'
>         d['2'] = 2
>         d[0.25] = None
>         l.reverse()

>     if __name__ == '__main__':
>         with Manager() as manager:
>             d = manager.dict()
>             l = manager.list(range(10))

>             p = Process(target=f, args=(d, l))
>             p.start()
>             p.join()

>             print(d)
>             print(l)

> надрукую

>     {0.25: None, 1: '1', '2': 2}
>     [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

> Менеджери серверних процесів є більш гнучкими, ніж використання спільних об’єктів пам’яті, оскільки вони можуть підтримувати довільні типи об’єктів. Крім того, один менеджер може спільно використовуватися процесами на різних комп’ютерах у мережі. Однак вони повільніші, ніж використання спільної пам’яті.

### Використання пулу працівників

Клас `Pool` представляє пул робочих процесів. Він має методи, які дозволяють розвантажувати завдання на робочі процеси кількома різними способами.

Наприклад:

    from multiprocessing import Pool, TimeoutError
    import time
    import os

    def f(x):
        return x*x

    if __name__ == '__main__':
        # start 4 worker processes
        with Pool(processes=4) as pool:

            # print "[0, 1, 4,..., 81]"
            print(pool.map(f, range(10)))

            # print same numbers in arbitrary order
            for i in pool.imap_unordered(f, range(10)):
                print(i)

            # evaluate "f(20)" asynchronously
            res = pool.apply_async(f, (20,))      # runs in *only* one process
            print(res.get(timeout=1))             # prints "400"

            # evaluate "os.getpid()" asynchronously
            res = pool.apply_async(os.getpid, ()) # runs in *only* one process
            print(res.get(timeout=1))             # prints the PID of that process

            # launching multiple evaluations asynchronously *may* use more processes
            multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
            print([res.get(timeout=1) for res in multiple_results])

            # make a single worker sleep for 10 seconds
            res = pool.apply_async(time.sleep, (10,))
            try:
                print(res.get(timeout=1))
            except TimeoutError:
                print("We lacked patience and got a multiprocessing.TimeoutError")

            print("For the moment, the pool remains available for more work")

        # exiting the 'with'-block has stopped the pool
        print("Now the pool is closed and no longer available")

Зверніть увагу, що методи пулу повинні використовуватися тільки тим процесом, який його створив.

Примітка

Функціональність цього пакета вимагає, щоб модуль `__main__` міг імпортуватися дочірніми елементами. Це описано в Інструкції з програмування, проте тут варто звернути увагу на це. Це означає, що деякі приклади, такі як приклади `multiprocessing.pool.Pool` не працюватимуть в інтерактивному інтерпретаторі. Наприклад:

    >>> from multiprocessing import Pool
    >>> p = Pool(5)
    >>> def f(x):
    ...     return x*x
    ...
    >>> with p:
    ...     p.map(f, [1,2,3])
    Process PoolWorker-1:
    Process PoolWorker-2:
    Process PoolWorker-3:
    Traceback (most recent call last):
    Traceback (most recent call last):
    Traceback (most recent call last):
    AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>
    AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>
    AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>

(Якщо ви спробуєте це, він фактично виведе три повні трасування, чергувані напіввипадковим чином, і тоді вам, можливо, доведеться якось зупинити батьківський процес.)

## довідка

Пакет `multiprocessing` здебільшого повторює API модуля `threading`.

### `Process` і винятки

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

  Об’єкти процесу представляють діяльність, яка виконується в окремому процесі. Клас `Process` має еквіваленти всіх методів `threading.Thread`.

  The constructor should always be called with keyword arguments. group should always be `None`; it exists solely for compatibility with `threading.Thread`. target is the callable object to be invoked by the `run()` method. It defaults to `None`, meaning nothing is called. name is the process name (see `name` for more details). args is the argument tuple for the target invocation. kwargs is a dictionary of keyword arguments for the target invocation. If provided, the keyword-only daemon argument sets the process `daemon` flag to `True` or `False`. If `None` (the default), this flag will be inherited from the creating process.

  By default, no arguments are passed to target. The args argument, which defaults to `()`, can be used to specify a list or tuple of the arguments to pass to target.

  Якщо підклас перевизначає конструктор, він повинен переконатися, що він викликає конструктор базового класу (`Process.__init__()`), перш ніж щось робити з процесом.

  Змінено в версії 3.3: Added the daemon parameter.

  run()

    Метод, що представляє діяльність процесу.

    Ви можете перевизначити цей метод у підкласі. Стандартний метод `run()` викликає викликаний об’єкт, переданий конструктору об’єкта як цільовий аргумент, якщо такий є, з послідовними аргументами та ключовими аргументами, взятими з аргументів args і kwargs відповідно.

    Using a list or tuple as the args argument passed to `Process` achieves the same effect.

    Приклад:

        >>> from multiprocessing import Process
        >>> p = Process(target=print, args=[1])
        >>> p.run()
        1
        >>> p = Process(target=print, args=(1,))
        >>> p.run()
        1

  start()

    Запустіть процес.

    Це має бути викликано щонайбільше один раз на об’єкт процесу. Він організовує виклик методу `run()` об’єкта в окремому процесі.

  join([timeout])

    Якщо додатковий аргумент timeout має значення `None` (за замовчуванням), метод блокується, доки процес, чий метод `join()` викликається, не завершиться. Якщо timeout є додатним числом, воно блокує щонайбільше timeout секунд. Зауважте, що метод повертає `None`, якщо його процес завершується або якщо метод закінчився. Перевірте `exitcode` процесу, щоб визначити, чи він завершився.

    До процесу можна приєднуватися багато разів.

    Процес не може приєднатися до себе, оскільки це спричинить тупикову блокування. Спроба приєднатися до процесу до його запуску є помилкою.

  name

    Назва процесу. Ім’я - це рядок, який використовується лише для ідентифікації. Він не має семантики. Кілька процесів можуть мати однакові назви.

    Початкове ім’я задається конструктором. Якщо конструктору не надано явної назви, ім’я у формі „Process-N1:N2:…:Nk“ побудований, де кожен Nk є N-м дочірнім елементом свого батька.

  is_alive()

    Повернути, чи процес активний.

    Приблизно, об’єкт процесу живий з моменту повернення методу `start()` до завершення дочірнього процесу.

  daemon

    Прапор демона процесу, логічне значення. Це має бути встановлено перед викликом `start()`.

    Початкове значення успадковується від процесу створення.

    Коли процес завершується, він намагається завершити всі свої демонічні дочірні процеси.

    Зауважте, що демонічному процесу не дозволяється створювати дочірні процеси. Інакше демонічний процес залишив би своїх нащадків сиротами, якщо він буде припинений під час завершення процесу батьківського процесу. Крім того, це не демони чи служби Unix, це звичайні процеси, які будуть припинені (і не приєднані), якщо недемонічні процеси вийшли.

  Окрім API `threading.Thread`, об’єкти `Process` також підтримують такі атрибути та методи:

  pid

    Поверніть ідентифікатор процесу. До того, як процес буде створено, це буде `None`.

  exitcode

    Код виходу дитини. Це буде `None`, якщо процес ще не завершено.

    Якщо дочірній метод `run()` повернувся нормально, код виходу буде 0. Якщо він закінчився через `sys.exit()` із цілим аргументом N, код виходу буде N.

    Якщо дочірній процес завершився через виняток, який не було перехоплено в `run()`, код виходу буде 1. Якщо його було завершено сигналом N, код виходу матиме від’ємне значення -N.

  authkey

    Ключ автентифікації процесу (рядок байтів).

    Коли `multiprocessing` ініціалізовано, головному процесу призначається випадковий рядок за допомогою `os.urandom()`.

    Коли об’єкт `Process` створюється, він успадковує ключ автентифікації свого батьківського процесу, хоча це можна змінити, встановивши `authkey` інший байтовий рядок.

    Перегляньте Ключі автентифікації.

  sentinel

    Числовий дескриптор системного об’єкта, який стане «готовим» після завершення процесу.

    You can use this value if you want to wait on several events at once using `multiprocessing.connection.wait()`. Otherwise calling `join()` is simpler.

    On Windows, this is an OS handle usable with the `WaitForSingleObject` and `WaitForMultipleObjects` family of API calls. On Unix, this is a file descriptor usable with primitives from the `select` module.

    Нове в версії 3.3.

  terminate()

    Terminate the process. On POSIX this is done using the `SIGTERM` signal; on Windows `TerminateProcess()` is used. Note that exit handlers and finally clauses, etc., will not be executed.

    Зверніть увагу, що процеси-нащадки процесу не будуть припинені – вони просто стануть сиротами.

    Попередження

    Якщо цей метод використовується, коли пов’язаний процес використовує канал або чергу, канал або чергу можуть бути пошкоджені та можуть стати непридатними для використання іншим процесом. Подібним чином, якщо процес отримав блокування або семафор тощо, його завершення може призвести до блокування інших процесів.

  kill()

    Same as `terminate()` but using the `SIGKILL` signal on Unix.

    Нове в версії 3.7.

  close()

    Закрийте об’єкт `Process`, звільнивши всі пов’язані з ним ресурси. `ValueError` виникає, якщо основний процес все ще виконується. Після успішного повернення `close()` більшість інших методів і атрибутів об’єкта `Process` викличуть `ValueError`.

    Нове в версії 3.7.

  Зауважте, що методи `start()`, `join()`, `is_alive()`, `terminate()` і `exitcode` має викликати лише процес, який створив об’єкт процесу .

  Приклад використання деяких методів `Process`:

      >>> import multiprocessing, time, signal
      >>> p = multiprocessing.Process(target=time.sleep, args=(1000,))
      >>> print(p, p.is_alive())
      <Process ... initial> False
      >>> p.start()
      >>> print(p, p.is_alive())
      <Process ... started> True
      >>> p.terminate()
      >>> time.sleep(0.1)
      >>> print(p, p.is_alive())
      <Process ... stopped exitcode=-SIGTERM> False
      >>> p.exitcode == -signal.SIGTERM
      True

exception multiprocessing.ProcessError

  Базовий клас усіх винятків `multiprocessing`.

exception multiprocessing.BufferTooShort

  Exception raised by `Connection.recv_bytes_into()` when the supplied buffer object is too small for the message read.

  Якщо `e` є екземпляром `BufferTooShort`, то `e.args[0]` видасть повідомлення як рядок байтів.

exception multiprocessing.AuthenticationError

  Викликається, коли виникає помилка автентифікації.

exception multiprocessing.TimeoutError

  Викликається методами з тайм-аутом, коли час очікування закінчується.

### Труби та черги

При використанні кількох процесів зазвичай використовується передача повідомлень для зв’язку між процесами та уникається використання будь-яких примітивів синхронізації, таких як блокування.

Для передачі повідомлень можна використовувати `Pipe()` (для з’єднання між двома процесами) або чергу (що дозволяє використовувати кілька виробників і споживачів).

Типи `Queue`, `SimpleQueue` і `JoinableQueue` є чергами багатьох виробників і споживачів FIFO, змодельованими на основі `queue.Queue` клас у стандартній бібліотеці. Вони відрізняються тим, що `Queue` не має методів `task_done()` і `join()`, представлених у `queue.Queue` Python 2.5 клас.

Якщо ви використовуєте `JoinableQueue`, тоді ви має викликати `JoinableQueue.task_done()` для кожного завдання, вилученого з черги, інакше семафор, який використовується для підрахунку кількості незавершених завдань, може зрештою переповнитися, викликаючи виняток.

Зауважте, що можна також створити спільну чергу за допомогою об’єкта менеджера – див. Менеджери.

Примітка

`multiprocessing` використовує звичайні винятки `queue.Empty` і `queue.Full`, щоб сигналізувати про час очікування. Вони недоступні в просторі імен `multiprocessing`, тому їх потрібно імпортувати з `queue`.

Примітка

Коли об’єкт ставиться в чергу, об’єкт очищається, а фоновий потік пізніше скидає обрані дані в базовий канал. Це має деякі наслідки, які є трохи дивними, але не повинні викликати жодних практичних труднощів - якщо вони дійсно вас турбують, ви можете натомість використати чергу, створену за допомогою manager.

1. Після розміщення об’єкта в порожній черзі може виникнути нескінченно мала затримка, перш ніж метод `empty()` черги поверне `False`, а `get_nowait()` зможе повернутися без виклику `queue.Empty`.

2. Якщо кілька процесів ставлять об’єкти в чергу, об’єкти можуть бути отримані на іншому кінці не за порядком. Однак об’єкти, поставлені в чергу одним і тим же процесом, завжди будуть в очікуваному порядку один відносно одного.

Попередження

Якщо процес зупинено за допомогою `Process.terminate()` або `os.kill()` під час спроби використання `Queue`, то дані в черзі, ймовірно, будуть пошкоджені. Це може призвести до того, що будь-який інший процес отримає виняток, коли він спробує використати чергу пізніше.

Попередження

Як згадувалося вище, якщо дочірній процес поставив елементи в чергу (і він не використовував `JoinableQueue.cancel_join_thread`), тоді цей процес не завершиться, доки всі буферизовані елементи не будуть скинуті в канал.

Це означає, що якщо ви спробуєте приєднатися до цього процесу, ви можете отримати тупикову блокування, якщо ви не впевнені, що всі елементи, які були поставлені в чергу, використано. Так само, якщо дочірній процес є недемонічним, тоді батьківський процес може зависнути при виході, коли він намагається приєднатися до всіх своїх недемонічних дочірніх процесів.

Зауважте, що черга, створена за допомогою менеджера, не має цієї проблеми. Дивіться Інструкції з програмування.

Для прикладу використання черг для міжпроцесного зв’язку див. Приклади.

multiprocessing.Pipe([duplex])

  Повертає пару об’єктів `(conn1, conn2)` `Connection`, що представляють кінці труби.

  Якщо duplex має значення `True` (за замовчуванням), тоді канал є двонаправленим. Якщо duplex має значення `False`, тоді канал є односпрямованим: `conn1` можна використовувати лише для отримання повідомлень, а `conn2` — лише для надсилання повідомлень.

class multiprocessing.Queue([maxsize])

  Повертає спільну чергу процесу, реалізовану за допомогою каналу та кількох блокувань/семафорів. Коли процес вперше ставить елемент у чергу, запускається потік, який передає об’єкти з буфера в канал.

  Звичайні винятки `queue.Empty` і `queue.Full` із модуля `queue` стандартної бібліотеки створюються, щоб повідомити про час очікування.

  `Queue` реалізує всі методи `queue.Queue`, крім `task_done()` і `join()`.

  qsize()

    Повертає приблизний розмір черги. Через семантику багатопоточності/багатопроцесорності це число ненадійне.

    Note that this may raise `NotImplementedError` on Unix platforms like macOS where `sem_getvalue()` is not implemented.

  empty()

    Повертає `True`, якщо черга порожня, `False` інакше. Через семантику багатопоточності/багатопроцесорності це ненадійно.

  full()

    Повертає `True`, якщо черга заповнена, `False` інакше. Через семантику багатопоточності/багатопроцесорності це ненадійно.

  put(obj[, block[, timeout]])

    Помістіть obj у чергу. Якщо необов’язковий аргумент block має значення `True` (за замовчуванням), а timeout має значення `None` (за замовчуванням), за потреби блокуйте, доки не з’явиться вільний слот. Якщо timeout є додатним числом, він блокує щонайбільше timeout секунд і викликає виняток `queue.Full`, якщо протягом цього часу не було вільного місця. В іншому випадку (block має значення `False`), помістіть елемент у чергу, якщо вільний слот є негайно доступним, інакше викликайте виняток `queue.Full` (timeout у цьому випадку ігнорується).

    Змінено в версії 3.8: Якщо чергу закрито, замість `AssertionError` виникає `ValueError`.

  put_nowait(obj)

    Еквівалент `put(obj, False)`.

  get([block[, timeout]])

    Видалити та повернути елемент із черги. Якщо необов’язкові аргументи block мають значення `True` (за замовчуванням), а timeout — `None` (за замовчуванням), за потреби блокуйте, доки елемент не стане доступним. Якщо timeout є додатним числом, він блокує щонайбільше timeout секунд і викликає виняток `queue.Empty`, якщо жоден елемент не був доступний протягом цього часу. В іншому випадку (блок має значення `False`), повертає елемент, якщо він одразу доступний, інакше викликає виняток `queue.Empty` (у цьому випадку timeout ігнорується).

    Змінено в версії 3.8: Якщо чергу закрито, замість `OSError` виникає `ValueError`.

  get_nowait()

    Еквівалент `get(False)`.

  `multiprocessing.Queue` має кілька додаткових методів, яких немає в `queue.Queue`. Ці методи зазвичай непотрібні для більшості коду:

  close()

    Укажіть, що поточний процес більше не додаватиме дані до цієї черги. Фоновий потік завершиться, коли всі буферизовані дані буде скинуто в канал. Це викликається автоматично, коли в черзі збирається сміття.

  join_thread()

    Приєднайтеся до фонової нитки. Це можна використовувати лише після виклику `close()`. Він блокується, доки не завершиться фоновий потік, гарантуючи, що всі дані в буфері скинуті в канал.

    За замовчуванням, якщо процес не є творцем черги, після виходу він спробує приєднатися до фонового потоку черги. Процес може викликати `cancel_join_thread()`, щоб змусити `join_thread()` нічого не робити.

  cancel_join_thread()

    Запобігти блокуванню `join_thread()`. Зокрема, це запобігає автоматичному приєднанню фонового потоку під час завершення процесу – див. `join_thread()`.

    Кращою назвою для цього методу може бути `allow_exit_without_flush()`. Ймовірно, це спричинить втрату даних у черзі, і вам майже напевно не потрібно буде їх використовувати. Насправді він доступний лише тоді, коли вам потрібно, щоб поточний процес завершився негайно, не чекаючи, щоб скинути дані з черги в основний канал, і ви не дбаєте про втрачені дані.

  Примітка

  Функціональність цього класу вимагає функціонуючої спільної реалізації семафора в головній операційній системі. Без нього функціональні можливості цього класу будуть вимкнені, а спроби створити екземпляр `Queue` призведуть до `ImportError`. Додаткову інформацію див. bpo-3770. Те саме стосується будь-якого зі спеціалізованих типів черги, перелічених нижче.

class multiprocessing.SimpleQueue

  Це спрощений тип `Queue`, дуже схожий на заблокований `Pipe`.

  close()

    Закрийте чергу: звільніть внутрішні ресурси.

    Після закриття чергу більше не можна використовувати. Наприклад, методи `get()`, `put()` і `empty()` більше не можна викликати.

    Нове в версії 3.9.

  empty()

    Повертає `True`, якщо черга порожня, `False` інакше.

  get()

    Видалити та повернути елемент із черги.

  put(item)

    Поставте товар в чергу.

class multiprocessing.JoinableQueue([maxsize])

  `JoinableQueue`, підклас `Queue`, це черга, яка додатково має методи `task_done()` і `join()`.

  task_done()

    Вказує на те, що завдання, яке раніше було в черзі, виконано. Використовується споживачами черги. Для кожного `get()`, який використовується для отримання завдання, наступний виклик `task_done()` повідомляє черзі, що обробку завдання завершено.

    Якщо `join()` зараз блокується, воно відновиться, коли всі елементи будуть оброблені (це означає, що виклик `task_done()` отримано для кожного елемента, який був `put()` в чергу).

    Викликає `ValueError`, якщо викликається стільки разів, скільки було елементів у черзі.

  join()

    Блокуйте, доки не буде отримано та оброблено всі елементи в черзі.

    Кількість незавершених завдань зростає щоразу, коли елемент додається до черги. Підрахунок зменшується щоразу, коли споживач викликає `task_done()`, щоб вказати, що елемент отримано та вся робота над ним завершена. Коли кількість незавершених завдань падає до нуля, `join()` розблоковується.

### Різне

multiprocessing.active_children()

  Повернути список усіх живих дітей поточного процесу.

  Виклик цього має побічний ефект «приєднання» до будь-яких процесів, які вже завершилися.

multiprocessing.cpu_count()

  Повертає кількість процесорів у системі.

  This number is not equivalent to the number of CPUs the current process can use. The number of usable CPUs can be obtained with `len(os.sched_getaffinity(0))`

  Коли кількість ЦП не може бути визначена, виникає `NotImplementedError`.

  Дивись також

  `os.cpu_count()`

multiprocessing.current_process()

  Повертає об’єкт `Process`, що відповідає поточному процесу.

  Аналог `threading.current_thread()`.

multiprocessing.parent_process()

  Повертає об’єкт `Process`, що відповідає батьківському процесу `current_process()`. Для основного процесу `parent_process` буде `None`.

  Нове в версії 3.8.

multiprocessing.freeze_support()

  Додайте підтримку, коли програма, яка використовує `multiprocessing`, була заморожена для створення виконуваного файлу Windows. (Було перевірено за допомогою py2exe, PyInstaller і cx_Freeze.)

  Цю функцію потрібно викликати відразу після рядка `if __name__ == '__main__'` головного модуля. Наприклад:

      from multiprocessing import Process, freeze_support

      def f():
          print('hello world!')

      if __name__ == '__main__':
          freeze_support()
          Process(target=f).start()

  Якщо рядок `freeze_support()` пропущено, спроба запустити заморожений виконуваний файл викличе `RuntimeError`.

  Виклик `freeze_support()` не має ефекту під час виклику в будь-якій операційній системі, крім Windows. Крім того, якщо модуль нормально запускається інтерпретатором Python у Windows (програма не була заморожена), то `freeze_support()` не має ефекту.

multiprocessing.get_all_start_methods()

  Returns a list of the supported start methods, the first of which is the default. The possible start methods are `'fork'`, `'spawn'` and `'forkserver'`. On Windows only `'spawn'` is available. On Unix `'fork'` and `'spawn'` are always supported, with `'fork'` being the default.

  Нове в версії 3.4.

multiprocessing.get_context(method=None)

  Повертає об’єкт контексту, який має ті самі атрибути, що й модуль `multiprocessing`.

  If method is `None` then the default context is returned. Otherwise method should be `'fork'`, `'spawn'`, `'forkserver'`. `ValueError` is raised if the specified start method is not available.

  Нове в версії 3.4.

multiprocessing.get_start_method(allow_none=False)

  Повертає назву методу запуску, який використовується для запуску процесів.

  Якщо метод запуску не було виправлено і allow_none має значення false, тоді метод запуску фіксується за замовчуванням і повертається ім’я. Якщо метод запуску не було виправлено і allow_none має значення true, тоді повертається `None`.

  The return value can be `'fork'`, `'spawn'`, `'forkserver'` or `None`. `'fork'` is the default on Unix, while `'spawn'` is the default on Windows and macOS.

  Нове в версії 3.4.

  Змінено в версії 3.8: У macOS метод запуску spawn тепер є типовим. Метод запуску fork слід вважати небезпечним, оскільки він може призвести до збою підпроцесу. Див. bpo-33725.

multiprocessing.set_executable(executable)

  Встановіть шлях інтерпретатора Python для використання під час запуску дочірнього процесу. (За замовчуванням використовується `sys.executable`). Вбудовувачі, ймовірно, повинні будуть зробити щось на зразок:

      set_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))

  перш ніж вони зможуть створити дочірні процеси.

  Змінено в версії 3.4: Now supported on Unix when the `'spawn'` start method is used.

  Змінено в версії 3.11: Приймає path-like object.

multiprocessing.set_start_method(method, force=False)

  Set the method which should be used to start child processes. The method argument can be `'fork'`, `'spawn'` or `'forkserver'`. Raises `RuntimeError` if the start method has already been set and force is not `True`. If method is `None` and force is `True` then the start method is set to `None`. If method is `None` and force is `False` then the context is set to the default context.

  Зверніть увагу, що це має бути викликано щонайбільше один раз, і його слід захистити всередині пропозиції `if __name__ == '__main__'` головного модуля.

  Нове в версії 3.4.

Примітка

`multiprocessing` не містить аналогів `threading.active_count()`, `threading.enumerate()`, `threading.settrace()`, `threading.setprofile()`, `threading.Timer` або `threading.local`.

### Об’єкти підключення

Об’єкти підключення дозволяють надсилати та отримувати об’єкти або рядки, які можна вибрати. Їх можна розглядати як підключені сокети, орієнтовані на повідомлення.

Об’єкти підключення зазвичай створюються за допомогою `Pipe` – див. також Слухачі та клієнти.

class multiprocessing.connection.Connection
  send(obj)

    Надішліть об’єкт на інший кінець з’єднання, який слід прочитати за допомогою `recv()`.

    Об’єкт має бути маринованим. Дуже великі pickles (приблизно 32 MiB+, хоча це залежить від ОС) можуть викликати виняток `ValueError`.

  recv()

    Повернути об’єкт, надісланий з іншого кінця з’єднання за допомогою `send()`. Блокує, поки не буде що отримати. Викликає `EOFError`, якщо не залишилося нічого для отримання, а інший кінець був закритий.

  fileno()

    Повертає дескриптор файлу або дескриптор, який використовується підключенням.

  close()

    Закрийте з’єднання.

    Це викликається автоматично, коли підключення збирається сміттям.

  poll([timeout])

    Повернути, чи є дані для читання.

    Якщо timeout не вказано, він негайно повернеться. Якщо timeout є числом, це вказує максимальний час у секундах для блокування. Якщо timeout має значення `None`, тоді використовується нескінченний тайм-аут.

    Зауважте, що кілька об’єктів з’єднання можна опитувати одночасно за допомогою `multiprocessing.connection.wait()`.

  send_bytes(buffer[, offset[, size]])

    Надіслати байтові дані з bytes-like object як повне повідомлення.

    Якщо вказано зсув, дані зчитуються з цієї позиції в буфері. Якщо задано size, то стільки байтів буде прочитано з буфера. Дуже великі буфери (приблизно 32 MiB+, хоча це залежить від ОС) можуть викликати виключення `ValueError`

  recv_bytes([maxlength])

    Повертає повне повідомлення байтових даних, надісланих з іншого кінця з’єднання, у вигляді рядка. Блокує, поки не буде що отримати. Викликає `EOFError`, якщо не залишилося нічого для отримання, а інший кінець закрито.

    Якщо вказано maxlength і повідомлення довше, ніж maxlength, тоді виникає `OSError` і з’єднання більше не читається.

    Змінено в версії 3.3: Раніше ця функція викликала `IOError`, який тепер є псевдонімом `OSError`.

  recv_bytes_into(buffer[, offset])

    Прочитайте в buffer повне повідомлення байтових даних, надісланих з іншого кінця з’єднання, і поверніть кількість байтів у повідомленні. Блокує, поки не буде що отримати. Викликає `EOFError`, якщо не залишилося нічого для отримання, а інший кінець був закритий.

    buffer має бути доступним для запису bytes-like object. Якщо задано offset, повідомлення буде записано в буфер із цієї позиції. Зсув має бути невід’ємним цілим числом, меншим за довжину буфера (у байтах).

    Якщо буфер закороткий, виникає виняток `BufferTooShort`, і повне повідомлення доступне як `e.args[0]`, де `e` є винятком.

  Змінено в версії 3.3: Самі об’єкти підключення тепер можна передавати між процесами за допомогою `Connection.send()` і `Connection.recv()`.

  Connection objects also now support the context management protocol – see Типи менеджера контексту. `__enter__()` returns the connection object, and `__exit__()` calls `close()`.

Наприклад:

    >>> from multiprocessing import Pipe
    >>> a, b = Pipe()
    >>> a.send([1, 'hello', None])
    >>> b.recv()
    [1, 'hello', None]
    >>> b.send_bytes(b'thank you')
    >>> a.recv_bytes()
    b'thank you'
    >>> import array
    >>> arr1 = array.array('i', range(5))
    >>> arr2 = array.array('i', [0] * 10)
    >>> a.send_bytes(arr1)
    >>> count = b.recv_bytes_into(arr2)
    >>> assert count == len(arr1) * arr1.itemsize
    >>> arr2
    array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])

Попередження

Метод `Connection.recv()` автоматично видаляє отримані дані, що може становити загрозу безпеці, якщо ви не можете довіряти процесу, який надіслав повідомлення.

Таким чином, якщо об’єкт підключення не було створено за допомогою `Pipe()`, ви повинні використовувати лише методи `recv()` і `send()` після виконання певної автентифікації. Перегляньте Ключі автентифікації.

Попередження

Якщо процес зупиняється під час спроби читання або запису в канал, то дані в каналі, ймовірно, будуть пошкоджені, тому що може стати неможливо точно визначити, де пролягають межі повідомлення.

### Примітиви синхронізації

Зазвичай примітиви синхронізації не такі необхідні в багатопроцесовій програмі, як у багатопотоковій програмі. Перегляньте документацію для модуля `threading`.

Зауважте, що можна також створити примітиви синхронізації за допомогою об’єкта менеджера – див. Менеджери.

class multiprocessing.Barrier(parties[, action[, timeout]])

  Бар’єрний об’єкт: клон `threading.Barrier`.

  Нове в версії 3.3.

class multiprocessing.BoundedSemaphore([value])

  Обмежений семафорний об’єкт: близький аналог `threading.BoundedSemaphore`.

  Існує єдина відмінність від його близького аналога: перший аргумент методу `acquire` має назву block, що узгоджується з `Lock.acquire()`.

  Примітка

  У macOS це неможливо відрізнити від `Semaphore`, оскільки `sem_getvalue()` не реалізовано на цій платформі.

class multiprocessing.Condition([lock])

  Змінна умови: псевдонім для `threading.Condition`.

  Якщо вказано lock, це має бути об’єкт `Lock` або `RLock` із `multiprocessing`.

  Змінено в версії 3.3: Додано метод `wait_for()`.

class multiprocessing.Event

  Клон `threading.Event`.

class multiprocessing.Lock

  Нерекурсивний об’єкт блокування: близький аналог `threading.Lock`. Після того, як процес або потік отримав блокування, наступні спроби отримати його від будь-якого процесу або потоку будуть блокуватися, доки його не буде звільнено; будь-який процес або потік може його звільнити. Концепції та поведінка `threading.Lock`, які застосовуються до потоків, відтворені тут у `multiprocessing.Lock`, оскільки вони застосовуються до процесів або потоків, за винятком зазначених випадків.

  Зауважте, що `Lock` насправді є фабричною функцією, яка повертає примірник `multiprocessing.synchronize.Lock`, ініціалізований контекстом за замовчуванням.

  `Lock` підтримує протокол context manager і тому може використовуватися в `with` операторах.

  acquire(block=True, timeout=None)

    Отримайте блокування, блокування або неблокування.

    Якщо для аргументу block встановлено значення `True` (за замовчуванням), виклик методу блокуватиметься, доки блокування не перейде в розблокований стан, а потім встановлюватиме для нього значення `True` і повертатиме `True`. Зауважте, що назва цього першого аргументу відрізняється від такої в `threading.Lock.acquire()`.

    Якщо для аргументу block встановлено значення `False`, виклик методу не блокується. Якщо блокування зараз у заблокованому стані, поверніть `False`; інакше встановіть блокування в заблокований стан і поверніть `True`.

    При виклику з додатним значенням з плаваючою комою для timeout, блокувати щонайбільше на кількість секунд, визначену timeout, доки не вдасться отримати блокування. Виклики з від’ємним значенням для timeout еквівалентні timeout рівному нулю. Виклики зі значенням timeout `None` (за замовчуванням) встановлюють період очікування як нескінченний. Зауважте, що обробка негативних значень або значень `None` для timeout відрізняється від реалізованої поведінки в `threading.Lock.acquire()`. Аргумент timeout не має практичного значення, якщо для аргументу block встановлено значення `False` і, таким чином, він ігнорується. Повертає `True`, якщо блокування отримано, або `False`, якщо період очікування минув.

  release()

    Відпустіть блокування. Це може бути викликано з будь-якого процесу або потоку, а не тільки процесу або потоку, який спочатку отримав блокування.

    Поведінка така ж, як у `threading.Lock.release()`, за винятком того, що під час виклику для розблокованого блокування виникає `ValueError`.

class multiprocessing.RLock

  Об’єкт рекурсивного блокування: близький аналог `threading.RLock`. Рекурсивне блокування має бути звільнено процесом або потоком, який його отримав. Після того як процес або потік отримав рекурсивне блокування, той самий процес або потік може отримати його знову без блокування; цей процес або потік повинен випускати його один раз за кожен раз, коли його було отримано.

  Зауважте, що `RLock` насправді є фабричною функцією, яка повертає екземпляр `multiprocessing.synchronize.RLock`, ініціалізований контекстом за замовчуванням.

  `RLock` підтримує протокол context manager і тому може використовуватися в `with` операторах.

  acquire(block=True, timeout=None)

    Отримайте блокування, блокування або неблокування.

    При виклику з аргументом block, встановленим у значення `True`, блокувати, доки блокування не буде в розблокованому стані (не належить жодному процесу або потоку), якщо блокування вже не належить поточному процесу або потоку. Тоді поточний процес або потік отримує право власності на блокування (якщо він ще не має права власності), а рівень рекурсії всередині блокування збільшується на одиницю, що призводить до повернення значення `True`. Зауважте, що є кілька відмінностей у поведінці цього першого аргументу порівняно з реалізацією `threading.RLock.acquire()`, починаючи з назви самого аргументу.

    При виклику з аргументом block, встановленим на `False`, не блокувати. Якщо блокування вже було отримано (і, отже, ним володіє) інший процес або потік, поточний процес або потік не приймає права власності, а рівень рекурсії в межах блокування не змінюється, що призводить до повернення значення `False` . Якщо блокування знаходиться в розблокованому стані, поточний процес або потік приймає право власності, і рівень рекурсії збільшується, що призводить до повернення значення `True`.

    Використання та поведінка аргументу timeout такі ж, як і в `Lock.acquire()`. Зауважте, що деякі з цих дій timeout відрізняються від реалізованих у `threading.RLock.acquire()`.

  release()

    Зніміть блокування, зменшивши рівень рекурсії. Якщо після зменшення рівень рекурсії дорівнює нулю, скиньте блокування до розблокованого (не належить жодному процесу чи потоку), а якщо будь-які інші процеси чи потоки заблоковані в очікуванні розблокування блокування, дозвольте рівно одному з них продовжити. Якщо після декременту рівень рекурсії все ще ненульовий, блокування залишається заблокованим і належить процесу або потоку, що викликає.

    Викликайте цей метод лише тоді, коли процес або потік, що викликає, володіє блокуванням. Помилка `AssertionError` виникає, якщо цей метод викликається процесом або потоком, відмінним від власника, або якщо блокування знаходиться в розблокованому стані (не належить). Зауважте, що тип винятку, викликаного в цій ситуації, відрізняється від реалізованої поведінки в `threading.RLock.release()`.

class multiprocessing.Semaphore([value])

  Об’єкт семафор: близький аналог `threading.Semaphore`.

  Існує єдина відмінність від його близького аналога: перший аргумент методу `acquire` має назву block, що узгоджується з `Lock.acquire()`.

Примітка

У macOS `sem_timedwait` не підтримується, тому виклик `acquire()` із тайм-аутом буде емулювати поведінку цієї функції за допомогою циклу сну.

Примітка

If the SIGINT signal generated by ``Ctrl`-`C`` arrives while the main thread is blocked by a call to `BoundedSemaphore.acquire()`, `Lock.acquire()`, `RLock.acquire()`, `Semaphore.acquire()`, `Condition.acquire()` or `Condition.wait()` then the call will be immediately interrupted and `KeyboardInterrupt` will be raised.

This differs from the behaviour of `threading` where SIGINT will be ignored while the equivalent blocking calls are in progress.

Примітка

Для деяких функцій цього пакета необхідна функціональна реалізація спільного семафора в головній операційній системі. Без нього модуль `multiprocessing.synchronize` буде вимкнено, а спроби його імпортувати призведуть до `ImportError`. Додаткову інформацію див. bpo-3770.

### Спільні об’єкти `ctypes`

Можна створювати спільні об’єкти, використовуючи спільну пам’ять, яку можуть успадковувати дочірні процеси.

multiprocessing.Value(typecode_or_type, *args, lock=True)

  Повертає об’єкт `ctypes`, виділений зі спільної пам’яті. За замовчуванням значення, що повертається, фактично є синхронізованою оболонкою для об’єкта. До самого об’єкта можна отримати доступ через атрибут value `Value`.

  typecode_or_type визначає тип повернутого об’єкта: це або тип ctypes, або односимвольний код типу, який використовується модулем `array`. *args передається конструктору для типу.

  Якщо lock має значення `True` (за замовчуванням), тоді створюється новий об’єкт рекурсивного блокування для синхронізації доступу до значення. Якщо lock є об’єктом `Lock` або `RLock`, то він використовуватиметься для синхронізації доступу до значення. Якщо lock має значення `False`, тоді доступ до повернутого об’єкта не буде автоматично захищений блокуванням, тому він не обов’язково буде «безпечним для процесу».

  Такі операції, як `+=`, які передбачають читання та запис, не є атомарними. Отже, якщо, наприклад, ви хочете атомарно збільшити спільне значення, недостатньо просто зробити

      counter.value += 1

  Якщо припустити, що пов’язане блокування є рекурсивним (що є за замовчуванням), ви можете замість цього зробити:

      with counter.get_lock():
          counter.value += 1

  Зверніть увагу, що lock є аргументом лише для ключового слова.

multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)

  Повертає масив ctypes, виділений зі спільної пам’яті. За замовчуванням значення, що повертається, фактично є синхронізованою оболонкою для масиву.

  typecode_or_type визначає тип елементів повернутого масиву: це або тип ctypes, або односимвольний код типу, який використовується модулем `array`. Якщо size_or_initializer є цілим числом, воно визначає довжину масиву, і масив буде спочатку обнулений. В іншому випадку size_or_initializer — це послідовність, яка використовується для ініціалізації масиву, довжина якої визначає довжину масиву.

  Якщо lock має значення `True` (за замовчуванням), тоді створюється новий об’єкт блокування для синхронізації доступу до значення. Якщо lock є об’єктом `Lock` або `RLock`, то він використовуватиметься для синхронізації доступу до значення. Якщо lock має значення `False`, тоді доступ до повернутого об’єкта не буде автоматично захищений блокуванням, тому він не обов’язково буде «безпечним для процесу».

  Зауважте, що lock є лише ключовим аргументом.

  Зауважте, що масив `ctypes.c_char` має атрибути value і raw, які дозволяють використовувати його для зберігання та отримання рядків.

#### Модуль `multiprocessing.sharedctypes`

Модуль `multiprocessing.sharedctypes` надає функції для виділення об’єктів `ctypes` зі спільної пам’яті, які можуть бути успадковані дочірніми процесами.

Примітка

Хоча можна зберігати вказівник у спільній пам’яті, пам’ятайте, що це посилатиметься на розташування в адресному просторі певного процесу. Однак вказівник, швидше за все, буде недійсним у контексті другого процесу, і спроба розіменувати вказівник з другого процесу може спричинити збій.

multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer)

  Повертає масив ctypes, виділений зі спільної пам’яті.

  typecode_or_type визначає тип елементів повернутого масиву: це або тип ctypes, або односимвольний код типу, який використовується модулем `array`. Якщо size_or_initializer є цілим числом, воно визначає довжину масиву, і масив буде спочатку обнулено. В іншому випадку size_or_initializer — це послідовність, яка використовується для ініціалізації масиву і довжина якої визначає довжину масиву.

  Зауважте, що встановлення й отримання елемента потенційно не є атомарним — замість цього використовуйте `Array()`, щоб переконатися, що доступ автоматично синхронізується за допомогою блокування.

multiprocessing.sharedctypes.RawValue(typecode_or_type, *args)

  Повертає об’єкт ctypes, виділений зі спільної пам’яті.

  typecode_or_type визначає тип повернутого об’єкта: це або тип ctypes, або односимвольний код типу, який використовується модулем `array`. *args передається конструктору для типу.

  Зауважте, що встановлення й отримання значення потенційно не є атомарним — замість цього використовуйте `Value()`, щоб переконатися, що доступ автоматично синхронізується за допомогою блокування.

  Зауважте, що масив `ctypes.c_char` має атрибути `value` і `raw`, які дозволяють використовувати його для зберігання та отримання рядків - дивіться документацію для `ctypes`.

multiprocessing.sharedctypes.Array(typecode_or_type, size_or_initializer, *, lock=True)

  Те саме, що `RawArray()`, за винятком того, що залежно від значення lock замість необробленого масиву ctypes може повертатися обгортка синхронізації, безпечна для процесу.

  Якщо lock має значення `True` (за замовчуванням), тоді створюється новий об’єкт блокування для синхронізації доступу до значення. Якщо lock є об’єктом `Lock` або `RLock`, то він використовуватиметься для синхронізації доступу до значення. Якщо lock має значення `False`, тоді доступ до повернутого об’єкта не буде автоматично захищений блокуванням, тому він не обов’язково буде «безпечним для процесу».

  Зверніть увагу, що lock є аргументом лише для ключового слова.

multiprocessing.sharedctypes.Value(typecode_or_type, *args, lock=True)

  Те саме, що `RawValue()`, за винятком того, що залежно від значення lock замість необробленого об’єкта ctypes може повертатися безпечна для процесу оболонка синхронізації.

  Якщо lock має значення `True` (за замовчуванням), тоді створюється новий об’єкт блокування для синхронізації доступу до значення. Якщо lock є об’єктом `Lock` або `RLock`, то він використовуватиметься для синхронізації доступу до значення. Якщо lock має значення `False`, тоді доступ до повернутого об’єкта не буде автоматично захищений блокуванням, тому він не обов’язково буде «безпечним для процесу».

  Зверніть увагу, що lock є аргументом лише для ключового слова.

multiprocessing.sharedctypes.copy(obj)

  Повертає об’єкт ctypes, виділений зі спільної пам’яті, який є копією об’єкта ctypes obj.

multiprocessing.sharedctypes.synchronized(obj[, lock])

  Повертає безпечний для процесу об’єкт оболонки для об’єкта ctypes, який використовує lock для синхронізації доступу. Якщо lock має значення `None` (за замовчуванням), то об’єкт `multiprocessing.RLock` створюється автоматично.

  Синхронізована оболонка матиме два методи на додаток до методів об’єкта, який вона обертає: `get_obj()` повертає обгорнутий об’єкт, а `get_lock()` повертає об’єкт блокування, який використовується для синхронізації.

  Зверніть увагу, що доступ до об’єкта ctypes через оболонку може бути набагато повільнішим, ніж доступ до необробленого об’єкта ctypes.

  Змінено в версії 3.5: Синхронізовані об’єкти підтримують протокол context manager.

У наведеній нижче таблиці порівнюється синтаксис для створення спільних об’єктів ctypes зі спільної пам’яті зі звичайним синтаксисом ctypes. (У таблиці `MyStruct` є деякий підклас `ctypes.Structure`.)

ctypes | sharedctypes за допомогою типу | sharedctypes з використанням коду типу
--- | --- | ---
c_double (2,4) | RawValue(c_double, 2,4) | RawValue(„d“, 2,4)
MyStruct(4, 6) | RawValue(MyStruct, 4, 6) |
(c_short * 7)() | RawArray(c_short, 7) | RawArray(„h“, 7)
(c_int * 3)(9, 2, 8) | RawArray(c_int, (9, 2, 8)) | RawArray(„i“, (9, 2, 8))

Нижче наведено приклад, коли кілька об’єктів ctypes змінено дочірнім процесом:

    from multiprocessing import Process, Lock
    from multiprocessing.sharedctypes import Value, Array
    from ctypes import Structure, c_double

    class Point(Structure):
        _fields_ = [('x', c_double), ('y', c_double)]

    def modify(n, x, s, A):
        n.value **= 2
        x.value **= 2
        s.value = s.value.upper()
        for a in A:
            a.x **= 2
            a.y **= 2

    if __name__ == '__main__':
        lock = Lock()

        n = Value('i', 7)
        x = Value(c_double, 1.0/3.0, lock=False)
        s = Array('c', b'hello world', lock=lock)
        A = Array(Point, [(1.875,-6.25), (-5.75,2.0), (2.375,9.5)], lock=lock)

        p = Process(target=modify, args=(n, x, s, A))
        p.start()
        p.join()

        print(n.value)
        print(x.value)
        print(s.value)
        print([(a.x, a.y) for a in A])

Надруковані результати:

    49
    0.1111111111111111
    HELLO WORLD
    [(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]

### Менеджери

Менеджери надають можливість створювати дані, якими можна ділитися між різними процесами, включно з загальним доступом через мережу між процесами, що виконуються на різних машинах. Об’єкт менеджера контролює серверний процес, який керує спільними об’єктами. Інші процеси можуть отримати доступ до спільних об’єктів за допомогою проксі-серверів.

multiprocessing.Manager()

  Повертає запущений об’єкт `SyncManager`, який можна використовувати для спільного використання об’єктів між процесами. Повернений об’єкт менеджера відповідає породженому дочірньому процесу та має методи, які створюватимуть спільні об’єкти та повертатимуть відповідні проксі-сервери.

Процеси менеджера буде закрито, щойно їх буде зібрано сміття або їхній батьківський процес завершиться. Класи менеджерів визначені в модулі `multiprocessing.managers`:

class multiprocessing.managers.BaseManager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)

  Створіть об’єкт BaseManager.

  Після створення потрібно викликати `start()` або `get_server().serve_forever()`, щоб переконатися, що об’єкт менеджера посилається на запущений процес менеджера.

  адреса — це адреса, на якій процес менеджера прослуховує нові підключення. Якщо адреса має значення `None`, тоді вибирається довільна адреса.

  authkey — це ключ автентифікації, який використовуватиметься для перевірки дійсності вхідних підключень до процесу сервера. Якщо authkey має значення `None`, тоді використовується `current_process().authkey`. В іншому випадку використовується authkey, і це має бути рядок байтів.

  serializer must be `'pickle'` (use `pickle` serialization) or `'xmlrpclib'` (use `xmlrpc.client` serialization).

  ctx is a context object, or `None` (use the current context). See the `get_context()` function.

  shutdown_timeout is a timeout in seconds used to wait until the process used by the manager completes in the `shutdown()` method. If the shutdown times out, the process is terminated. If terminating the process also times out, the process is killed.

  Змінено в версії 3.11: Added the shutdown_timeout parameter.

  start([initializer[, initargs]])

    Запустіть підпроцес, щоб запустити менеджер. Якщо initializer не `None`, тоді підпроцес викличе `initializer(*initargs)` під час запуску.

  get_server()

    Повертає об’єкт `Server`, який представляє фактичний сервер під керуванням менеджера. Об’єкт `Server` підтримує метод `serve_forever()`:

        >>> from multiprocessing.managers import BaseManager
        >>> manager = BaseManager(address=('', 50000), authkey=b'abc')
        >>> server = manager.get_server()
        >>> server.serve_forever()

    `Server` додатково має атрибут `address`.

  connect()

    Підключіть об’єкт локального менеджера до віддаленого процесу менеджера:

        >>> from multiprocessing.managers import BaseManager
        >>> m = BaseManager(address=('127.0.0.1', 50000), authkey=b'abc')
        >>> m.connect()

  shutdown()

    Зупиніть процес, який використовує менеджер. Це доступно, лише якщо `start()` було використано для запуску процесу сервера.

    Це можна викликати кілька разів.

  register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])

    Метод класу, який можна використовувати для реєстрації типу або виклику в класі менеджера.

    typeid — це «ідентифікатор типу», який використовується для ідентифікації певного типу спільного об’єкта. Це має бути рядок.

    callable — це виклик, який використовується для створення об’єктів для ідентифікатора цього типу. Якщо екземпляр менеджера буде підключено до сервера за допомогою методу `connect()` або якщо аргумент create_method має значення `False`, тоді це можна залишити як `None`.

    proxytype є підкласом `BaseProxy`, який використовується для створення проксі для спільних об’єктів із цим typeid. Якщо `None`, то проксі-клас створюється автоматично.

    exposed використовується для визначення послідовності назв методів, до яких проксі-серверам для цього typeid має бути дозволено доступ за допомогою `BaseProxy._callmethod()`. (Якщо exposed має значення `None`, тоді замість нього використовується `proxytype._exposed_`, якщо він існує.) У випадку, коли відкритий список не вказано, усі «загальнодоступні методи» спільного об’єкта будуть доступними. . (Тут «публічний метод» означає будь-який атрибут, який має метод `__call__()` і ім’я якого не починається з `'_'`.)

    method_to_typeid — це зіставлення, яке використовується для визначення типу повернення тих відкритих методів, які мають повертати проксі. Він відображає назви методів у рядках typeid. (Якщо method_to_typeid має значення `None`, тоді замість нього використовується `proxytype._method_to_typeid_`, якщо він існує.) Якщо назва методу не є ключем цього відображення або якщо відображення має значення `None`, тоді об’єкт, повернутий методом, буде скопійовано за значенням.

    create_method визначає, чи слід створювати метод з іменем typeid, яке можна використовувати, щоб наказати серверному процесу створити новий спільний об’єкт і повернути для нього проксі. За замовчуванням це `True`.

  Екземпляри `BaseManager` також мають одну властивість лише для читання:

  address

    Адреса, яку використовує менеджер.

  Змінено в версії 3.3: Об’єкти менеджера підтримують протокол керування контекстом – див. Типи менеджера контексту. `__enter__()` запускає серверний процес (якщо він ще не запущений), а потім повертає об’єкт менеджера. `__exit__()` викликає `shutdown()`.

  У попередніх версіях `__enter__()` не запускав серверний процес менеджера, якщо він ще не був запущений.

class multiprocessing.managers.SyncManager

  Підклас `BaseManager`, який можна використовувати для синхронізації процесів. Об’єкти цього типу повертає `multiprocessing.Manager()`.

  Його методи створюють і повертають Проксі об’єкти для ряду типів даних, які зазвичай використовуються, щоб синхронізувати між процесами. Це, зокрема, включає спільні списки та словники.

  Barrier(parties[, action[, timeout]])

    Створіть спільний об’єкт `threading.Barrier` і поверніть для нього проксі.

    Нове в версії 3.3.

  BoundedSemaphore([value])

    Створіть спільний об’єкт `threading.BoundedSemaphore` і поверніть для нього проксі.

  Condition([lock])

    Створіть спільний об’єкт `threading.Condition` і поверніть для нього проксі.

    Якщо вказано lock, це має бути проксі для об’єкта `threading.Lock` або `threading.RLock`.

    Змінено в версії 3.3: Додано метод `wait_for()`.

  Event()

    Створіть спільний об’єкт `threading.Event` і поверніть для нього проксі.

  Lock()

    Створіть спільний об’єкт `threading.Lock` і поверніть для нього проксі.

  Namespace()

    Створіть спільний об’єкт `Namespace` і поверніть для нього проксі.

  Queue([maxsize])

    Створіть спільний об’єкт `queue.Queue` і поверніть для нього проксі.

  RLock()

    Створіть спільний об’єкт `threading.RLock` і поверніть для нього проксі.

  Semaphore([value])

    Створіть спільний об’єкт `threading.Semaphore` і поверніть для нього проксі.

  Array(typecode, sequence)

    Створіть масив і поверніть для нього проксі.

  Value(typecode, value)

    Створіть об’єкт із доступним для запису атрибутом «значення» та поверніть для нього проксі-сервер.

  dict()
  dict(mapping)
  dict(sequence)

    Створіть спільний об’єкт `dict` і поверніть для нього проксі.

  list()
  list(sequence)

    Створіть спільний об’єкт `list` і поверніть для нього проксі.

  Змінено в версії 3.6: Спільні об’єкти можуть бути вкладеними. Наприклад, спільний об’єкт-контейнер, такий як спільний список, може містити інші спільні об’єкти, якими керуватиме та синхронізуватиме `SyncManager`.

class multiprocessing.managers.Namespace

  Тип, який можна зареєструвати в `SyncManager`.

  Об’єкт простору імен не має відкритих методів, але має атрибути, доступні для запису. Його подання показує значення його атрибутів.

  Однак, коли використовується проксі для об’єкта простору імен, атрибут, що починається з `'_''` буде атрибутом проксі, а не атрибутом референта:

      >>> manager = multiprocessing.Manager()
      >>> Global = manager.Namespace()
      >>> Global.x = 10
      >>> Global.y = 'hello'
      >>> Global._z = 12.3    # this is an attribute of the proxy
      >>> print(Global)
      Namespace(x=10, y='hello')

#### Індивідуальні менеджери

Щоб створити власний менеджер, потрібно створити підклас `BaseManager` і використовувати метод класу `register()` для реєстрації нових типів або викликів у класі менеджера. Наприклад:

    from multiprocessing.managers import BaseManager

    class MathsClass:
        def add(self, x, y):
            return x + y
        def mul(self, x, y):
            return x * y

    class MyManager(BaseManager):
        pass

    MyManager.register('Maths', MathsClass)

    if __name__ == '__main__':
        with MyManager() as manager:
            maths = manager.Maths()
            print(maths.add(4, 3))         # prints 7
            print(maths.mul(7, 8))         # prints 56

#### Використання віддаленого менеджера

Можна запустити керуючий сервер на одній машині, а клієнти використовуватимуть його з інших машин (за умови, що задіяні брандмауери дозволяють це).

Виконання наступних команд створює сервер для однієї спільної черги, до якої мають доступ віддалені клієнти:

    >>> from multiprocessing.managers import BaseManager
    >>> from queue import Queue
    >>> queue = Queue()
    >>> class QueueManager(BaseManager): pass
    >>> QueueManager.register('get_queue', callable=lambda:queue)
    >>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
    >>> s = m.get_server()
    >>> s.serve_forever()

Один клієнт може отримати доступ до сервера наступним чином:

    >>> from multiprocessing.managers import BaseManager
    >>> class QueueManager(BaseManager): pass
    >>> QueueManager.register('get_queue')
    >>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
    >>> m.connect()
    >>> queue = m.get_queue()
    >>> queue.put('hello')

Інший клієнт також може використовувати його:

    >>> from multiprocessing.managers import BaseManager
    >>> class QueueManager(BaseManager): pass
    >>> QueueManager.register('get_queue')
    >>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
    >>> m.connect()
    >>> queue = m.get_queue()
    >>> queue.get()
    'hello'

Локальні процеси також можуть отримати доступ до цієї черги, використовуючи код вище на клієнті для доступу до неї віддалено:

    >>> from multiprocessing import Process, Queue
    >>> from multiprocessing.managers import BaseManager
    >>> class Worker(Process):
    ...     def __init__(self, q):
    ...         self.q = q
    ...         super().__init__()
    ...     def run(self):
    ...         self.q.put('local hello')
    ...
    >>> queue = Queue()
    >>> w = Worker(queue)
    >>> w.start()
    >>> class QueueManager(BaseManager): pass
    ...
    >>> QueueManager.register('get_queue', callable=lambda: queue)
    >>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
    >>> s = m.get_server()
    >>> s.serve_forever()

### Проксі об’єкти

Проксі — це об’єкт, який посилається на спільний об’єкт, який живе (імовірно) в іншому процесі. Спільний об’єкт називається референтом проксі. Кілька проксі-об’єктів можуть мати один і той же референт.

Проксі-об’єкт має методи, які викликають відповідні методи його референта (хоча не кожен метод референта обов’язково буде доступним через проксі). Таким чином, проксі можна використовувати так само, як і його референт:

    >>> from multiprocessing import Manager
    >>> manager = Manager()
    >>> l = manager.list([i*i for i in range(10)])
    >>> print(l)
    [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
    >>> print(repr(l))
    <ListProxy object, typeid 'list' at 0x...>
    >>> l[4]
    16
    >>> l[2:5]
    [4, 9, 16]

Зауважте, що застосування `str()` до проксі поверне подання референта, тоді як застосування `repr()` поверне подання проксі.

Важливою особливістю проксі-об’єктів є те, що їх можна вибирати, тому їх можна передавати між процесами. Таким чином, референт може містити Проксі об’єкти. Це дозволяє вкладати ці керовані списки, dicts та інші Проксі об’єкти:

    >>> a = manager.list()
    >>> b = manager.list()
    >>> a.append(b)         # referent of a now contains referent of b
    >>> print(a, b)
    [<ListProxy object, typeid 'list' at ...>] []
    >>> b.append('hello')
    >>> print(a[0], b)
    ['hello'] ['hello']

Подібним чином, проксі dict і список можуть бути вкладені один в одного:

    >>> l_outer = manager.list([ manager.dict() for i in range(2) ])
    >>> d_first_inner = l_outer[0]
    >>> d_first_inner['a'] = 1
    >>> d_first_inner['b'] = 2
    >>> l_outer[1]['c'] = 3
    >>> l_outer[1]['z'] = 26
    >>> print(l_outer[0])
    {'a': 1, 'b': 2}
    >>> print(l_outer[1])
    {'c': 3, 'z': 26}

Якщо стандартні (не проксі) `list` або `dict` об’єкти містяться в референті, модифікації цих змінних значень не поширюватимуться через менеджер, оскільки проксі не може дізнатися, коли значення що містяться в них, змінено. Однак збереження значення в проксі-контейнері (що запускає `__setitem__` в проксі-об’єкті) поширюється через менеджер, тому для ефективної зміни такого елемента можна повторно призначити змінене значення проксі-серверу контейнера: :

    # create a list proxy and append a mutable object (a dictionary)
    lproxy = manager.list()
    lproxy.append({})
    # now mutate the dictionary
    d = lproxy[0]
    d['a'] = 1
    d['b'] = 2
    # at this point, the changes to d are not yet synced, but by
    # updating the dictionary, the proxy is notified of the change
    lproxy[0] = d

Цей підхід, можливо, менш зручний, ніж використання вкладених Проксі об’єкти для більшості випадків використання, але також демонструє рівень контролю над синхронізацією.

Примітка

Проксі-типи в `multiprocessing` не підтримують порівняння за значенням. Так, наприклад, ми маємо:

    >>> manager.list([1,2,3]) == [1,2,3]
    False

Під час порівнянь слід просто використовувати копію референта.

class multiprocessing.managers.BaseProxy

  Проксі-об’єкти є екземплярами підкласів `BaseProxy`.

  _callmethod(methodname[, args[, kwds]])

    Виклик і повернення результату методу референта проксі.

    Якщо `proxy` є проксі, референтом якого є `obj`, тоді вираз

        proxy._callmethod(methodname, args, kwds)

    обчислить вираз

        getattr(obj, methodname)(*args, **kwds)

    в процесі менеджера.

    Поверненим значенням буде копія результату виклику або проксі для нового спільного об’єкта – див. документацію щодо аргументу method_to_typeid `BaseManager.register()`.

    Якщо виклик викликає виняток, він повторно викликається `_callmethod()`. Якщо в процесі менеджера виникає інший виняток, він перетворюється на виняток `RemoteError` і викликається `_callmethod()`.

    Зокрема, зауважте, що виняток буде створено, якщо methodname не було виявлено.

    Приклад використання `_callmethod()`:

        >>> l = manager.list(range(10))
        >>> l._callmethod('__len__')
        10
        >>> l._callmethod('__getitem__', (slice(2, 7),)) # equivalent to l[2:7]
        [2, 3, 4, 5, 6]
        >>> l._callmethod('__getitem__', (20,))          # equivalent to l[20]
        Traceback (most recent call last):
        ...
        IndexError: list index out of range

  _getvalue()

    Повернути копію референту.

    Якщо референт неможливо вибрати, це спричинить виняток.

  __repr__()

    Повертає представлення проксі-об’єкта.

  __str__()

    Повернути представлення референта.

#### Прибирати

Проксі-об’єкт використовує зворотний виклик weakref, щоб, коли він збирає сміття, він скасовує реєстрацію в менеджері, якому належить його референт.

Спільний об’єкт видаляється з процесу менеджера, коли більше немає проксі-серверів, які посилаються на нього.

### Пули процесів

Можна створити пул процесів, які виконуватимуть передані йому завдання за допомогою класу `Pool`.

class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

  Об’єкт пулу процесів, який керує пулом робочих процесів, до яких можна надсилати завдання. Він підтримує асинхронні результати з тайм-аутами та зворотними викликами та має реалізацію паралельної карти.

  processes is the number of worker processes to use. If processes is `None` then the number returned by `os.cpu_count()` is used.

  Якщо initializer не `None`, тоді кожен робочий процес викличе `initializer(*initargs)` під час свого запуску.

  maxtasksperchild — це кількість завдань, які робочий процес може виконати, перш ніж він вийде та буде замінений новим робочим процесом, щоб звільнити невикористані ресурси. Типовим значенням maxtasksperchild є `None`, що означає, що робочі процеси живуть стільки ж, скільки пул.

  context можна використовувати для визначення контексту, який використовується для запуску робочих процесів. Зазвичай пул створюється за допомогою функції `multiprocessing.Pool()` або методу `Pool()` контекстного об’єкта. В обох випадках контекст встановлено належним чином.

  Зауважте, що методи об’єкта пулу має викликати тільки процес, який створив пул.

  Попередження

  Об’єкти `multiprocessing.pool` мають внутрішні ресурси, якими потрібно належним чином керувати (як і будь-яким іншим ресурсом), використовуючи пул як контекстний менеджер або викликаючи `close()` і `terminate()` вручну. Якщо цього не зробити, процес може призупинити завершення.

  Зауважте, що некоректно покладатися на збирач сміття для знищення пулу, оскільки CPython не гарантує, що буде викликано фіналізатор пулу (див. `object.__del__()` для отримання додаткової інформації).

  Змінено в версії 3.2: Added the maxtasksperchild parameter.

  Змінено в версії 3.4: Added the context parameter.

  Примітка

  Робочі процеси в межах `Pool` зазвичай живі протягом повної тривалості робочої черги пулу. Частий шаблон, який зустрічається в інших системах (таких як Apache, mod_wsgi тощо) для звільнення ресурсів, які зберігаються робочими засобами, полягає в тому, щоб дозволити робочому в межах пулу завершити лише певний обсяг роботи перед виходом, очищенням і породженням нового процесу на заміну старого. Аргумент maxtasksperchild для `Pool` надає цю можливість кінцевому користувачеві.

  apply(func[, args[, kwds]])

    Виклик func з аргументами args і ключовими аргументами kwds. Блокується, поки не буде готовий результат. Враховуючи ці блоки, `apply_async()` краще підходить для виконання роботи паралельно. Крім того, func виконується лише в одному з воркерів пулу.

  apply_async(func[, args[, kwds[, callback[, error_callback]]]])

    Варіант методу `apply()`, який повертає об’єкт `AsyncResult`.

    Якщо вказано callback, це має бути виклик, який приймає один аргумент. Коли результат стає готовим, до нього застосовується callback, якщо тільки виклик не вдався, у цьому випадку замість нього застосовується error_callback.

    Якщо вказано error_callback, це має бути виклик, який приймає один аргумент. Якщо цільова функція дає збій, то error_callback викликається з екземпляром винятку.

    Зворотні виклики мають завершитися негайно, інакше потік, який обробляє результати, буде заблоковано.

  map(func, iterable[, chunksize])

    Паралельний еквівалент вбудованої функції `map()` (хоча вона підтримує лише один аргумент iterable, для кількох ітерацій див. `starmap()`). Блокується, поки не буде готовий результат.

    Цей метод розбиває iterable на декілька фрагментів, які він надсилає до пулу процесів як окремі завдання. (Приблизний) розмір цих фрагментів можна вказати, встановивши для chunksize додатне ціле число.

    Зауважте, що це може спричинити велике використання пам’яті для дуже довгих ітерацій. Розгляньте можливість використання `imap()` або `imap_unordered()` з явним параметром chunksize для кращої ефективності.

  map_async(func, iterable[, chunksize[, callback[, error_callback]]])

    Варіант методу `map()`, який повертає об’єкт `AsyncResult`.

    Якщо вказано callback, це має бути виклик, який приймає один аргумент. Коли результат стає готовим, до нього застосовується callback, якщо тільки виклик не вдався, у цьому випадку замість нього застосовується error_callback.

    Якщо вказано error_callback, це має бути виклик, який приймає один аргумент. Якщо цільова функція дає збій, то error_callback викликається з екземпляром винятку.

    Зворотні виклики мають завершитися негайно, інакше потік, який обробляє результати, буде заблоковано.

  imap(func, iterable[, chunksize])

    Ленича версія `map()`.

    Аргумент chunksize такий самий, як той, який використовується методом `map()`. Для дуже довгих ітерацій використання великого значення для chunksize може зробити роботу завершеною набагато швидше, ніж використання значення за замовчуванням `1`.

    Крім того, якщо chunksize дорівнює `1`, тоді метод `next()` ітератора, який повертає метод `imap()`, має додатковий параметр timeout: `next(timeout)` викличе `multiprocessing.TimeoutError`, якщо результат не може бути повернутий протягом часу очікування секунд.

  imap_unordered(func, iterable[, chunksize])

    Те саме, що `imap()`, за винятком того, що порядок результатів від повернутого ітератора слід вважати довільним. (Тільки коли є лише один робочий процес, порядок гарантовано буде «правильним».)

  starmap(func, iterable[, chunksize])

    Подібно до `map()`, за винятком того, що елементи iterable мають бути ітерованими, які розпаковуються як аргументи.

    Тому ітерація `[(1,2), (3, 4)]` призводить до `[func(1,2), func(3,4)]`.

    Нове в версії 3.3.

  starmap_async(func, iterable[, chunksize[, callback[, error_callback]]])

    Комбінація `starmap()` і `map_async()`, яка виконує ітерацію по iterable ітерацій і викликає func з розпакованими ітераціями. Повертає об’єкт результату.

    Нове в версії 3.3.

  close()

    Запобігає надсиланню додаткових завдань до пулу. Після виконання всіх завдань робочі процеси завершаться.

  terminate()

    Негайно зупиняє робочі процеси, не завершуючи незавершену роботу. Коли об’єкт пулу збирається як сміття, негайно буде викликано `terminate()`.

  join()

    Зачекайте, поки робочі процеси завершаться. Потрібно викликати `close()` або `terminate()` перед використанням `join()`.

  Змінено в версії 3.3: Об’єкти пулу тепер підтримують протокол керування контекстом – див. Типи менеджера контексту. `__enter__()` повертає об’єкт пулу, а `__exit__()` викликає `terminate()`.

class multiprocessing.pool.AsyncResult

  Клас результату, який повертають `Pool.apply_async()` і `Pool.map_async()`.

  get([timeout])

    Поверніть результат, коли він надійде. Якщо timeout не `None` і результат не надходить протягом timeout секунд, тоді виникає `multiprocessing.TimeoutError`. Якщо віддалений виклик викликав виняток, цей виняток буде повторно викликано `get()`.

  wait([timeout])

    Зачекайте, поки буде доступний результат або поки не мине тайм-аут секунди.

  ready()

    Повідомити, чи завершено виклик.

  successful()

    Повертає, чи завершено виклик без виклику винятку. Викличе `ValueError`, якщо результат не готовий.

    Змінено в версії 3.7: Якщо результат не готовий, замість `AssertionError` виникає `ValueError`.

Наступний приклад демонструє використання пулу:

    from multiprocessing import Pool
    import time

    def f(x):
        return x*x

    if __name__ == '__main__':
        with Pool(processes=4) as pool:         # start 4 worker processes
            result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously in a single process
            print(result.get(timeout=1))        # prints "100" unless your computer is *very* slow

            print(pool.map(f, range(10)))       # prints "[0, 1, 4,..., 81]"

            it = pool.imap(f, range(10))
            print(next(it))                     # prints "0"
            print(next(it))                     # prints "1"
            print(it.next(timeout=1))           # prints "4" unless your computer is *very* slow

            result = pool.apply_async(time.sleep, (10,))
            print(result.get(timeout=1))        # raises multiprocessing.TimeoutError

### Слухачі та клієнти

Зазвичай передача повідомлень між процесами здійснюється за допомогою черг або за допомогою об’єктів `Connection`, які повертає `Pipe()`.

Однак модуль `multiprocessing.connection` забезпечує додаткову гнучкість. По суті, це надає API високого рівня, орієнтований на повідомлення, для роботи з сокетами або іменованими каналами Windows. Він також підтримує дайджест-автентифікацію за допомогою модуля `hmac` і для опитування кількох з’єднань одночасно.

multiprocessing.connection.deliver_challenge(connection, authkey)

  Надішліть випадково згенероване повідомлення на інший кінець з’єднання та дочекайтеся відповіді.

  Якщо відповідь відповідає дайджесту повідомлення з використанням authkey як ключа, тоді на інший кінець з’єднання надсилається вітальне повідомлення. Інакше виникає `AuthenticationError`.

multiprocessing.connection.answer_challenge(connection, authkey)

  Отримайте повідомлення, обчисліть дайджест повідомлення, використовуючи authkey як ключ, а потім надішліть дайджест назад.

  Якщо вітальне повідомлення не отримано, виникає `AuthenticationError`.

multiprocessing.connection.Client(address[, family[, authkey]])

  Спроба встановити з’єднання зі слухачем, який використовує адресу address, повертаючи `Connection`.

  Тип з’єднання визначається аргументом family, але зазвичай його можна опустити, оскільки його зазвичай можна визначити з формату address. (Див. Формати адрес)

  If authkey is given and not None, it should be a byte string and will be used as the secret key for an HMAC-based authentication challenge. No authentication is done if authkey is None. `AuthenticationError` is raised if authentication fails. See Ключі автентифікації.

class multiprocessing.connection.Listener([address[, family[, backlog[, authkey]]]])

  Обгортка для пов’язаного сокета або каналу з іменем Windows, який «слухає» з’єднання.

  address — це адреса, яка буде використовуватися зв’язаним сокетом або іменованим каналом об’єкта слухача.

  Примітка

  Якщо використовується адреса «0.0.0.0», ця адреса не буде кінцевою точкою підключення в Windows. Якщо вам потрібна підключена кінцева точка, вам слід використовувати «127.0.0.1».

  сімейство — це тип розетки (або названої труби), яку слід використовувати. Це може бути один із рядків `'AF_INET'` (для сокета TCP), `'AF_UNIX'` (для сокета домену Unix) або `'AF_PIPE''` (для іменованого каналу Windows) . З них лише перший гарантовано доступний. Якщо family має значення `None`, тоді сім’я виводиться з формату address. Якщо адреса також `None`, тоді вибрано значення за замовчуванням. Це за замовчуванням сімейство, яке вважається найшвидшим із доступних. Дивіться Формати адрес. Зауважте, що якщо сімейство має значення `'AF_UNIX`, а адреса `None`, то сокет буде створено в приватному тимчасовому каталозі, створеному за допомогою `tempfile.mkstemp()`.

  Якщо об’єкт слухача використовує сокет, тоді backlog (1 за замовчуванням) передається в метод `listen()` сокета після того, як його буде зв’язано.

  If authkey is given and not None, it should be a byte string and will be used as the secret key for an HMAC-based authentication challenge. No authentication is done if authkey is None. `AuthenticationError` is raised if authentication fails. See Ключі автентифікації.

  accept()

    Прийняти підключення до зв’язаного сокета або іменованого каналу об’єкта слухача та повернути об’єкт `Connection`. Якщо спроба автентифікації не вдається, виникає `AuthenticationError`.

  close()

    Закрийте прив’язаний сокет або іменований канал об’єкта слухача. Це викликається автоматично, коли слухач збирає сміття. Однак бажано називати це явно.

  Об’єкти слухача мають такі властивості лише для читання:

  address

    Адреса, яка використовується об’єктом Listener.

  last_accepted

    Адреса, з якої надійшло останнє прийняте підключення. Якщо це недоступно, це `None`.

  Змінено в версії 3.3: Об’єкти слухача тепер підтримують протокол керування контекстом – див. Типи менеджера контексту. `__enter__()` повертає об’єкт слухача, а `__exit__()` викликає `close()`.

multiprocessing.connection.wait(object_list, timeout=None)

  Зачекайте, поки об’єкт у object_list буде готовий. Повертає список тих об’єктів у object_list, які готові. Якщо timeout є числом з плаваючою точкою, виклик блокується щонайбільше на стільки секунд. Якщо timeout має значення `None`, тоді він блокуватиметься на необмежений період. Від’ємний тайм-аут еквівалентний нульовому тайм-ауту.

  For both Unix and Windows, an object can appear in object_list if it is

  - читабельний об’єкт `Connection`;

  - підключений і читабельний об’єкт `socket.socket`; або

  - атрибут `sentinel` об’єкта `Process`.

  Об’єкт з’єднання або сокета готовий, коли є доступні дані для читання з нього, або інший кінець закрито.

  Unix: `wait(object_list, timeout)` almost equivalent `select.select(object_list, [], [], timeout)`. The difference is that, if `select.select()` is interrupted by a signal, it can raise `OSError` with an error number of `EINTR`, whereas `wait()` will not.

  Windows: An item in object_list must either be an integer handle which is waitable (according to the definition used by the documentation of the Win32 function `WaitForMultipleObjects()`) or it can be an object with a `fileno()` method which returns a socket handle or pipe handle. (Note that pipe handles and socket handles are not waitable handles.)

  Нове в версії 3.3.

Приклади

Наступний код сервера створює прослуховувач, який використовує `'секретний пароль` як ключ автентифікації. Потім він очікує з’єднання та надсилає деякі дані клієнту:

    from multiprocessing.connection import Listener
    from array import array

    address = ('localhost', 6000)     # family is deduced to be 'AF_INET'

    with Listener(address, authkey=b'secret password') as listener:
        with listener.accept() as conn:
            print('connection accepted from', listener.last_accepted)

            conn.send([2.25, None, 'junk', float])

            conn.send_bytes(b'hello')

            conn.send_bytes(array('i', [42, 1729]))

Наступний код підключається до сервера та отримує деякі дані з сервера:

    from multiprocessing.connection import Client
    from array import array

    address = ('localhost', 6000)

    with Client(address, authkey=b'secret password') as conn:
        print(conn.recv())                  # => [2.25, None, 'junk', float]

        print(conn.recv_bytes())            # => 'hello'

        arr = array('i', [0, 0, 0, 0, 0])
        print(conn.recv_bytes_into(arr))    # => 8
        print(arr)                          # => array('i', [42, 1729, 0, 0, 0])

Наступний код використовує `wait()` для очікування повідомлень від кількох процесів одночасно:

    import time, random
    from multiprocessing import Process, Pipe, current_process
    from multiprocessing.connection import wait

    def foo(w):
        for i in range(10):
            w.send((i, current_process().name))
        w.close()

    if __name__ == '__main__':
        readers = []

        for i in range(4):
            r, w = Pipe(duplex=False)
            readers.append(r)
            p = Process(target=foo, args=(w,))
            p.start()
            # We close the writable end of the pipe now to be sure that
            # p is the only process which owns a handle for it.  This
            # ensures that when p closes its handle for the writable end,
            # wait() will promptly report the readable end as being ready.
            w.close()

        while readers:
            for r in wait(readers):
                try:
                    msg = r.recv()
                except EOFError:
                    readers.remove(r)
                else:
                    print(msg)

#### Формати адрес

- Адреса «AF_INET» — це кортеж у формі «(ім’я хоста, порт)», де ім’я хоста — рядок, а порт — ціле число.

- Адреса «AF_UNIX» — це рядок, що представляє назву файлу у файловій системі.

- An `'AF_PIPE'` address is a string of the form `r'\\.\pipe\PipeName'`. To use `Client()` to connect to a named pipe on a remote computer called ServerName one should use an address of the form `r'\\ServerName\pipe\PipeName'` instead.

Зауважте, що будь-який рядок, який починається двома зворотними похилими рисками, за замовчуванням вважається адресою `'AF_PIPE'`, а не адресою `'AF_UNIX'`.

### Ключі автентифікації

Коли використовується `Connection.recv`, отримані дані автоматично видаляються. На жаль, видалення даних із ненадійного джерела становить загрозу безпеці. Тому `Listener` і `Client()` використовують модуль `hmac` для забезпечення автентифікації дайджесту.

Ключ автентифікації — це рядок байтів, який можна розглядати як пароль: коли з’єднання встановлено, обидва кінці вимагатимуть підтвердження того, що інший знає ключ автентифікації. (Демонстрація того, що обидві сторони використовують той самий ключ, не передбачає надсилання ключа через з’єднання.)

Якщо автентифікація запитується, але ключ автентифікації не вказано, тоді використовується значення, що повертається `current_process().authkey` (див. `Process`). Це значення буде автоматично успадковано будь-яким об’єктом `Process`, який створює поточний процес. Це означає, що (за замовчуванням) усі процеси багатопроцесної програми спільно використовуватимуть один ключ автентифікації, який можна використовувати під час встановлення з’єднань між собою.

Відповідні ключі автентифікації також можна згенерувати за допомогою `os.urandom()`.

### Лісозаготівля

Доступна певна підтримка журналювання. Однак зауважте, що пакунок `logging` не використовує спільні блокування процесів, тому (залежно від типу обробника) повідомлення від різних процесів можуть переплутатися.

multiprocessing.get_logger()

  Повертає реєстратор, який використовується `multiprocessing`. За потреби буде створено новий.

  When first created the logger has level `logging.NOTSET` and no default handler. Messages sent to this logger will not by default propagate to the root logger.

  Зауважте, що у Windows дочірні процеси успадковуватимуть лише рівень реєстратора батьківського процесу – будь-які інші налаштування реєстратора не успадковуватимуться.

multiprocessing.log_to_stderr(level=None)

  Ця функція виконує виклик `get_logger()`, але окрім повернення реєстратора, створеного get_logger, вона додає обробник, який надсилає вихідні дані до `sys.stderr` у форматі `'[%(levelname)s/%(processName)s] %(message)s ''`. Ви можете змінити `levelname` реєстратора, передавши аргумент `level`.

Нижче наведено приклад сеансу з увімкненим журналюванням:

    >>> import multiprocessing, logging
    >>> logger = multiprocessing.log_to_stderr()
    >>> logger.setLevel(logging.INFO)
    >>> logger.warning('doomed')
    [WARNING/MainProcess] doomed
    >>> m = multiprocessing.Manager()
    [INFO/SyncManager-...] child process calling self.run()
    [INFO/SyncManager-...] created temp directory /.../pymp-...
    [INFO/SyncManager-...] manager serving at '/.../listener-...'
    >>> del m
    [INFO/MainProcess] sending shutdown message to manager
    [INFO/SyncManager-...] manager exiting with exitcode 0

Щоб отримати повну таблицю рівнів журналювання, перегляньте модуль `logging`.

### Модуль `multiprocessing.dummy`

`multiprocessing.dummy` повторює API `multiprocessing`, але є не більше ніж обгорткою модуля `threading`.

Зокрема, функція `Pool`, надана `multiprocessing.dummy`, повертає екземпляр `ThreadPool`, який є підкласом `Pool`, який підтримує всі виклики методів, але використовує пул робочих потоків, а не робочих процесів.

class multiprocessing.pool.ThreadPool([processes[, initializer[, initargs]]])

  Об’єкт пулу потоків, який керує пулом робочих потоків, до яких можна надсилати завдання. Екземпляри `ThreadPool` повністю сумісні з інтерфейсом екземплярів `Pool`, і їхніми ресурсами також потрібно правильно керувати, використовуючи пул як контекстний менеджер або викликаючи `close()` і `terminate()` вручну.

  processes is the number of worker threads to use. If processes is `None` then the number returned by `os.cpu_count()` is used.

  Якщо initializer не `None`, тоді кожен робочий процес викличе `initializer(*initargs)` під час свого запуску.

  На відміну від `Pool`, maxtasksperchild і context не можна надати.

  Примітка

  `ThreadPool` має той самий інтерфейс, що й `Pool`, який розроблено навколо пулу процесів і передує появі модуля `concurrent.futures`. Таким чином, він успадковує деякі операції, які не мають сенсу для пулу, що підтримується потоками, і має власний тип для представлення статусу асинхронних завдань, `AsyncResult`, який не розуміється жодною іншою бібліотекою.

  Зазвичай користувачі мають віддавати перевагу використанню `concurrent.futures.ThreadPoolExecutor`, який має простіший інтерфейс, розроблений навколо потоків із самого початку та повертає `concurrent.futures.Future` екземпляри, сумісні з багатьма інші бібліотеки, включаючи `asyncio`.

## Інструкції з програмування

Існують певні вказівки та ідіоми, яких слід дотримуватися під час використання `multiprocessing`.

### Всі методи запуску

Наступне стосується всіх методів запуску.

Уникайте спільного стану

> Наскільки це можливо, слід намагатися уникати переміщення великих обсягів даних між процесами.

> Ймовірно, найкраще використовувати черги або канали для зв’язку між процесами, а не використовувати примітиви синхронізації нижчого рівня.

Пробірність

> Переконайтеся, що аргументи методів проксі-серверів можна вибрати.

Безпека потоків проксі

> Не використовуйте проксі-об’єкт із кількох потоків, якщо ви не захистите його за допомогою блокування.

> (Ніколи не виникає проблем із різними процесами, які використовують той самий проксі.)

Приєднання до зомбованих процесів

> On Unix when a process finishes but has not been joined it becomes a zombie. There should never be very many because each time a new process starts (or `active_children()` is called) all completed processes which have not yet been joined will be joined. Also calling a finished process’s `Process.is_alive` will join the process. Even so it is probably good practice to explicitly join all the processes that you start.

Краще успадкувати, ніж маринувати/розмаринувати

> Під час використання методів запуску spawn або forkserver багато типів із `multiprocessing` мають бути доступними для вибору, щоб дочірні процеси могли їх використовувати. Однак зазвичай слід уникати надсилання спільних об’єктів іншим процесам за допомогою каналів або черг. Натомість ви повинні організувати програму так, щоб процес, якому потрібен доступ до спільного ресурсу, створеного в іншому місці, міг успадкувати його від процесу-предка.

Уникайте завершення процесів

> Використання методу `Process.terminate` для зупинки процесу може призвести до того, що будь-які спільні ресурси (такі як блокування, семафори, канали та черги), які зараз використовуються цим процесом, стануть несправними або недоступними для інших процесів.

> Тому, ймовірно, найкраще використовувати `Process.terminate` лише для процесів, які ніколи не використовують спільні ресурси.

Приєднання до процесів, які використовують черги

> Майте на увазі, що процес, який поставив елементи в чергу, чекатиме перед завершенням, доки всі буферизовані елементи не будуть передані потоком «фідера» до основного каналу. (Дочірній процес може викликати метод `Queue.cancel_join_thread` черги, щоб уникнути такої поведінки.)

> Це означає, що щоразу, коли ви використовуєте чергу, вам потрібно переконатися, що всі елементи, які було поставлено в чергу, зрештою буде видалено перед приєднанням до процесу. Інакше ви не можете бути впевнені, що процеси, які поставили елементи в чергу, завершаться. Пам’ятайте також, що недемонічні процеси будуть приєднані автоматично.

> Прикладом, який призведе до взаємоблокування, є наступний:

>     from multiprocessing import Process, Queue

>     def f(q):
>         q.put('X' * 1000000)

>     if __name__ == '__main__':
>         queue = Queue()
>         p = Process(target=f, args=(queue,))
>         p.start()
>         p.join()                    # this deadlocks
>         obj = queue.get()

> Виправити тут можна було б поміняти місцями останні два рядки (або просто видалити рядок `p.join()`).

Явно передати ресурси дочірнім процесам

> On Unix using the fork start method, a child process can make use of a shared resource created in a parent process using a global resource. However, it is better to pass the object as an argument to the constructor for the child process.

> Окрім того, що код (потенційно) сумісний із Windows та іншими методами запуску, це також гарантує, що поки дочірній процес живий, об’єкт не збиратиме сміття в батьківському процесі. Це може бути важливо, якщо якийсь ресурс звільняється, коли об’єкт збирається як сміття в батьківському процесі.

> Так наприклад

>     from multiprocessing import Process, Lock

>     def f():
>         ... do something using "lock" ...

>     if __name__ == '__main__':
>         lock = Lock()
>         for i in range(10):
>             Process(target=f).start()

> слід переписати як

>     from multiprocessing import Process, Lock

>     def f(l):
>         ... do something using "l" ...

>     if __name__ == '__main__':
>         lock = Lock()
>         for i in range(10):
>             Process(target=f, args=(lock,)).start()

Остерігайтеся заміни `sys.stdin` на «файлоподібний об’єкт»

> `multiprocessing` спочатку безумовно називався:

>     os.close(sys.stdin.fileno())

> у методі `multiprocessing.Process._bootstrap()` — це призвело до проблем із процесами в процесах. Це було змінено на:

>     sys.stdin.close()
>     sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)

> Which solves the fundamental issue of processes colliding with each other resulting in a bad file descriptor error, but introduces a potential danger to applications which replace `sys.stdin()` with a «file-like object» with output buffering. This danger is that if multiple processes call `close()` on this file-like object, it could result in the same data being flushed to the object multiple times, resulting in corruption.

> Якщо ви пишете файлоподібний об’єкт і використовуєте власне кешування, ви можете зробити його безпечним для розгалуження, зберігаючи pid кожного разу, коли ви додаєте його до кешу, і відкидаючи кеш, коли pid змінюється. Наприклад:

>     @property
>     def cache(self):
>         pid = os.getpid()
>         if pid != self._pid:
>             self._pid = pid
>             self._cache = []
>         return self._cache

> Для отримання додаткової інформації перегляньте bpo-5155, bpo-5313 та bpo-5331

### Методи запуску spawn і forkserver

There are a few extra restrictions which don’t apply to the fork start method.

Більше маринування

> Переконайтеся, що всі аргументи `Process.__init__()` можна вибрати. Крім того, якщо ви створите підклас `Process`, то переконайтеся, що екземпляри можна вибрати під час виклику методу `Process.start`.

Глобальні змінні

> Майте на увазі, що якщо код, запущений у дочірньому процесі, намагається отримати доступ до глобальної змінної, тоді значення, яке він бачить (якщо таке є), може не збігатися зі значенням у батьківському процесі під час `Process.start викликано`.

> Однак глобальні змінні, які є лише константами рівня модуля, не викликають проблем.

Безпечне імпортування основного модуля

> Make sure that the main module can be safely imported by a new Python interpreter without causing unintended side effects (such as starting a new process).

> Наприклад, використання методу запуску spawn або forkserver під час запуску наступного модуля призведе до помилки з `RuntimeError`:

>     from multiprocessing import Process

>     def foo():
>         print('hello')

>     p = Process(target=foo)
>     p.start()

> Натомість слід захистити «точку входу» програми за допомогою `if __name__ == '__main__':` наступним чином:

>     from multiprocessing import Process, freeze_support, set_start_method

>     def foo():
>         print('hello')

>     if __name__ == '__main__':
>         freeze_support()
>         set_start_method('spawn')
>         p = Process(target=foo)
>         p.start()

> (Рядок `freeze_support()` можна опустити, якщо програма буде працювати нормально, а не зависати.)

> Це дозволяє щойно створеному інтерпретатору Python безпечно імпортувати модуль, а потім запускати функцію foo() модуля.

> Подібні обмеження застосовуються, якщо пул або менеджер створено в основному модулі.

## Приклади

Демонстрація створення та використання налаштованих менеджерів і проксі-серверів:

    from multiprocessing import freeze_support
    from multiprocessing.managers import BaseManager, BaseProxy
    import operator

    ##

    class Foo:
        def f(self):
            print('you called Foo.f()')
        def g(self):
            print('you called Foo.g()')
        def _h(self):
            print('you called Foo._h()')

    # A simple generator function
    def baz():
        for i in range(10):
            yield i*i

    # Proxy type for generator objects
    class GeneratorProxy(BaseProxy):
        _exposed_ = ['__next__']
        def __iter__(self):
            return self
        def __next__(self):
            return self._callmethod('__next__')

    # Function to return the operator module
    def get_operator_module():
        return operator

    ##

    class MyManager(BaseManager):
        pass

    # register the Foo class; make `f()` and `g()` accessible via proxy
    MyManager.register('Foo1', Foo)

    # register the Foo class; make `g()` and `_h()` accessible via proxy
    MyManager.register('Foo2', Foo, exposed=('g', '_h'))

    # register the generator function baz; use `GeneratorProxy` to make proxies
    MyManager.register('baz', baz, proxytype=GeneratorProxy)

    # register get_operator_module(); make public functions accessible via proxy
    MyManager.register('operator', get_operator_module)

    ##

    def test():
        manager = MyManager()
        manager.start()

        print('-' * 20)

        f1 = manager.Foo1()
        f1.f()
        f1.g()
        assert not hasattr(f1, '_h')
        assert sorted(f1._exposed_) == sorted(['f', 'g'])

        print('-' * 20)

        f2 = manager.Foo2()
        f2.g()
        f2._h()
        assert not hasattr(f2, 'f')
        assert sorted(f2._exposed_) == sorted(['g', '_h'])

        print('-' * 20)

        it = manager.baz()
        for i in it:
            print('<%d>' % i, end=' ')
        print()

        print('-' * 20)

        op = manager.operator()
        print('op.add(23, 45) =', op.add(23, 45))
        print('op.pow(2, 94) =', op.pow(2, 94))
        print('op._exposed_ =', op._exposed_)

    ##

    if __name__ == '__main__':
        freeze_support()
        test()

Використання `Pool`:

    import multiprocessing
    import time
    import random
    import sys

    #
    # Functions used by test code
    #

    def calculate(func, args):
        result = func(*args)
        return '%s says that %s%s = %s' % (
            multiprocessing.current_process().name,
            func.__name__, args, result
            )

    def calculatestar(args):
        return calculate(*args)

    def mul(a, b):
        time.sleep(0.5 * random.random())
        return a * b

    def plus(a, b):
        time.sleep(0.5 * random.random())
        return a + b

    def f(x):
        return 1.0 / (x - 5.0)

    def pow3(x):
        return x ** 3

    def noop(x):
        pass

    #
    # Test code
    #

    def test():
        PROCESSES = 4
        print('Creating pool with %d processes\n' % PROCESSES)

        with multiprocessing.Pool(PROCESSES) as pool:
            #
            # Tests
            #

            TASKS = [(mul, (i, 7)) for i in range(10)] + \
                    [(plus, (i, 8)) for i in range(10)]

            results = [pool.apply_async(calculate, t) for t in TASKS]
            imap_it = pool.imap(calculatestar, TASKS)
            imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)

            print('Ordered results using pool.apply_async():')
            for r in results:
                print('\t', r.get())
            print()

            print('Ordered results using pool.imap():')
            for x in imap_it:
                print('\t', x)
            print()

            print('Unordered results using pool.imap_unordered():')
            for x in imap_unordered_it:
                print('\t', x)
            print()

            print('Ordered results using pool.map() --- will block till complete:')
            for x in pool.map(calculatestar, TASKS):
                print('\t', x)
            print()

            #
            # Test error handling
            #

            print('Testing error handling:')

            try:
                print(pool.apply(f, (5,)))
            except ZeroDivisionError:
                print('\tGot ZeroDivisionError as expected from pool.apply()')
            else:
                raise AssertionError('expected ZeroDivisionError')

            try:
                print(pool.map(f, list(range(10))))
            except ZeroDivisionError:
                print('\tGot ZeroDivisionError as expected from pool.map()')
            else:
                raise AssertionError('expected ZeroDivisionError')

            try:
                print(list(pool.imap(f, list(range(10)))))
            except ZeroDivisionError:
                print('\tGot ZeroDivisionError as expected from list(pool.imap())')
            else:
                raise AssertionError('expected ZeroDivisionError')

            it = pool.imap(f, list(range(10)))
            for i in range(10):
                try:
                    x = next(it)
                except ZeroDivisionError:
                    if i == 5:
                        pass
                except StopIteration:
                    break
                else:
                    if i == 5:
                        raise AssertionError('expected ZeroDivisionError')

            assert i == 9
            print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
            print()

            #
            # Testing timeouts
            #

            print('Testing ApplyResult.get() with timeout:', end=' ')
            res = pool.apply_async(calculate, TASKS[0])
            while 1:
                sys.stdout.flush()
                try:
                    sys.stdout.write('\n\t%s' % res.get(0.02))
                    break
                except multiprocessing.TimeoutError:
                    sys.stdout.write('.')
            print()
            print()

            print('Testing IMapIterator.next() with timeout:', end=' ')
            it = pool.imap(calculatestar, TASKS)
            while 1:
                sys.stdout.flush()
                try:
                    sys.stdout.write('\n\t%s' % it.next(0.02))
                except StopIteration:
                    break
                except multiprocessing.TimeoutError:
                    sys.stdout.write('.')
            print()
            print()

    if __name__ == '__main__':
        multiprocessing.freeze_support()
        test()

Приклад, який показує, як використовувати черги для передачі завдань у колекцію робочих процесів і збору результатів:

    import time
    import random

    from multiprocessing import Process, Queue, current_process, freeze_support

    #
    # Function run by worker processes
    #

    def worker(input, output):
        for func, args in iter(input.get, 'STOP'):
            result = calculate(func, args)
            output.put(result)

    #
    # Function used to calculate result
    #

    def calculate(func, args):
        result = func(*args)
        return '%s says that %s%s = %s' % \
            (current_process().name, func.__name__, args, result)

    #
    # Functions referenced by tasks
    #

    def mul(a, b):
        time.sleep(0.5*random.random())
        return a * b

    def plus(a, b):
        time.sleep(0.5*random.random())
        return a + b

    #
    #
    #

    def test():
        NUMBER_OF_PROCESSES = 4
        TASKS1 = [(mul, (i, 7)) for i in range(20)]
        TASKS2 = [(plus, (i, 8)) for i in range(10)]

        # Create queues
        task_queue = Queue()
        done_queue = Queue()

        # Submit tasks
        for task in TASKS1:
            task_queue.put(task)

        # Start worker processes
        for i in range(NUMBER_OF_PROCESSES):
            Process(target=worker, args=(task_queue, done_queue)).start()

        # Get and print results
        print('Unordered results:')
        for i in range(len(TASKS1)):
            print('\t', done_queue.get())

        # Add more tasks using `put()`
        for task in TASKS2:
            task_queue.put(task)

        # Get and print some more results
        for i in range(len(TASKS2)):
            print('\t', done_queue.get())

        # Tell child processes to stop
        for i in range(NUMBER_OF_PROCESSES):
            task_queue.put('STOP')

    if __name__ == '__main__':
        freeze_support()
        test()