2
\$\begingroup\$

I put together a ProducerConsumer class that wraps the C#'s BlockingCollection class with some bells and whistles (producers can have multiple consumers, consumers can have lambdas, built in parallel execution, exceptions thrown on cycles, etc). As you can imagine it uses generics.

var producer = new GhostLine<int>();
var consumerDoubler = new GhostLine<int>();
var consumerTripler = new GhostLine<int>();
producer.AddConsumer(consumerDoubler);
producer.AddConsumer(consumerTripler);

...

consumerTripler.Process = w =>
{
    int x = w * 3;
    Console.WriteLine("Tripled value: " + x);
    return x;
};

The whole thing is (or seems to be working) and both of my unit tests succeed when I run them.

Some areas I am interested in: performance, parallelism (obviously ... including potential deadlocks, best mechanisms between tasks/Parallel.For/threads/etc), ease of use, convenience methods / features, encapsulation, and hosting on github (long time programmer but new to github).

You can find the project here.

The class I am most interested in getting feedback on is GhostLine.cs:

    public void AddConsumer(GhostLine<T> consumer) {
        if (AncestorDuplicate(consumer) || DescendantDuplicate(consumer))
        {
            throw new CycleException("Adding this consumer would create a cycle.");
        }

        consumer.Parent = this;
        Consumers.Add(consumer);
    }

    ///<summary>
    ///Iterates through all the WorkUnits until the producer calls CompleteAdding() and all added WorkUnits have been processed.
    ///</summary>
    public void Begin()
    {
        // if the external process doesn't set up a process, just pass the value along
        if (ProcessCompleted.GetInvocationList().Count() == 0 && PassValuesIfNoProcessDelegate)
        {
            Process = w =>
            {
                return w;
            };
        }

        Task.Run(() =>
        {
            InnerWorkLoop();
        });
    }

    private void InnerWorkLoop()
    {
        foreach (var workUnit in WorkUnits.GetConsumingEnumerable())
        {
            // requires WorkUnits to be empty and IsCompleted to be true
            var transformedWorkUnit = Process(workUnit);

            if (!EqualityComparer<T>.Default.Equals(transformedWorkUnit, default(T)))
            {
                foreach (var consumer in Consumers)
                {
                    consumer.WorkUnits.Add(transformedWorkUnit);
                }
            }

            if (ExecuteInParallel)
            {
                Parallel.ForEach(Consumers, (consumer) =>
                {
                    consumer.Begin();
                });
            }
            else
            {
                foreach (var consumer in Consumers)
                {
                    consumer.Begin();
                }
            }
        }
        OnProcessCompleted(EventArgs.Empty);
    }

    ///<summary>
    ///This method is called when the Begin() method completes.
    ///</para>If the ProcessCompleted handler has any delegates they will be called here.</para>
    ///</summary>
    protected virtual void OnProcessCompleted(EventArgs e)
    {
        if (ProcessCompleted != null)
            ProcessCompleted(this, e);
    }
\$\endgroup\$
  • 2
    \$\begingroup\$ Please include all code required for us to make an assesment of this code. I feel at least the entire class should be posted here, including source code of classes you call. Without this code, this question is off-topic for this site. \$\endgroup\$ – dfhwze Aug 24 '19 at 21:47