Suspended event handling with C#

Question

I have 2 classes:

public class A
{
    private const int MAXCOUNTER = 100500;
    private Thread m_thrd;
    public event Action<string> ItemStarted;
    public event Action<string> ItemFinished;

    private void OnItemStarted(string name)
    {
        if (ItemStarted != null) ItemStarted(name);
    }

    private void OnItemFinished(string name)
    {
        if (ItemFinished != null) ItemFinished(name);
    }

    public A()
    {
        m_thrd = new Thread(this.Run);
        m_thrd.Start();
    }

    private void Run()
    {
        for (int i = 0; i < MAXCOUNTER; i++)
        {
            OnItemStarted(i.ToString());
            // some long term operations
            OnItemFinished(i.ToString());
        }
    }
}

public class B
{
    private Thread m_thrd;
    private Queue<string> m_data;
    public B()
    {
        m_thrd = new Thread(this.ProcessData);
        m_thrd.Start();
    }

    public void ItemStartedHandler(string str)
    {
        m_data.Enqueue(str);
    }

    public void ItemFinishedHandler(string str)
    {
        if (m_data.Dequeue() != str)
            throw new Exception("dequeued element is not the same as finish one!");
    }

    private void ProcessData()
    {
        lock (m_data)
        {
            while (m_data.Count != 0)
            {
                var item = m_data.Peek();
                //make some long term operations on the item
            }
        }
    }
}

also we have somewhere else in code

A a = new A();
B b = new B();
a.ItemStarted += b.ItemStartedHandler;
a.ItemFinished += b.ItemFinishedHandler;

• So, if ItemFinished is raised while ProcessData() is still working, what will happen?
• Should I use something like AutoResetEvent to make class A wait on class B finishes ProcessData?
• Is it necessary to use lock in ProcessData?
• Is it ok to invoke thread of class B with m_thrd = new Thread(this.ProcessData);? The thing is confusing me here - won't ProcessData finish before any ItemStarted event is raised (won't it lead to situation when thread in B already is finished when ItemStarted is generated first time)?

Answer 1

Your code is a bit of a threading mess.

Your first issue is a race condition if ItemFinished is raised before ProcessData() is finished.

Don't worry about using AutoResetEvent. The simple thing is to lock every access to m_data. So that leads to the next quertion - yes, the lock is necessary and it's necessary everywhere.

But your final point is the most important. You need to change each of the constructors to a .Start() method to allow time to wire up before you begin.

But, even then you have a massive race condition.

Here's what you should do to make this work. NuGet "Rx-Main" to add Microsoft's Reactive Framework to your code. Then do this:

var scheduler1 = new EventLoopScheduler();
var scheduler2 = new EventLoopScheduler();

Observable
    .Range(0, 100500)
    .ObserveOn(scheduler1)
    .Do(x => { /* some long term operations */ })
    .ObserveOn(scheduler2)
    .Subscribe(x =>
    {
        //make some long term operations on the item
    });

Job done.