Independently constraining task parallelism at different levels of a nested loop

Question

I would like to performance test a REST API using C#. The API offers a list of endpoints, and each endpoint accepts page and size parameters. My code needs to measure the following over the course of a test run:

• Multiple API endpoints
• Multiple attempts per endpoint, so that I can generate mix/max/avg values for each endpoint
• Multiple pages per endpoint attempt

That means I'll have a nest of three for loops.

The base case is that everything runs in serial - one request at a time. I want then to add parallelism to different loops, independently. I want to create different test cases that place bounds on the concurrency for each of the loops, in order to compare results and observe how performance degrades as (for example) the number of endpoints queried concurrently increases, or as the total number of queries increases.

I have written this test code:

public void MeasureResponseTimes(int endpointConcurrency, int measurementConcurrency, int pageConcurrency, int requestConcurrency, int measurementsPerEndpoint, int pageSize)
{
 var endpointSemaphore = new SemaphoreSlim(endpointConcurrency);
 var requestSemaphore = new SemaphoreSlim(requestConcurrency);
 // The number of pages in the response would be calculated/known at this point, but let's say each response is 10 pages long
 var responseSizeInPages = endpoints.Select(e => new KeyValuePair<string, int>(e.Path, 10)).ToDictionary(kvp => kvp.Key, kvp => kvp.Value);
 var taskFactory = new TaskFactory(new ThreadPerTaskScheduler());
 var timings = new ConcurrentBag<Tuple<string, int, int, long>>();
 var tasks = new ConcurrentBag<Task>();
 foreach (var endpoint in endpoints)
 {
 endpointSemaphore.Wait();
 var measurementSemaphore = new SemaphoreSlim(measurementConcurrency);
 for (var attempt = 0; attempt < measurementsPerEndpoint; attempt++)
 {
 var attempt1 = attempt;
 tasks.Add(taskFactory.StartNew(() =>
 {
 measurementSemaphore.Wait();
 var pageSemaphore = new SemaphoreSlim(pageConcurrency);
 for (var page = 1; page <= responseSizeInPages[endpoint.Path]; page++)
 {
 var attempt2 = attempt1;
 var page1 = page;
 tasks.Add(taskFactory.StartNew(() =>
 {
 pageSemaphore.Wait();
 requestSemaphore.Wait();
 MakeRequest(pageSize, endpoint, timings, attempt2, page1);
 requestSemaphore.Release();
 pageSemaphore.Release();
 }));
 }
 measurementSemaphore.Release();
 }));
 }
 endpointSemaphore.Release();
 }
 Task.WaitAll(tasks.ToArray());
 var attemptMeasurements = timings.GroupBy(t => new {t.Item1, t.Item2}, t => t.Item4); // group by endpoint/attempt
 var endpointMeasurements = attemptMeasurements.GroupBy(a => a.Key.Item1, a => a.Sum()); // sum timing of all pages
 foreach (var endpointMeasurement in endpointMeasurements)
 {
 Console.WriteLine($"{endpointConcurrency},{measurementConcurrency},{pageConcurrency},{requestConcurrency},{measurementsPerEndpoint},{pageSize}," +
 $"{endpointMeasurement.Key},{endpointMeasurement.Average()},{endpointMeasurement.Max()},{endpointMeasurement.Min()}");
 }
}
private static void MakeRequest(int pageSize, SwaggerEndpoint endpoint, ConcurrentBag<Tuple<string, int, int, long>> timings, int attempt, int page)
{
 var client = new RestClient(IdwApiUrl);
 var request = new RestRequest($"{endpoint.Path}?page={page}&size={pageSize}");
 IRestResponse response = null;
 var timing = PerformanceTimer.Measure(() => response = client.Execute(request));
 if ((int) response.StatusCode == 200)
 {
 timings.Add(Tuple.Create(endpoint.Path, attempt, page, timing));
 }
}

I make the request on the inside of the nested loop, and print the timings to CSV so I can draw the results on a chart.

Using NUnit test cases, I plan to vary these concurrency parameters:

• endpointSemaphore: how many endpoints to query at once
• measurementSemaphore: how many parallel attempts on the same endpoint
• pageSemaphore: how many pages of a request to query at once
• requestSemaphore: how many simultaneous requests, across all endpoints/measurements/pages

This works, but I'm interested to know if anyone can see an easier way to accomplish the general problem of wanting to independently constrain parallelism at different levels of a nested loop. I don't really like my use of TPL inside the loops, but haven't found a better way (yet). Thanks in advance.

---

EDIT: I like this better for the nested loop:

