#region Copyright notice and license
// Copyright 2015 gRPC authors.
//
// 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.
// See the License for the specific language governing permissions and
// limitations under the License.
#endregion
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
using Grpc.Core.Internal;
using Grpc.Core.Logging;
using Grpc.Core.Utils;
namespace Grpc.Core
{
///
/// Encapsulates initialization and shutdown of gRPC library.
///
public class GrpcEnvironment
{
const int MinDefaultThreadPoolSize = 4;
const int DefaultBatchContextPoolSharedCapacity = 10000;
const int DefaultBatchContextPoolThreadLocalCapacity = 64;
const int DefaultRequestCallContextPoolSharedCapacity = 10000;
const int DefaultRequestCallContextPoolThreadLocalCapacity = 64;
static object staticLock = new object();
static GrpcEnvironment instance;
static int refCount;
static int? customThreadPoolSize;
static int? customCompletionQueueCount;
static bool inlineHandlers;
static int batchContextPoolSharedCapacity = DefaultBatchContextPoolSharedCapacity;
static int batchContextPoolThreadLocalCapacity = DefaultBatchContextPoolThreadLocalCapacity;
static int requestCallContextPoolSharedCapacity = DefaultRequestCallContextPoolSharedCapacity;
static int requestCallContextPoolThreadLocalCapacity = DefaultRequestCallContextPoolThreadLocalCapacity;
static readonly HashSet registeredChannels = new HashSet();
static readonly HashSet registeredServers = new HashSet();
static ILogger logger = new LogLevelFilterLogger(new ConsoleLogger(), LogLevel.Off, true);
readonly IObjectPool batchContextPool;
readonly IObjectPool requestCallContextPool;
readonly GrpcThreadPool threadPool;
readonly DebugStats debugStats = new DebugStats();
readonly AtomicCounter cqPickerCounter = new AtomicCounter();
bool isShutdown;
///
/// Returns a reference-counted instance of initialized gRPC environment.
/// Subsequent invocations return the same instance unless reference count has dropped to zero previously.
///
internal static GrpcEnvironment AddRef()
{
ShutdownHooks.Register();
lock (staticLock)
{
refCount++;
if (instance == null)
{
instance = new GrpcEnvironment();
}
return instance;
}
}
///
/// Decrements the reference count for currently active environment and asynchronously shuts down the gRPC environment if reference count drops to zero.
///
internal static async Task ReleaseAsync()
{
GrpcEnvironment instanceToShutdown = null;
lock (staticLock)
{
GrpcPreconditions.CheckState(refCount > 0);
refCount--;
if (refCount == 0)
{
instanceToShutdown = instance;
instance = null;
}
}
if (instanceToShutdown != null)
{
await instanceToShutdown.ShutdownAsync().ConfigureAwait(false);
}
}
internal static int GetRefCount()
{
lock (staticLock)
{
return refCount;
}
}
internal static void RegisterChannel(Channel channel)
{
lock (staticLock)
{
GrpcPreconditions.CheckNotNull(channel);
registeredChannels.Add(channel);
}
}
internal static void UnregisterChannel(Channel channel)
{
lock (staticLock)
{
GrpcPreconditions.CheckNotNull(channel);
GrpcPreconditions.CheckArgument(registeredChannels.Remove(channel), "Channel not found in the registered channels set.");
}
}
internal static void RegisterServer(Server server)
{
lock (staticLock)
{
GrpcPreconditions.CheckNotNull(server);
registeredServers.Add(server);
}
}
internal static void UnregisterServer(Server server)
{
lock (staticLock)
{
GrpcPreconditions.CheckNotNull(server);
GrpcPreconditions.CheckArgument(registeredServers.Remove(server), "Server not found in the registered servers set.");
}
}
///
/// Requests shutdown of all channels created by the current process.
///
public static Task ShutdownChannelsAsync()
{
HashSet snapshot = null;
lock (staticLock)
{
snapshot = new HashSet(registeredChannels);
}
return Task.WhenAll(snapshot.Select((channel) => channel.ShutdownAsync()));
}
///
/// Requests immediate shutdown of all servers created by the current process.
///
public static Task KillServersAsync()
{
HashSet snapshot = null;
lock (staticLock)
{
snapshot = new HashSet(registeredServers);
}
return Task.WhenAll(snapshot.Select((server) => server.KillAsync()));
}
///
/// Gets application-wide logger used by gRPC.
///
/// The logger.
public static ILogger Logger
{
get
{
return logger;
}
}
///
/// Sets the application-wide logger that should be used by gRPC.
///
public static void SetLogger(ILogger customLogger)
{
GrpcPreconditions.CheckNotNull(customLogger, "customLogger");
logger = customLogger;
}
///
/// Sets the number of threads in the gRPC thread pool that polls for internal RPC events.
/// Can be only invoked before the GrpcEnviroment is started and cannot be changed afterwards.
/// Setting thread pool size is an advanced setting and you should only use it if you know what you are doing.
/// Most users should rely on the default value provided by gRPC library.
/// Note: this method is part of an experimental API that can change or be removed without any prior notice.
///
public static void SetThreadPoolSize(int threadCount)
{
lock (staticLock)
{
GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
GrpcPreconditions.CheckArgument(threadCount > 0, "threadCount needs to be a positive number");
customThreadPoolSize = threadCount;
}
}
///
/// Sets the number of completion queues in the gRPC thread pool that polls for internal RPC events.
/// Can be only invoked before the GrpcEnviroment is started and cannot be changed afterwards.
/// Setting the number of completions queues is an advanced setting and you should only use it if you know what you are doing.
/// Most users should rely on the default value provided by gRPC library.
/// Note: this method is part of an experimental API that can change or be removed without any prior notice.
///
public static void SetCompletionQueueCount(int completionQueueCount)
{
lock (staticLock)
{
GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
GrpcPreconditions.CheckArgument(completionQueueCount > 0, "threadCount needs to be a positive number");
customCompletionQueueCount = completionQueueCount;
}
}
///
/// By default, gRPC's internal event handlers get offloaded to .NET default thread pool thread (inlineHandlers=false).
/// Setting inlineHandlers to true will allow scheduling the event handlers directly to
/// GrpcThreadPool internal threads. That can lead to significant performance gains in some situations,
/// but requires user to never block in async code (incorrectly written code can easily lead to deadlocks).
/// Inlining handlers is an advanced setting and you should only use it if you know what you are doing.
/// Most users should rely on the default value provided by gRPC library.
/// Note: this method is part of an experimental API that can change or be removed without any prior notice.
/// Note: inlineHandlers=true was the default in gRPC C# v1.4.x and earlier.
///
public static void SetHandlerInlining(bool inlineHandlers)
{
lock (staticLock)
{
GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
GrpcEnvironment.inlineHandlers = inlineHandlers;
}
}
///
/// Sets the parameters for a pool that caches batch context instances. Reusing batch context instances
/// instead of creating a new one for every C core operation helps reducing the GC pressure.
/// Can be only invoked before the GrpcEnviroment is started and cannot be changed afterwards.
/// This is an advanced setting and you should only use it if you know what you are doing.
/// Most users should rely on the default value provided by gRPC library.
/// Note: this method is part of an experimental API that can change or be removed without any prior notice.
///
public static void SetBatchContextPoolParams(int sharedCapacity, int threadLocalCapacity)
{
lock (staticLock)
{
GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
GrpcPreconditions.CheckArgument(sharedCapacity >= 0, "Shared capacity needs to be a non-negative number");
GrpcPreconditions.CheckArgument(threadLocalCapacity >= 0, "Thread local capacity needs to be a non-negative number");
batchContextPoolSharedCapacity = sharedCapacity;
batchContextPoolThreadLocalCapacity = threadLocalCapacity;
}
}
///
/// Sets the parameters for a pool that caches request call context instances. Reusing request call context instances
/// instead of creating a new one for every requested call in C core helps reducing the GC pressure.
/// Can be only invoked before the GrpcEnviroment is started and cannot be changed afterwards.
/// This is an advanced setting and you should only use it if you know what you are doing.
/// Most users should rely on the default value provided by gRPC library.
/// Note: this method is part of an experimental API that can change or be removed without any prior notice.
///
public static void SetRequestCallContextPoolParams(int sharedCapacity, int threadLocalCapacity)
{
lock (staticLock)
{
GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
GrpcPreconditions.CheckArgument(sharedCapacity >= 0, "Shared capacity needs to be a non-negative number");
GrpcPreconditions.CheckArgument(threadLocalCapacity >= 0, "Thread local capacity needs to be a non-negative number");
requestCallContextPoolSharedCapacity = sharedCapacity;
requestCallContextPoolThreadLocalCapacity = threadLocalCapacity;
}
}
///
/// Occurs when GrpcEnvironment is about the start the shutdown logic.
/// If GrpcEnvironment is later initialized and shutdown, the event will be fired again (unless unregistered first).
///
public static event EventHandler ShuttingDown;
///
/// Creates gRPC environment.
///
private GrpcEnvironment()
{
GrpcNativeInit();
batchContextPool = new DefaultObjectPool(() => BatchContextSafeHandle.Create(this.batchContextPool), batchContextPoolSharedCapacity, batchContextPoolThreadLocalCapacity);
requestCallContextPool = new DefaultObjectPool(() => RequestCallContextSafeHandle.Create(this.requestCallContextPool), requestCallContextPoolSharedCapacity, requestCallContextPoolThreadLocalCapacity);
threadPool = new GrpcThreadPool(this, GetThreadPoolSizeOrDefault(), GetCompletionQueueCountOrDefault(), inlineHandlers);
threadPool.Start();
}
///
/// Gets the completion queues used by this gRPC environment.
///
internal IReadOnlyCollection CompletionQueues
{
get
{
return this.threadPool.CompletionQueues;
}
}
internal IObjectPool BatchContextPool => batchContextPool;
internal IObjectPool RequestCallContextPool => requestCallContextPool;
internal bool IsAlive
{
get
{
return this.threadPool.IsAlive;
}
}
///
/// Picks a completion queue in a round-robin fashion.
/// Shouldn't be invoked on a per-call basis (used at per-channel basis).
///
internal CompletionQueueSafeHandle PickCompletionQueue()
{
var cqIndex = (int) ((cqPickerCounter.Increment() - 1) % this.threadPool.CompletionQueues.Count);
return this.threadPool.CompletionQueues.ElementAt(cqIndex);
}
///
/// Gets the completion queue used by this gRPC environment.
///
internal DebugStats DebugStats
{
get
{
return this.debugStats;
}
}
///
/// Gets version of gRPC C core.
///
internal static string GetCoreVersionString()
{
var ptr = NativeMethods.Get().grpcsharp_version_string(); // the pointer is not owned
return Marshal.PtrToStringAnsi(ptr);
}
internal static void GrpcNativeInit()
{
NativeMethods.Get().grpcsharp_init();
}
internal static void GrpcNativeShutdown()
{
NativeMethods.Get().grpcsharp_shutdown();
}
///
/// Shuts down this environment.
///
private async Task ShutdownAsync()
{
if (isShutdown)
{
throw new InvalidOperationException("ShutdownAsync has already been called");
}
await Task.Run(() => ShuttingDown?.Invoke(this, null)).ConfigureAwait(false);
await threadPool.StopAsync().ConfigureAwait(false);
batchContextPool.Dispose();
GrpcNativeShutdown();
isShutdown = true;
debugStats.CheckOK();
}
private int GetThreadPoolSizeOrDefault()
{
if (customThreadPoolSize.HasValue)
{
return customThreadPoolSize.Value;
}
// In systems with many cores, use half of the cores for GrpcThreadPool
// and the other half for .NET thread pool. This heuristic definitely needs
// more work, but seems to work reasonably well for a start.
return Math.Max(MinDefaultThreadPoolSize, Environment.ProcessorCount / 2);
}
private int GetCompletionQueueCountOrDefault()
{
if (customCompletionQueueCount.HasValue)
{
return customCompletionQueueCount.Value;
}
// by default, create a completion queue for each thread
return GetThreadPoolSizeOrDefault();
}
private static class ShutdownHooks
{
static object staticLock = new object();
static bool hooksRegistered;
public static void Register()
{
lock (staticLock)
{
if (!hooksRegistered)
{
#if NETSTANDARD1_5
System.Runtime.Loader.AssemblyLoadContext.Default.Unloading += (assemblyLoadContext) => { HandleShutdown(); };
#else
AppDomain.CurrentDomain.ProcessExit += (sender, eventArgs) => { HandleShutdown(); };
AppDomain.CurrentDomain.DomainUnload += (sender, eventArgs) => { HandleShutdown(); };
#endif
}
hooksRegistered = true;
}
}
///
/// Handler for AppDomain.DomainUnload, AppDomain.ProcessExit and AssemblyLoadContext.Unloading hooks.
///
private static void HandleShutdown()
{
Task.WaitAll(GrpcEnvironment.ShutdownChannelsAsync(), GrpcEnvironment.KillServersAsync());
}
}
}
}