Thread-safety bug for the pipeline's stopping state

Hi, I receive exceptions when trying to dispose a data replay pipeline close to its end. Two threads enter Pipeline.Stop at a similar time, which leads to one thread thinking the pipeline is in the Completed state while the other is still trying to stop it.

Context

I am implementing a seeking capability for replaying data from a store using the PsiStoreStreamReader and Importer. The importer reads data from a store and disposes of the pipeline when there are no more messages. The user can stop the replay at any time, which calls Pipeline.Dispose from another thread.

When (T1) the replay stops on its own and (T2) user disposes of the pipeline at a similar time, there are cases where two threads (T1 + T2) have entered Pipeline.Stop simultaneously. The current way of handling this isn't sufficient as the following IsStopping check evaluates to false for both threads, since its state is set to Stopping after the check:
https://github.com/microsoft/psi/blob/35cb04ce24569677d3a8da0216e9450e66cbbd83/Sources/Runtime/Microsoft.Psi/Executive/Pipeline.cs#L836

Solution

In terms of keeping consistent with your state pattern, I'm not sure if you want to ensure thread safety for every pipeline state. The most pressing fix was to change:
if (this.IsStopping) -> if (Interlocked.CompareExchange(ref this.isPipelineStopping, 1, 0) != 0)

Which is similar to how you check if a pipeline's been disposed: https://github.com/microsoft/psi/blob/35cb04ce24569677d3a8da0216e9450e66cbbd83/Sources/Runtime/Microsoft.Psi/Executive/Pipeline.cs#L682

You could also have this private property private long isPipelineStopping = 0; as a long type so you can keep using the IsStopping getter as IsStopping => Interlocked.Read(ref this.isPipelineStopping) == 1;

Start in Sources/Runtime/Microsoft.Psi/Executive/Pipeline.cs at Pipeline.Stop and compare its stopping-state handling with the disposal check in Pipeline.Dispose. Confirm the fix prevents concurrent stop calls from observing inconsistent state; done means disposing near replay completion no longer produces the reported exceptions.