Завершение в циклах потока данных TPL

Question

У меня возникла проблема с определением того, как обнаружить завершение в потоке потока данных TCL.

У меня есть цикл обратной связи в части потока данных, который делает запросы GET на удаленный сервер и обрабатывает ответы данных (преобразуя их с большим количеством потоков данных, а затем фиксируя результаты).

Источник данных разбивает результаты на страницы 1000 записей и не укажет, сколько у него страниц доступно. Я должен просто читать, пока не получу полную страницу данных.

Обычно количество страниц 1, часто это до 10, каждый раз и снова у нас есть 1000.

У меня есть много запросов на получение в начале.
Я хочу иметь возможность использовать пул потоков, чтобы справиться с этим, все это прекрасно, я могу ставить в очередь несколько запросов на данные и запрашивать их одновременно. Если я наткнулся на экземпляр, где мне нужно получить большое количество страниц, я хочу использовать все свои потоки для этого. Я не хочу, чтобы меня оставила одна нить, отбитая, пока остальные закончили.

Проблема у меня есть, когда я бросаю эту логику в поток данных, таких как:

//generate initial requests for activity 
var request = new TransformManyBlock<int, DataRequest>(cmp => QueueRequests(cmp)); 

//fetch the initial requests and feedback more requests to our input buffer if we need to 
TransformBlock<DataRequest, DataResponse> fetch = null; 
fetch = new TransformBlock<DataRequest, DataResponse>(async req => 
{ 
    var resp = await Fetch(req); 

    if (resp.Results.Count == 1000) 
     await fetch.SendAsync(QueueAnotherRequest(req)); 

    return resp; 
} 
, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 }); 

//commit each type of request 
var commit = new ActionBlock<DataResponse>(async resp => await Commit(resp)); 

request.LinkTo(fetch); 
fetch.LinkTo(commit); 

//when are we complete? 

QueueRequests производит IEnumerable<DataRequest>. Я заказываю очередные запросы N-страницы сразу, принимая это, что я посылаю несколько больше вызовов, чем мне нужно. В экземплярах DataRequest используется счетчик LastPage, чтобы избежать неудобного выполнения запросов, которые мы знаем, после последней страницы. Все это прекрасно.

Проблема:
Если я зациклился, возвращая больше запросов в входной буфер ввода, как я показал в этом примере, у меня возникла проблема с сигналом (или даже обнаружением) завершения. Я не могу установить завершение на выборку из запроса, так как после завершения установки я больше не могу получать обратную связь.

Я могу следить за тем, чтобы буферы ввода и вывода были пустыми при извлечении, но я думаю, что я бы рискнул извлечения, все еще занятой запросом, когда я устанавливал завершение, тем самым предотвращая запросы на очередность для дополнительных страниц.

Я мог бы с некоторой долей вероятности узнать, что выборка занята (либо имеет вход, либо занят обработкой ввода).

У меня отсутствует очевидный/простой способ решить эту проблему?

• Я мог бы зацикливаться на выбор, а не на очередность запросов. Проблема заключается в том, что я хочу иметь возможность использовать максимальное количество потоков для ограничения того, что я делаю на удаленном сервере. Может ли параллельный цикл внутри блока совместно использовать планировщик с самим блоком, а итоговое количество потоков контролируется через планировщик?

• Я мог бы создать собственный блок преобразования для извлечения для обработки сигнализации завершения. Кажется, очень много работы для такого простого сценария.

Большое спасибо за предоставленную помощь!

Answer 1

Сейчас я добавил простой занятое состояние счетчика выборки блок: -

int fetch_busy = 0; 

TransformBlock<DataRequest, DataResponse> fetch_activity=null; 
fetch = new TransformBlock<DataRequest, ActivityResponse>(async req => 
    { 
     try 
     { 
      Interlocked.Increment(ref fetch_busy); 
      var resp = await Fetch(req); 

      if (resp.Results.Count == 1000) 
      { 
       await fetch.SendAsync(QueueAnotherRequest(req)); 
      } 

      Interlocked.Decrement(ref fetch_busy); 
      return resp; 
     } 
     catch (Exception ex) 
     { 
      Interlocked.Decrement(ref fetch_busy); 
      throw ex; 
     } 
    } 
    , new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 }); 

который я затем использовать для сигнала завершения следующим образом: -

request.Completion.ContinueWith(async _ => 
    { 
     while (fetch.InputCount > 0 || fetch_busy > 0) 
     { 
      await Task.Delay(100); 
     } 

     fetch.Complete(); 
    }); 

Который не кажется очень элегантно , но я должен работать.

Answer 2

В TPL DataFlow, вы можете link the blocks с DataflowLinkOptions с указанием propagation of completion of the block:

request.LinkTo(fetch, new DataflowLinkOptions { PropagateCompletion = true }); 
fetch.LinkTo(commit, new DataflowLinkOptions { PropagateCompletion = true }); 

После этого, вы просто вызвать метод Complete() для request блока, и вы сделали!

// the completion will be propagated to all the blocks 
request.Complete(); 

Последнее, что вы должны использовать это Completion задачу свойства последнего блока:

commit.Completion.ContinueWith(t => 
    { 
     /* check the status of the task and correctness of the requests handling */ 
    });