Multithreading for making http post requests to web service

Question

I want to send multiple HTTP post requests to a Web Service in C# .For example , if n=3 then http post requests from 3 xml files should be made and also the response should be written in a file.Once the first 3 requests are made then the next 3 requests will be made . So i made the following code but i was getting random outputs at first. But now i am getting either out of index range exception in the inner for loop or Internal server error (500). Plz suggest appropriate changes. I am using .NET4.0

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading;
using System.Xml;
using System.Net;
using System.Threading.Tasks;
namespace ConsoleApplication5
{
class Program
{
    static void Main(string[] args)
    {
        int n = 0;
        Console.WriteLine("Enter the number");
        string s = Console.ReadLine();
        int.TryParse(s, out n);
        string path = "C:\\";
        string[] files = null;
        files = Directory.GetFiles(path, "*.xml", SearchOption.TopDirectoryOnly);


        List<Task> tasks = new List<Task>(files.Length);

        for (int i = 0; i < files.Length; i += n)
        {
            for (int j = 0; j < n; j++)
            {
                int x = i + j;

                if (x < files.Length && files[x] != null)
                {
                    Task t = new Task(() => function(files[x]));
                    t.Start();
                    tasks.Add(t);
                }
            }

            if (tasks.Count > 0)
            {
                Task.WaitAll(tasks.ToArray(), Timeout.Infinite); // or less than infinite
                tasks.Clear();
            }
        }
    }
    public static void function(string temp)
    {
        XmlDocument doc = new XmlDocument();
        doc.Load(temp);
        HttpWebRequest request = (HttpWebRequest)WebRequest.Create("http://10.76.22.135/wpaADws/ADService.asmx");

        request.ContentType = "text/xml;charset=\"utf-8\"";
        request.Accept = "text/xml";
        request.Method = "POST";
        Stream stream = request.GetRequestStream();
        doc.Save(stream);
        stream.Close();
        HttpWebResponse response = (HttpWebResponse)request.GetResponse();
        using (StreamReader rd = new StreamReader(response.GetResponseStream()))
        {
            string soapResult = rd.ReadToEnd();
            doc.LoadXml(soapResult);
            File.WriteAllText(temp, doc.DocumentElement.InnerText);

            //XmlTextWriter xml=new XmlTextWriter(
            Console.WriteLine(soapResult);
            Console.ReadKey();
        }

    }

}

}

Answer 1

This code works . Explaination :

• Firstly the user gives the source and destination paths for the .xml files.

• Directory.getFiles() helps us to get the .xml files in the string array . (we have to pass .xml as a parameter) .

• SO now what basically happens is for each file we get at the source pat , a thread is created .

• But say if the user wants to send "n" requests at a time , then n threads are created at a time.
• And the next set of threads are not created unless the previous threads are finished executing.
• This is ensured by thread.Join().

