CODEWITHSUNDEEP
c#tpl-dataflow
Zeus82
Zeus82

Reputation: 6375

C TPL Dataflow BatchBlock await Completion never returns

I created a small test program that illustrates my issue

class Item
{
    public string Name;
}

class Program
{
    static BatchBlock<Item> items = new BatchBlock<Item>(2);
    static ConcurrentBag<Item> bag = new ConcurrentBag<Item>();

    public static async Task Main(string[] args)
    {
        var insertItems = new ActionBlock<Item[]>(b => BatchProcessor(b));
        items.LinkTo(insertItems);
        var finished = items.Completion.ContinueWith(delegate { insertItems.Complete(); }).ConfigureAwait(false);

        Console.WriteLine("From main 1");
        items.Post(new Item() { Name = "1" });

        Console.WriteLine("From main 2");
        items.Post(new Item() { Name = "2" });

        Console.WriteLine("From main 3");
        items.Post(new Item() { Name = "3" });

        Console.WriteLine("From main 4");
        items.Post(new Item() { Name = "4" });

        Console.WriteLine("From main 5");
        items.Post(new Item() { Name = "5" });

        Console.WriteLine("From main 6");
        items.Post(new Item() { Name = "6" });

        Console.WriteLine("From main 7");
        items.Post(new Item() { Name = "7" });

        Console.WriteLine("Finishing");
        items.TriggerBatch();

        await items.Completion; // Never completes!!!
    }

    static void BatchProcessor(Item[] items)
    {
        Console.WriteLine("Callback");
        foreach(var i in items)
        {
            Console.WriteLine("From Callback " + i.Name);
            bag.Add(i);
        }
    }
}

The console output is as expected:

From main 1
From main 2
From main 3
From main 4
From main 5
From main 6
From main 7
Finishing
Callback
From Callback 1
From Callback 2
Callback
From Callback 3
From Callback 4
Callback
From Callback 5
From Callback 6
Callback
From Callback 7

What am I missing?

Upvotes: 1

Views: 503

Answers (1)

JSteward
JSteward

Reputation: 7091

You need to call Complete when you're done sending data to the blocks.

This:

await items.Completion;

In order to await completion of the entire flow, should really be:

items.Complete()
await insertItems.Completion;

Finally, you can link block to propagate completion so the ContinueWith in this case is not necessary.

This:

items.Completion.ContinueWith(delegate { insertItems.Complete(); }).ConfigureAwait(false);

Can be replaced with:

items.LinkTo(insertItems, new DataflowLinkOptions() { PropagateCompletion = true });

Upvotes: 7

Related Questions