Exceptions in Parallel.ForEach, Parallel.ForEachAsync, Task.Run and Task.Run.ContinueWith

Details
Written by: Stanko Milosev
Category: C#
Hits: 1450
First example Task.Run: 
Task task1 = Task.Run(() => throw new SystemException("Test exception in Task.Run"));
Task task2 = Task.Run(async () =>
{
	await Task.Delay(1000);
	WriteToTextBox("Completed");
});

try
{
	await Task.WhenAll(task1, task2);
}
catch (Exception ex)
{
	WriteToTextBox(ex.Message);
}
This will output: 
Completed
Test exception in Task.Run
Second, Task.Run.ContinueWith, exception will be swallowed: 
Task task1 = Task.Run(() => throw new SystemException("Test exception in Task.Run"))
	.ContinueWith(t => WriteToTextBox("ContinueWith exception test"));

Task task2 = Task.Run(async () =>
{
	await Task.Delay(1000);
	WriteToTextBox("Completed");
});

try
{
	await Task.WhenAll(task1, task2);
}
catch (Exception ex)
{
	WriteToTextBox(ex.Message);
}
Output: 
ContinueWith exception test
Completed
Parallel.ForEach, exception will be thrown, but not catched outside of Parallel.ForEach, and program will end: 
string[] myItems = ["item1", "item2", "item3", "item4", "item5"];

Task task1;
Task task2;

try
{
	Parallel.ForEach(myItems, async void (myItem) =>
	{
		task1 = Task.Run(() => throw new SystemException("Test exception in Task.Run"));
		task2 = Task.Run(async () =>
		{
			await Task.Delay(1000);
			WriteToTextBox(myItem);
		});
		await Task.WhenAll(task1, task2);
	});
}
catch (Exception ex)
{
	WriteToTextBox(ex.Message);
}
Parallel.ForEachAsync: 
string[] myItems = ["item1", "item2", "item3", "item4", "item5"];

Task task1;
Task task2;

using CancellationTokenSource cts = new();
CancellationToken token = cts.Token;

try
{
	int i = 0;
	await Parallel.ForEachAsync(myItems, new ParallelOptions
		{
			CancellationToken = token
		}
		, async (myItem, ct) =>
		{
			i = Interlocked.Increment(ref i);
			task1 = Task.Run(() => throw new SystemException($"Test exception in Task.Run {i}"), ct);
			task2 = Task.Run(async () =>
			{
				await Task.Delay(1000, ct);
				WriteToTextBox(myItem);
			}, ct);
			await Task.WhenAll(task1, task2);
		});
}
catch (Exception ex)
{
	WriteToTextBox(ex.Message);
}
Only one exception will be thrown in output: 
item4
item2
item1
item5
item3
Test exception in Task.Run 5
Example download from here

Update UI from Parallel.ForEach

Details
Written by: Stanko Milosev
Category: C#
Hits: 863
Here is one my list of possibilities, how to update UI from Parallel.ForEach.

First, and probably the best with IProgress: 

private readonly IProgress<int> _recordCountProgress;
private readonly IProgress<string> _fileNameProgress;
...
Parallel.ForEach(Directory.EnumerateFiles(path, "*.*", SearchOption.AllDirectories), options, fileName =>
{
	RecordCount = Interlocked.Increment(ref _recordCount);
	if (RecordCount % progressStep == 0)
	{
		_fileNameProgress.Report(fileName);
		_recordCountProgress.Report(RecordCount);
	}
});
...
IProgress<int> recordCountProgress = new Progress<int>(NumberOfFilesProcessedIProgress);
IProgress<string> fileNameProgress = new Progress<string>(FileProcessedIProgress);
notice: 
	if (RecordCount % progressStep == 0)
If there are only few files _recordCount will be always 0, because it is to fast, if there are too many files, without UI would be blocked.

Second with SynchronizationContext: 

Parallel.ForEach(Directory.EnumerateFiles(path, "*.*", SearchOption.AllDirectories), options, fileName =>
{
	RecordCount = Interlocked.Increment(ref _recordCount);
	if (RecordCount % progressStep == 0)
	{
			_syncContext?.Post(_ =>
			{
				FileProcessed?.Invoke(this,
					new FilesProcessedSynchronizationContextEventArgs(fileName));
				NumberOfFilesProcessed?.Invoke(this,
					new RecordCountSynchronizationContextEventArgs(_recordCount));
			}, null);
	}
});
Same as previous, problem with the line: 
	if (RecordCount % progressStep == 0)
Third with System.Threading.Timer: 
CancellationTokenSource = cancellationTokenSource;
_timer = new System.Threading.Timer(_ =>
{
	if (_queue.TryDequeue(out string? fileName))
	{
		if (form.InvokeRequired)
		{
			form.BeginInvoke(() =>
			{
				FileProcessed?.Invoke(this, new FilesProcessedEventArgs(fileName));
				NumberOfFilesProcessed?.Invoke(this, new RecordCountEventArgs(_recordCount));
			});
		}
		else
		{
			FileProcessed?.Invoke(this, new FilesProcessedEventArgs(fileName));
			NumberOfFilesProcessed?.Invoke(this, new RecordCountEventArgs(_recordCount));
		}
	}
}, null, 0, 100);
Example download from here

Copy CSV file to MS SQL with SqlBulkCopy and IDataReader

