support adding profilers to C# qps clients

pull/6643/head
Jan Tattermusch 9 years ago
parent d0f5f72035
commit 09c5e056f9
  1. 55
      src/csharp/Grpc.IntegrationTesting/ClientRunners.cs

@ -32,6 +32,7 @@
#endregion #endregion
using System; using System;
using System.Collections.Concurrent;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.IO; using System.IO;
@ -41,7 +42,9 @@ using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Google.Protobuf; using Google.Protobuf;
using Grpc.Core; using Grpc.Core;
using Grpc.Core.Internal;
using Grpc.Core.Logging; using Grpc.Core.Logging;
using Grpc.Core.Profiling;
using Grpc.Core.Utils; using Grpc.Core.Utils;
using NUnit.Framework; using NUnit.Framework;
using Grpc.Testing; using Grpc.Testing;
@ -55,6 +58,15 @@ namespace Grpc.IntegrationTesting
{ {
static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<ClientRunners>(); static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<ClientRunners>();
// Profilers to use for clients.
static readonly BlockingCollection<BasicProfiler> profilers = new BlockingCollection<BasicProfiler>();
internal static void AddProfiler(BasicProfiler profiler)
{
GrpcPreconditions.CheckNotNull(profiler);
profilers.Add(profiler);
}
/// <summary> /// <summary>
/// Creates a started client runner. /// Creates a started client runner.
/// </summary> /// </summary>
@ -83,7 +95,8 @@ namespace Grpc.IntegrationTesting
config.OutstandingRpcsPerChannel, config.OutstandingRpcsPerChannel,
config.LoadParams, config.LoadParams,
config.PayloadConfig, config.PayloadConfig,
config.HistogramParams); config.HistogramParams,
() => GetNextProfiler());
} }
private static List<Channel> CreateChannels(int clientChannels, IEnumerable<string> serverTargets, SecurityParams securityParams) private static List<Channel> CreateChannels(int clientChannels, IEnumerable<string> serverTargets, SecurityParams securityParams)
@ -110,9 +123,16 @@ namespace Grpc.IntegrationTesting
} }
return result; return result;
} }
private static BasicProfiler GetNextProfiler()
{
BasicProfiler result = null;
profilers.TryTake(out result);
return result;
}
} }
public class ClientRunnerImpl : IClientRunner internal class ClientRunnerImpl : IClientRunner
{ {
const double SecondsToNanos = 1e9; const double SecondsToNanos = 1e9;
@ -125,8 +145,9 @@ namespace Grpc.IntegrationTesting
readonly List<Task> runnerTasks; readonly List<Task> runnerTasks;
readonly CancellationTokenSource stoppedCts = new CancellationTokenSource(); readonly CancellationTokenSource stoppedCts = new CancellationTokenSource();
readonly WallClockStopwatch wallClockStopwatch = new WallClockStopwatch(); readonly WallClockStopwatch wallClockStopwatch = new WallClockStopwatch();
readonly AtomicCounter statsResetCount = new AtomicCounter();
public ClientRunnerImpl(List<Channel> channels, ClientType clientType, RpcType rpcType, int outstandingRpcsPerChannel, LoadParams loadParams, PayloadConfig payloadConfig, HistogramParams histogramParams) public ClientRunnerImpl(List<Channel> channels, ClientType clientType, RpcType rpcType, int outstandingRpcsPerChannel, LoadParams loadParams, PayloadConfig payloadConfig, HistogramParams histogramParams, Func<BasicProfiler> profilerFactory)
{ {
GrpcPreconditions.CheckArgument(outstandingRpcsPerChannel > 0, "outstandingRpcsPerChannel"); GrpcPreconditions.CheckArgument(outstandingRpcsPerChannel > 0, "outstandingRpcsPerChannel");
GrpcPreconditions.CheckNotNull(histogramParams, "histogramParams"); GrpcPreconditions.CheckNotNull(histogramParams, "histogramParams");
@ -142,7 +163,8 @@ namespace Grpc.IntegrationTesting
for (int i = 0; i < outstandingRpcsPerChannel; i++) for (int i = 0; i < outstandingRpcsPerChannel; i++)
{ {
var timer = CreateTimer(loadParams, 1.0 / this.channels.Count / outstandingRpcsPerChannel); var timer = CreateTimer(loadParams, 1.0 / this.channels.Count / outstandingRpcsPerChannel);
this.runnerTasks.Add(RunClientAsync(channel, timer)); var optionalProfiler = profilerFactory();
this.runnerTasks.Add(RunClientAsync(channel, timer, optionalProfiler));
} }
} }
} }
@ -152,6 +174,11 @@ namespace Grpc.IntegrationTesting
var histogramData = histogram.GetSnapshot(reset); var histogramData = histogram.GetSnapshot(reset);
var secondsElapsed = wallClockStopwatch.GetElapsedSnapshot(reset).TotalSeconds; var secondsElapsed = wallClockStopwatch.GetElapsedSnapshot(reset).TotalSeconds;
if (reset)
{
statsResetCount.Increment();
}
// TODO: populate user time and system time // TODO: populate user time and system time
return new ClientStats return new ClientStats
{ {
@ -175,14 +202,28 @@ namespace Grpc.IntegrationTesting
} }
} }
private void RunUnary(Channel channel, IInterarrivalTimer timer) private void RunUnary(Channel channel, IInterarrivalTimer timer, BasicProfiler optionalProfiler)
{ {
if (optionalProfiler != null)
{
Profilers.SetForCurrentThread(optionalProfiler);
}
bool profilerReset = false;
var client = BenchmarkService.NewClient(channel); var client = BenchmarkService.NewClient(channel);
var request = CreateSimpleRequest(); var request = CreateSimpleRequest();
var stopwatch = new Stopwatch(); var stopwatch = new Stopwatch();
while (!stoppedCts.Token.IsCancellationRequested) while (!stoppedCts.Token.IsCancellationRequested)
{ {
// after the first stats reset, also reset the profiler.
if (optionalProfiler != null && !profilerReset && statsResetCount.Count > 0)
{
optionalProfiler.Reset();
profilerReset = true;
}
stopwatch.Restart(); stopwatch.Restart();
client.UnaryCall(request); client.UnaryCall(request);
stopwatch.Stop(); stopwatch.Stop();
@ -268,7 +309,7 @@ namespace Grpc.IntegrationTesting
} }
} }
private Task RunClientAsync(Channel channel, IInterarrivalTimer timer) private Task RunClientAsync(Channel channel, IInterarrivalTimer timer, BasicProfiler optionalProfiler)
{ {
if (payloadConfig.PayloadCase == PayloadConfig.PayloadOneofCase.BytebufParams) if (payloadConfig.PayloadCase == PayloadConfig.PayloadOneofCase.BytebufParams)
{ {
@ -282,7 +323,7 @@ namespace Grpc.IntegrationTesting
{ {
GrpcPreconditions.CheckArgument(rpcType == RpcType.UNARY, "Sync client can only be used for Unary calls in C#"); GrpcPreconditions.CheckArgument(rpcType == RpcType.UNARY, "Sync client can only be used for Unary calls in C#");
// create a dedicated thread for the synchronous client // create a dedicated thread for the synchronous client
return Task.Factory.StartNew(() => RunUnary(channel, timer), TaskCreationOptions.LongRunning); return Task.Factory.StartNew(() => RunUnary(channel, timer, optionalProfiler), TaskCreationOptions.LongRunning);
} }
else if (clientType == ClientType.ASYNC_CLIENT) else if (clientType == ClientType.ASYNC_CLIENT)
{ {

Loading…
Cancel
Save