Design Pattern for multithreaded observers

Question

In a digital signal acquisition system, often data is pushed into an observer in the system by one thread.

example from Wikipedia/Observer_pattern:

foreach (IObserver observer in observers)
    observer.Update(message);

When e.g. a user action from e.g. a GUI-thread requires the data to stop flowing, you want to break the subject-observer connection, and even dispose of the observer alltogether.

One may argue: you should just stop the data source, and wait for a sentinel value to dispose of the connection. But that would incur more latency in the system.

Of course, if the data pumping thread has just asked for the address of the observer, it might find it's sending a message to a destroyed object.

Has someone created an 'official' Design Pattern countering this situation? Shouldn't they?

Answer 1

You could send a message to all observers informing them the data source is terminating and let the observers remove themselves from the list.

Skizz

In response to the comment, the implementation of the subject-observer pattern should allow for dynamic addition / removal of observers. In C#, the event system is a subject/observer pattern where observers are added using 'event += observer' and removed using 'event -= observer'.

Answer 2

If you want to have the data source to always be on the safe side of concurrency, you should have at least one pointer that is always safe for him to use. So the Observer object should have a lifetime that isn't ended before that of the data source.

This can be done by only adding Observers, but never removing them. You could have each observer not do the core implementation itself, but have it delegate this task to an ObserverImpl object. You lock access to this impl object. This is no big deal, it just means the GUI unsubscriber would be blocked for a little while in case the observer is busy using the ObserverImpl object. If GUI responsiveness would be an issue, you can use some kind of concurrent job-queue mechanism with an unsubscription job pushed onto it. ( like PostMessage in Windows )

When unsubscribing, you just substitute the core implementation for a dummy implementation. Again this operation should grab the lock. This would indeed introduce some waiting for the data source, but since it's just a [ lock - pointer swap - unlock ] you could say that this is fast enough for real-time applications.

If you want to avoid stacking Observer objects that just contain a dummy, you have to do some kind of bookkeeping, but this could boil down to something trivial like an object holding a pointer to the Observer object he needs from the list.

Optimization : If you also keep the implementations ( the real one + the dummy ) alive as long as the Observer itself, you can do this without an actual lock, and use something like InterlockedExchangePointer to swap the pointers. Worst case scenario : delegating call is going on while pointer is swapped --> no big deal all objects stay alive and delegating can continue. Next delegating call will be to new implementation object. ( Barring any new swaps of course )