1

I have the following operations:

  1. User submits event
  2. We store event in a queue
  3. Wait for events and store them in db for redundancy
  4. Wait for events and process them
  5. Remove events from queue and db

For performance reasons, op 3 runs on a different thread.

The queue is always accessed under a lock, since events are submitted by multiple threads.

What I am struggling with is how to ensure that events are stored in the db before I process and remove them, since I could theoretically end up with 1-2-4-5-3 instead of 1-2-3-4-5. Operations 4 and 5 run sequentially, so no worries there.

What I hoped to find was some Monitor implementation that would prioritize the wait queue. This way, we would enter the monitor as part of ops 3 and 4 and wait for new events. Once op 1 is executed, you would PulseAll and ops 3 and 4 would be first to take the monitor.

Unfortunately, I was unable to find such an implementation in C# and I was hoping I wouldn’t have to roll my own synchronization primitive.

I should also mention that I actually use two queues to store events, as I assumed a second queue just for the db would be simpler to handle than a single queue with extra synchronization.

Do you have any suggestions on what I could try? At the moment the only two solutions I see are rolling my own monitor (not impossible, but I’d rather not) or just taking quite a performance hit and waiting for the events to be written to the database.

4
  • You are correct -- when you PulseAll, all the threads marked as waiting become marked as ready; which of the ready threads get to take the monitor next is an implementation detail of the monitor. There's no guarantee that any thread of any particular priority will be the chosen one. For some interesting discussion on the quirks of the .NET monitor see the comments to ericlippert.com/2015/11/16/monitor-madness-part-one and ericlippert.com/2015/11/19/monitor-madness-part-two. Commented Mar 16 at 1:26
  • unclear on what's not linear about your flow? if step 3 is run in parallel for performance, in parallel with what?
    – Ewan
    Commented Mar 16 at 22:30
  • @Ewan You run Steps 1+2, 3 and 4 in parallel.
    – Cristi
    Commented Mar 17 at 9:44
  • @Xzenon that doesnt seem to match your description, 2 has to come after 1, 3 has to come after 2, 4 has to come after 2 and possibly 3? 5 has to come after 3
    – Ewan
    Commented Mar 17 at 9:53

2 Answers 2

2

Firstly, if step 4 must wait for step 3 to complete, your execution diagram is indeed fully linearized - there is no overlapped execution allowed. If this is the case, continuation passing style should be used for simplicity and clarity.


My answer below assumes that steps 3 and 4 are allowed to execute independently and potentially simultaneously.

Here is my assumption described in terms of dependency:

  • step 1 --> step 2
  • step 2 --> step 3
  • step 2 --> step 4
  • step 3 AND step 4 --> step 5

My solution is to create a wrapper for each event, consisting of:

  1. A reference to the event
  2. A flag used to indicate the completion of the persistence step (your step 3)
  3. Another flag used to indicate the completion of the processing step (your step 4)

The requirements for the two flags are that:

  • Each flag can only transition from unset to set.
  • The flag needs to be accessible from anywhere, i.e. set from any thread
  • There must be no consistency issues between the setting of that flag and the reading of the same (from other threads).
  • When both flags are set, some cleanup action (your step 5) needs to be kicked off.

Data types suitable for implementing these flags:

  • An atomic integer
  • TaskCompletionSource

TaskCompletionSource has the benefit of working nicely with Task.WhenAll:

TaskCompletionSource tcsStep3;
TaskCompletionSource tcsStep4;
// ...
TaskCompletionSource[] tcs34 = { tcsStep3, tcsStep4 };
Tasks.WhenAll(tcs34).ContinueWith( _ => eventWrapper.DoCleanup() )
2

I'm going to echo @rwongs answer, it doesn't seem like you have any parallel processing in your description of the events.

Assuming that you have a web API end point with a controller for the initial user event, and a single queue worker for the rest of the processing, wouldn't it look like this?

public async Task AddEvent(Event e) //step 1
{
   await SendToQueue(e) //step 2
}


public async Task HandleEvent(Event e) //in queue worker
{
    await SaveToDb(e); //step 3
    await ProcessEvent(e) //Step 4 (dependency on the event being in the db)
    await DeleteFromDb(e) //step 5
}

If step 4 doesn't need the database entry, and assuming that its network transfer time that makes it slow, then..

public async Task HandleEvent(Event e) //in queue worker
{
    t1 = SaveToDb(e); //step 3
    t2 =  ProcessEvent(e) //Step 4 (**NO** dependency on the event being in the db)
    await Task.WhenAll({t1, t2}) //wait for 3 and 4 to both complete
    await DeleteFromDb(e) //step 5
}

If you have two queues, because both step 3 and 4 are cpu bound, and no dependency...

public async Task HandleEventDB(Event e) //in queue worker 1
{
    await SaveToDb(e); //step 3
}

public async Task HandleEventProcess(Event e) //in queue worker 2
{
    await ProcessEvent(e) //step 4
    addToQueue(Step4CompletedEvent);
}

public async Task HandleEventProcess(Step4CompletedEvent e) //in queue worker 3
{
   if(CheckEventIsOnDb())
   {
      await DeleteFromDb(e) //step 5
   }
   else
   {
      //put back on queue;
   }
}

If there is no dependency of 4 on 3 then, adding to the db and then immediately removing it doesn't make any sense?

If there IS a dependency then you don't need two queues and can just do all the steps after 2 in one queue worker?

Finally, this talk of thread communication and queues doesn't make sense to me in terms of performance. If you are using queues its to process work on different machines or at least processes, and you don't have thread communication.

If you have CPU bound work in step 4, It doesn't seem like saving to the DB in a different thread is going to save you any time? especially if you have to sync up afterwards. Your bottleneck will be on how many Step4s are running at the same time.

The parallelization should be done within step 4, or by running more than one queue processor on the same box, or having some maxConcurrentListeners setting in your worker so it processes X events at concurrently

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.