improve App Host shutdown (#3727)

Improve sequence and exception handling for outpost shut down

dotnet/samples/Hello/Program.cs

@@ -47,7 +47,6 @@ namespace Hello
                 Message = goodbye
             }.ToCloudEvent(this.AgentId.Key);
             await PublishEvent(evt).ConfigureAwait(false);
-            await Task.Delay(60000);
             await App.ShutdownAsync();
         }
         public async Task<string> SayHello(string ask)

dotnet/src/Microsoft.AutoGen.Agents/Client/AgentWorkerRuntime.cs

@@ -99,11 +99,21 @@ public sealed class AgentWorkerRuntime : IHostedService, IDisposable, IAgentWork
                     }
                 }
             }
-            catch (Exception ex)
+            catch (OperationCanceledException)
+            {
+                // Time to shut down.
+                break;
+            }
+            catch (Exception ex) when (!_shutdownCts.IsCancellationRequested)
             {
                 _logger.LogError(ex, "Error reading from channel.");
                 channel = RecreateChannel(channel);
             }
+            catch
+            {
+                // Shutdown requested.
+                break;
+            }
         }
     }
 
@@ -113,34 +123,39 @@ public sealed class AgentWorkerRuntime : IHostedService, IDisposable, IAgentWork
         var outboundMessages = _outboundMessagesChannel.Reader;
         while (!_shutdownCts.IsCancellationRequested)
         {
-            await outboundMessages.WaitToReadAsync().ConfigureAwait(false);
-
-            // Read the next message if we don't already have an unsent message
-            // waiting to be sent.
-            if (!outboundMessages.TryRead(out var message))
+            try
             {
-                break;
-            }
+                await outboundMessages.WaitToReadAsync().ConfigureAwait(false);
 
-            while (!_shutdownCts.IsCancellationRequested)
-            {
-                try
+                // Read the next message if we don't already have an unsent message
+                // waiting to be sent.
+                if (!outboundMessages.TryRead(out var message))
                 {
-                    await channel.RequestStream.WriteAsync(message, _shutdownCts.Token).ConfigureAwait(false);
                     break;
                 }
-                catch (Exception ex) when (!_shutdownCts.IsCancellationRequested)
-                {
-                    _logger.LogError(ex, "Error writing to channel.");
-                    channel = RecreateChannel(channel);
-                    continue;
-                }
-                catch
+
+                while (!_shutdownCts.IsCancellationRequested)
                 {
-                    // Shutdown requested.
+                    await channel.RequestStream.WriteAsync(message, _shutdownCts.Token).ConfigureAwait(false);
                     break;
                 }
             }
+            catch (OperationCanceledException)
+            {
+                // Time to shut down.
+                break;
+            }
+            catch (Exception ex) when (!_shutdownCts.IsCancellationRequested)
+            {
+                _logger.LogError(ex, "Error writing to channel.");
+                channel = RecreateChannel(channel);
+                continue;
+            }
+            catch
+            {
+                // Shutdown requested.
+                break;
+            }
         }
     }
 
@@ -286,10 +301,6 @@ public sealed class AgentWorkerRuntime : IHostedService, IDisposable, IAgentWork
     public async Task StopAsync(CancellationToken cancellationToken)
     {
         _shutdownCts.Cancel();
-        lock (_channelLock)
-        {
-            _channel?.Dispose();
-        }
 
         _outboundMessagesChannel.Writer.TryComplete();
 
@@ -302,6 +313,10 @@ public sealed class AgentWorkerRuntime : IHostedService, IDisposable, IAgentWork
         {
             await writeTask.ConfigureAwait(false);
         }
+        lock (_channelLock)
+        {
+            _channel?.Dispose();
+        }
     }
 
     public ValueTask SendRequest(RpcRequest request)

dotnet/src/Microsoft.AutoGen.Agents/Client/App.cs

@@ -49,7 +49,7 @@ public static class App
         {
             throw new InvalidOperationException("Client not started");
         }
-        await RuntimeApp!.StopAsync();
         await ClientApp.StopAsync();
+        await RuntimeApp!.StopAsync();
     }
 }

dotnet/src/Microsoft.AutoGen.Agents/Runtime/WorkerProcessConnection.cs

@@ -87,6 +87,9 @@ internal sealed class WorkerProcessConnection : IAsyncDisposable
                 _gateway.OnReceivedMessageAsync(this, message).Ignore();
             }
         }
+        catch (OperationCanceledException)
+        {
+        }
         finally
         {
             _shutdownCancellationToken.Cancel();
@@ -104,6 +107,9 @@ internal sealed class WorkerProcessConnection : IAsyncDisposable
                 await ResponseStream.WriteAsync(message);
             }
         }
+        catch (OperationCanceledException)
+        {
+        }
         finally
         {
             _shutdownCancellationToken.Cancel();