Red Architecture - red help button for complex and complicated systems - part 3 (multithreading to help us)

The final part of the Red Architecture description will be dedicated to multithreading. For the sake of fairness, it is worth saying that the initial version of class v cannot be considered optimal, since there is nothing in it to solve one of the main problems that developers of real world applications inevitably come to. To fully understand the current article, you need to get acquainted with the concept of Red Architecture here .



Looking ahead, I’ll say that we can solve all the problems of multithreading without going beyond the limits of class v. Moreover, there will be much fewer changes than it might seem, and as a result, class v code with fully resolved multithreading problems will consist of a little more than 50 lines! Moreover, these 50 with a few lines will be more optimal than the version of the class v described in the first part . In this case, a specific code that solves the problems of thread synchronization will take only 20 lines!

In the course of the text, we will parse individual lines from the listing of the completed classes v and Tests, which are given at the end of this article.

Where can I apply Red Architecture?

I want to emphasize that the examples presented here, like the whole concept of Red Architecture, are offered for use on all possible languages ​​and platforms . C # / Xamarin and the .NET platform were chosen to demonstrate Red Architecture based on my personal preferences, nothing more.

Two variants of class v

We will have two variants of class v. The second option, identical in functionality and way of using the first, will be arranged somewhat more complicated. But it will be possible to use it not only in the “standard” C # .NET environment, but also in the PCL environment of Xamarin, which means for mobile development under three platforms at once: iOS, Android, Windows 10 Mobile. The fact is that thread safe collections are not available in the PCL environment of the Xamarin framework, so the version of class v for Xamarin / PCL will contain more code for synchronizing threads. This is what we will consider in this article, since a simplified version of class v (also at the end of this article) is of less value from the point of view of understanding multithreading problems and how to solve them.

A bit of optimization

First of all, we will get rid of the base class and make class v self-sufficient. We do not need the notification mechanism of the base class that we have used up to the present moment. The inherited mechanism does not allow solving multithreading problems in an optimal way. Therefore, now we ourselves will send events to handler functions:

static Dictionary> handlersMap = new Dictionary>(); 
// ...
foreach (var handlr in new List(handlersMap[key]))
    lock(handlr)
        try
        {
            handlr.Invoke(key, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, new List>(){ new KeyValuePair(key, o) })); 

In the Add () method in the foreach loop, we copy the elements from HashSet 'a to Listand the iteration is already over the sheet, not the hashset. We need to do this, because the value returned by the handlersMap [key] expression is a global variable accessible from public state mutating methods such as m () and h (), therefore, it is possible that the HashMap returned by the handlersMap [key] expression will be modified by another thread during iteration over it in the Add () method, and this will cause runtime execution, since until iteration over the collection inside the foreach has been completed, its (collection) modification is prohibited. That is why we "substitute" for iteration not a global variable, but a List into which elements of the global HashSet are copied.

But this protection is not enough. In expression

new List(handlersMap[key]) 

a copy operation is implicitly applied to the value (hashset) of handlersMap [key]. This will definitely cause problems if, between the start and end of the copy operation, some other thread tries to add or remove an item in the copied hashset. Therefore, we put the lock (Monitor.Enter (handlersMap [key])) on this hash just before the start of foreach

