Ensuring that events raised by system.diagnostics process class happen in the parent thread

Question

I'm quite new to threading primitives in C# and was hoping you might be able to suggest improvements to this. I need to ensure that the XXX call below happens within the calling thread (XXX is a foreign call into a thread-unsafe library), so I used a queue here. It seems a bit like there should be a better primitive for this. Maybe delegates are applicable somehow? I don't understand delegates.

I also have to wonder if I've gotten this whole scheme right in the first place! Maybe there's a deadlock I'm not seeing. Threading is so tricky.

As an additional restriction, it's very important that this works on .NET 3.5.

    public void RunProc(AutoResetEvent killSubProc)
    {
        using (Process process = new Process())
        {
            var timeout = 8000;
            var channel = new Queue<string> {};

            process.StartInfo.FileName = "blah.exe";
            process.StartInfo.Arguments = @"stuff";

            process.StartInfo.UseShellExecute = false;
            process.StartInfo.RedirectStandardOutput = true;
            process.StartInfo.RedirectStandardError = true;
            process.EnableRaisingEvents = true;

            using (AutoResetEvent channelWaitHandle = new AutoResetEvent(false))
            {
                process.OutputDataReceived += (sender, e) => {
                    if (e.Data != null)
                    {
                        lock (channel) { channel.Enqueue("STDOUT"); channel.Enqueue(e.Data); }
                        channelWaitHandle.Set();
                    }
                };
                process.ErrorDataReceived += (sender, e) =>
                {
                    if (e.Data != null)
                    {
                        lock (channel) { channel.Enqueue("STDERR"); channel.Enqueue(e.Data); }
                        channelWaitHandle.Set();
                    }
                };
                process.Exited += (sender, e) =>
                {
                    lock (channel) { channel.Enqueue("EXIT"); }
                    channelWaitHandle.Set();
                };

                process.Start();

                process.BeginOutputReadLine();
                process.BeginErrorReadLine();

                bool running = true;
                while (running)
                {
                    int idx = WaitHandle.WaitAny(new WaitHandle[] {killSubProc, channelWaitHandle});
                    if (idx == 0)
                    {
                        process.Kill();
                        running = false;
                    }
                    else
                    {
                        lock (channel)
                        {
                            while (channel.Count > 0)
                            {
                                var item = channel.Dequeue();
                                XXX(item);

                                if (item == "EXIT")
                                {
                                    running = false;
                                }
                            }
                        } 
                    }
                }
            }
        }
    }

Answer 1

If you want a pretty solution, you would have to separate your data source (process in your case) from data processor (your while loop), hide implementations behind interfaces and then wire those interfaces together via events or aggregation. This is actually a pretty common problem called "Producer-Consumer", you should be able to find multiple decent implementations on the net.

If you want a simple solution though, yours is probably fine. A few things you might want to improve:

1. Reduce nesting in your while loop to avoid arrow code. This can be done by using break/continue/return statements or by extracting logic to separate method or entity.
2. Depending on how often you receive messages and on how much time XXXtakes, you might want to release lock before calling XXX.
3. You might want to wait for process to exit before returning form your method. This will ensure, that everything went smoothly, and there is no process left hanging somewhere due to some error in your code.

4. If XXX signature is not set in stone, you should probably replace strings with complex object:

   class Message
   {
      //use enum instead of hardcoded strings, such as "STDERR"
      public MessageType Type { get; set; }
      //actual data from process
      public string Data { get; set; }
   } 
   

   You should remove hardcoded "magic" strings either way.

Answer 2

In a context where thread affinity is required and a SynchronizationContext is not available (as in console applications or windows services), a blocking implementation is required.

Our main consideration for a console application could be, to execute the XXX operations in a background thread instead of the main thread. I don't know what your XXX method is doing, but, if it requires execution on the thread where the library instance is created, create that instance in that background thread and process the queueu in that same thread so that your UI is still responsive.

Anyways, I will go with the main thread.

I have just a few recommendations:

WaitHandle.WaitAny(new WaitHandle[] {killSubProc, channelWaitHandle});

This wait can cause an issue. If the remote application sends a message and while execution of xxx() continues on the main thread, another message is received, that new message will not be processed until another new message is received.

Main Thread   -> Wait               xxx()----------------- WaitAny       xxx()
Thread X      ->         Message Set
Thread Y      ->                          Message Set                     
Thread Z      ->                                              Message Set

Another suggestion:

Move the queue and queueing/dequeueing logic into a class to de-couple it's implementation from invocation of XXX (Make a reusable blocking queue)

Here we have a very generic Message class to store the callback handler and the payload (e.data in your case)

internal class Message
{
    private readonly SendOrPostCallback handler;
    private readonly object payload;

    internal Message(SendOrPostCallback handler, object payload)
    {
        this.handler = handler;
        this.payload = payload;
    }

    internal void Execute()
    {
        handler(payload);
    }
}

And another very short class for keeping track of the queue and to execute the operations (XXX in your case). Please check the use of Queue.Synchronized

public class BlockingMessageSynchronizer
{
    private Queue synchronizationQueue;
    private bool stopRequested;

    public void Start()
    {
        lock (this)
        {
            if (synchronizationQueue != null)
            {
                throw new InvalidOperationException("The synchronization has already started.");
            }
            synchronizationQueue = Queue.Synchronized(new Queue());
            stopRequested = false;
        }

        while (!stopRequested)
        {
            // Thread.Sleep(1);
            ProcessQueue();
        }

        ProcessQueue();
        synchronizationQueue = null;
    }

    public void Post(SendOrPostCallback messageHandler, object payload)
    {
        if (synchronizationQueue == null || stopRequested)
        {
            lock (this)
            {
                if (synchronizationQueue == null || stopRequested)
                {
                    throw new InvalidOperationException("The synhronization is not started or is stopped");
                }
            }
        }
        synchronizationQueue.Enqueue(new Message(messageHandler, payload));
    }

    public void Stop()
    {
        stopRequested = true;
    }

    private void ProcessQueue()
    {
        while (synchronizationQueue.Count > 0)
        {
            ((Message)synchronizationQueue.Dequeue()).Execute();
        }
    }
}

This class can be used on ay thread to force execution of the callback handler on that thread, by blocking that thread. This implementation will cause 100% cpu since there is no Wait operations, you can consider Thread.Sleep(1) before the call to ProcessQueue() in the while loop in Start()

You should have a method with the following signature to execute XXX(); because XXX has a signature accepting a string and this is no common abstraction:

private static void HandleCrossThreadMessage(object state)
{
    XXX(state as string);
}

And finally, your stdout or stderr handlers will look like:

process.OutputDataReceived += (sender, e) =>
{
    if (e.Data != null)
    {
        synchronizer.Post(HandleCrossThreadMessage, e.Data);
    }
};

process.ErrorDataReceived += (sender, e) =>
{
    if (e.Data != null)
    {
        synchronizer.Post(HandleCrossThreadMessage, "STDERR");
        synchronizer.Post(HandleCrossThreadMessage, e.Data);
    }
};

process.Exited += (sender, e) =>
{
    synchronizer.Post(HandleCrossThreadMessage, "EXIT");
    synchronizer.Stop();
};

BlockingMessageSynchronizer synchronizer = new BlockingMessageSynchronizer();

process.Start();

process.BeginOutputReadLine();
process.BeginErrorReadLine();

synchronizer.Start();

When you call synchronizer.Stop(); inside the exit event, the remaining messages in the queue will be processed and the execution will get out of the synchronizer.Start(); blocking call.