var taskScheduler = new ThreadPerTaskScheduler();
Parallel.ForEach(substitutedEndpoints, new ParallelOptions {MaxDegreeOfParallelism = endpointConcurrency, TaskScheduler = taskScheduler },
 endpoint =>
 {
 Parallel.For(0, measurementsPerEndpoint, new ParallelOptions {MaxDegreeOfParallelism = measurementConcurrency, TaskScheduler = taskScheduler},
 attempt =>
 {
 Parallel.For(1, responseSizeInPages[endpoint.Path]+1, new ParallelOptions {MaxDegreeOfParallelism = pageConcurrency, TaskScheduler = taskScheduler }, page =>
 {
 requestSemaphore.Wait();
 MakeRequest(pageSize, endpoint, timings, attempt, page);
 requestSemaphore.Release();
 });
 });
 });

• just one semaphore, and nested Parallel.For.

Answer 1

I see a public method with a long parameter list inside a class with only one high level responsibility and I always feel it's too much procedural. Change those parameters to class members:

public sealed class RestEndpointBenchmark
{
 public int EndpointConcurrency
 {
 get => _endpointConcurrency;
 set
 {
 if (value <= 0)
 throw new ArgumentOutOfRangeException();
 if (_isTestInProgress)
 throw new InvalidOperationException("Cannot change...");
 _endpointConcurrency = value;
 }
 }
 // All the other relevant properties
 public void MeasureResponseTimes()
 {
 // ...
 }
 private int _endpointConcurrency = 1;
}

I think you have too many nested loops, they make code harder to read and to understand. You may move them to separate methods:

public void MeasureResponseTimes()
{
 EnsureBenchmarkIsNotInProgress();
 _isTestInProgress = true;
 try
 {
 InitializeBenchmark();
 Task.WaitAll(CreateBenchmarkTasks());
 }
 finally
 {
 _isTestInProgress = false;
 }
}

Now that we're fully using a class we can use instance members instead of local variables. Repeat this recursively until you will have simple methods with a single simple responsibility.

We will then need to fix few more things:

• SemaphoreSlim implements IDisposable then it must be properly disposed. Add IDisposable to your class and add the relevant code (with the usual pattern). When it's a local variable use it with using statement.
• MakeRequest() still has too many parameters. Let's introduce a Request class, if in one month you will need to add a new option it will be pretty smooth.
• I don't know RestClient class and its RestRequest but I guess they implement IDisposable, use them with using statement. Just ignore this if it doesn't apply.
• I didn't test your code but I feel that it should NOT have a list of tasks which is populated from within inner tasks...use of async and await should simplify your logic (I imagine a list of tasks only for outer loop over Endpoints) and inner tasks are simply awaited. This will also make the list a simple List<Task> instead of ConcurrentBag<Task>.
• Overall design seems too complicate (I strongly believe that breaking down to small functions will make improvements slightly more obvious). There are really too many semaphores where I'd would like to see only few Parallel.For() or Parallel.ForEach() (given that you're already using ThreadPerTaskScheduler and MaxDegreeOfParallelism).

Now I'd want to introduce a separate class to format results. Now you need CSV but it may be pretty easy to move to XLSX directly or maybe to prepare a nicely formatted HTML output. Move that responsibility to a separate method. Even better (is it a command line utility? do you want to make it configurable?) let's introduce a completely separate abstract class BenchmarkResultFormatter and a new BenchmarkResult which contains data you generated with those GroupBy().

Caller will then be free to:

var benchmark = new RestEndpointBenchmark
{
 // Configure the benchmark
 EndpointConcurrency = 4,
 MeasurementConcurrency = 4,
 PageConcurrency = 2,
};
benchmark.MeasureResponseTimes();
var formatter = new CsvBenchmarkResultFormatter(benchmark);
Console.WriteLine(formatter);

Now that output is to console is just a detail and no one else but the outer caller knows this. Also the format is just a detail and it may be even configured.

One note about CSV format: if you want to use it with Microsoft Excel do not hard-code the , because it uses the list separator from the current locale (System.Globalization.CultureInfo.CurrentCulture.TextInfo.ListSeparator).

var formatter = new CsvBenchmarkResultFormatter(benchmark)
{
 Separator = CultureInfo.CurrentCulture.TextInfo.ListSeparator
};
Console.WriteLine(formatter);

• Chris B: I've been through this and implemented almost all of your suggestions. This was originally implemented as an NUnit test (as it just gave me an easy visual entrypoint for multiple test cases), and I think there's perhaps a temptation (at least with me) to "think procedurally" when writing tests. RestClient and RestRequest are provided by the RestSharp library and interestingly do not implement IDisposable. Thank you so much for taking the time to go through this and provide such outstanding constructive feedback, I appreciated it.
• Adriano Repetti: @Chris I understand that feeling when writing tests, it's the temptation of "I won't reuse this elsewhere then..."