Monitor.Enter(handlersMap[key]);
foreach (var handlr in new List(handlersMap[key]))
{ 
// ...

and “release” (Monitor.Exit (handlersMap [key])) lock immediately after entering the foreach loop

            
foreach (var handlr in new List(handlersMap[key]))
{
    if (Monitor.IsEntered(handlersMap[key]))
    {
        Monitor.PulseAll(handlersMap[key]);
        Monitor.Exit(handlersMap[key]);
    }
    // ...

According to the rules of the Monitor object, the number of calls to Enter () should correspond to the number of calls to Exit (), so we have an if check (Monitor.IsEntered (handlersMap [key])) which ensures that if the lock was installed, we will exit only one times, at the start of the first iteration of the foreach loop. Immediately after the line Monitor.Exit (handlersMap [key]) the hash handlersMap [key] will again be available for use by other threads. Thus, we limit the hash lock to the minimum possible time, we can say that in this case the hash will be blocked literally for a moment.

Immediately after the foreach loop, we see a repetition of the lock release code.

// ...
if (Monitor.IsEntered(handlersMap[key]))
{
     Monitor.PulseAll(handlersMap[key]);
     Monitor.Exit(handlersMap[key]);
}
// ...

This code is necessary in case there is no iteration in foreach, which is possible when there will be no handlers for any of the keys in the corresponding hashset.

The following code requires extensive explanation:

 lock(handlr)
    try {
        // ...

The fact is that in the Red Architecture concept, the only objects created outside the class v and requiring thread synchronization are handler functions. If we couldn’t manage the code that the handlers call our functions, we would have to “fence” something like in each handler

void OnEvent(object sender, NotifyCollectionChangedEventArgs e)
{
    lock(OnEvent);
    // полезный код метода
    unlock(OnEvent);
}

Pay attention to the lock () unlock () lines between which the useful method code is located. If data inside the handler that is external to it are modified, then lock () and unlock () would need to be added. Because at the same time the flows entering this function will change the values ​​of external variables in a chaotic order.

But instead, we added just one line to the entire program - lock (handlr), and did it inside the v class without touching anything outside it! Now we can write any number of handler functions without thinking about their thread safety, since the implementation of the v class guarantees that only one thread can enter this particular handler, other threads will “stand” on lock (handlr) and wait for the completion of work in this the handler of the previous thread that entered it.

foreach, for (;;) and multithreading

In the Tests listing (at the end of the article), there is a foreachTest (string [] a) method that checks the operation of the for (;;) loop during the simultaneous entry into this method and, therefore, in the for (;;) loop of two streams. The following is a possible part of the output of this method:

// ...
~: string20
~: string21
~: string22
~: astring38
~: astring39
~: string23
~: string24
~: astring40
~: astring41
~: string25
~: astring42
~: string26
~: astring43
~: astring44
~: string27
~: astring45
~: string28
// ...

We see that in spite of the mixed output of the “string” and “astring” lines, the numerical suffix for each of the lines goes in order, i.e. to output each of the lines, the local variable i is taken true. This conclusion suggests that the simultaneous entry of two threads in for (;;) is safe. Probably, all the variables declared in the framework of the for (;;) construct, for example, the variable int i, are created on the stack of the stream that went into for (;;). That is why access to the variables created inside for (;;) does not need “manual” synchronization, since they are already available only to the thread in whose stack they were created. This is the case on C # and the .NET platform. In other languages, although it is unlikely, there may be a different behavior, so such a test will not be superfluous.

try ... catch is the norm, not an exception

try ... catch At first glance, this construction seems unnecessary, but it is important. It is intended to protect us from the situation when, at the time of the call to handlr.Invoke (), the object in which handlr was defined was destroyed. Destruction of an object can be done by another thread or garbage collector at any time between the lines

foreach (var handlr in new List(handlersMap[key]))

and

handlr.Invoke();

In the exception handling - catch block, we check if the handler refers to a null (deleted) object, we just delete it from the handler lists.

lock (handlr)
try
{
    handlr.Invoke(key, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, new List>(){ new KeyValuePair(key, o) }));
#if __tests__
    /* check modification of global collection of handlers for a key while iteration through its copy */
    handlersMap[key].Add((object sender, NotifyCollectionChangedEventArgs e) => { });
#endif
}
catch (Exception e)
{
	// because exception can be thrown inside handlr.Invoke(), but before handler was destroyied.
	if (ReferenceEquals(null,handlr) && e is NullReferenceException)
		// handler invalid, remove it
		m(handlr);
	else
		// exception in handler's body
		throw e;
}

Class v initialization

A static constructor is one of the hallmarks of C #. It cannot be called directly. It is called automatically only once, before creating the first object of this class. We use it to initialize handlersMap - for all keys from k we prepare for use empty HashSets intended for storing handler functions of each of the keys. In the absence of a static constructor in other languages, any method that initializes the object is suitable.

   
static v()
{
    foreach (k e in Enum.GetValues(typeof(k)))
    handlersMap[e] = new HashSet();
    new Tests().run();
}

What to do with thread unsafe collection?

The C # class HashSet does not provide synchronization when modifying from multiple threads (not thread safe), so we must synchronize the modification of this object, namely deleting and adding elements. In our case, for this it is enough to add one line lock (handlersMap [key]) immediately before the operation of deleting / adding an element in the methods m (), h () of class v. In this case, the object blocking the stream will be the HashMap object associated with this particular key. This will provide the ability to modify this particular hashset with only one thread.

Side Effects of Multithreading

It is worth mentioning some of the “side effects” of multithreading. In particular, the code of the handler functions should be prepared for the fact that in some cases it will be called after the "unsubscribe" of the handler function from receiving events. That is, after calling the m (key, handler) handler for some time (probably a fraction of a second) it can still be called. This is possible because at the time of calling handlersMap [key] .Remove (handler) in the m () method, this handler may already be copied by another thread in the foreach line (var handlr in new List (handlersMap [key])) , and will be called in the Add () method of class v after it is deleted in the m () method.

Simple rules to solve complex problems

In conclusion, I want to draw attention to the fact that, being diligent developers, we do not violate agreements on the use of locks. In particular, such agreements are listed on this page in the Remarks section of docs.microsoft.com/en-us/dotnet/csharp/language-reference/keywords/lock-statement . They are common to all languages, not just C #. The essence of these agreements is as follows:

• Do not use public types as an object for a lock.

We use 2 types of objects for locks and both are private. The first type is a HashSet object that is private to class v. The second type is an object of type function handler. Functions handlers are declared private in all objects that declare them and use them to receive events. In the case of Red Architecture, only the v class should call the handlers directly, and nothing else.

Listings

Below is the completed code for the v and Tests classes. In C #, you can use them directly by copying from here. Translating this code into other languages ​​will be a small and entertaining task for you.

Below is the code for the “universal” class v, which can also be used in mobile application projects based on the Xamarin / C # platform.

using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Threading;
namespace Common
{
    public enum k {OnMessageEdit, MessageEdit, MessageReply, Unused, MessageSendProgress, OnMessageSendProgress, OnIsTyping, IsTyping, MessageSend, JoinRoom, OnMessageReceived, OnlineStatus, OnUpdateUserOnlineStatus }
    public class v
    {
        static Dictionary> handlersMap = new Dictionary>();
        public static void h(k[] keys, NotifyCollectionChangedEventHandler handler)
        {
            foreach (var key in keys)
            lock(handlersMap[key])
            handlersMap[key].Add(handler);
        }
        public static void m(NotifyCollectionChangedEventHandler handler)
        {
            foreach (k key in Enum.GetValues(typeof(k)))
            lock(handlersMap[key])
            handlersMap[key].Remove(handler);
        }
        public static void Add(k key, object o)
        {
            Monitor.Enter(handlersMap[key]);
            foreach (var handlr in new List(handlersMap[key]))
            {
                if (Monitor.IsEntered(handlersMap[key]))
                {
                    Monitor.PulseAll(handlersMap[key]);
                    Monitor.Exit(handlersMap[key]);
                }
                lock (handlr)
                try
                {
                    handlr.Invoke(key, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, new List>(){ new KeyValuePair(key, o) }));
#if __tests__
                    /* check modification of global collection of handlers for a key while iteration through its copy */
                    handlersMap[key].Add((object sender, NotifyCollectionChangedEventArgs e) => { });
#endif
                }
                catch (Exception e)
                {
	               // because exception can be thrown inside handlr.Invoke(), but before handler was destroyied.
	               if (ReferenceEquals(null,handlr) && e is NullReferenceException)
		            // handler invalid, remove it
			    m(handlr);
		       else
			    // exception in handler's body
			    throw e;
                 }
            }
            if (Monitor.IsEntered(handlersMap[key]))
            {
                Monitor.PulseAll(handlersMap[key]);
                Monitor.Exit(handlersMap[key]);
            }
        }
        static v()
        {
            foreach (k e in Enum.GetValues(typeof(k)))
            handlersMap[e] = new HashSet();
            new Tests().run();
        }
    }
}

Below is the code for the “simplified” class v, which can be used on the “standard” C # .NET platform. Its only difference from the “universal” counterpart is the use of the ConcurrentBag collection instead of the HashMap, a type that provides out-of-the-box synchronization of flows when accessing yourself. Using ConcurrentBag instead of HashSet allowed to remove from the class v most of the code that synchronizes the threads.

using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Collections.Concurrent;
using System.Threading;
namespace Common
{
    public enum k { OnMessageEdit, MessageEdit, MessageReply, Unused, MessageSendProgress, OnMessageSendProgress, OnIsTyping, IsTyping, MessageSend, JoinRoom, OnMessageReceived, OnlineStatus, OnUpdateUserOnlineStatus }
    public class v
    {
        static Dictionary> handlersMap = new Dictionary>();
        public static void h(k[] keys, NotifyCollectionChangedEventHandler handler)
        {
            foreach (var key in keys)
            handlersMap[key].Add(handler);
        }
        public static void m(NotifyCollectionChangedEventHandler handler)
        {
            foreach (k key in Enum.GetValues(typeof(k)))
            handlersMap[key].Remove(handler);
        }
        public static void Add(k key, object o)
        {
            foreach (var handlr in new List(handlersMap[key]))
            {
                lock (handlr)
                try
                {
                    handlr.Invoke(key, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, new List>(){ new KeyValuePair(key, o) }));
#if __tests__
                    /* check modification of global collection of handlers for a key while iteration through its copy */
                    handlersMap[key].Add((object sender, NotifyCollectionChangedEventArgs e) => { });
#endif
                }
                catch (Exception e)
                {
	               // because exception can be thrown inside handlr.Invoke(), but before handler was destroyied.
	               if (ReferenceEquals(null,handlr) && e is NullReferenceException)
		            // handler invalid, remove it
			    m(handlr);
		       else
			    // exception in handler's body
			    throw e;
                }
            }
        }
        static v()
        {
            foreach (k e in Enum.GetValues(typeof(k)))
            handlersMap[e] = new ConcurrentBag();
            new Tests().run();
        }
    }
}