• And after a request is made to the web service , we get the response by getResponse() and the response is written in .xml files which are stored at the destination paths.

   using System;
   using System.Collections.Generic;
   using System.Linq;
   using System.Text;
   using System.IO;
   using System.Threading;
   using System.Xml;
   using System.Net;
   namespace ConsoleApplication4
   {
       class Program
       {
        int flag = 1;
        string destination;
        string source;
        static void Main(string[] args)
      {
      Console.ForegroundColor = ConsoleColor.Red;
  
      Console.WriteLine("**************************** Send HTTP Post Requests **************************");
      int n = 0;
      Program p = new Program();
      Console.WriteLine("Enter the number of requests you want to send at a time");
      string s = Console.ReadLine();
      int.TryParse(s, out n);
      Console.WriteLine("Enter Source");
      p.source = Console.ReadLine();
      Console.WriteLine("Enter Destination");
      p.destination = Console.ReadLine();
  
      string[] files = null;
      files = Directory.GetFiles(p.source, "*.xml", SearchOption.TopDirectoryOnly);
  
      Thread[] thread = new Thread[files.Length];
  
      int len = files.Length;
      for (int i = 0; i<len; i+=n)
      {
          int x = i;
          //Thread.Sleep(5000);
          for (int j = 0; j < n && x < len; j++)
          {
  
              var localx = x;
              thread[x] = new Thread(() => function(files[localx], p));
              thread[x].Start();
              Thread.Sleep(50);
              //thread[x].Join();
              x++;
          }
          int y = x - n;
          for (; y < x; y++)
          {
              int t = y;
              thread[t].Join();
  
          }
  
      }
  
      // thread[0] = new Thread(() => function(files[0]));
      //thread[0].Start();
      Console.ReadKey();
  
  }
  public static void function(string temp,Program p)
  {
  
      XmlDocument doc = new XmlDocument();
      doc.Load(temp);
  
      string final_d=p.destination + "response " + p.flag + ".xml";
      p.flag++;
      HttpWebRequest request = (HttpWebRequest)WebRequest.Create("http://10.76.22.135/wpaADws/ADService.asmx");
      request.ContentType = "text/xml;charset=\"utf-8\"";
      request.Accept = "text/xml";
      request.Method = "POST";
      Stream stream = request.GetRequestStream();
      doc.Save(stream);
      stream.Close();
  
      HttpWebResponse response = (HttpWebResponse)request.GetResponse();
      using (StreamReader rd = new StreamReader(response.GetResponseStream()))
      {
          string soapResult = rd.ReadToEnd();
          doc.LoadXml(soapResult);
          File.WriteAllText(final_d, doc.DocumentElement.InnerText);
  
          //XmlTextWriter xml=new XmlTextWriter(
          Console.WriteLine(soapResult);
          //Console.ReadKey();
      }
  }
  

  } }

Answer 2

The IndexOutOfRangeException you experienced in your original post was due to the improper index handling on the last batch of files you were processing. That last batch can be incomplete and you treated that as a regular batch of set size

(n=3 in your post)

Since you're moving to TPL and Tasks, I suggest Parallel Programming with Microsoft .NET, and the pipeline pattern which seems very appropriate to your scenario. You can harness the power of concurrent collections and the producer/consumer pattern together with the pipeline, like below. BlockingCollection ensures concurrent adding of items and the BlockingCollection.GetConsumingEnumerable call produces a consuming blocking enumerator for your collection.

const int BUFFER_SIZE = 3; // no concurrent items to process
const string XML_FOLDER_PATH = "<whatever>";


