1
\$\begingroup\$

This is a request-response model over sockets/websockets (like HTTP) where you technically match request id to response id and return the response. In addition to that, there is timeout in case that message is never matched. It does its job but the implementation is not disposing the CancellationTokenSource and TaskCompletionSource. It may even be done in a better way, so here is why I want a code review.

Example usage

var requestManager = new RequestManager();
var input = JsonSerializer.Serialize(new Test { Message = "Hello World" });

var pending = requestManager.AddPendingRequest(x =>
{
    var text = x.Deserialize<Test>();
    return text?.Message == "Hello World";
}, TimeSpan.FromSeconds(5));

await Task.Delay(4000);

var jsonElement = JsonSerializer.Deserialize<JsonElement>(input);

if (requestManager.TryMatchRequest(jsonElement))
{
    Console.WriteLine("Found");
}

public class Test
{
    public string Message { get; set; }
}

Code to review

public class PendingRequest
{
    private readonly CancellationTokenSource _cts;

    public PendingRequest(Func<JsonElement, bool> handler, TimeSpan timeout)
    {
        Handler = handler;
        Event = new AsyncResetEvent(false, false);
        Timeout = timeout;

        _cts = new CancellationTokenSource(timeout);
        _cts.Token.Register(Fail, false);
    }

    public Func<JsonElement, bool> Handler { get; }
    public JsonElement? Result { get; private set; }
    public bool Completed { get; private set; }
    public AsyncResetEvent Event { get; }
    public TimeSpan Timeout { get; }

    public bool CheckData(JsonElement data)
    {
        if (Handler(data))
        {
            Result = data;
            Completed = true;
            Event.Set();
            return true;
        }

        return false;
    }

    public void Fail()
    {
        Completed = true;
        Event.Set();
    }
}

public class RequestManager
{
    private readonly IList<PendingRequest> _pendingRequests = new List<PendingRequest>();

    public PendingRequest AddPendingRequest(Func<JsonElement, bool> handler, TimeSpan timeout)
    {
        var pending = new PendingRequest(handler, timeout);

        lock (_pendingRequests)
        {
            _pendingRequests.Add(pending);
        }

        return pending;
    }

    public void FailAllPendingRequests()
    {
        lock (_pendingRequests)
        {
            foreach (var pendingRequest in _pendingRequests.ToList())
            {
                pendingRequest.Fail();
                _pendingRequests.Remove(pendingRequest);
            }
        }
    }

    public IList<PendingRequest> GetAllPendingRequests()
    {
        lock (_pendingRequests)
        {
            return _pendingRequests;
        }
    }

    public bool TryMatchRequest(JsonElement tokenData)
    {
        lock (_pendingRequests)
        {
            foreach (var request in _pendingRequests)
            {
                if (!request.CheckData(tokenData))
                {
                    continue;
                }

                _pendingRequests.Remove(request);
                return true;
            }
        }

        return false;
    }

    public void RemoveTimedOutRequests()
    {
        PendingRequest[] requests;
        lock (_pendingRequests)
        {
            requests = _pendingRequests.ToArray();
        }

        foreach (var request in requests.Where(r => r.Completed))
        {
            lock (_pendingRequests)
            {
                _pendingRequests.Remove(request);
            }
        }
    }
}

/// <summary>
///     Async auto reset based on Stephen Toub`s implementation
///     https://devblogs.microsoft.com/pfxteam/building-async-coordination-primitives-part-2-asyncautoresetevent/
/// </summary>
public sealed class AsyncResetEvent : IDisposable
{
    private static readonly Task<bool> Completed = Task.FromResult(true);
    private readonly Queue<TaskCompletionSource<bool>> _waits = new();
    private readonly bool _reset;
    private bool _signaled;

    /// <summary>
    ///     New AsyncResetEvent
    /// </summary>
    /// <param name="initialState"></param>
    /// <param name="reset"></param>
    public AsyncResetEvent(bool initialState = false, bool reset = true)
    {
        _signaled = initialState;
        _reset = reset;
    }

