I was recently asked a question in an interview and it really got me thinking.
I am trying to understand and learn more about multithreading, parallelism and concurrency, and performance.
The scenario is that you have a list of file paths. Files are saved on your HDD or on blob storage. You have read the files and store them in a database. How would you do it in the most optimal manner?
The following are some of the ways that I could think of:
The simplest way is to loop through the list and perform this task sequentially.
Foreach(var filePath in filePaths)
{
ProcessFile(filePath);
}
public void ProcessFile(string filePath)
{
var file = readFile(filePath);
storeInDb(file);
}
2nd way I could think of is creating multiple threads perhaps:
Foreach(var filePath in filePaths)
{
Thread t = new Thread(ProcessFIle(filePath));
t.Start();
}
(not sure if the above code is correct.)
3rd way is using async await
List<Tasks> listOfTasks;
Foreach(var filePath in filePaths)
{
var task = ProcessFile(filePath);
listOfTasks.Add(task);
}
Task.WhenAll(listOftasks);
public async void ProcessFile(string filePath)
{
var file = readFile(filePath);
storeInDb(file);
}
4th way is Parallel.For:
Parallel.For(0,filePaths.Count , new ParallelOptions { MaxDegreeOfParallelism = 10 }, i =>
{
ProcessFile(filePaths[i]);
});
What are the differences between them. Which one would be better suited for the job and is there anything better?
You could also use Microsoft's Reactive Framework (aka Rx) - NuGet System.Reactive and add using System.Reactive.Linq; - then you can do this:
IObservable<string> query =
from filePath in filePaths.ToObservable()
from file in Observable.Start(() => ReadFile(filePath))
from db in Observable.Start(() => StoreInDb(file))
select filePath;
IDisposable subscription =
query
.Subscribe(
filePath => Console.WriteLine($"{filePath} Processed."),
() => Console.WriteLine("Done."));
I wrote a simple extension method to help start async tasks, limit the amount of concurrency, and wait for them all to complete;
public static async Task WhenAll(this IEnumerable<Task> tasks, int batchSize)
{
var started = new List<Task>();
foreach(var t in tasks)
{
started.Add(t);
if (started.Count >= batchSize)
{
var ended = await Task.WhenAny(started);
started.Remove(ended);
}
}
await Task.WhenAll(started);
}
Then you'd want a method to stream the file contents directly into the database. For example;
async Task Process(string filename){
using var stream = File.OpenRead(filename)
// TODO connect to the database
var sqlCommand = ...;
sqlCommand.CommandText = "update [table] set [column] = @stream";
sqlCommand.Parameters.Add(new SqlParameter("@stream", SqlDbType.VarBinary)
{
Value = stream
});
await sqlCommand.ExecuteNonQueryAsync();
}
IEnumerable<string> files = ...;
await files.Select(f => Process(f)).WhenAll(20);
Is this the best approach? Probably not. Since it's too easy to misuse this extension. Accidently starting tasks multiple times, or starting them all at once.