public static void Pipeline()
{
  var bufferXmlFileNames = new BlockingCollection<string>(BUFFER_SIZE);
  var bufferInputXmlDocuments = new BlockingCollection<XmlDocument>(BUFFER_SIZE);
  var bufferWebRequests = new BlockingCollection<HttpWebRequest>(BUFFER_SIZE);
  var bufferSoapResults = new BlockingCollection<string>(BUFFER_SIZE);

  var f = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);

  // Stage 1: get xml file paths
  var stage1 = f.StartNew(() => {
  try
  {
    foreach (var phrase in Directory.GetFiles(XML_FOLDER_PATH, "*.xml", SearchOption.TopDirectoryOnly))
    { // build concurrent collection
      bufferXmlFileNames.Add(phrase);
    }
  }
  finally
  { // no more additions acceptedin
    bufferXmlFileNames.CompleteAdding();
  }
});

  // Stage 2: ProduceInputXmlDocuments(bufferXmlFileNames, bufferInputXmlDocuments)
  var stage2 = f.StartNew(() =>  {
  try
  {
    foreach (var xmlFileName in bufferXmlFileNames.GetConsumingEnumerable())
    {
      XmlDocument doc = new XmlDocument();
      doc.Load(xmlFileName);
      bufferInputXmlDocuments.Add(doc);          
    }
  }
  finally
  {
    bufferInputXmlDocuments.CompleteAdding();
  }
});

  // Stage 3:  PostRequests(BlockingCollection<XmlDocument> xmlDocs, BlockingCollection<HttpWebRequest> posts)
  var stage3 = f.StartNew(() =>  {
  try
  {
    foreach (var xmlDoc in bufferInputXmlDocuments.GetConsumingEnumerable())
    {
      HttpWebRequest request = (HttpWebRequest)WebRequest.Create("http://10.76.22.135/wpaADws/ADService.asmx");
      request.ContentType = "text/xml;charset=\"utf-8\"";
      request.Accept = "text/xml";
      request.Method = "POST";
      //
      Stream stream = request.GetRequestStream();
      xmlDoc.Save(stream);
      stream.Close();
      //
      bufferWebRequests.Add(request);
    }
  }
  finally
  {
    bufferWebRequests.CompleteAdding();
  }
});

  // Stage 4: ProcessResponses(bufferWebRequests, bufferSoapResults)
  var stage4 = f.StartNew(() =>
  {
    try
    {
      foreach (var postRequest in bufferWebRequests.GetConsumingEnumerable())
      {
        HttpWebResponse response = (HttpWebResponse)postRequest.GetResponse();
        using (StreamReader rd = new StreamReader(response.GetResponseStream()))
        {
          string soapResult = rd.ReadToEnd();
          bufferSoapResults.Add(soapResult);
        }
      }
    }
    finally
    {
      bufferSoapResults.CompleteAdding();
    }
  });

  // stage 5: update UI
  var stage5 = f.StartNew(() =>
  {
    foreach (var soapResult in bufferSoapResults.GetConsumingEnumerable())
    {
      Console.WriteLine(soapResult);
    }
  });

  // display blocking collection load state, 
  // the number of elements in each blocking collection of the pipeline stages
  // you can supress this call completely, because it is informational only
  var stageDisplay = f.StartNew(
    () =>
    {
      while (true)
      {
        Console.WriteLine("{0,10} {1,10} {2,10} {3,10}", bufferXmlFileNames.Count, bufferInputXmlDocuments.Count, bufferWebRequests.Count, bufferSoapResults.Count);
        //check last stage completion
        if (stage5.IsCompleted)
          return;
      }
    }
      );
  Task.WaitAll(stage1, stage2, stage3, stage4, stage5); //or
  //Task.WaitAll(stage1, stage2, stage3, stage4, stage5, stageDisplay);
}

Answer 3

How about using tasks like this:

    List<Task> tasks = new List<Task>(n);

    for (int i = 0; i < files.Length; i += n)
    {
        for (int j = 0; j < n; j++)
        {
            int x = i + j;

            if (x < files.Length && files[x] != null)
            {
                Task t = new Task(() => function(files[x]));
                t.Start();
                tasks.Add(t);
            }
        }

        if (tasks.Count > 0)
        {
            Task.WaitAll(tasks.ToArray(), Timeout.Infinite); // or less than infinite
            tasks.Clear();
        }
    }

I tried to be a little tidier on the indexing...

Also, note that the int x = i + j; in the inner loop is important due to how C# captures variables for the lambda.

If the problem is tracing down indexing arithmetic, maybe use indexing variables with meaningful names?

    List<Task> tasks = new List<Task>(taskCount);

    for (int filesIdx = 0; filesIdx < files.Length; filesIdx += taskCount)
    {
        for (int tasksIdx = 0; tasksIdx < taskCount; tasksIdx++)
        {
            int index = filesIdx + tasksIdx;

            if (index < files.Length && files[index] != null)
            {
                Task task = new Task(() => function(files[index]));
                task.Start();
                tasks.Add(task);
            }
        }

        if (tasks.Count > 0)
        {
            Task.WaitAll(tasks.ToArray(), Timeout.Infinite); // or less than infinite
            tasks.Clear();
        }
    }