Multi-Threaded Processor allows you to efficiently process data in parallel across multiple threads, using both simple and advanced approaches for scatter-gather operations.
Unlike a traditional "scatter-gather", MTP allows you to separate the loading and processing across threads and then await the results. This is incredibly powerful when the loading is done from a remote source where latency is involved.
An Example
PerigeeApplication.ApplicationNoInit("MTP", (c) => {
c.Add("Multi-Threaded Processor", (ct, l) => {
//Let's declare a list of fruits to work on
var Fruits = new List<string>()
{
"Apples",
"Bananas",
"Cherries",
"Dates",
"Elderberries",
"Figs",
"Grapes",
"Honeydew",
"Indian Fig",
"Jackfruit",
"Kiwi",
"Lemon",
"Mango",
"Nectarine",
"Oranges",
"Papaya",
"Quince",
"Raspberries",
"Strawberries",
"Tangerines",
"Ugli Fruit",
"Vanilla Bean",
"Watermelon",
"Xigua",
"Yellow Passion Fruit"
};
l.LogInformation("--------============ Example 1 ============--------");
//Example 1: Using the IEnumerable extension method to simple "scatter gather" across a number of defined threads. In this case, 5 threads are used
var mtpResult = Fruits.ParallelProcessMultiThread((s) =>
{
l.LogInformation("Processing {fruit}", s);
Task.Delay(400).Wait(); //Artificial delay
return s.GetHashCode();
}, null, concurrency: 5);
//Every object in this case is wrapped in a result object, giving you exceptions thrown, process time, and access to the input/output
foreach (var rs in mtpResult)
{
l.LogInformation("Passed in {in}, got {out}, in {time}ms", rs.InData, rs.OutData, rs.ProcessTime?.TotalMilliseconds.ToString("N0") ?? "");
}
//Example 2: Separate the loading and processing of the scatter gather across threads
// In this example, we're declaring the multi-threaded processor directly, along with the input type, output type, and the method to be called.
// Notice the 3 on the ThreadCount? This will only process and declare 3 threads to work on input items
l.LogInformation("--------============ Example 2 ============--------");
var mtp = new MultiThreadedProcessor<string, int>((s) =>
{
l.LogInformation("Processing {fruit}", s);
return s.GetHashCode();
}, ct, ThreadCount: 3, l);
//To actively watch for processed items, attach to the event
mtp.OnDataProcessed += (sender, args) =>
{
l.LogInformation("Passed in {in}, got {out}, in {time}ms", args.InData, args.OutData, args.ProcessTime?.TotalMilliseconds.ToString("N0") ?? "");
};
//Now let's simulate a delayed loading of the items by using Enqueue
foreach (var fruit in Fruits)
{
mtp.Enqueue(fruit);
Task.Delay(Random.Shared.Next(50, 200)).Wait(); //Artificial random delay
}
//Once all of the items have been enqueued, wait for the last item to be processed
mtp.AwaitProcessed(ct);
//Done! Everything has been processed and we have successfully awaited a multi-threaded, separated "scatter gather"
while (PerigeeApplication.delayOrCancel(1000, ct)) { };
});
});
SDK
ParallelProcessMultiThread
Processes a collection of items in parallel using a specified processing function.
Parameters:
items (IEnumerable<TIn>): The collection of items to process.
Process (Func<TIn, TOut>): The function to process each item.
cancelToken (CancellationToken?, optional): An optional cancellation token to cancel the processing.
concurrency (int, optional): The number of concurrent threads to use. Default is 3.
Logger (ILogger, optional): An optional logger for logging.
Returns:
List<MultiProcessResult<TIn, TOut>>: A list of processing results.
Example:
var results = myItems.ParallelProcessMultiThread(item => ProcessItem(item));
Enqueue
Enqueue an item to be processed.
Parameters:
item (TIn): The item to be processed.
Example:
processor.Enqueue(myItem);
AwaitProcessed
Await until all pending executions have been processed. Optionally accepts a cancellation token or a time span to limit the wait.
Example:
processor.AwaitProcessed();
// With CancellationToken
processor.AwaitProcessed(cancellationToken);
// With TimeSpan
processor.AwaitProcessed(TimeSpan.FromMinutes(1));
Class Reference
MultiProcessResult<TIn, TOut>
A class that represents the result of processing an individual item.
InData
The input data that was processed.
OutData
The output data produced by the process.
Exception
An optional exception that occurred during processing.
ProcessTime
An optional timespan representing the processing time.
