feat: more async (#264)

.github/workflows/build.yml

@@ -6,10 +6,6 @@ on:
       - main
       - "[0-9]+.[0-9]+.x"
   pull_request:
-    branches:
-      - main
-      - "[0-9]+.[0-9]+.x"
-
 
 jobs:
   versionning:

Client/src/Common/Submitter/ClientServiceConnector.cs

@@ -14,8 +14,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using System.Threading.Tasks;
+
 using ArmoniK.Api.Client.Options;
 using ArmoniK.Api.Client.Submitter;
+using ArmoniK.Utils;
+
+using Grpc.Net.Client;
 
 using Microsoft.Extensions.Logging;
 
@@ -33,8 +38,8 @@ public class ClientServiceConnector
   /// <param name="properties">Configuration Properties</param>
   /// <param name="loggerFactory">Optional logger factory</param>
   /// <returns>The connection pool</returns>
-  public static ChannelPool ControlPlaneConnectionPool(Properties      properties,
-                                                       ILoggerFactory? loggerFactory = null)
+  public static ObjectPool<GrpcChannel> ControlPlaneConnectionPool(Properties      properties,
+                                                                   ILoggerFactory? loggerFactory = null)
   {
     var options = new GrpcClient
                   {
@@ -52,7 +57,32 @@ public static ChannelPool ControlPlaneConnectionPool(Properties      properties,
                     ProxyPassword         = properties.ProxyPassword,
                   };
 
-    return new ChannelPool(() => GrpcChannelFactory.CreateChannel(options,
-                                                                  loggerFactory?.CreateLogger(typeof(ClientServiceConnector))));
+    return new ObjectPool<GrpcChannel>(ct => new ValueTask<GrpcChannel>(GrpcChannelFactory.CreateChannel(options,
+                                                                                                         loggerFactory?.CreateLogger(typeof(ClientServiceConnector)))),
+
+
+#if NET5_0_OR_GREATER
+                                       async (channel, ct) =>
+                                       {
+switch (channel.State)
+          {
+            case ConnectivityState.TransientFailure:
+              await channel.ShutdownAsync()
+                  .ConfigureAwait(false);
+              return false;
+            case ConnectivityState.Shutdown:
+              return false;
+            case ConnectivityState.Idle:
+            case ConnectivityState.Connecting:
+            case ConnectivityState.Ready:
+            default:
+              return true;
+          }
+                                       }
+#else
+                                       (_,
+                                        _) => new ValueTask<bool>(true)
+#endif
+                                      );
   }
 }

Client/src/Common/Submitter/TasksClientExt.cs

@@ -1,6 +1,6 @@
 // This file is part of the ArmoniK project
 //
-// Copyright (C) ANEO, 2021-$CURRENT_YEAR$. All rights reserved.
+// Copyright (C) ANEO, 2021-2024. All rights reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License")
 // you may not use this file except in compliance with the License.
@@ -17,6 +17,8 @@
 
 using System.Collections.Generic;
 using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
 
 using ArmoniK.Api.gRPC.V1;
 using ArmoniK.Api.gRPC.V1.Tasks;
@@ -133,21 +135,25 @@ public static FiltersAnd TaskStatusFilter(TaskStatus status,
   /// <param name="filters"> filters to apply on the tasks </param>
   /// <param name="sort"> sorting order </param>
   /// <param name="pageSize"> page size </param>
+  /// <param name="cancellationToken"></param>
   /// <returns></returns>
-  public static IEnumerable<TaskSummary> ListTasks(this Tasks.TasksClient      tasksClient,
-                                                   Filters                     filters,
-                                                   ListTasksRequest.Types.Sort sort,
-                                                   int                         pageSize = 50)
+  public static async IAsyncEnumerable<TaskSummary> ListTasksAsync(this Tasks.TasksClient                     tasksClient,
+                                                                   Filters                                    filters,
+                                                                   ListTasksRequest.Types.Sort                sort,
+                                                                   int                                        pageSize          = 50,
+                                                                   [EnumeratorCancellation] CancellationToken cancellationToken = default)
   {
     var               page = 0;
     ListTasksResponse res;
-    while ((res = tasksClient.ListTasks(new ListTasksRequest
-                                        {
-                                          Filters  = filters,
-                                          Sort     = sort,
-                                          PageSize = pageSize,
-                                          Page     = page,
-                                        })).Tasks.Any())
+    while ((res = await tasksClient.ListTasksAsync(new ListTasksRequest
+                                                   {
+                                                     Filters  = filters,
+                                                     Sort     = sort,
+                                                     PageSize = pageSize,
+                                                     Page     = page,
+                                                   },
+                                                   cancellationToken: cancellationToken)
+                                   .ConfigureAwait(false)).Tasks.Any())
     {
       foreach (var taskSummary in res.Tasks)
       {
@@ -157,4 +163,22 @@ public static IEnumerable<TaskSummary> ListTasks(this Tasks.TasksClient      tas
       page++;
     }
   }
+
+  /// <summary>
+  ///   List tasks while handling page size
+  /// </summary>
+  /// <param name="tasksClient"> the tasks client </param>
+  /// <param name="filters"> filters to apply on the tasks </param>
+  /// <param name="sort"> sorting order </param>
+  /// <param name="pageSize"> page size </param>
+  /// <returns></returns>
+  public static IEnumerable<TaskSummary> ListTasks(this Tasks.TasksClient      tasksClient,
+                                                   Filters                     filters,
+                                                   ListTasksRequest.Types.Sort sort,
+                                                   int                         pageSize = 50)
+    => ListTasksAsync(tasksClient,
+                      filters,
+                      sort,
+                      pageSize)
+      .ToEnumerable();
 }

Client/src/Symphony/ArmonikSymphonyClient.cs

@@ -18,6 +18,9 @@
 using ArmoniK.DevelopmentKit.Client.Common;
 using ArmoniK.DevelopmentKit.Client.Common.Submitter;
 using ArmoniK.DevelopmentKit.Common;
+using ArmoniK.Utils;
+
+using Grpc.Net.Client;
 
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
@@ -63,7 +66,7 @@ public ArmonikSymphonyClient(IConfiguration configuration,
   /// </summary>
   public string SectionGrpc { get; set; } = "Grpc";
 
-  private ChannelPool GrpcPool { get; set; }
+  private ObjectPool<GrpcChannel> GrpcPool { get; set; }
 
 
   private IConfiguration Configuration { get; }