Многопроцессорная синхронизация

Question

Предположим, мне нужно запустить 5 процессов параллельно, но процессы с 2 по 5 зависят от первого процесса. Как я могу убедиться, что процесс 1 будет запущен раньше остальных? Должен ли я использовать многопроцессорное событие Python () или Lock () или оба?

Пример 1:

process 1
process 2 or 3 or 4 or 5
process 2 or 3 or 4 or 5
process 2 or 3 or 4 or 5
process 2 or 3 or 4 or 5

Пример 2:

process 3
process 1 or 2 or 4 or 5
process 1 or 2 or 4 or 5
process 1 or 2 or 4 or 5
process 1 or 2 or 4 or 5

Пример 3 с 2 зависимостями:

process 1
process 2 or 3 (run in parallel after 1)
process 4
process 5 or 6 (run in parallel after 1 and after 4)

Все процессы вызывают одну и ту же функцию (msg), но все возвращают разные значения.

Мне нужно некоторое руководство, не обязательно код, если вы можете предоставить, спасибо.

Псевдокод:

import Multiprocessing as mp

function(msg):
    return 1 if msg == "one"
    return 2 if msg == "two"
    return 3 if msg == "three"
    return 4 if msg == "four"
    return 5 if msg == "five"

msgs = ['one', 'two', 'three', 'four', 'five']

jobs = []
for msg in msgs:
    p = Process(target=function, args=(msg,))
    p.start()
    jobs.append(p)

for job in jobs:
    job.join()

В этом случае все процессы будут работать неупорядоченно.

Если я хочу, чтобы процесс 1 был выполнен до того, как я смог сделать:

Возможное решение:

import Multiprocessing as mp

function(msg):
    return 1 if msg == "one"
    return 2 if msg == "two"
    return 3 if msg == "three"
    return 4 if msg == "four"
    return 5 if msg == "five"

msg = ['one']
p1 = Process(target=function, args=(msg,))
p1.start()
p1.join()


msgs = ['two', 'three', 'four', 'five']

jobs = []
for msg in msgs:
    p = Process(target=function, args=(msg,))
    p.start()
    jobs.append(p)

for job in jobs:
    job.join()

Есть ли лучшее решение или как это хорошо? Это работает, но это не значит, что это не может быть сделано лучше (например, меньше повторений кода).

Answer 1

Не уверен, что было сделано в конце, но в конце концов вы можете использовать Event s для этой цели:

import multiprocessing as mp

def function(msg,events):
  if msg == "one":
    print(1)
    events[0].set()
  if msg == "two":
    print("2 waiting")
    events[0].wait()
    events[1].wait()
    print("2 done")
  if msg == "three":
    print(3)
    events[1].set()
  if msg == "four":
    print(4)
  if msg == "five":
    print("5 waiting")
    events[0].wait()
    print("5 done")

if __name__ == '__main__':
  events = [mp.Event(),mp.Event()]
  jobs = []
  for item in ['one','two','three','four','five']:
    job = mp.Process(target=function, args=(item,events))
    job.start()
    jobs.append(job)
  for job in jobs:
    job.join()

Здесь я намеренно ввел вторую зависимость: p2 зависит как от p1, так и от p3(и p5 все еще зависит от p1). Таким образом, если вы запустите его пару раз, он покажет больше вариаций (чем с одной зависимостью):

python procy.py
2 waiting
4
1
5 waiting
5 done
3
2 done

python procy.py
1
5 waiting
2 waiting
4
5 done
3
2 done

python procy.py
1
4
3
5 waiting
5 done
2 waiting
2 done