Below is the code for the Tests class, which tests the multi-threaded use of v, as well as handler functions. Pay attention to the comments. They have a lot of useful information about how the testing and testing code works.

using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Linq;
namespace ChatClient.Core.Common
{
    class DeadObject
    {
        void OnEvent(object sender, NotifyCollectionChangedEventArgs e)
        {
            var newItem = (KeyValuePair)e.NewItems[0];
            Debug.WriteLine(String.Format("~ OnEvent() of dead object: key: {0} value: {1}", newItem.Key.ToString(), newItem.Value));
        }
        public DeadObject()
        {
            v.h(new k[] { k.OnlineStatus }, OnEvent);
        }
        ~DeadObject()
        {
            // Accidentally we forgot to call v.m(OnEvent) here, and now v.handlersMap contains reference to "dead" handler
        }
    }
    public class Tests
    {
        void OnEvent(object sender, NotifyCollectionChangedEventArgs e)
        {
            var newItem = (KeyValuePair)e.NewItems[0];
            Debug.WriteLine(String.Format("~ OnEvent(): key: {0} value: {1}", newItem.Key.ToString(), newItem.Value));
            if (newItem.Key == k.Unused)
            {
                // v.Add(k.Unused, "stack overflow crash"); // reentrant call in current thread causes stack overflow crash. Deadlock doesn't happen, because lock mechanism allows reentrancy for a thread that already has a lock on a particular object
                // Task.Run(() => v.Add(k.Unused, "deadlock")); // the same call in a separate thread don't overflow, but causes infinite recursive loop
            }
        }
        void OnEvent2(object sender, NotifyCollectionChangedEventArgs e)
        {
            var newItem = (KeyValuePair)e.NewItems[0];
            Debug.WriteLine(String.Format("~ OnEvent2(): key: {0} value: {1}", newItem.Key.ToString(), newItem.Value));
        }
        void foreachTest(string[] a)
        {
            for (int i = 0; i < a.Length; i++)
            {
                Debug.WriteLine(String.Format("~ : {0}{1}", a[i], i));
            }
        }
        async void HandlersLockTester1(object sender, NotifyCollectionChangedEventArgs e)
        {
            var newItem = (KeyValuePair)e.NewItems[0];
            Debug.WriteLine(String.Format("~ HandlersLockTester1(): key: {0} value: {1}", newItem.Key.ToString(), newItem.Value));
            await Task.Delay(300);
        }
        async void HandlersLockTester2(object sender, NotifyCollectionChangedEventArgs e)
        {
            var newItem = (KeyValuePair)e.NewItems[0];
            Debug.WriteLine(String.Format("~ HandlersLockTester2(): key: {0} value: {1}", newItem.Key.ToString(), newItem.Value));
        }
        public async void run()
        {
            // Direct call for garbage collector - should be called for testing purposes only, not recommended for a business logic of an application
            GC.Collect();
            /*
             * == test v.Add()::foreach (var handlr in new List(handlersMap[key]))
             * for two threads entering the foreach loop at the same time and iterating handlers only of its key
             */
            Task t1 = Task.Run(() => { v.Add(k.OnMessageReceived, "this key"); });
            Task t2 = Task.Run(() => { v.Add(k.MessageEdit, "that key"); });
            // wait for both threads to complete before executing next test
            await Task.WhenAll(new Task[] { t1, t2 });
            /* For now DeadObject may be already destroyed, so we may test catch block in v class */
            v.Add(k.OnlineStatus, "for dead object");
            /* test reentrant calls - causes stack overflow or infinite loop, depending on code at OnEvent::if(newItem.Key == k.Unused) clause */
            v.Add(k.Unused, 'a');
            /* testing foreach loop entering multiple threads */
            var s = Enumerable.Repeat("string", 200).ToArray();
            var n = Enumerable.Repeat("astring", 200).ToArray();
            t1 = Task.Run(() => { foreachTest(s); });
            t2 = Task.Run(() => { foreachTest(n); });
            // wait for both threads to complete before executing next test
            await Task.WhenAll(new Task[] { t1, t2 });
            /* testing lock(handlr) in Add() method of class v */
            v.h(new k[] { k.IsTyping }, HandlersLockTester1);
            v.h(new k[] { k.JoinRoom }, HandlersLockTester2);
            // line 1
            Task.Run(() => { v.Add(k.IsTyping, "first thread for the same handler"); });
            // line 2
            Task.Run(() => { v.Add(k.IsTyping, "second thread for the same handler"); });
            // line below will MOST OF THE TIMES complete executing before the line 2 above, because line 2 will wait completion of line 1
            // since both previous lines 1 and 2 are calling the same handler, access to which is synchronized by lock(handlr) in Add() method of class v
            Task.Run(() => { v.Add(k.JoinRoom, "third thread for other handler"); });
        }
        public Tests()
        {
            // add OnEvent for each key
            v.h(new k[] { k.OnMessageReceived, k.MessageEdit, k.Unused }, OnEvent);
            // add OnEvent2 for each key
            v.h(new k[] { k.Unused, k.OnMessageReceived, k.MessageEdit }, OnEvent2);
            /* == test try catch blocks in v class, when handler is destroyed before handlr.Invoke() called */
            var ddo = new DeadObject();
            // then try to delete object, setting its value to null. We are in a managed environment, so we can't directly manage life cicle of an object.
            ddo = null;
        }
    }
}

The code registering the handler function, as well as the handler function itself for such a class v could look like this:

the registration code of the handler function

// add OnEvent for each key
v.h(new k[] { k.OnMessageReceived, k.MessageEdit, k.Unused }, OnEvent); 

handler function code

void OnEvent(object sender, NotifyCollectionChangedEventArgs e)
{
    var newItem = (KeyValuePair)e.NewItems[0];
    Debug.Write("~ OnEvent(): key {0} value {1}", newItem.Key.ToString(), newItem.Value);
}

A general description of Red Architecture is here .