-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #97 from aneoconsulting/fl/connection-pool
Fl/connection pool
- Loading branch information
Showing
10 changed files
with
631 additions
and
433 deletions.
There are no files selected for viewing
160 changes: 83 additions & 77 deletions
160
Client/src/ArmoniK.DevelopmentKit.Client.Common/Submitter/BaseClientSubmitter.cs
Large diffs are not rendered by default.
Oops, something went wrong.
154 changes: 154 additions & 0 deletions
154
Client/src/ArmoniK.DevelopmentKit.Client.Common/Submitter/ChannelPool.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
|
|
||
Oops, something went wrong.