Use spawn start method in multiprocessing programs (#11391)

* Use `spawn` start method in multiprocessing programs

* Set `spawn` start method in doctest

* Use `with` statement for locks

* Pass multiprocessing context explicitly
This commit is contained in:
Xuehai Pan 2024-05-03 00:43:59 +08:00 committed by GitHub
parent 5131e3145d
commit ea53051576
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -11,11 +11,11 @@ They are synchronized with locks and message passing but other forms of
synchronization could be used. synchronization could be used.
""" """
from multiprocessing import Lock, Pipe, Process import multiprocessing as mp
# lock used to ensure that two processes do not access a pipe at the same time # lock used to ensure that two processes do not access a pipe at the same time
# NOTE This breaks testing on build runner. May work better locally # NOTE This breaks testing on build runner. May work better locally
# process_lock = Lock() # process_lock = mp.Lock()
""" """
The function run by the processes that sorts the list The function run by the processes that sorts the list
@ -29,8 +29,17 @@ resultPipe = the pipe used to send results back to main
""" """
def oe_process(position, value, l_send, r_send, lr_cv, rr_cv, result_pipe): def oe_process(
process_lock = Lock() position,
value,
l_send,
r_send,
lr_cv,
rr_cv,
result_pipe,
multiprocessing_context,
):
process_lock = multiprocessing_context.Lock()
# we perform n swaps since after n swaps we know we are sorted # we perform n swaps since after n swaps we know we are sorted
# we *could* stop early if we are sorted already, but it takes as long to # we *could* stop early if we are sorted already, but it takes as long to
@ -38,27 +47,23 @@ def oe_process(position, value, l_send, r_send, lr_cv, rr_cv, result_pipe):
for i in range(10): for i in range(10):
if (i + position) % 2 == 0 and r_send is not None: if (i + position) % 2 == 0 and r_send is not None:
# send your value to your right neighbor # send your value to your right neighbor
process_lock.acquire() with process_lock:
r_send[1].send(value) r_send[1].send(value)
process_lock.release()
# receive your right neighbor's value # receive your right neighbor's value
process_lock.acquire() with process_lock:
temp = rr_cv[0].recv() temp = rr_cv[0].recv()
process_lock.release()
# take the lower value since you are on the left # take the lower value since you are on the left
value = min(value, temp) value = min(value, temp)
elif (i + position) % 2 != 0 and l_send is not None: elif (i + position) % 2 != 0 and l_send is not None:
# send your value to your left neighbor # send your value to your left neighbor
process_lock.acquire() with process_lock:
l_send[1].send(value) l_send[1].send(value)
process_lock.release()
# receive your left neighbor's value # receive your left neighbor's value
process_lock.acquire() with process_lock:
temp = lr_cv[0].recv() temp = lr_cv[0].recv()
process_lock.release()
# take the higher value since you are on the right # take the higher value since you are on the right
value = max(value, temp) value = max(value, temp)
@ -94,39 +99,60 @@ def odd_even_transposition(arr):
>>> odd_even_transposition(unsorted_list) == sorted(unsorted_list + [1]) >>> odd_even_transposition(unsorted_list) == sorted(unsorted_list + [1])
False False
""" """
# spawn method is considered safer than fork
multiprocessing_context = mp.get_context("spawn")
process_array_ = [] process_array_ = []
result_pipe = [] result_pipe = []
# initialize the list of pipes where the values will be retrieved # initialize the list of pipes where the values will be retrieved
for _ in arr: for _ in arr:
result_pipe.append(Pipe()) result_pipe.append(multiprocessing_context.Pipe())
# creates the processes # creates the processes
# the first and last process only have one neighbor so they are made outside # the first and last process only have one neighbor so they are made outside
# of the loop # of the loop
temp_rs = Pipe() temp_rs = multiprocessing_context.Pipe()
temp_rr = Pipe() temp_rr = multiprocessing_context.Pipe()
process_array_.append( process_array_.append(
Process( multiprocessing_context.Process(
target=oe_process, target=oe_process,
args=(0, arr[0], None, temp_rs, None, temp_rr, result_pipe[0]), args=(
0,
arr[0],
None,
temp_rs,
None,
temp_rr,
result_pipe[0],
multiprocessing_context,
),
) )
) )
temp_lr = temp_rs temp_lr = temp_rs
temp_ls = temp_rr temp_ls = temp_rr
for i in range(1, len(arr) - 1): for i in range(1, len(arr) - 1):
temp_rs = Pipe() temp_rs = multiprocessing_context.Pipe()
temp_rr = Pipe() temp_rr = multiprocessing_context.Pipe()
process_array_.append( process_array_.append(
Process( multiprocessing_context.Process(
target=oe_process, target=oe_process,
args=(i, arr[i], temp_ls, temp_rs, temp_lr, temp_rr, result_pipe[i]), args=(
i,
arr[i],
temp_ls,
temp_rs,
temp_lr,
temp_rr,
result_pipe[i],
multiprocessing_context,
),
) )
) )
temp_lr = temp_rs temp_lr = temp_rs
temp_ls = temp_rr temp_ls = temp_rr
process_array_.append( process_array_.append(
Process( multiprocessing_context.Process(
target=oe_process, target=oe_process,
args=( args=(
len(arr) - 1, len(arr) - 1,
@ -136,6 +162,7 @@ def odd_even_transposition(arr):
temp_lr, temp_lr,
None, None,
result_pipe[len(arr) - 1], result_pipe[len(arr) - 1],
multiprocessing_context,
), ),
) )
) )