Я пытался разработать конвейер метода с использованием асинхронного вызова метода. Логика для конвейера выглядит следующим образом
- В коллекции имеется n данных, которые должны быть переданы в число методов m в конвейере
- Перечислите коллекцию Т
- Введите первый элемент в первый метод
- Получить вывод, передать его во второй метод асинхронно
- Одновременно передать второй элемент коллекции первому методу
- После завершения первого метода передайте результат во второй метод (если второй метод все еще выполняется, поместите результат в его очередь и начните выполнение третьего элемента в первом методе)
- Когда второй метод завершает выполнение, берите первый элемент из очереди и выполняйте его и т. Д. (Каждый метод должен выполняться асинхронно, никто не должен ждать завершения следующего)
- В методе mth, после выполнения данных, сохранить результат в списке
- После завершения n-го элемента в m-м методе верните список результатов (n-количество результатов) на самый первый уровень.
Я придумал следующий код, но он не работал должным образом, результат никогда не возвращается и, более того, он не выполняется в том порядке, в каком он должен быть.
static class Program
{
static void Main(string[] args)
{
var list = new List<int> { 1, 2, 3, 4 };
var result = list.ForEachPipeline(Add, Square, Add, Square);
foreach (var element in result)
{
Console.WriteLine(element);
Console.WriteLine("---------------------");
}
Console.ReadLine();
}
private static int Add(int j)
{
return j + 1;
}
private static int Square(int j)
{
return j * j;
}
internal static void AddNotify<T>(this List<T> list, T item)
{
Console.WriteLine("Adding {0} to the list", item);
list.Add(item);
}
}
internal class Function<T>
{
private readonly Func<T, T> _func;
private readonly List<T> _result = new List<T>();
private readonly Queue<T> DataQueue = new Queue<T>();
private bool _isBusy;
static readonly object Sync = new object();
readonly ManualResetEvent _waitHandle = new ManualResetEvent(false);
internal Function(Func<T, T> func)
{
_func = func;
}
internal Function<T> Next { get; set; }
internal Function<T> Start { get; set; }
internal int Count;
internal IEnumerable<T> Execute(IEnumerable<T> source)
{
var isSingle = true;
foreach (var element in source) {
var result = _func(element);
if (Next != null)
{
Next.ExecuteAsync(result, _waitHandle);
isSingle = false;
}
else
_result.AddNotify(result);
}
if (!isSingle)
_waitHandle.WaitOne();
return _result;
}
internal void ExecuteAsync(T element, ManualResetEvent resetEvent)
{
lock(Sync)
{
if(_isBusy)
{
DataQueue.Enqueue(element);
return;
}
_isBusy = true;
_func.BeginInvoke(element, CallBack, resetEvent);
}
}
internal void CallBack(IAsyncResult result)
{
bool set = false;
var worker = (Func<T, T>) ((AsyncResult) result).AsyncDelegate;
var resultElement = worker.EndInvoke(result);
var resetEvent = result.AsyncState as ManualResetEvent;
lock(Sync)
{
_isBusy = false;
if(Next != null)
Next.ExecuteAsync(resultElement, resetEvent);
else
Start._result.AddNotify(resultElement);
if(DataQueue.Count > 1)
{
var element = DataQueue.Dequeue();
ExecuteAsync(element, resetEvent);
}
if(Start._result.Count == Count)
set = true;
}
if(set)
resetEvent.Set();
}
}
public static class Pipe
{
public static IEnumerable<T> ForEachPipeline<T>(this IEnumerable<T> source, params Func<T, T>[] pipes)
{
Function<T> start = null, previous = null;
foreach (var function in pipes.Select(pipe => new Function<T>(pipe){ Count = source.Count()}))
{
if (start == null)
{
start = previous = function;
start.Start = function;
continue;
}
function.Start = start;
previous.Next = function;
previous = function;
}
return start != null ? start.Execute(source) : null;
}
}
Не могли бы вы, ребята, помочь мне заставить эту штуку работать? Если этот дизайн не подходит для реального метода конвейера, пожалуйста, не стесняйтесь предложить другой.
Редактировать : Я должен строго придерживаться .Net 3.5.