I am writing a new web service that will be performing a lot of large data load operations. To do so I am moving the data to a temporary table then merging the data in to the live data via a stored procedure.
That part I have down and working and am very confident in. What I am doing now is trying to optimize the loading code to help take the load off of the IIS server that will be performing these operations.
Here is what I am currently doing synchronously and works fine.
public void BulkLoadData(IDataReaderWithMappings reader,
string createTempTableStatement,
string loadingStoredProcedureName)
{
using (var connection = new SqlConnection(_connectionString))
using (var bulkCopy = new SqlBulkCopy(connection))
using (var command = new SqlCommand(createTempTableStatement, connection))
{
connection.Open();
command.ExecuteNonQuery();
bulkCopy.DestinationTableName = "#temp";
var mappings = reader.GetMappings();
foreach (var sqlBulkCopyColumnMapping in mappings)
{
bulkCopy.ColumnMappings.Add(sqlBulkCopyColumnMapping);
}
bulkCopy.WriteToServer(reader);
command.CommandType = CommandType.StoredProcedure;
command.CommandText = loadingStoredProcedureName;
command.ExecuteNonQuery();
}
}
And here is my attempt at translating that in to asynchronous code.
public async Task BulkLoadDataAsync(IDataReaderWithMappings reader,
string createTempTableStatement,
string loadingStoredProcedureName)
{
using (var connection = new SqlConnection(_connectionString))
using (var bulkCopy = new SqlBulkCopy(connection))
using (var command = new SqlCommand(createTempTableStatement))
{
var firstStage = connection.OpenAsync()
.ContinueWith(async delegate
{
await command.ExecuteNonQueryAsync().ConfigureAwait(false);
});
bulkCopy.DestinationTableName = "#temp";
var mappings = reader.GetMappings();
foreach (var sqlBulkCopyColumnMapping in mappings)
{
bulkCopy.ColumnMappings.Add(sqlBulkCopyColumnMapping);
}
//Wait for the connection opening and the create table task to finish.
await firstStage.ConfigureAwait(false);
await bulkCopy.WriteToServerAsync(reader).ConfigureAwait(false);
command.CommandType = CommandType.StoredProcedure;
command.CommandText = loadingStoredProcedureName;
await command.ExecuteNonQueryAsync().ConfigureAwait(false);
}
}
The part I am not sure of my self about is that .ContinueWith(
section. I thought it would be good to start opening the connection and creating the temporary table while I populate the bulk copy column mappings, and I have used async/await many times before but this is the first time I have used .ContinueWith(
in conjunction with the async/await syntax.
Did I do it correctly, or should that await firstStage;
be await firstStage.Unwrap();
? I still do not exactly understand when I should be using Unwrap()
. EDIT: Ok, Unwrap()
is not listed so I could not use it even if I wanted to, but if anyone wants to include a explanation in their answer of when I should I would appreciate it.
(P.S. Do not worry about SQL injection from createTempTableStatement
the strings being passed in are hard-coded resource strings with no user input.)
With the suggestions from svick and some other tweaks I made while waiting for an answer here is the final version
public async Task BulkLoadDataAsync(IDataReader reader,
string createTempTableStatement,
string loadingStoredProcedureName,
params SqlParameter[] args)
{
using (var connection = new SqlConnection(_connectionString))
using (var bulkCopy = new SqlBulkCopy(connection))
using (var command = new SqlCommand(createTempTableStatement))
{
Func<Task> firstStageFunc = async () =>
{
await connection.OpenAsync().ConfigureAwait(false);
await command.ExecuteNonQueryAsync().ConfigureAwait(false);
};
var firstStage = firstStageFunc();
bulkCopy.DestinationTableName = "#temp";
var mappedReader = reader as IDataReaderWithMappings;
if (mappedReader != null)
{
var mappings = mappedReader.GetMappings();
foreach (var sqlBulkCopyColumnMapping in mappings)
{
bulkCopy.ColumnMappings.Add(sqlBulkCopyColumnMapping);
}
}
//Wait for the connection opening and the create table task to finish.
await firstStage.ConfigureAwait(false);
var bulkCopyTask = bulkCopy.WriteToServerAsync(reader);
command.CommandType = CommandType.StoredProcedure;
command.CommandText = loadingStoredProcedureName;
if (args != null)
command.Parameters.AddRange(args);
//Wait for the bulk copy to finish.
await bulkCopyTask.ConfigureAwait(false);
await command.ExecuteNonQueryAsync().ConfigureAwait(false);
}
}