Merge pull request #129 from OctopusDeploy/richev/async-dynamic-worker-timeouts-on-octopus-cloud

richev-o · web-flow · commit 45bbfd516160 · 2020-12-08T14:18:50.000+11:00
Async rework
diff --git a/source/Halibut/Transport/Protocol/MessageExchangeStream.cs b/source/Halibut/Transport/Protocol/MessageExchangeStream.cs
@@ -57,8 +57,8 @@ void SendControlMessage(string message)
 
         async Task SendControlMessageAsync(string message)
         {
-            await streamWriter.WriteLineAsync(message);
-            await streamWriter.FlushAsync();
+            await streamWriter.WriteLineAsync(message).ConfigureAwait(false);
+            await streamWriter.FlushAsync().ConfigureAwait(false);
         }
 
         void SendIdentityMessage(string identityLine)
@@ -82,7 +82,7 @@ public void SendProceed()
 
         public async Task SendProceedAsync()
         {
-            await SendControlMessageAsync(Proceed);
+            await SendControlMessageAsync(Proceed).ConfigureAwait(false);
         }
 
         public void SendEnd()
@@ -109,7 +109,7 @@ public bool ExpectNextOrEnd()
 
         public async Task<bool> ExpectNextOrEndAsync()
         {
-            var line = await ReadLineAsync();
+            var line = await ReadLineAsync().ConfigureAwait(false);
             switch (line)
             {
                 case Next:
@@ -146,10 +146,10 @@ string ReadLine()
 
         async Task<string> ReadLineAsync()
         {
-            var line = await streamReader.ReadLineAsync();
+            var line = await streamReader.ReadLineAsync().ConfigureAwait(false);
             while (line == string.Empty)
             {
-                line = await streamReader.ReadLineAsync();
+                line = await streamReader.ReadLineAsync().ConfigureAwait(false);
             }
 
             return line;
diff --git a/source/Halibut/Transport/Protocol/WebSocketStream.cs b/source/Halibut/Transport/Protocol/WebSocketStream.cs
@@ -38,9 +38,9 @@ public override int Read(byte[] buffer, int offset, int count)
         {
             AssertCanReadOrWrite();
             var segment = new ArraySegment<byte>(buffer, offset, count);
-            var recieveResult = context.ReceiveAsync(segment, CancellationToken.None)
+            var receiveResult = context.ReceiveAsync(segment, CancellationToken.None)
                 .ConfigureAwait(false).GetAwaiter().GetResult();
-            return recieveResult.Count;
+            return receiveResult.Count;
         }
 
         public async Task<string> ReadTextMessage()
@@ -53,11 +53,11 @@ public async Task<string> ReadTextMessage()
                 using(var cts = new CancellationTokenSource(HalibutLimits.TcpClientReceiveTimeout))
                 using(var combined = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, cancel.Token))
                 {
-                    var result = await context.ReceiveAsync(buffer, combined.Token);
+                    var result = await context.ReceiveAsync(buffer, combined.Token).ConfigureAwait(false);
                     if (result.MessageType == WebSocketMessageType.Close)
                     {
                         using(var sendCancel = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
-                            await context.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Close received", sendCancel.Token);
+                            await context.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Close received", sendCancel.Token).ConfigureAwait(false);
                         return null;
                     }
 
diff --git a/source/Halibut/Transport/SecureListener.cs b/source/Halibut/Transport/SecureListener.cs
@@ -143,16 +143,15 @@ void WaitForPendingConnectionOrCancellation()
                         }
 
                         var client = listener.AcceptTcpClient();
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-                        HandleClient(client);
-#pragma warning restore CS4014
+                        Task.Run(async () => await HandleClient(client).ConfigureAwait(false)).ConfigureAwait(false);
                         numberOfFailedAttemptsInRow = 0;
                     }
                     catch (SocketException e) when (e.SocketErrorCode == SocketError.Interrupted)
                     {
                     }
                     catch (ObjectDisposedException)
                     {
+                        // Happens on shutdown
                     }
                     catch (Exception ex)
                     {
@@ -190,7 +189,7 @@ async Task HandleClient(TcpClient client)
                 client.ReceiveTimeout = (int)HalibutLimits.TcpClientReceiveTimeout.TotalMilliseconds;
 
                 log.Write(EventType.ListenerAcceptedClient, "Accepted TCP client: {0}", client.Client.RemoteEndPoint);
-                await ExecuteRequest(client);
+                await ExecuteRequest(client).ConfigureAwait(false);
             }
             catch (ObjectDisposedException)
             {
@@ -210,7 +209,7 @@ async Task ExecuteRequest(TcpClient client)
                 try
                 {
                     log.Write(EventType.SecurityNegotiation, "Performing TLS server handshake");
-                    await ssl.AuthenticateAsServerAsync(serverCertificate, true, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false);
+                    await ssl.AuthenticateAsServerAsync(serverCertificate, true, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false).ConfigureAwait(false);
 
                     log.Write(EventType.SecurityNegotiation, "Secure connection established, client is not yet authenticated, client connected with {0}", ssl.SslProtocol.ToString());
 
@@ -247,7 +246,7 @@ async Task ExecuteRequest(TcpClient client)
                         });
 
                         tcpClientManager.AddActiveClient(thumbprint, client);
-                        await ExchangeMessages(ssl);
+                        await ExchangeMessages(ssl).ConfigureAwait(false);
                     }
                 }
                 catch (AuthenticationException ex)
