title: Read a specific file type with a single thread
body: <p>I have a folder which has multiple files, I intend to read those files and insert into their type speificed tables.</p>
  
  <p>Problem:</p>
  
  <p>I am reading the files with mutliple threads, but everytime a thread tries to read a type (string) of a file that is being read by another thread, it crashes with the error:</p>
  
  <blockquote>
    <p>Transaction (Process ID 69) was deadlocked on lock | communication
    buffer resources with another process and has been chosen as the
    deadlock victim. Rerun the transaction.    at
    System.Data.SqlClient.SqlConnection.OnError(SqlException exception,
    Boolean breakConnection, Action<code>1 wrapCloseInAction)    at
    System.Data.SqlClient.TdsParser.ThrowExceptionAndWarning(TdsParserStateObject
    stateObj, Boolean callerHasConnectionLock, Boolean asyncClose)    at
    System.Data.SqlClient.TdsParser.TryRun(RunBehavior runBehavior,
    SqlCommand cmdHandler, SqlDataReader dataStream,
    BulkCopySimpleResultSet bulkCopyHandler, TdsParserStateObject
    stateObj, Boolean&amp; dataReady)    at
    System.Data.SqlClient.SqlCommand.RunExecuteNonQueryTds(String
    methodName, Boolean async, Int32 timeout, Boolean asyncWrite)    at
    System.Data.SqlClient.SqlCommand.InternalExecuteNonQuery(TaskCompletionSource</code>1
    completion, String methodName, Boolean sendToPipe, Int32 timeout,
    Boolean&amp; usedCache, Boolean asyncWrite, Boolean inRetry)    at
    System.Data.SqlClient.SqlCommand.ExecuteNonQuery()    at
    Microsoft.Practices.EnterpriseLibrary.Data.Database.DoExecuteNonQuery(DbCommand
    command)    at
    Microsoft.Practices.EnterpriseLibrary.Data.Database.ExecuteNonQuery(DbCommand
    command)</p>
  </blockquote>
  
  <p>I could do it with a single thread but that takes alot of time, with multi-threading, How could I possible avoid the crash?</p>
  
  <p>Code:</p>
  
  <pre><code> public int InsertData(string data, string tableName)
          {
              string query;
              var token = JToken.Parse(data)["data"];
              JArray jArrayData;
              if (token is JArray)
              {
                  jArrayData = token as JArray;// JArray.Parse(data) as JArray;
                  foreach (var item in jArrayData.Children())
                  {
                      var itemProperties = item.Children&lt;JProperty&gt;();
                      query = ServerHelper.CreateInsertQuery(itemProperties, tableName);
                      ExecuteQuery(query);
                  }
              }
  
              return 0;
          }
  
     private int ExecuteQuery(string query)
          {
              if(conn.State == ConnectionState.Closed)
              {
                  conn.Open();
              }
              SqlCommand cmd = new SqlCommand(query, conn);
              return cmd.ExecuteNonQuery();
          }
  
  
   public static string CreateInsertQuery(JEnumerable&lt;JProperty&gt; itemProperties, string tableName)
          {
              string[] nameArray = itemProperties.Select(x =&gt; x.Name).ToArray();
              string[] valuesArray = itemProperties.Select(x =&gt; "'" + x.Value.ToString().Replace("'","''") + "'").ToArray();
              //check for null
                string query = "Insert Into " + tableName + " (" + string.Join(",", nameArray) + @")"
           + " values(" + string.Join(",", valuesArray) + ")";
          return query;
          }
  </code></pre>
  
  <p>The code that tries to execute queries in parallel is :</p>
  
  <pre><code>   public void PerformInitialization()
          {
              ts = new CancellationTokenSource();
              ct = ts.Token;
  
              ChangeProcToDat();
  
              string strMaxThreads = ConfigurationManager.AppSettings["MaxThreads"].ToString();
              if(!string.IsNullOrEmpty(strMaxThreads))
              {
                  bool parseSuccess = Int32.TryParse(strMaxThreads, out maxThreads);
                  if(parseSuccess == false)
                  {
                      maxThreads = 1;
                  }
              }
  
              processQueueTasks = new Task[maxThreads];
  
              processData = new ProcessData();
  
              for (int i = 0; i &lt; maxThreads; i++)
              {
                  processQueueTasks[i] = Task.Factory.StartNew(() =&gt; ProcessQueue(), ct);
              }
          }
  
   public void ProcessQueue()
      {
          bool cancelled = false;
          try
          {
  
              while (!ct.IsCancellationRequested)
              {
                  try
                  {
  
                      Directory.CreateDirectory(queueFolderPath);
  
                      DirectoryInfo queueFolderDi = new DirectoryInfo(queueFolderPath);
  
                      FileInfo datFile = queueFolderDi.GetFiles("*.dat", SearchOption.AllDirectories).OrderBy(f =&gt; f.LastWriteTime).ToList().FirstOrDefault();
                      while (datFile != null)
                      {
                          string processingFileName = datFile.FullName.Replace(".dat", ".proc");
                          FileInfo file = new FileInfo(processingFileName);
                          File.Move(datFile.FullName, processingFileName);
  
                          try
                          {                                
  
                              if (file.Name.Contains("_ABC_"))
                              {
                                  var fileNameArr = file.Name.Split('_', '.');
  
                                  var fileType = "";  # this is the type of the file
                                  if (fileNameArr.Length &gt;= 3)
                                  {
                                      fileType = fileNameArr[3];
                                  }
  
                                  // the db part
  
                                 var content = File.ReadAllText(file.FullName);
  
                                  int responseCode = 0;
                                  InsertData(content, reportType);
  
                                  File.Delete(file.FullName);
  
  
                              }
  
                          }
                          catch (Exception ex)
                          {
                              Log.WriteToFailed("Error processing queue file " + file.Name + " "
                                  + Environment.NewLine + ex.Message + Environment.NewLine + ex.StackTrace + Environment.NewLine);
  
                          }
  
  
                          datFile = queueFolderDi.GetFiles("*.dat", SearchOption.AllDirectories).OrderBy(f =&gt; f.LastWriteTime).ToList().FirstOrDefault();
  
  
                      }
  
  
                  }
                  catch (Exception ex)
                  {
                      Log.WriteToFailed("Error processing files in the queue folder."
                                  + Environment.NewLine + ex.Message + Environment.NewLine + ex.StackTrace + Environment.NewLine);
  
                  }
                  finally
                  {
                      cancelled = ct.WaitHandle.WaitOne(1000);
                  }
  
  
              }
          }
          catch (Exception ex)
          {
              Log.WriteToFailed("Error occured in send thread. Message:" + ex.Message + "\n Stack trace:" + ex.StackTrace);
          }
      }
  </code></pre>