0

I'll be using C# for my example since this is what I am using, but the problem is sort of generic.

My app code takes a collection of Foos (externally populated and edited) to do some processing, and needs to process it again every time the collection is edited (an element is added, removed or changed).

The simplest way to write this would be:

interface IBar //< wraps the external mechanism that populates the collection
{
  Foo[] Foos { get; }
  event EventHandler<FooEdit> FoosChanged;
}

public void MyAppLogic(IBar myBar)
{
  ProcessAllInitialFoos(myBar.Foos);
  myBar.FoosChanged += (FooEdit edit) => { ProcessFooEdit(edit); }
}

public void ProcessAllInitialFoos(Foo[] foos)
{
  foreach(Foo foo in foos) { ProcessFooEdit(new FooEdit(FooAdded, foo)); }
}

The reason why this doesn't work in my case is that edits to the collection happen in a different thread ("update thread") from the one that runs MyAppLogic ("app thread"). So if the IBar implementation simply invokes FoosChanged on the update thread, the code above is incorrect - it will miss updates that happen between the call to ProcessAllInitialFoos and the subscription.

Trying to solve the problem

This variant:

myBar.FoosChanged += (FooEdit edit) => { ProcessFooEdit(edit); }
ProcessAllInitialFoos(myBar.Foos);

solves that problem, but has its own: now ProcessFooEdit can now be called on different threads concurrently and needs synchronization.

This can be mitigated by using a dirty flag and postponing the actual processing:

private bool areFoosDirty = false;

public void MyAppLogic(IBar myBar)
{
  myBar.FoosChanged += (FooEdit edit) => { areFoosDirty = true; }
  areFoosDirty = true;
}

public void MyAppUpdateLoop(IBar myBar)
{
  //...
  if (areFoosDirty)
  {
    ProcessAllFoosFromScratch(myBar.Foos);
    areFoosDirty = false;
  }
}

This doesn't need locking - only concurrent access to a bool - but is only convenient if the app already has an update loop running somewhere.

Alternative patterns

Here a couple of ideas I had (which don't satisfy me completely).

Start receiving edit notification explicitly

The idea is delaying all edits notifications until the initial collection data is processed.

interface IBar
{
  Foo[] InitialFoos { get; }
  event EventHandler<FooEdit> FoosChanged;
  void StartInvokingFoosChangedEvent();
}
public void MyAppLogic(IBar myBar)
{
  ProcessAllInitialFoos(myBar.InitialFoos);
  // Edits to Foos between here and StartInvokingFoosChangedEvent are enqueued internally.
  myBar.FoosChanged += (FooEdit edit) => { ProcessFooEdit(edit); }
  void StartInvokingFoosChangedEvent();
  // Now all enqueued edits are dispatched on the update thread before any following edit
}

This avoids data loss and concurrency, but is clunky and only fits the case where IBar has a single subscriber (it's not clear what should happen if a handler is added to FooChanged after StartInvokingFoosChangedEvent is called).

Dispatch initial Foos when an event handler is added

interface IBar
{
  // When a new handler is added, it is invoked once with "FooAdded" for every Foo 
  // in the collection at that moment.
  event EventHandler<FooEdit> FoosChanged;
}
public void MyAppLogic(IBar myBar)
{
  myBar.FoosChanged += (FooEdit edit) => { ProcessFooEdit(edit); }
}

This produces the neatest client code (all the processing logic can go in the event handler) but it's quite surprising (a "Changed" event is called on existing items that have not technically changed) and difficult to implement (in order to avoid locking, the implementation of IBar needs to dispatch all the handler invocations on the update thread, so it must keep track of which handlers have been recently added and which edits they need to handle).

Is there some better alternative I haven't thought to? Am I missing some potential pitfalls here?

0

This might be a place to consider splitting up the code into separate artifacts/executables for producer and consumer in order to avoid the complexity of handling all the threading in a single executable.

Putting that aside for a moment and assuming we absolutely must do this in a single multithreaded process, I would probably shift the focus away from using event handlers as this will almost certainly cause more headaches than it gains, and towards using concurrent data structures (a queue, most likely).

Assume you have something like the following in your code base:

public class FooBroker
{
   // See: https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentqueue-1?view=netcore-3.1
   public static ConcurrentQueue<FooEdit> EditQueue {get; protected set; }
}

Wherever you are discovering the edits to Foo you could simply queue up the edit message:

FooEdit fooEdit = new FooEdit(); // Fill out data as needed
FooBroker.EditQueue.Enqueue(fooEdit);

At this point the producer is done. You can then have multiple consumer threads that are basically doing something like this:

while(true)
{
      FooEdit result;
      if (!FooBroker.TryDequeue(out result))
      {
         // Do processing here on result
      }
      else
      {
         // Unable to get message, sleep for a little bit so you don't run too hot.
      }
}

At this point you've avoided all the threading pitfalls of event handlers without hand-written locking and allowed yourself to scale up these two components independently.

As a nice side benefit if you later decide to split these into multiple executables, the code is based on a distinct producer-consumer model that will make all of this simpler.

7
  • Replacing the event with queues does make things clearer. The downside is that you need to explicitly poll the queues. Also two minor things. First, you need to use a Queue under a lock I think. On subscription, the implementation needs to fill the queue with FooAdded edits for the current elements while the update thread is possibly dispatching more edits, and the latter edits must be enqueued after the former. A ConcurrentQueue alone won't enforce that. Second, if you want multiple consumers you need to create a FooBroker and consumer thread for each of them. Oct 15 '20 at 10:15
  • Those are valid things to consider. You shouldn't need to lock the ConcurrentQueue, though. The docs say: "ConcurrentQueue<T> handles all synchronization internally. If two threads call TryDequeue at precisely the same moment, neither operation is blocked. When a conflict is detected between two threads, one thread has to try again to retrieve the next element, and the synchronization is handled internally." (docs.microsoft.com/en-us/dotnet/api/…) This would imply that you can leave the thread safety to it.
    – Michael
    Oct 15 '20 at 21:44
  • The best way for maintaining sequence of operations within a Foo would largely depend on how you are discovering the edits. If you can keep all of the updates for a Foo flowing through one producer thread, you can enforce that ordering fairly easily.
    – Michael
    Oct 15 '20 at 21:46
  • docs.microsoft.com/en-us/dotnet/standard/collections/… also gives some general information about the threadsafe collections. As a quick summary, the library functions will pretty much handle what you're describing.
    – Michael
    Oct 15 '20 at 21:47
  • TBC on subscription the collection already contains some elements. The app thread needs to process those elements before other edits. To do this, it needs to atomically get the existing elements and push a FooAdded edit for each of them in the queue. Only then the update thread should push more edits. You need a lock for this. If you leave both threads pushing their edits concurrently in the queue, they'll interleave and possibly be out of order (if an element is removed just after the subscription you might get the FooRemoved for that element in the queue before the FooAdded for it). Oct 16 '20 at 10:17

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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