Fl/connection pool

Client/src/ArmoniK.DevelopmentKit.Client.Common/Submitter/ChannelPool.cs

@@ -0,0 +1,154 @@
+// This file is part of the ArmoniK project
+//
+// Copyright (C) ANEO, 2021-2022. All rights reserved.
+//   W. Kirschenmann   <[email protected]>
+//   J. Gurhem         <[email protected]>
+//   D. Dubuc          <[email protected]>
+//   L. Ziane Khodja   <[email protected]>
+//   F. Lemaitre       <[email protected]>
+//   S. Djebbar        <[email protected]>
+//   J. Fonseca        <[email protected]>
+//   D. Brasseur       <[email protected]>
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY, without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+using System;
+using System.Collections.Concurrent;
+
+using Grpc.Core;
+
+using JetBrains.Annotations;
+
+using Microsoft.Extensions.Logging;
+
+namespace ArmoniK.DevelopmentKit.Client.Common.Submitter;
+
+/// <summary>
+///   Helper to have a connection pool for gRPC services
+/// </summary>
+public sealed class ChannelPool
+{
+  private readonly Func<ChannelBase> channelFactory_;
+
+  [CanBeNull]
+  private readonly ILogger<ChannelPool> logger_;
+
+  private readonly ConcurrentBag<ChannelBase> pool_;
+
+  /// <summary>
+  ///   Constructs a new channelPool
+  /// </summary>
+  /// <param name="channelFactory">Function used to create new channels</param>
+  /// <param name="loggerFactory">loggerFactory used to instantiate a logger for the pool</param>
+  public ChannelPool(Func<ChannelBase>          channelFactory,
+                     [CanBeNull] ILoggerFactory loggerFactory = null)
+  {
+    channelFactory_ = channelFactory;
+    pool_           = new ConcurrentBag<ChannelBase>();
+    logger_         = loggerFactory?.CreateLogger<ChannelPool>();
+  }
+
+  /// <summary>
+  ///   Get a channel from the pool. If the pool is empty, create a new channel
+  /// </summary>
+  /// <returns>A ChannelBase used by nobody else</returns>
+  private ChannelBase AcquireChannel()
+  {
+    if (pool_.TryTake(out var channel))
+    {
+      logger_?.LogDebug("Acquired already existing channel {channel} from pool",
+                        channel);
+      return channel;
+    }
+
+    channel = channelFactory_();
+    logger_?.LogInformation("Created and acquired new channel {channel} from pool",
+                            channel);
+    return channel;
+  }
+
+  /// <summary>
+  ///   Release a ChannelBase to the pool that could be reused later by someone else
+  /// </summary>
+  /// <param name="channel">Channel to release</param>
+  private void ReleaseChannel(ChannelBase channel)
+  {
+    logger_?.LogDebug("Released channel {channel} to pool",
+                      channel);
+    pool_.Add(channel);
+  }
+
+  /// <summary>
+  ///   Get a channel that will be automatically released when disposed
+  /// </summary>
+  /// <returns></returns>
+  public ChannelGuard GetChannel()
+    => new(this);
+
+  /// <summary>
+  ///   Call f with an acquired channel
+  /// </summary>
+  /// <param name="f">Function to be called</param>
+  /// <typeparam name="T">Type of the return type of f</typeparam>
+  /// <returns>Value returned by f</returns>
+  public T WithChannel<T>(Func<ChannelBase, T> f)
+  {
+    using var channel = GetChannel();
+    return f(channel.Channel);
+  }
+
+  /// <summary>
+  ///   Call f with an acquired channel
+  /// </summary>
+  /// <param name="f">Function to be called</param>
+  public void WithChannel(Action<ChannelBase> f)
+  {
+    using var channel = GetChannel();
+    f(channel.Channel);
+  }
+
+  /// <summary>
+  ///   Helper class that acquires a channel from a pool when constructed, and releases it when disposed
+  /// </summary>
+  public sealed class ChannelGuard : IDisposable
+  {
+    /// <summary>
+    ///   Channel that is used by nobody else
+    /// </summary>
+    public readonly ChannelBase Channel;
+
+    private readonly ChannelPool pool_;
+
+    /// <summary>
+    ///   Acquire a channel that will be released when disposed
+    /// </summary>
+    /// <param name="channelPool"></param>
+    public ChannelGuard(ChannelPool channelPool)
+    {
+      pool_   = channelPool;
+      Channel = channelPool.AcquireChannel();
+    }
+
+    /// <inheritdoc />
+    public void Dispose()
+      => pool_.ReleaseChannel(Channel);
+
+    /// <summary>
+    ///   Implicit convert a ChannelGuard into a ChannelBase
+    /// </summary>
+    /// <param name="guard">ChannelGuard</param>
+    /// <returns>ChannelBase</returns>
+    public static implicit operator ChannelBase(ChannelGuard guard)
+      => guard.Channel;
+  }
+}