    /// <summary>
    ///     Dispose
    /// </summary>
    public void Dispose()
    {
        _waits.Clear();
    }

    /// <summary>
    ///     Wait for the AutoResetEvent to be set
    /// </summary>
    /// <returns></returns>
    public Task<bool> WaitAsync(TimeSpan? timeout = null)
    {
        lock (_waits)
        {
            if (_signaled)
            {
                if (_reset)
                {
                    _signaled = false;
                }

                return Completed;
            }

            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            if (timeout != null)
            {
                var cancellationSource = new CancellationTokenSource(timeout.Value);
                cancellationSource.Token.Register(() =>
                {
                    tcs.TrySetResult(false);
                }, false);
            }

            _waits.Enqueue(tcs);
            return tcs.Task;
        }
    }

    /// <summary>
    ///     Signal a waiter
    /// </summary>
    public void Set()
    {
        lock (_waits)
        {
            if (!_reset)
            {
                // Act as ManualResetEvent. Once set keep it signaled and signal everyone who is waiting
                _signaled = true;
                while (_waits.Count > 0)
                {
                    var toRelease = _waits.Dequeue();
                    toRelease.TrySetResult(true);
                }
            }
            else
            {
                // Act as AutoResetEvent. When set signal 1 waiter
                if (_waits.Count > 0)
                {
                    var toRelease = _waits.Dequeue();
                    toRelease.TrySetResult(true);
                }
                else if (!_signaled)
                {
                    _signaled = true;
                }
            }
        }
    }
}
```
\$\endgroup\$
4
  • 1
    \$\begingroup\$ I think no one is brave enough here to review Stephen Toub's code :P That's why I would suggest to mention it in the question but remove his work from the shared code fragment. \$\endgroup\$ Feb 13, 2023 at 8:13
  • \$\begingroup\$ True^ It's a bit modified tho devblogs.microsoft.com/pfxteam/… (the part with the CancellationTokenSource) \$\endgroup\$
    – nop
    Feb 13, 2023 at 8:23
  • 1
    \$\begingroup\$ Then IMHO it would make sense to have a separate review for that... But that's my opinion \$\endgroup\$ Feb 13, 2023 at 8:30
  • \$\begingroup\$ Tbh, I don't mind if anything else is used instead, e.g. Nito.AsyncEx. \$\endgroup\$
    – nop
    Feb 13, 2023 at 8:35

1 Answer 1

1
\$\begingroup\$

PendingRequest

Public interface

I would suggest to try to minimize the public surface of this class. I think only Completed and Result should be exposed as properties, others could remain as private fields.

private readonly CancellationTokenSource _cancellationSignal;
private readonly Func<JsonElement, bool> _handler;
private readonly AsyncResetEvent _completionSignal = new(false, false);
  
public JsonElement? Result { get; private set; }
public bool Completed { get; private set; }

As in the above example, I would suggest to use better naming than _cts and Event.

Guard expression vs Early exit

In the CheckData it might make sense to use an early exit to streamline the code

if (!_handler(data))
    return false;
        
Result = data;
Completed = true;
_completionSignal.Set();
return true;

Code duplication

These two lines are executed both inside the CheckData and in the Fail

Completed = true;
_completionSignal.Set();

Depending on the use case I might rename the Fail to CompleteImmediately/CompleteNow and use it inside the CheckData OR extract that code and use it multiple places


For the sake of completeness here is the entire class

public class PendingRequest
{
    private readonly CancellationTokenSource _cancellationSignal;
    private readonly Func<JsonElement, bool> _handler;
    private readonly AsyncResetEvent _completionSignal = new(false, false);
  
    public JsonElement? Result { get; private set; }
    public bool Completed { get; private set; }

    public PendingRequest(Func<JsonElement, bool> handler, TimeSpan timeout)
    {
        _handler = handler;
        _cancellationSignal = new CancellationTokenSource(timeout);
        _cancellationSignal.Token.Register(CompleteNow, false);
    }

    public bool CheckData(JsonElement data)
    {
        if (!_handler(data))
            return false;
        
        Result = data;
        CompleteNow();
        return true;
    }

    public void Fail() => CompleteNow();

    private void CompleteNow()
    {
        Completed = true;
        _completionSignal.Set();
    }
}

RequestManager

Method naming

IMHO adding everywhere Request or PendingRequest as a method suffix is unnecessary. From the class name and from the methods' parameters it should clear what we are talking about.

Thread-safe collection vs lock

If possible I would suggest to use one of the concurrent collections because they are thread-safe for sure and they might utilize lock-free algorithms.

Because you need to add and remove items from the collection I would suggest to use ConcurrentDictionary. (ConcurrentBag might be tempting at the first glance, but it does not really support arbitrary item removal.)

private ConcurrentDictionary<Guid, PendingRequest> _requests = new();

I've removed readonly, because we will set a new value for it in multiple methods.

Add

public PendingRequest Add(Func<JsonElement, bool> handler, TimeSpan timeout)
{
    var pending = new PendingRequest(handler, timeout);
    _requests.TryAdd(Guid.NewGuid(), pending);
    return pending;
}

The key does not matter here, we just need to make sure that it is unique.

FailAll / CompleteAllNow

public void FailAll()
{
    foreach (var (_ ,pendingRequest) in _requests)
        pendingRequest.Fail();
        
    _requests = new();
}

First we complete all requests right away then we empty the _requests collection.

GetAll

public IList<PendingRequest> GetAll()
    => _requests.Select(kv => kv.Value).ToList();

Because we use a Dictionary that's why we need to transform the KeyValuePair elements.

TryMatch

If my understanding is correct then you want to remove the first match (not all the matches)

public bool TryMatch(JsonElement tokenData)
{
    var request = _requests.FirstOrDefault(r => r.Value.CheckData(tokenData));
    if (request.Equals(default(KeyValuePair<Guid, int>)))
        return false;

    return _requests.TryRemove(request);
}

Here you can't use the == default or is default for the KeyValuePair struct.

RemoveTimedOut

public void RemoveTimedOut()
{
    var completed = _requests.Where(r => r.Value.Completed);
    _requests = new(_requests.Except(completed));
}

Here we create a new collection without those requests which are already completed.


For the sake of completeness here is the entire class

public class RequestManager
{
    private ConcurrentDictionary<Guid, PendingRequest> _requests = new();

    public PendingRequest Add(Func<JsonElement, bool> handler, TimeSpan timeout)
    {
        var pending = new PendingRequest(handler, timeout);
        _requests.TryAdd(Guid.NewGuid(), pending);
        return pending;
    }

    public void FailAll()
    {
        foreach (var (_ ,pendingRequest) in _requests)
            pendingRequest.Fail();
        
        _requests = new();
    }

    public IList<PendingRequest> GetAll()
        => _requests.Select(kv => kv.Value).ToList();

    public bool TryMatch(JsonElement tokenData)
    {
        var request = _requests.FirstOrDefault(r => r.Value.CheckData(tokenData));
        if (request.Equals(default(KeyValuePair<Guid, int>)))
            return false;

        return _requests.TryRemove(request);
    }

    public void RemoveTimedOut()
    {
        var completed = _requests.Where(r => r.Value.Completed);
        _requests = new(_requests.Except(completed));
    }
}
\$\endgroup\$
3
  • \$\begingroup\$ Thank you very much for the descriptive answer! Where do you think is alright to dispose the CancellationTokenSource instance? \$\endgroup\$
    – nop
    Feb 14, 2023 at 10:29
  • 1
    \$\begingroup\$ @nop I think inside the CompleteNow it would make sense for me. \$\endgroup\$ Feb 14, 2023 at 10:47
  • 1
    \$\begingroup\$ Thank you very much once again! ^^ \$\endgroup\$
    – nop
    Feb 14, 2023 at 11:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.