Add record in Db from Parallel.ForEachAsync and Ef Core

Details
Written by: Stanko Milosev
Category: C#
Hits: 1157
Here one my example to insert new record in DB using entity framework core, and Parallel.ForEachAsync.

First install Microsoft.EntityFrameworkCore, then install Microsoft.EntityFrameworkCore.SqlServer

FileNamesEntity, or how the table in DB looks like: 

public class FileNamesTestEntity
{
    [Key] public int Id { get; set; }
    public string? FileName { get; set; }
}
FileNamesTestDbContext: 
public class FileNamesTestDbContext : DbContext
{
    public DbSet<FileNamesTestEntity> FileNamesTest { get; set; }
    public FileNamesTestDbContext(DbContextOptions<FileNamesTestDbContext> options) : base(options) { }
}
And the code: 
using Microsoft.EntityFrameworkCore;

namespace UpdateOrInsertRecordInDbFromParallelForEachAsync
{
    public partial class Form1 : Form
    {
        private CancellationTokenSource CancellationTokenSource { get; } = new();
        public Form1()
        {
            InitializeComponent();
        }

        private async void btnStartIProgress_Click(object sender, EventArgs e)
        {
            int recordCount = 0;
            IProgress<int> recordCountProgress = new Progress<int>(NumberOfFilesProcessedIProgress);
            IProgress<string> fileNameProgress = new Progress<string>(FileProcessedIProgress);
            int progressStep = 100;

            var options = new DbContextOptionsBuilder<FileNamesTestDbContext>()
                .UseSqlServer("Server=myServer;Database=MyDb;User Id=myUser;Password=myPass;Encrypt=True;TrustServerCertificate=True;")
                .Options;

            await Parallel.ForEachAsync(Directory.EnumerateFiles(textBox1.Text, "*.*", SearchOption.AllDirectories), new ParallelOptions
            {
                CancellationToken = CancellationTokenSource.Token
            }
                , async (fileName, ct) =>
                {
                    await using var dbContext = new FileNamesTestDbContext(options);
                    FileNamesTestEntity fileNamesTestEntity = new FileNamesTestEntity();
                    fileNamesTestEntity.FileName = fileName;
                    dbContext.FileNamesTest.Add(fileNamesTestEntity);
                    await dbContext.SaveChangesAsync(ct);

                    recordCount = Interlocked.Increment(ref recordCount);
                    if (recordCount % progressStep == 0)
                    {
                        fileNameProgress.Report(fileName);
                        recordCountProgress.Report(recordCount);
                    }
                });

            MessageBox.Show("Done!");
        }

        private void FileProcessedIProgress(string fileName)
        {
            lblIProgressUiFileName.Text = fileName;
        }

        private void NumberOfFilesProcessedIProgress(int recordCount)
        {
            lblIProgressUiRecordCount.Text = $"Processed files: {recordCount}";
        }

        private void btnStopIProgress_Click(object sender, EventArgs e)
        {
            CancellationTokenSource.Cancel();
        }
    }
}
Example download from here

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

Details
Written by: Stanko Milosev
Category: C#
Hits: 1517
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: 934
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: 982
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
  1. Producer-Consumer Pattern
  2. Tasks.Run vs Thread
  3. Logger wrapper for Windows Forms
  4. Validating XML with XSL in .Net Core

Subcategories

WPF

Beginning

Code snippets

NUnit

LINQ

Windows Forms

Page 3 of 39