Client/src/ArmoniK.DevelopmentKit.Client.Common/Submitter/ClientServiceConnector.cs

@@ -1,19 +1,18 @@
-#if NET5_0_OR_GREATER
-using Grpc.Net.Client;
-using Grpc.Core;
-
-using System.Runtime.InteropServices;
-using System.Security.Cryptography.X509Certificates;
-#else
-using Grpc.Core;
-#endif
 using System;
 using System.IO;
 using System.Net.Http;
 
+using Grpc.Core;
+
 using JetBrains.Annotations;
 
 using Microsoft.Extensions.Logging;
+#if NET5_0_OR_GREATER
+using Grpc.Net.Client;
+
+using System.Runtime.InteropServices;
+using System.Security.Cryptography.X509Certificates;
+#endif
 
 namespace ArmoniK.DevelopmentKit.Client.Common.Submitter;
 
@@ -30,9 +29,9 @@ public class ClientServiceConnector
   /// <param name="sslValidation">Optional : Check if the ssl must have a strong validation (default true)</param>
   /// <param name="loggerFactory">Optional : the logger factory to create the logger</param>
   /// <returns></returns>
-  public static ChannelBase ControlPlaneConnection(string         endPoint,
-                                                   bool           sslValidation = true,
-                                                   ILoggerFactory loggerFactory = null)
+  private static ChannelBase ControlPlaneConnection(string         endPoint,
+                                                    bool           sslValidation = true,
+                                                    ILoggerFactory loggerFactory = null)
     => ControlPlaneConnection(endPoint,
                               "",
                               "",
@@ -42,17 +41,17 @@ public static ChannelBase ControlPlaneConnection(string         endPoint,
   /// <summary>
   ///   Open Connection with the control plane with mTLS authentication
   /// </summary>
-  /// <param name="endPoint"></param>
+  /// <param name="endPoint">The address and port of control plane</param>
   /// <param name="clientCertFilename">The certificate filename in a pem format</param>
   /// <param name="clientKeyFilename">The client key filename in a pem format</param>
   /// <param name="sslValidation">Check if the ssl must have a strong validation</param>
   /// <param name="loggerFactory">Optional logger factory</param>
   /// <returns></returns>
-  public static ChannelBase ControlPlaneConnection(string                     endPoint,
-                                                   string                     clientCertFilename,
-                                                   string                     clientKeyFilename,
-                                                   bool                       sslValidation = true,
-                                                   [CanBeNull] ILoggerFactory loggerFactory = null)
+  private static ChannelBase ControlPlaneConnection(string                     endPoint,
+                                                    string                     clientCertFilename,
+                                                    string                     clientKeyFilename,
+                                                    bool                       sslValidation = true,
+                                                    [CanBeNull] ILoggerFactory loggerFactory = null)
   {
     var logger = loggerFactory?.CreateLogger<ClientServiceConnector>();
     if ((!string.IsNullOrEmpty(clientCertFilename) && string.IsNullOrEmpty(clientKeyFilename)) ||
@@ -99,15 +98,15 @@ public static ChannelBase ControlPlaneConnection(string                     endP
   /// <summary>
   ///   Open Connection with the control plane with mTLS authentication
   /// </summary>
-  /// <param name="endPoint"></param>
+  /// <param name="endPoint">The address and port of control plane</param>
   /// <param name="clientPem">The pair certificate + key data in a pem format</param>
   /// <param name="sslValidation">Check if the ssl must have a strong validation</param>
   /// <param name="loggerFactory">Optional logger factory</param>
   /// <returns></returns>
-  public static ChannelBase ControlPlaneConnection(string                     endPoint,
-                                                   Tuple<string, string>      clientPem     = null,
-                                                   bool                       sslValidation = true,
-                                                   [CanBeNull] ILoggerFactory loggerFactory = null)
+  private static ChannelBase ControlPlaneConnection(string                     endPoint,
+                                                    Tuple<string, string>      clientPem     = null,
+                                                    bool                       sslValidation = true,
+                                                    [CanBeNull] ILoggerFactory loggerFactory = null)
   {
     var uri = new Uri(endPoint);
 
@@ -168,4 +167,59 @@ public static ChannelBase ControlPlaneConnection(string                     endP
 #endif
     return channel;
   }
+
+  /// <summary>
+  ///   Create a connection pool to the control plane with or without SSL and no mTLS
+  /// </summary>
+  /// <param name="endPoint">The address and port of control plane</param>
+  /// <param name="sslValidation">Optional : Check if the ssl must have a strong validation (default true)</param>
+  /// <param name="loggerFactory">Optional : the logger factory to create the logger</param>
+  /// <returns></returns>
+  public static ChannelPool ControlPlaneConnectionPool(string         endPoint,
+                                                       bool           sslValidation = true,
+                                                       ILoggerFactory loggerFactory = null)
+    => new(() => ControlPlaneConnection(endPoint,
+                                        sslValidation,
+                                        loggerFactory),
+           loggerFactory);
+
+
+  /// <summary>
+  ///   Create a connection pool to the control plane with mTLS authentication
+  /// </summary>
+  /// <param name="endPoint">The address and port of control plane</param>
+  /// <param name="clientCertFilename">The certificate filename in a pem format</param>
+  /// <param name="clientKeyFilename">The client key filename in a pem format</param>
+  /// <param name="sslValidation">Check if the ssl must have a strong validation</param>
+  /// <param name="loggerFactory">Optional logger factory</param>
+  /// <returns></returns>
+  public static ChannelPool ControlPlaneConnectionPool(string                     endPoint,
+                                                       string                     clientCertFilename,
+                                                       string                     clientKeyFilename,
+                                                       bool                       sslValidation = true,
+                                                       [CanBeNull] ILoggerFactory loggerFactory = null)
+    => new(() => ControlPlaneConnection(endPoint,
+                                        clientCertFilename,
+                                        clientKeyFilename,
+                                        sslValidation,
+                                        loggerFactory),
+           loggerFactory);
+
+  /// <summary>
+  ///   Create a connection pool to the control plane with mTLS authentication
+  /// </summary>
+  /// <param name="endPoint">The address and port of control plane</param>
+  /// <param name="clientPem">The pair certificate + key data in a pem format</param>
+  /// <param name="sslValidation">Check if the ssl must have a strong validation</param>
+  /// <param name="loggerFactory">Optional logger factory</param>
+  /// <returns>The connection pool</returns>
+  public static ChannelPool ControlPlaneConnectionPool(string                     endPoint,
+                                                       Tuple<string, string>      clientPem     = null,
+                                                       bool                       sslValidation = true,
+                                                       [CanBeNull] ILoggerFactory loggerFactory = null)
+    => new(() => ControlPlaneConnection(endPoint,
+                                        clientPem,
+                                        sslValidation,
+                                        loggerFactory),
+           loggerFactory);
 }

DataSynapseApi/ArmoniK.DevelopmentKit.GridServer.Client/ArmonikDataSynapseClientService.cs

@@ -1,5 +1,5 @@
 // This file is part of the ArmoniK project
-// 
+//
 // Copyright (C) ANEO, 2021-2022.
 //   W. Kirschenmann   <[email protected]>
 //   J. Gurhem         <[email protected]>
@@ -8,13 +8,13 @@
 //   F. Lemaitre       <[email protected]>
 //   S. Djebbar        <[email protected]>
 //   J. Fonseca        <[email protected]>
-// 
+//
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
-// 
+//
 //     http://www.apache.org/licenses/LICENSE-2.0
-// 
+//
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -27,8 +27,6 @@
 
 using Google.Protobuf.WellKnownTypes;
 
-using Grpc.Core;
-
 using JetBrains.Annotations;
 
 using Microsoft.Extensions.Logging;
@@ -66,7 +64,7 @@ public ArmonikDataSynapseClientService(Properties                 properties,
   [CanBeNull]
   private ILogger<ArmonikDataSynapseClientService> Logger { get; }
 
-  private ChannelBase GrpcChannel { get; set; }
+  private ChannelPool GrpcPool { get; set; }
 
 
   /// <summary>
@@ -92,24 +90,24 @@ public SessionService CreateSession(TaskOptions taskOptions = null)
 
     Logger?.LogDebug("Creating Session... ");
 
-    return new SessionService(GrpcChannel,
+    return new SessionService(GrpcPool,
                               LoggerFactory,
                               TaskOptions);
   }
 
   private void ControlPlaneConnection()
   {
-    if (GrpcChannel != null)
+    if (GrpcPool != null)
     {
       return;
     }
 
 
-    GrpcChannel = ClientServiceConnector.ControlPlaneConnection(properties_.ConnectionString,
-                                                                properties_.ClientCertFilePem,
-                                                                properties_.ClientKeyFilePem,
-                                                                properties_.ConfSSLValidation,
-                                                                LoggerFactory);
+    GrpcPool = ClientServiceConnector.ControlPlaneConnectionPool(properties_.ConnectionString,
+                                                                 properties_.ClientCertFilePem,
+                                                                 properties_.ClientKeyFilePem,
+                                                                 properties_.ConfSSLValidation,
+                                                                 LoggerFactory);
   }
 
   /// <summary>
@@ -122,7 +120,7 @@ public SessionService OpenSession(string      sessionId,
   {
     ControlPlaneConnection();
 
-    return new SessionService(GrpcChannel,
+    return new SessionService(GrpcPool,
                               LoggerFactory,
                               clientOptions ?? SessionService.InitializeDefaultTaskOptions(),
                               new Session