Event-based producer/consumer in C#

Question

I want to create a production-ready producer/consumer that could help me avoid thread synchronization hell. Is this thread-safe? The main issue is to be safe with exceptions that can arrive.

public class AsynchSimpleProducer<T> : IDisposable
{
    private readonly Action<Exception> error;
    private readonly BlockingCollection<T> blockingCollection = new BlockingCollection<T>(50);

    public event Func<T, Task> NewItem;

    public AsynchSimpleProducer(Action<Exception> error)
    {
        this.error = error;

        Task.Factory.StartNew(() =>
                              {
                                  //this loop is ends only when blockingCollection.CompleteAdding called
                                  Parallel.ForEach(blockingCollection.GetConsumingPartitioner(), SendItem);
                              });
    }

    private void SendItem(T item)
    {
        try
        {
            if (NewItem !=null)
                NewItem(item).Wait();
        }
        catch (Exception ex)
        {
            error(ex);
        }
    }

    public void Send(T newValue)
    {
        blockingCollection.Add(newValue);
    }

    public void Dispose()
    {
        blockingCollection.CompleteAdding();
    }
}

Example of usage:

simpleProducer = new AsynchSimpleProducer<int>(Error);
simpleProducer.NewItem += simpleProducer_NewItem;
//this part can be in task
var next = random.Next(0, 1000);
simpleProducer.Send(next);

Can you explain a bit more about the specific problem you are trying to solve? What do you mean by "avoiding thread synchronisation hell"? — RobH
– RobH, Commented Jun 11, 2015 at 8:35
@Robh locks,two separate functions that depends on each other — Avram
– Avram, Commented Jun 11, 2015 at 8:47
You need to add an extended example because that's not enough detail to provide a meaningful answer - I could review what you've written but I don't understand the point of it so it won't be very meaningful — RobH
– RobH, Commented Jun 11, 2015 at 8:53
I have rolled back the last edit. Please see what you may and may not do after receiving answers. — Heslacher
– Heslacher, Commented Jun 11, 2015 at 9:27

RobH · Accepted Answer · 2015-06-11 11:48:44Z

I don't really understand how you want to use this so I'm just going to add a couple of comments for you to ponder.

Naming

It's more common to abbreviate Asynchronous to Async not Asynch when used in identifers: AsynchSimpleProducer<T> -> AsyncSimpleProducer<T>.

NewItem isn't a great name for an event, events generally need a verb to make sense e.g. Closed, Clicked, Closing are all good names for an event. Consider something like ItemAdded or similar.

Thread safety

You aren't raising your event safely:

if (NewItem != null)
    NewItem(item).Wait();

NewItem can change between the check and using it. If a listener unsubscribes in this time you'll get an Exception! The proper way to do this is:

var newItem = NewItem;
if (newItem != null)
{
    newItem(item).Wait();
}

Note that I have also added in braces as I personally think they should always be included, even for single line blocks.

Asynchronicity

I had to type something that made me very sad.

newItem(item).Wait();

It's not async if you immediately wait on it. It would be interesting to know what sort of work you're passing in here as it's not obvious whether you need to try to make this asynchronous or not. I'm concerned that you have confused threading and asynchronicity.

Edit

I can't believe I didn't mention this before, I must have been distracted...

If your event handlers throw (the things listening on the NewItem event), you're toast. Say you have 2 listeners, if the first one throws, the second wont be called.

You need to go through each one in the invocation list:

 // untested...
 var runningTasks = new List<Task>();
 foreach (var delegate in newItem.GetInvocationList())
 {
     runningTasks.Add(delegate(item));
 }
 // catch exceptions here.
 Task.WaitAll(runningTasks.ToArray());

I wish I could suggest something useful but I still haven't figure out what you intend on doing with this thing.

The code is async, i am using parallel. The wait function call, is used for error handling. — Avram
– Avram, Commented Jun 11, 2015 at 9:15
@Avram - that comment confirms that you don't understand async vs parallel... Read this and see if it makes sense: stackoverflow.com/a/14099602/1402923 — RobH
– RobH, Commented Jun 11, 2015 at 9:38
lets talk first about the purpose of the code. this code is receiving data by Send function put it in collection and forget (the producer forget.). on same time, there is "forever loop" of taking data from collection and sending the data, by event (ideally, fire and forget). I think that correct definition of this code, is non blocking Producer/Consumer (that contains event notification feature) — Avram
– Avram, Commented Jun 11, 2015 at 22:57
also the code must support error handling, especially on the part of fire and forget. — Avram
– Avram, Commented Jun 11, 2015 at 22:59

Stack Exchange Network

Event-based producer/consumer in C#

1 Answer 1

Naming

Thread safety

Asynchronicity

Edit

You must log in to answer this question.

Hot Network Questions

Event-based producer/consumer in C#

1 Answer 1

Naming

Thread safety

Asynchronicity

Edit

You must log in to answer this question.

Related

Hot Network Questions