diff --git a/source/Halibut/Transport/SecureWebSocketListener.cs b/source/Halibut/Transport/SecureWebSocketListener.cs
@@ -14,7 +14,6 @@ namespace Halibut.Transport
     public class SecureWebSocketListener : IDisposable
     {
         readonly string endPoint;
-        readonly X509Certificate2 serverCertificate;
         readonly Func<MessageExchangeProtocol, Task> protocolHandler;
         readonly Predicate<string> verifyClientThumbprint;
         readonly Func<string, string, UnauthorizedClientConnectResponse> unauthorizedClientConnect;
@@ -53,7 +52,6 @@ public SecureWebSocketListener(string endPoint, X509Certificate2 serverCertifica
                 endPoint += "/";
 
             this.endPoint = endPoint;
-            this.serverCertificate = serverCertificate;
             this.protocolHandler = protocolHandler;
             this.verifyClientThumbprint = verifyClientThumbprint;
             this.unauthorizedClientConnect = unauthorizedClientConnect;
@@ -75,7 +73,7 @@ public void Start()
 
             log = logFactory.ForPrefix(endPoint);
             log.Write(EventType.ListenerStarted, "Listener started");
-            Task.Run(async () => await Accept());
+            Task.Run(async () => await Accept().ConfigureAwait(false)).ConfigureAwait(false);
         }
 
         async Task Accept()
@@ -86,12 +84,12 @@ async Task Accept()
                 {
                     try
                     {
-                        var context = await listener.GetContextAsync();
+                        var context = await listener.GetContextAsync().ConfigureAwait(false);
 
                         if (context.Request.IsWebSocketRequest)
                         {
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-                            HandleClient(context);
+                            Task.Run(async () => await HandleClient(context).ConfigureAwait(false)).ConfigureAwait(false);
 #pragma warning restore CS4014
                         }
                         else
@@ -115,7 +113,7 @@ async Task HandleClient(HttpListenerContext context)
             try
             {
                 log.Write(EventType.ListenerAcceptedClient, "Accepted Web Socket client: {0}", context.Request.RemoteEndPoint);
-                await ExecuteRequest(context);
+                await ExecuteRequest(context).ConfigureAwait(false);
             }
             catch (ObjectDisposedException)
             {
@@ -136,10 +134,10 @@ async Task ExecuteRequest(HttpListenerContext listenerContext)
             WebSocketStream webSocketStream = null;
             try
             {
-                var webSocketContext = await listenerContext.AcceptWebSocketAsync("Octopus");
+                var webSocketContext = await listenerContext.AcceptWebSocketAsync("Octopus").ConfigureAwait(false);
                 webSocketStream = new WebSocketStream(webSocketContext.WebSocket);
 
-                var req = await webSocketStream.ReadTextMessage(); // Initial message
+                var req = await webSocketStream.ReadTextMessage().ConfigureAwait(false); // Initial message
                 if (string.IsNullOrEmpty(req))
                 {
                     log.Write(EventType.Diagnostic, "Ignoring empty request");
@@ -152,10 +150,12 @@ async Task ExecuteRequest(HttpListenerContext listenerContext)
                     return;
                 }
 
-                if (await Authorize(listenerContext, clientName))
+                var authorized = await Authorize(listenerContext, clientName).ConfigureAwait(false);
+
+                if (authorized)
                 {
                     // Delegate the open stream to the protocol handler - we no longer own the stream lifetime
-                    await ExchangeMessages(webSocketStream);
+                    await ExchangeMessages(webSocketStream).ConfigureAwait(false);
 
                     // Mark the stream as delegated once everything has succeeded
                     keepConnection = true;
@@ -202,7 +202,7 @@ void SendFriendlyHtmlPage(HttpListenerResponse response)
         async Task<bool> Authorize(HttpListenerContext context, EndPoint clientName)
         {
             log.Write(EventType.Diagnostic, "Begin authorization");
-            var certificate = await context.Request.GetClientCertificateAsync();
+            var certificate = await context.Request.GetClientCertificateAsync().ConfigureAwait(false);
             if (certificate == null)
             {
                 log.Write(EventType.ClientDenied, "A client at {0} connected, and attempted a message exchange, but did not present a client certificate", clientName);

Original file line number	Diff line number	Diff line change
`@@ -57,8 +57,8 @@ void SendControlMessage(string message)`
`57`	`57`
`58`	`58`	`async Task SendControlMessageAsync(string message)`
`59`	`59`	`{`
`60`		`- await streamWriter.WriteLineAsync(message);`
`61`		`- await streamWriter.FlushAsync();`
	`60`	`+ await streamWriter.WriteLineAsync(message).ConfigureAwait(false);`
	`61`	`+ await streamWriter.FlushAsync().ConfigureAwait(false);`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`void SendIdentityMessage(string identityLine)`
`@@ -82,7 +82,7 @@ public void SendProceed()`
`82`	`82`
`83`	`83`	`public async Task SendProceedAsync()`
`84`	`84`	`{`
`85`		`- await SendControlMessageAsync(Proceed);`
	`85`	`+ await SendControlMessageAsync(Proceed).ConfigureAwait(false);`
`86`	`86`	`}`
`87`	`87`
`88`	`88`	`public void SendEnd()`
`@@ -109,7 +109,7 @@ public bool ExpectNextOrEnd()`
`109`	`109`
`110`	`110`	`public async Task<bool> ExpectNextOrEndAsync()`
`111`	`111`	`{`
`112`		`- var line = await ReadLineAsync();`
	`112`	`+ var line = await ReadLineAsync().ConfigureAwait(false);`
`113`	`113`	`switch (line)`
`114`	`114`	`{`
`115`	`115`	`case Next:`
`@@ -146,10 +146,10 @@ string ReadLine()`
`146`	`146`
`147`	`147`	`async Task<string> ReadLineAsync()`
`148`	`148`	`{`
`149`		`- var line = await streamReader.ReadLineAsync();`
	`149`	`+ var line = await streamReader.ReadLineAsync().ConfigureAwait(false);`
`150`	`150`	`while (line == string.Empty)`
`151`	`151`	`{`
`152`		`- line = await streamReader.ReadLineAsync();`
	`152`	`+ line = await streamReader.ReadLineAsync().ConfigureAwait(false);`
`153`	`153`	`}`
`154`	`154`
`155`	`155`	`return line;`
Original file line number	Diff line number	Diff line change
`@@ -38,9 +38,9 @@ public override int Read(byte[] buffer, int offset, int count)`
`38`	`38`	`{`
`39`	`39`	`AssertCanReadOrWrite();`
`40`	`40`	`var segment = new ArraySegment<byte>(buffer, offset, count);`
`41`		`- var recieveResult = context.ReceiveAsync(segment, CancellationToken.None)`
	`41`	`+ var receiveResult = context.ReceiveAsync(segment, CancellationToken.None)`
`42`	`42`	`.ConfigureAwait(false).GetAwaiter().GetResult();`
`43`		`- return recieveResult.Count;`
	`43`	`+ return receiveResult.Count;`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`public async Task<string> ReadTextMessage()`
`@@ -53,11 +53,11 @@ public async Task<string> ReadTextMessage()`
`53`	`53`	`using(var cts = new CancellationTokenSource(HalibutLimits.TcpClientReceiveTimeout))`
`54`	`54`	`using(var combined = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, cancel.Token))`
`55`	`55`	`{`
`56`		`- var result = await context.ReceiveAsync(buffer, combined.Token);`
	`56`	`+ var result = await context.ReceiveAsync(buffer, combined.Token).ConfigureAwait(false);`
`57`	`57`	`if (result.MessageType == WebSocketMessageType.Close)`
`58`	`58`	`{`
`59`	`59`	`using(var sendCancel = new CancellationTokenSource(TimeSpan.FromSeconds(1)))`
`60`		`- await context.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Close received", sendCancel.Token);`
	`60`	`+ await context.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "Close received", sendCancel.Token).ConfigureAwait(false);`
`61`	`61`	`return null;`
`62`	`62`	`}`
`63`	`63`
Original file line number	Diff line number	Diff line change
`@@ -143,16 +143,15 @@ void WaitForPendingConnectionOrCancellation()`
`143`	`143`	`}`
`144`	`144`
`145`	`145`	`var client = listener.AcceptTcpClient();`
`146`		`-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed`
`147`		`- HandleClient(client);`
`148`		`-#pragma warning restore CS4014`
	`146`	`+ Task.Run(async () => await HandleClient(client).ConfigureAwait(false)).ConfigureAwait(false);`
`149`	`147`	`numberOfFailedAttemptsInRow = 0;`
`150`	`148`	`}`
`151`	`149`	`catch (SocketException e) when (e.SocketErrorCode == SocketError.Interrupted)`
`152`	`150`	`{`
`153`	`151`	`}`
`154`	`152`	`catch (ObjectDisposedException)`
`155`	`153`	`{`
	`154`	`+ // Happens on shutdown`
`156`	`155`	`}`
`157`	`156`	`catch (Exception ex)`
`158`	`157`	`{`
`@@ -190,7 +189,7 @@ async Task HandleClient(TcpClient client)`
`190`	`189`	`client.ReceiveTimeout = (int)HalibutLimits.TcpClientReceiveTimeout.TotalMilliseconds;`
`191`	`190`
`192`	`191`	`log.Write(EventType.ListenerAcceptedClient, "Accepted TCP client: {0}", client.Client.RemoteEndPoint);`
`193`		`- await ExecuteRequest(client);`
	`192`	`+ await ExecuteRequest(client).ConfigureAwait(false);`
`194`	`193`	`}`
`195`	`194`	`catch (ObjectDisposedException)`
`196`	`195`	`{`
`@@ -210,7 +209,7 @@ async Task ExecuteRequest(TcpClient client)`
`210`	`209`	`try`
`211`	`210`	`{`
`212`	`211`	`log.Write(EventType.SecurityNegotiation, "Performing TLS server handshake");`
`213`		`- await ssl.AuthenticateAsServerAsync(serverCertificate, true, SslProtocols.Tls \| SslProtocols.Tls11 \| SslProtocols.Tls12, false);`
	`212`	`+ await ssl.AuthenticateAsServerAsync(serverCertificate, true, SslProtocols.Tls \| SslProtocols.Tls11 \| SslProtocols.Tls12, false).ConfigureAwait(false);`
`214`	`213`
`215`	`214`	`log.Write(EventType.SecurityNegotiation, "Secure connection established, client is not yet authenticated, client connected with {0}", ssl.SslProtocol.ToString());`
`216`	`215`
`@@ -247,7 +246,7 @@ async Task ExecuteRequest(TcpClient client)`
`247`	`246`	`});`
`248`	`247`
`249`	`248`	`tcpClientManager.AddActiveClient(thumbprint, client);`
`250`		`- await ExchangeMessages(ssl);`
	`249`	`+ await ExchangeMessages(ssl).ConfigureAwait(false);`
`251`	`250`	`}`
`252`	`251`	`}`
`253`	`252`	`catch (AuthenticationException ex)`
Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,6 @@ namespace Halibut.Transport`
`14`	`14`	`public class SecureWebSocketListener : IDisposable`
`15`	`15`	`{`
`16`	`16`	`readonly string endPoint;`
`17`		`- readonly X509Certificate2 serverCertificate;`
`18`	`17`	`readonly Func<MessageExchangeProtocol, Task> protocolHandler;`
`19`	`18`	`readonly Predicate<string> verifyClientThumbprint;`
`20`	`19`	`readonly Func<string, string, UnauthorizedClientConnectResponse> unauthorizedClientConnect;`
`@@ -53,7 +52,6 @@ public SecureWebSocketListener(string endPoint, X509Certificate2 serverCertifica`
`53`	`52`	`endPoint += "/";`
`54`	`53`
`55`	`54`	`this.endPoint = endPoint;`
`56`		`- this.serverCertificate = serverCertificate;`
`57`	`55`	`this.protocolHandler = protocolHandler;`
`58`	`56`	`this.verifyClientThumbprint = verifyClientThumbprint;`
`59`	`57`	`this.unauthorizedClientConnect = unauthorizedClientConnect;`
`@@ -75,7 +73,7 @@ public void Start()`
`75`	`73`
`76`	`74`	`log = logFactory.ForPrefix(endPoint);`
`77`	`75`	`log.Write(EventType.ListenerStarted, "Listener started");`
`78`		`- Task.Run(async () => await Accept());`
	`76`	`+ Task.Run(async () => await Accept().ConfigureAwait(false)).ConfigureAwait(false);`
`79`	`77`	`}`
`80`	`78`
`81`	`79`	`async Task Accept()`
`@@ -86,12 +84,12 @@ async Task Accept()`
`86`	`84`	`{`
`87`	`85`	`try`
`88`	`86`	`{`
`89`		`- var context = await listener.GetContextAsync();`
	`87`	`+ var context = await listener.GetContextAsync().ConfigureAwait(false);`
`90`	`88`
`91`	`89`	`if (context.Request.IsWebSocketRequest)`
`92`	`90`	`{`
`93`	`91`	`#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed`
`94`		`- HandleClient(context);`
	`92`	`+ Task.Run(async () => await HandleClient(context).ConfigureAwait(false)).ConfigureAwait(false);`
`95`	`93`	`#pragma warning restore CS4014`
`96`	`94`	`}`
`97`	`95`	`else`
`@@ -115,7 +113,7 @@ async Task HandleClient(HttpListenerContext context)`
`115`	`113`	`try`
`116`	`114`	`{`
`117`	`115`	`log.Write(EventType.ListenerAcceptedClient, "Accepted Web Socket client: {0}", context.Request.RemoteEndPoint);`
`118`		`- await ExecuteRequest(context);`
	`116`	`+ await ExecuteRequest(context).ConfigureAwait(false);`
`119`	`117`	`}`
`120`	`118`	`catch (ObjectDisposedException)`
`121`	`119`	`{`
`@@ -136,10 +134,10 @@ async Task ExecuteRequest(HttpListenerContext listenerContext)`
`136`	`134`	`WebSocketStream webSocketStream = null;`
`137`	`135`	`try`
`138`	`136`	`{`
`139`		`- var webSocketContext = await listenerContext.AcceptWebSocketAsync("Octopus");`
	`137`	`+ var webSocketContext = await listenerContext.AcceptWebSocketAsync("Octopus").ConfigureAwait(false);`
`140`	`138`	`webSocketStream = new WebSocketStream(webSocketContext.WebSocket);`
`141`	`139`
`142`		`- var req = await webSocketStream.ReadTextMessage(); // Initial message`
	`140`	`+ var req = await webSocketStream.ReadTextMessage().ConfigureAwait(false); // Initial message`
`143`	`141`	`if (string.IsNullOrEmpty(req))`
`144`	`142`	`{`
`145`	`143`	`log.Write(EventType.Diagnostic, "Ignoring empty request");`
`@@ -152,10 +150,12 @@ async Task ExecuteRequest(HttpListenerContext listenerContext)`
`152`	`150`	`return;`
`153`	`151`	`}`
`154`	`152`
`155`		`- if (await Authorize(listenerContext, clientName))`
	`153`	`+ var authorized = await Authorize(listenerContext, clientName).ConfigureAwait(false);`
	`154`	`+`
	`155`	`+ if (authorized)`
`156`	`156`	`{`
`157`	`157`	`// Delegate the open stream to the protocol handler - we no longer own the stream lifetime`
`158`		`- await ExchangeMessages(webSocketStream);`
	`158`	`+ await ExchangeMessages(webSocketStream).ConfigureAwait(false);`
`159`	`159`
`160`	`160`	`// Mark the stream as delegated once everything has succeeded`
`161`	`161`	`keepConnection = true;`
`@@ -202,7 +202,7 @@ void SendFriendlyHtmlPage(HttpListenerResponse response)`
`202`	`202`	`async Task<bool> Authorize(HttpListenerContext context, EndPoint clientName)`
`203`	`203`	`{`
`204`	`204`	`log.Write(EventType.Diagnostic, "Begin authorization");`
`205`		`- var certificate = await context.Request.GetClientCertificateAsync();`
	`205`	`+ var certificate = await context.Request.GetClientCertificateAsync().ConfigureAwait(false);`
`206`	`206`	`if (certificate == null)`
`207`	`207`	`{`
`208`	`208`	`log.Write(EventType.ClientDenied, "A client at {0} connected, and attempted a message exchange, but did not present a client certificate", clientName);`