How can I make sure a thread gets dibs after a certain Task

Question

I'm in a bit of a conundrum regarding multithreading. I'm currently working on a real-time service using SinglaR. The idea is that a connected user can request data from another. Below is a gist of what the request and response functions look like.

Consider the following code:

private readonly ConcurrentBag _sharedObejcts= new ConcurrentBag();

The request:

[...]

var sharedObject = new MyObject();
_sharedObejcts.Add(sharedObject);

ForwardRequestFireAndForget();

try
{
    await Task.Delay(30000, sharedObject.myCancellationToken);
}
catch
{
    return sharedObject.ResponseProperty;
}

_myConcurrentBag.TryTake(sharedObject);

[...]

The response:

[...]

var result = DoSomePossiblyVeryLengthyTaskHere();

var sharedObject = ConcurrentBag 
    .Where(x)
    .FirstOrDefault();

// The request has timed out so the object isn't there anymore.
if(sharedObject == null)
{
    return someResponse;
}

sharedObject.ResponseProperty = result;

// triggers the cancellation source
sharedObject.Cancel();

return someOtherResponse;

[...]

So basically a request is made to the server, forwarded to the other host and the function waits for cancellation or it times out.

The other hosts call the respond function, which adds the repsonseObject and triggers myCancellationToken.

I am however unsure whether this represents a race condition. In theory, could the responding thread retrieve the sharedObject while the other thread still sits on the finally block? This would mean, the request timed out already, the task just hasn't gotten around to removing the object from the bag, which means the data is inconsistent.

What would be some guaranteed ways to make sure that the first thing that gets called after the Task.Delay() call is the TryTake()call?

Answer 1

You don't want to have the producer cancel the consumer's wait. That's way too much conflation of responsibilities.

Instead, what you really want is for the producer to send an asynchronous signal. This is done via TaskCompletionSource<T>. The consumer can add the object with an incomplete TCS, and then the consumer can (asynchronously) wait for that TCS to complete (or timeout). Then the producer just gives its value to the TCS.

Something like this:

class MyObject
{
  public TaskCompletionSource<MyProperty> ResponseProperty { get; } = new TaskCompletionSource<MyProperty>();
}


// request (consumer):

var sharedObject = new MyObject();
_sharedObejcts.Add(sharedObject);

ForwardRequestFireAndForget();

var responseTask = sharedObject.ResponseProperty.Task;
if (await Task.WhenAny(Task.Delay(30000), responseTask) != responseTask)
  return null;

_myConcurrentBag.TryTake(sharedObject);
return await responseTask;


// response (producer):

var result = DoSomePossiblyVeryLengthyTaskHere();
var sharedObject = ConcurrentBag 
    .Where(x)
    .FirstOrDefault();

// The request has timed out so the object isn't there anymore.
if(sharedObject == null)
  return someResponse;

sharedObject.ResponseProperty.TrySetResult(result);
return someOtherResponse;

The code above can be cleaned up a bit; specifically, it's not a bad idea to have the producer have a "producer view" of the shared object, and the consumer have a "consumer view", with both interfaces implemented by the same type. But the code above should give you the general idea.