Http Sync Connection calls no longer rely on Tasks so they use less resources

Author: Mpdreamz

src/Nest.Tests.Integration/ElasticsearchConfiguration.cs

@@ -12,7 +12,7 @@ public static IConnectionSettings Settings(int? port = null)
 		{
 			var host = Test.Default.Host;
 			if (port == null && Process.GetProcessesByName("fiddler").HasAny())
-				host = "local.localghost.io";
+				host = "ipv4.fiddler";
 
 			var uri = new UriBuilder("http", host, port.GetValueOrDefault(9200)).Uri;
 

src/Nest.Tests.Integration/Nest.Tests.Integration.csproj

@@ -143,6 +143,8 @@
       <DesignTimeSharedInput>True</DesignTimeSharedInput>
       <DependentUpon>Test.settings</DependentUpon>
     </Compile>
+    <Compile Include="Warmers\IndicesWarmersTests.cs" />
+    <Compile Include="Warmers\WarmersTests.cs" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Nest.Connection.Thrift\Nest.Connection.Thrift.csproj">

src/Nest/Domain/Connection/Connection.cs

@@ -163,14 +163,39 @@ private HttpWebRequest CreateWebRequest(string path, string method)
 		protected virtual ConnectionStatus DoSynchronousRequest(HttpWebRequest request, string data = null)
 		{
 			var timeout = this._ConnectionSettings.Timeout;
-			var task = this.DoAsyncRequest(request, data);
-			task.Wait(timeout);
-			if (task.Result == null && task.IsCanceled)
+			if (data != null)
+			{
+				using (var r = request.GetRequestStream())
+				{
+					byte[] buffer = Encoding.UTF8.GetBytes(data);
+					r.Write(buffer, 0, buffer.Length);
+				}
+			}
+			try
+			{
+				using (var response = (HttpWebResponse)request.GetResponse())
+				using (var responseStream = response.GetResponseStream())
+				using (var streamReader = new StreamReader(responseStream))
+				{
+					string result = streamReader.ReadToEnd();
+					var cs = new ConnectionStatus(result)
+					{
+						Request = data,
+						RequestUrl = request.RequestUri.ToString(),
+						RequestMethod = request.Method
+					};
+					return cs;
+				}
+			}
+			catch (WebException webException)
+			{
+				return new ConnectionStatus(webException) { Request = data, RequestUrl = request.RequestUri.ToString(), RequestMethod = request.Method };
+			}
+			catch 
 			{
-				var m = "Operation did not complete before the set timeout of " + timeout + "ms";
-				return new ConnectionStatus(new TimeoutException(m));
+				throw;
 			}
-			return task.Result;
+			
 		}
 
 		protected virtual Task<ConnectionStatus> DoAsyncRequest(HttpWebRequest request, string data = null)

src/ProtocolLoadTest/HttpManualAsyncTester.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using Nest;
+using System.Threading.Tasks;
+
+namespace ProtocolLoadTest
+{
+	internal class HttpManualAsyncTester : Tester, ITester
+    {
+        public void Run(string indexName, int port, int numMessages, int bufferSize)
+        {
+			var settings = this.CreateSettings(indexName, port);
+            var client = new ElasticClient(settings);
+            
+            Connect(client, settings);
+
+			HttpManualAsyncTester.GenerateAndIndex(client, indexName, numMessages, bufferSize);
+        }
+		protected static void GenerateAndIndex(ElasticClient client, string indexName, int numMessages, int bufferSize)
+		{
+			// refresh = false is default on elasticsearch's side.
+			var bulkParms = new SimpleBulkParameters() { Refresh = false };
+
+			var msgGenerator = new MessageGenerator();
+			var tasks = new List<Task>();
+			var partitionedMessages = msgGenerator.Generate(numMessages).Partition(bufferSize);
+			Interlocked.Exchange(ref NumSent, 0);
+			foreach (var messages in partitionedMessages)
+			{
+				var t = Task.Factory.StartNew(()=> client.IndexMany(messages, indexName, bulkParms), TaskCreationOptions.LongRunning);
+				tasks.Add(t);
+
+				Interlocked.Add(ref NumSent, bufferSize);
+				if (NumSent % 10000 == 0)
+				{
+					Console.WriteLine("Sent {0:0,0} messages to {1}", NumSent, indexName);
+				}
+			}
+			Task.WaitAll(tasks.ToArray());
+		}
+    }
+}

src/ProtocolLoadTest/Program.cs

@@ -2,6 +2,7 @@
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Linq;
 using Nest;
 
 namespace ProtocolLoadTest
@@ -21,10 +22,12 @@ class Program
 		static void Main(string[] args)
 		{
 			double httpRate = RunTest<HttpTester>(HTTP_PORT);
+			double manualAsyncHttpRate = RunTest<HttpManualAsyncTester>(HTTP_PORT);
 			//double thriftRate = RunTest<ThriftTester>(THRIFT_PORT);
 
 			Console.WriteLine();
-			Console.WriteLine("HTTP: {0:0,0}/s", httpRate);
+			Console.WriteLine("HTTP (IndexManyAsync): {0:0,0}/s", httpRate);
+			Console.WriteLine("HTTP (IndexMany + TaskFactory.StartNew): {0:0,0}/s", manualAsyncHttpRate);
 			//Console.WriteLine("Thrift: {0:0,0}/s", thriftRate);
 
 			Console.ReadLine();
@@ -58,9 +61,12 @@ private static double RunTest<T>(int port) where T : ITester
 
 		private static void RecreateIndex(string suffix)
 		{
+			var host = "localhost";
+			if (Process.GetProcessesByName("fiddler").Any())
+				host = "ipv4.fiddler";
 			string indexName = INDEX_PREFIX + suffix;
 
-			var connSettings = new ConnectionSettings(new Uri("http://ipv4.fiddler:9200"))
+			var connSettings = new ConnectionSettings(new Uri("http://"+host+":9200"))
 				.SetDefaultIndex(indexName);
 
 			var client = new ElasticClient(connSettings);
@@ -80,7 +86,7 @@ private static void RecreateIndex(string suffix)
 			var indexSettings = new IndexSettings();
 			indexSettings.NumberOfReplicas = 1;
 			indexSettings.NumberOfShards = 5;
-			indexSettings.Add("index.refresh_interval", "10s");
+			indexSettings.Add("index.refresh_interval", "-1");
 
 			var createResponse = client.CreateIndex(indexName, indexSettings);
 			client.MapFromAttributes<Message>();
@@ -90,7 +96,11 @@ private static void CloseIndex(string suffix)
 		{
 			string indexName = INDEX_PREFIX + suffix;
 
-			var connSettings = new ConnectionSettings(new Uri("http://ipv4.fiddler:9200"))
+			var host = "localhost";
+			if (Process.GetProcessesByName("fiddler").Any())
+				host = "ipv4.fiddler";
+
+			var connSettings = new ConnectionSettings(new Uri("http://" + host + ":9200"))
 				.SetDefaultIndex(indexName);
 
 			var client = new ElasticClient(connSettings);

src/ProtocolLoadTest/ProtocolLoadTest.csproj

@@ -64,6 +64,7 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="HttpManualAsyncTester.cs" />
     <Compile Include="HttpTester.cs" />
     <Compile Include="ITester.cs" />
     <Compile Include="MessageGenerator.cs" />

src/ProtocolLoadTest/Tester.cs

@@ -4,17 +4,22 @@
 using System.Threading;
 using System.Threading.Tasks;
 using System.Collections.ObjectModel;
+using System.Diagnostics;
+using System.Linq;
 
 namespace ProtocolLoadTest
 {
 	internal abstract class Tester
 	{
 		// Number of messages sent by all ThriftTester instances
-		private static int NumSent;
+		protected static int NumSent;
 
 		protected ConnectionSettings CreateSettings(string indexName, int port)
 		{
-			var uri = new UriBuilder("http", "ipv4.fiddler", port).Uri;
+			var host = "localhost";
+			if (Process.GetProcessesByName("fiddler").Any())
+				host = "ipv4.fiddler";
+			var uri = new UriBuilder("http", host, port).Uri;
 			return new ConnectionSettings(uri)
 			   .SetDefaultIndex(indexName)
 			   .SetMaximumAsyncConnections(2);
@@ -40,7 +45,6 @@ protected static void GenerateAndIndex(ElasticClient client, string indexName, i
 
 			var msgGenerator = new MessageGenerator();
 			var tasks = new List<Task>();
-
 			var partitionedMessages = msgGenerator.Generate(numMessages).Partition(bufferSize);
 			Interlocked.Exchange(ref NumSent, 0);
 			foreach (var messages in partitionedMessages)