Race condition in local flow launcher

Hello,

here's a small repro that demonstrates a race condition when launching parallel local runs with Metaflow 2.7.15

race_flow.py:

from metaflow import FlowSpec, Parameter, step


class RaceFlow(FlowSpec):

    message = Parameter("message")

    @step
    def start(self):
        self.message2 = self.message
        self.next(self.print_message)

    @step
    def print_message(self):
        print(self.message2)
        self.next(self.end)

    @step
    def end(self):
        print(f"printed message: {self.message2}")


if __name__ == "__main__":
    RaceFlow()

run_race_flows.sh:

#!/bin/sh

status="OK"

while [ "$status" == "OK" ]; do
    (
        python race_flow.py run --message AAA &
        python race_flow.py run --message BBB &
        wait
    ) > run_race_flows.log 2>&1

    status=$(grep -q AAA run_race_flows.log && grep -q BBB run_race_flows.log && echo OK)
done

output:

❯ sh run_race_flows.sh
❯ cat run_race_flows.log
Metaflow 2.7.15Metaflow 2.7.15 executing RaceFlow executing RaceFlow for user:jarno.seppanen for 
user:jarno.seppanen
Validating your flow...
Validating your flow...
    The graph looks good!
Running pylint...
    The graph looks good!
Running pylint...
    Pylint is happy!    Pylint is happy!

2023-01-18 11:27:35.695 Workflow starting (run-id 1674034055691361):
2023-01-18 11:27:35.695 Workflow starting (run-id 1674034055691361):
2023-01-18 11:27:35.699 [1674034055691361/start/1 (pid 12699)] Task is starting.
2023-01-18 11:27:35.699 [1674034055691361/start/1 (pid 12700)] Task is starting.
2023-01-18 11:27:36.098 [1674034055691361/start/1 (pid 12699)] Task finished successfully.
2023-01-18 11:27:36.098 [1674034055691361/start/1 (pid 12700)] Task finished successfully.
2023-01-18 11:27:36.102 [1674034055691361/print_message/2 (pid 12706)] Task is starting.
2023-01-18 11:27:36.102 [1674034055691361/print_message/2 (pid 12705)] Task is starting.
2023-01-18 11:27:36.424 [1674034055691361/print_message/2 (pid 12706)] 2023-01-18 11:27:36.424 BBB
[1674034055691361/print_message/2 (pid 12705)] BBB
2023-01-18 11:27:36.488 [1674034055691361/print_message/2 (pid 12705)] Task finished successfully.
2023-01-18 11:27:36.489 [1674034055691361/print_message/2 (pid 12706)] Task finished successfully.
2023-01-18 11:27:36.492 [1674034055691361/end/3 (pid 12711)] 2023-01-18 11:27:36.492 Task is starting.
[1674034055691361/end/3 (pid 12712)] Task is starting.
2023-01-18 11:27:36.815 [1674034055691361/end/3 (pid 12711)] printed message: BBB
2023-01-18 11:27:36.815 [1674034055691361/end/3 (pid 12712)] printed message: BBB
2023-01-18 11:27:36.878 2023-01-18 11:27:36.878 [1674034055691361/end/3 (pid 12711)] [1674034055691361/end/3 (pid 12712)] Task finished successfully.
Task finished successfully.
2023-01-18 11:27:36.879 2023-01-18 11:27:36.879 Done!
Done!

Expected output: the parallel jobs should print both AAA and BBB every time, but sometimes they print AAA twice or BBB twice.

Start by running race_flow.py through run_race_flows.sh with the two parallel messages and confirm the duplicate-output race. Trace the local flow launcher used by this reproduction; done means repeated parallel runs consistently print both AAA and BBB rather than one value twice.