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

Details
Written by: Stanko Milosev
Category: C#
Hits: 196
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: 128
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: 180
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: 170
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

C#

Azure

ASP.NET

JavaScript

Software Development Philosophy

MS SQL

IBM WebSphere MQ

MySQL

Joomla

Delphi

PHP

Windows

Life

Lazarus

Downloads

Android

CSS

Chrome

HTML

Linux

Eclipse

Page 2 of 163