C TPL Поток данных BatchBlock ожидает завершения никогда не возвращается - PullRequest
Новая
       10

C TPL Поток данных BatchBlock ожидает завершения никогда не возвращается

0 голосов
/ 17 ноября 2018

Я создал небольшую тестовую программу, которая иллюстрирует мою проблему 

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);
        }
    }
}

Вывод на консоль, как и ожидалось: 

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

Чего мне не хватает?

1 Ответ

0 голосов
/ 17 ноября 2018

Вам нужно позвонить Complete, когда вы закончите отправку данных в блоки.

Это: 

await items.Completion;

Чтобы дождаться завершения всего потока, действительно должно быть: 

items.Complete()
await insertItems.Completion;

Наконец, вы можете связать блок для распространения завершения, поэтому ContinueWith в этом случае не требуется.

Это: 

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

Может быть заменено на: 

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