aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/csharp/Grpc.Core/Internal
diff options
context:
space:
mode:
Diffstat (limited to 'src/csharp/Grpc.Core/Internal')
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCall.cs97
-rw-r--r--src/csharp/Grpc.Core/Internal/AsyncCallBase.cs41
-rw-r--r--src/csharp/Grpc.Core/Internal/CallSafeHandle.cs7
-rw-r--r--src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs13
-rw-r--r--src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs5
-rw-r--r--src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs22
-rw-r--r--src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs23
7 files changed, 95 insertions, 113 deletions
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 5e61e9ec12..da45c4829d 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -388,35 +388,29 @@ namespace Grpc.Core.Internal
private void Initialize(CompletionQueueSafeHandle cq)
{
- using (Profilers.ForCurrentThread().NewScope("AsyncCall.Initialize"))
- {
- var call = CreateNativeCall(cq);
+ var call = CreateNativeCall(cq);
- details.Channel.AddCallReference(this);
- InitializeInternal(call);
- RegisterCancellationCallback();
- }
+ details.Channel.AddCallReference(this);
+ InitializeInternal(call);
+ RegisterCancellationCallback();
}
private INativeCall CreateNativeCall(CompletionQueueSafeHandle cq)
{
- using (Profilers.ForCurrentThread().NewScope("AsyncCall.CreateNativeCall"))
- {
- if (injectedNativeCall != null)
- {
- return injectedNativeCall; // allows injecting a mock INativeCall in tests.
- }
+ if (injectedNativeCall != null)
+ {
+ return injectedNativeCall; // allows injecting a mock INativeCall in tests.
+ }
- var parentCall = details.Options.PropagationToken != null ? details.Options.PropagationToken.ParentCall : CallSafeHandle.NullInstance;
+ var parentCall = details.Options.PropagationToken != null ? details.Options.PropagationToken.ParentCall : CallSafeHandle.NullInstance;
- var credentials = details.Options.Credentials;
- using (var nativeCredentials = credentials != null ? credentials.ToNativeCredentials() : null)
- {
- var result = details.Channel.Handle.CreateCall(
- parentCall, ContextPropagationToken.DefaultMask, cq,
- details.Method, details.Host, Timespec.FromDateTime(details.Options.Deadline.Value), nativeCredentials);
- return result;
- }
+ var credentials = details.Options.Credentials;
+ using (var nativeCredentials = credentials != null ? credentials.ToNativeCredentials() : null)
+ {
+ var result = details.Channel.Handle.CreateCall(
+ parentCall, ContextPropagationToken.DefaultMask, cq,
+ details.Method, details.Host, Timespec.FromDateTime(details.Options.Deadline.Value), nativeCredentials);
+ return result;
}
}
@@ -456,47 +450,44 @@ namespace Grpc.Core.Internal
// NOTE: because this event is a result of batch containing GRPC_OP_RECV_STATUS_ON_CLIENT,
// success will be always set to true.
- using (Profilers.ForCurrentThread().NewScope("AsyncCall.HandleUnaryResponse"))
+ TaskCompletionSource<object> delayedStreamingWriteTcs = null;
+ TResponse msg = default(TResponse);
+ var deserializeException = TryDeserialize(receivedMessage, out msg);
+
+ lock (myLock)
{
- TaskCompletionSource<object> delayedStreamingWriteTcs = null;
- TResponse msg = default(TResponse);
- var deserializeException = TryDeserialize(receivedMessage, out msg);
+ finished = true;
- lock (myLock)
+ if (deserializeException != null && receivedStatus.Status.StatusCode == StatusCode.OK)
{
- finished = true;
-
- if (deserializeException != null && receivedStatus.Status.StatusCode == StatusCode.OK)
- {
- receivedStatus = new ClientSideStatus(DeserializeResponseFailureStatus, receivedStatus.Trailers);
- }
- finishedStatus = receivedStatus;
-
- if (isStreamingWriteCompletionDelayed)
- {
- delayedStreamingWriteTcs = streamingWriteTcs;
- streamingWriteTcs = null;
- }
-
- ReleaseResourcesIfPossible();
+ receivedStatus = new ClientSideStatus(DeserializeResponseFailureStatus, receivedStatus.Trailers);
}
+ finishedStatus = receivedStatus;
- responseHeadersTcs.SetResult(responseHeaders);
-
- if (delayedStreamingWriteTcs != null)
+ if (isStreamingWriteCompletionDelayed)
{
- delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly());
+ delayedStreamingWriteTcs = streamingWriteTcs;
+ streamingWriteTcs = null;
}
- var status = receivedStatus.Status;
- if (status.StatusCode != StatusCode.OK)
- {
- unaryResponseTcs.SetException(new RpcException(status));
- return;
- }
+ ReleaseResourcesIfPossible();
+ }
+
+ responseHeadersTcs.SetResult(responseHeaders);
- unaryResponseTcs.SetResult(msg);
+ if (delayedStreamingWriteTcs != null)
+ {
+ delayedStreamingWriteTcs.SetException(GetRpcExceptionClientOnly());
+ }
+
+ var status = receivedStatus.Status;
+ if (status.StatusCode != StatusCode.OK)
+ {
+ unaryResponseTcs.SetException(new RpcException(status));
+ return;
}
+
+ unaryResponseTcs.SetResult(msg);
}
/// <summary>
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
index 9f9d260e7e..8668903f6e 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -181,19 +181,16 @@ namespace Grpc.Core.Internal
/// </summary>
protected bool ReleaseResourcesIfPossible()
{
- using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.ReleaseResourcesIfPossible"))
+ if (!disposed && call != null)
{
- if (!disposed && call != null)
+ bool noMoreSendCompletions = streamingWriteTcs == null && (halfcloseRequested || cancelRequested || finished);
+ if (noMoreSendCompletions && readingDone && finished)
{
- bool noMoreSendCompletions = streamingWriteTcs == null && (halfcloseRequested || cancelRequested || finished);
- if (noMoreSendCompletions && readingDone && finished)
- {
- ReleaseResources();
- return true;
- }
+ ReleaseResources();
+ return true;
}
- return false;
}
+ return false;
}
protected abstract bool IsClient
@@ -229,28 +226,20 @@ namespace Grpc.Core.Internal
protected byte[] UnsafeSerialize(TWrite msg)
{
- using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.UnsafeSerialize"))
- {
- return serializer(msg);
- }
+ return serializer(msg);
}
protected Exception TryDeserialize(byte[] payload, out TRead msg)
{
- using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.TryDeserialize"))
+ try
{
- try
- {
-
- msg = deserializer(payload);
- return null;
-
- }
- catch (Exception e)
- {
- msg = default(TRead);
- return e;
- }
+ msg = deserializer(payload);
+ return null;
+ }
+ catch (Exception e)
+ {
+ msg = default(TRead);
+ return e;
}
}
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index 82361f5797..f817a61bce 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -76,11 +76,8 @@ namespace Grpc.Core.Internal
public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, MetadataArraySafeHandle metadataArray, WriteFlags writeFlags)
{
- using (Profilers.ForCurrentThread().NewScope("CallSafeHandle.StartUnary"))
- {
- Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags)
- .CheckOk();
- }
+ Native.grpcsharp_call_start_unary(this, ctx, payload, new UIntPtr((ulong)payload.Length), metadataArray, writeFlags)
+ .CheckOk();
}
public void StartClientStreaming(UnaryResponseClientHandler callback, MetadataArraySafeHandle metadataArray)
diff --git a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
index 62864dff0c..0fb6360a23 100644
--- a/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ChannelSafeHandle.cs
@@ -65,16 +65,13 @@ namespace Grpc.Core.Internal
public CallSafeHandle CreateCall(CallSafeHandle parentCall, ContextPropagationFlags propagationMask, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline, CallCredentialsSafeHandle credentials)
{
- using (Profilers.ForCurrentThread().NewScope("ChannelSafeHandle.CreateCall"))
+ var result = Native.grpcsharp_channel_create_call(this, parentCall, propagationMask, cq, method, host, deadline);
+ if (credentials != null)
{
- var result = Native.grpcsharp_channel_create_call(this, parentCall, propagationMask, cq, method, host, deadline);
- if (credentials != null)
- {
- result.SetCredentials(credentials);
- }
- result.Initialize(cq);
- return result;
+ result.SetCredentials(credentials);
}
+ result.Initialize(cq);
+ return result;
}
public ChannelState CheckConnectivityState(bool tryToConnect)
diff --git a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
index 46f5624223..6c9a31921e 100644
--- a/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CompletionQueueSafeHandle.cs
@@ -70,10 +70,7 @@ namespace Grpc.Core.Internal
public CompletionQueueEvent Pluck(IntPtr tag)
{
- using (Profilers.ForCurrentThread().NewScope("CompletionQueueSafeHandle.Pluck"))
- {
- return Native.grpcsharp_completion_queue_pluck(this, tag);
- }
+ return Native.grpcsharp_completion_queue_pluck(this, tag);
}
/// <summary>
diff --git a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
index a446c1f99f..25a6589f11 100644
--- a/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
+++ b/src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs
@@ -37,6 +37,7 @@ using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Grpc.Core.Logging;
+using Grpc.Core.Profiling;
using Grpc.Core.Utils;
namespace Grpc.Core.Internal
@@ -54,6 +55,8 @@ namespace Grpc.Core.Internal
readonly int poolSize;
readonly int completionQueueCount;
+ readonly List<BasicProfiler> threadProfilers = new List<BasicProfiler>(); // profilers assigned to threadpool threads
+
bool stopRequested;
IReadOnlyCollection<CompletionQueueSafeHandle> completionQueues;
@@ -82,7 +85,8 @@ namespace Grpc.Core.Internal
for (int i = 0; i < poolSize; i++)
{
- threads.Add(CreateAndStartThread(i));
+ var optionalProfiler = i < threadProfilers.Count ? threadProfilers[i] : null;
+ threads.Add(CreateAndStartThread(i, optionalProfiler));
}
}
}
@@ -111,6 +115,11 @@ namespace Grpc.Core.Internal
{
cq.Dispose();
}
+
+ for (int i = 0; i < threadProfilers.Count; i++)
+ {
+ threadProfilers[i].Dump(string.Format("grpc_trace_thread_{0}.txt", i));
+ }
});
}
@@ -137,12 +146,12 @@ namespace Grpc.Core.Internal
}
}
- private Thread CreateAndStartThread(int threadIndex)
+ private Thread CreateAndStartThread(int threadIndex, IProfiler optionalProfiler)
{
var cqIndex = threadIndex % completionQueues.Count;
var cq = completionQueues.ElementAt(cqIndex);
- var thread = new Thread(new ThreadStart(() => RunHandlerLoop(cq)));
+ var thread = new Thread(new ThreadStart(() => RunHandlerLoop(cq, optionalProfiler)));
thread.IsBackground = true;
thread.Name = string.Format("grpc {0} (cq {1})", threadIndex, cqIndex);
thread.Start();
@@ -153,8 +162,13 @@ namespace Grpc.Core.Internal
/// <summary>
/// Body of the polling thread.
/// </summary>
- private void RunHandlerLoop(CompletionQueueSafeHandle cq)
+ private void RunHandlerLoop(CompletionQueueSafeHandle cq, IProfiler optionalProfiler)
{
+ if (optionalProfiler != null)
+ {
+ Profilers.SetForCurrentThread(optionalProfiler);
+ }
+
CompletionQueueEvent ev;
do
{
diff --git a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
index dc9f62fdab..05dda5b148 100644
--- a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs
@@ -48,22 +48,19 @@ namespace Grpc.Core.Internal
public static MetadataArraySafeHandle Create(Metadata metadata)
{
- using (Profilers.ForCurrentThread().NewScope("MetadataArraySafeHandle.Create"))
+ if (metadata.Count == 0)
{
- if (metadata.Count == 0)
- {
- return new MetadataArraySafeHandle();
- }
+ return new MetadataArraySafeHandle();
+ }
- // TODO(jtattermusch): we might wanna check that the metadata is readonly
- var metadataArray = Native.grpcsharp_metadata_array_create(new UIntPtr((ulong)metadata.Count));
- for (int i = 0; i < metadata.Count; i++)
- {
- var valueBytes = metadata[i].GetSerializedValueUnsafe();
- Native.grpcsharp_metadata_array_add(metadataArray, metadata[i].Key, valueBytes, new UIntPtr((ulong)valueBytes.Length));
- }
- return metadataArray;
+ // TODO(jtattermusch): we might wanna check that the metadata is readonly
+ var metadataArray = Native.grpcsharp_metadata_array_create(new UIntPtr((ulong)metadata.Count));
+ for (int i = 0; i < metadata.Count; i++)
+ {
+ var valueBytes = metadata[i].GetSerializedValueUnsafe();
+ Native.grpcsharp_metadata_array_add(metadataArray, metadata[i].Key, valueBytes, new UIntPtr((ulong)valueBytes.Length));
}
+ return metadataArray;
}
/// <summary>
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-02 01:21:30 +0000