Details
Written by: Stanko Milosev
Category: C#
Hits: 906
Here is my example first main part: 
await using var sqlConnection = new SqlConnection(connectionString);
using var sqlBulkCopy = new SqlBulkCopy(sqlConnection);
using var csvDataReader = new CsvDataReader(csvPath);
await sqlConnection.OpenAsync();
sqlBulkCopy.DestinationTableName = "GpsInfo";
await sqlBulkCopy.WriteToServerAsync(csvDataReader);
Now IDataReader 
using System.Data;

namespace SaveCsvFileToSqServerWithBulkCopy;

public class CsvDataReader: IDataReader
{
    private readonly StreamReader _reader;
    private string[]? _currentRow;
    private readonly string[]? _headers;

    public CsvDataReader(string filePath)
    {
        _reader = new StreamReader(filePath);
        _headers = _reader.ReadLine()?.Split(';');
    }

    public bool Read()
    {
        if (_reader.EndOfStream) return false;
        _currentRow = _reader.ReadLine()?.Split(';');
        return true;
    }

    public object GetValue(int i) => _currentRow[i];
    public int FieldCount => _headers.Length;
    public void Dispose() => _reader.Dispose();

    public bool GetBoolean(int i)
    {
        throw new NotImplementedException();
    }

    public byte GetByte(int i)
    {
        throw new NotImplementedException();
    }

    public long GetBytes(int i, long fieldOffset, byte[]? buffer, int bufferoffset, int length)
    {
        throw new NotImplementedException();
    }

    public char GetChar(int i)
    {
        throw new NotImplementedException();
    }

    public long GetChars(int i, long fieldoffset, char[]? buffer, int bufferoffset, int length)
    {
        throw new NotImplementedException();
    }

    public IDataReader GetData(int i)
    {
        throw new NotImplementedException();
    }

    public string GetDataTypeName(int i)
    {
        throw new NotImplementedException();
    }

    public DateTime GetDateTime(int i)
    {
        throw new NotImplementedException();
    }

    public decimal GetDecimal(int i)
    {
        throw new NotImplementedException();
    }

    public double GetDouble(int i)
    {
        throw new NotImplementedException();
    }

    public Type GetFieldType(int i)
    {
        throw new NotImplementedException();
    }

    public float GetFloat(int i)
    {
        throw new NotImplementedException();
    }

    public Guid GetGuid(int i)
    {
        throw new NotImplementedException();
    }

    public short GetInt16(int i)
    {
        throw new NotImplementedException();
    }

    public int GetInt32(int i)
    {
        throw new NotImplementedException();
    }

    public long GetInt64(int i)
    {
        throw new NotImplementedException();
    }

    public string GetName(int i)
    {
        throw new NotImplementedException();
    }

    public int GetOrdinal(string name)
    {
        throw new NotImplementedException();
    }

    public string GetString(int i)
    {
        throw new NotImplementedException();
    }

    public int GetValues(object[] values)
    {
        throw new NotImplementedException();
    }

    public bool IsDBNull(int i)
    {
        throw new NotImplementedException();
    }

    public object this[int i] => throw new NotImplementedException();

    public object this[string name] => throw new NotImplementedException();


    public void Close()
    {
        throw new NotImplementedException();
    }

    public DataTable? GetSchemaTable()
    {
        throw new NotImplementedException();
    }

    public bool NextResult()
    {
        throw new NotImplementedException();
    }

    public int Depth { get; }
    public bool IsClosed { get; }
    public int RecordsAffected { get; }
}
Full example download from here

Producer-Consumer Pattern

Details
Written by: Stanko Milosev
Category: C#
Hits: 719
Hier is one my example of Producer-Consumer Pattern.

First I want to read all files from HDD and put them in the queue (producer): 

private void SearchForAllFilesAndPutThemInQueue(BlockingCollection<string> fileQueue, string path)
{
	foreach (string file in Directory.EnumerateFiles(path, "*.*", SearchOption.AllDirectories))
	{
		fileQueue.Add(file);
	}
}
Then I will read file names from queue and do something with (consumer): 
private void ReadFileNamesFromQueue
(
	BlockingCollection<string>? fileQueue
	, BlockingCollection<(LatLngModel, string)>? gpsInfoQueue
)
{
	int i = 0;

	if (fileQueue is null) throw new ArgumentNullException(nameof(fileQueue));
	if (gpsInfoQueue is null) throw new ArgumentNullException(nameof(gpsInfoQueue));

	foreach (string file in fileQueue.GetConsumingEnumerable())
	{
		try
		{
			ExtractGpsInfoFromImageCommand extractGpsInfoFromImageCommand = new ExtractGpsInfoFromImageCommand();
			extractGpsInfoFromImageCommand.ImageFileNameToReadGpsFrom = file;
			extractGpsInfoFromImage.Execute(extractGpsInfoFromImageCommand);

			if (extractGpsInfoFromImageCommand.LatLngModel is not null)
			{
				gpsInfoQueue.Add((extractGpsInfoFromImageCommand.LatLngModel, file));
			}
		}
		catch (Exception ex)
		{
			Debug.WriteLine(ex.Message());
		}
	}
}
Here you can download my example, little bit complicated, where I get list of file in one queue, in second I try to extract GPS info, and from third queue I am saving to CSV file file name and longitude / latitude.
  1. Tasks.Run vs Thread
  2. Logger wrapper for Windows Forms
  3. Validating XML with XSL in .Net Core
  4. Validating XML with Schematron

Subcategories

WPF

Beginning

Code snippets

NUnit

LINQ

Windows Forms

Page 3 of 39