TPL Dataflow blok zużywa całą dostępną pamięć

Question

mam TransformManyBlock z następującym wzorem:

• wejściowa: ścieżka do pliku
• wyjściowa: IEnumerable zawartości pliku, jeden wiersz naraz

Uruchomiłem ten blok na dużym pliku (61 GB), który jest zbyt duży, aby zmieścić się w pamięci RAM. Aby uniknąć przyrostu pamięci nieograniczonej, ustawiłem BoundedCapacity na bardzo niską wartość (na przykład 1) dla tego bloku i dla wszystkich bloków downstream. Mimo to blok pozornie iteruje IEnumerable łapczywie, co pochłania całą dostępną pamięć na komputerze, miażdżąc każdy proces do zatrzymania. OutputCount bloku nadal rośnie bez ograniczeń, dopóki nie zabiję procesu.

Co mogę zrobić, aby blok nie zużywał w ten sposób ?

EDIT: Oto przykładowy program, który ilustruje problem:

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Threading; 
using System.Threading.Tasks; 
using System.Threading.Tasks.Dataflow; 

class Program 
{ 
    static IEnumerable<string> GetSequence(char c) 
    { 
     for (var i = 0; i < 1024 * 1024; ++i) 
      yield return new string(c, 1024 * 1024); 
    } 

    static void Main(string[] args) 
    { 
     var options = new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 }; 
     var firstBlock = new TransformManyBlock<char, string>(c => GetSequence(c), options); 
     var secondBlock = new ActionBlock<string>(str => 
      { 
       Console.WriteLine(str.Substring(0, 10)); 
       Thread.Sleep(1000); 
      }, options); 

     firstBlock.LinkTo(secondBlock); 
     firstBlock.Completion.ContinueWith(task => 
      { 
       if (task.IsFaulted) ((IDataflowBlock) secondBlock).Fault(task.Exception); 
       else secondBlock.Complete(); 
      }); 

     firstBlock.Post('A'); 
     firstBlock.Complete(); 
     for (; ;) 
     { 
      Console.WriteLine("OutputCount: {0}", firstBlock.OutputCount); 
      Thread.Sleep(3000); 
     } 
    } 
} 

Jeśli jesteś na polu 64-bitowej, należy usunąć zaznaczenie opcji "Wolę 32-bit" w Visual Studio. Mam 16 GB pamięci RAM na moim komputerze, a ten program natychmiast zużywa każdy dostępny bajt.

Answer 1

Wydaje się, że źle rozumiesz działanie TPL Dataflow.

BoundedCapacity ogranicza liczbę przedmiotów, które można umieścić w bloku. W twoim przypadku oznacza to pojedynczy char do TransformManyBlock i pojedynczy string do ActionBlock.

Wysyłając pojedynczy element do TransformManyBlock, który następnie zwraca 1024*1024 ciągi i próbuje je przekazać do ActionBlock, które będą akceptować tylko jeden na raz. Pozostałe ciągi po prostu usiądą tam w kolejce wyjściowej TransformManyBlock.

Co prawdopodobnie chcesz zrobić, to stworzyć pojedyncze elementy blokowe i umieszczać w nim w sposób przesyłania strumieniowego przez oczekiwania (synchronicznie lub w inny sposób), kiedy to pojemność zostanie osiągnięty:

private static void Main() 
{ 
    MainAsync().Wait(); 
} 

private static async Task MainAsync() 
{ 
    var block = new ActionBlock<string>(async item => 
    { 
     Console.WriteLine(item.Substring(0, 10)); 
     await Task.Delay(1000); 
    }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); 

    foreach (var item in GetSequence('A')) 
    { 
     await block.SendAsync(item); 
    } 

    block.Complete(); 
    await block.Completion; 
}