Release SQLCommand.ExecuteNonQuery memory resources when executed in a threadpool.QueueUserWorkItem

[EDN Admin]

Hi to all of you,
I get an outofmemory exception when I run SQLCommand.ExecuteNonQuery (I insert lot of rows within several threads).
I am developping an udp log message system. Messages can contain huge array byte.
In single thread way, it works fine, all memory resources are released asap.
When I use threadpool.QueueUserWorkItem, resource is not released.
Process is the following:
- I listen a socket and as soon as I receive a byte array, I call threadpool.QueueUserWorkItem and pass it the method and byte array as argument
- In the multi threaded method, I instantiate a message that I fill with values from the byte array and then call the insert sql command.
- When I receive for instance 500 000 messages with a length of 60 000 bytes in 10 mn then I have an out of memory exception. My laptop has 4gb of RAM. it runs under winxp pro sp3 and I use VS2010, .net 4.0. SQLServer2008 (sqlserver memory size limited to
500 MB)
Below source code:
In this method, I listen a socket and once a message is received, I call threadpool.QueueUserWorkItem
<div style="color:Black;background-color:White; <pre>
<span style="color:Blue; public <span style="color:Blue; void Listen()
{
Int32 intBufferLength = 0;
<span style="color:Blue; try
{
<span style="color:Blue; if (UDPSocket == <span style="color:Blue; null)
<span style="color:Blue; throw <span style="color:Blue; new Exception(<span style="color:#A31515; "Null Socket.");

<span style="color:Blue; if (!UDPSocket.IsBound)
<span style="color:Blue; throw <span style="color:Blue; new Exception(<span style="color:#A31515; "Socket not connected.");
<span style="color:Blue; byte[] bytBuffer = <span style="color:Blue; new Byte[65535];
StartSignal = <span style="color:Blue; true;
<span style="color:Blue; while (StartSignal)
{
intBufferLength = UDPSocket.Receive(bytBuffer);
<span style="color:Blue; byte[] bytBuffOut = <span style="color:Blue; new <span style="color:Blue; byte[intBufferLength];
Buffer.BlockCopy(bytBuffer, 0, bytBuffOut, 0, intBufferLength);
System.Threading.ThreadPool.QueueUserWorkItem(OnDataReceived, bytBuffOut);
bytBuffOut = <span style="color:Blue; null;
}
}
<span style="color:Blue; catch (SocketException ex)
{
<span style="color:Blue; if (ex.ErrorCode != 10004) <span style="color:Green; //A blocking operation was interrupted by a call to WSACancelBlockingCall. Receive() is a blocking operation and the close method throw this error
<span style="color:Blue; throw <span style="color:Blue; new Exception(<span style="color:#A31515; "TraceCollector Listen error: " + ex.Message);
}
<span style="color:Blue; catch (System.Threading.ThreadAbortException)
{
}
<span style="color:Blue; catch (Exception ex)
{
<span style="color:Blue; throw <span style="color:Blue; new Exception(<span style="color:#A31515; "TraceCollector Listen error: " + ex.Message);
}
}
[/code]
In the multi threaded method, I parse the byte array and instantiate an object. Finally, I transmit this object to another method which copes vith the sqlserver insertion query
<div style="color:Black;background-color:White; <pre>
<span style="color:Blue; protected <span style="color:Blue; void OnDataReceived(<span style="color:Blue; object pbytBuffMsg)
{
<span style="color:Blue; try
{
TraceMessage TraceMessageObj = TraceMessage.ParseOTFormat(pbytBuffMsg);
<span style="color:Blue; if (TraceMessageObj != <span style="color:Blue; null)
{
InsertMsgToDB(TraceMessageObj);
}
pbytBuffMsg = <span style="color:Blue; null;
TraceMessageObj = <span style="color:Blue; null;
}
<span style="color:Blue; catch (Exception ex)
{
<span style="color:Blue; if (_log.IsErrorEnabled) _log.Error(<span style="color:#A31515; "OnDataReceived Error. Log Message: " + BitConverter.ToString((<span style="color:Blue; byte[])pbytBuffMsg, 0) + <span style="color:#A31515; ". Error: " + ex.Message);
}
}

<span style="color:Blue; public <span style="color:Blue; void InsertMsgToDB(TraceMessage sender)
{
<span style="color:Blue; if (_log.IsDebugEnabled) _log.Debug(<span style="color:#A31515; "InsertMsgToDB");

<span style="color:Blue; try
{
<span style="color:Blue; using (System.Data.SqlClient.SqlConnection sqlCon = <span style="color:Blue; new System.Data.SqlClient.SqlConnection(<span style="color:#A31515; @"Data Source=XXXXXSQLEXPRESS;Initial Catalog=MyCatalog;User Id=MyUsr; Password=MyPwd"))
{
sqlCon.Open();
<span style="color:Blue; using (System.Data.SqlClient.SqlCommand cmd = sqlCon.CreateCommand())
{
cmd.CommandText = <span style="color:#A31515; @"INSERT INTO OTEqptLog (MsgId, MsgVersion, MachineId, MachineName, Station, StationRank, SoftId, ProcessId, ThreadId, DateTime, MsgType, MsgSubType, MsgLevel, MsgLength, MsgText, Port, CompressMode) VALUES (@MsgId, @MsgVersion, @MachineId, @MachineName, @Station, @StationRank, @SoftId, @ProcessId, @ThreadId, @DateTime, @MsgType, @MsgSubType, @MsgLevel, @MsgLength, @MsgText, @Port, @CompressMode);";
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgId", sender.MsgId);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgVersion", sender.MsgVersion);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MachineId", sender.MachineId);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MachineName", sender.MachineName);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@Station", sender.Station);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@StationRank", sender.StationRank);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@SoftId", sender.SoftId);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@ProcessId", sender.ProcessId);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@ThreadId", sender.ThreadId);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@DateTime", sender.DateTimeDT);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgType", sender.MsgType);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgSubType", sender.MsgSubType);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgLevel", sender.MsgLevel);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgLength", Convert.ToInt32(sender.MsgLength));
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@MsgText", sender.MsgText);
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@Port", Convert.ToInt16(sender.Port));
cmd.Parameters.AddWithValue(<span style="color:#A31515; "@CompressMode", sender.Compression);
<span style="color:Blue; if (cmd.ExecuteNonQuery() == 0)
_log.Error(<span style="color:#A31515; "InsertMsgToDB. sQueryInsert: " + cmd.CommandText + <span style="color:#A31515; ". Err: Unable to insert in DB");
<span style="color:Green; //cmd.Parameters.Clear();
sqlCon.Close();
}
}
sender = <span style="color:Blue; null;
}
<span style="color:Blue; catch (Exception ex)
{
_log.Error(<span style="color:#A31515; "InsertMsgToDB. Message: " + sender.GetOTFormat() + <span style="color:#A31515; ". Err: " + ex.Message);
}
}
[/code]
If I just comment this line if (cmd.ExecuteNonQuery() == 0)<br/>
then I dont have issues.

Thank you for your support in advance
Mickael

View the full article