Multithreading in Unity by means of reactive extensions

This article will discuss the main problems encountered in the development of multi-threaded mobile games using Unity, as well as ways to solve them using UniRx (reactive extensions for Unity).

The article consists of two parts. The first is devoted to multithreading for the “smallest”, it tells in an accessible language about streams and how to create them, about stream synchronization. The second part is devoted to reactive extensions, their device, principle of operation and methods of application.

Since one of the languages ​​for writing scripts in Unity is C #, in which we develop applications, all the code will be written only in it. For an in-depth understanding of the principles of multithreading and reactive extensions, we recommend reading the basics of multithreading and what is reactive extensions . If the reader is familiar with this topic, then you can skip the first section.
')

Multithreading for the smallest

Multi-threaded applications are applications that perform several tasks simultaneously in separate threads. Applications that use multi-threading, more quickly respond to user actions, because the user interface remains active, while tasks that require intensive work of the processor are performed in other threads. Multi-threaded C # applications using Mono are developed using keywords: Thread, ThreadPool, and asynchronous delegates.

Let's look at a multithreaded application using the example of a build. Suppose that each worker performs his duties simultaneously with other workers. For example, one washes the floors, another washes the windows, etc. (and it all happens at the same time). These are our streams.



Thread is a class that allows you to create new threads within an existing application.
Asynchronous delegates - an asynchronous method call using a delegate that is defined with the same signature as the method being called. For an asynchronous method call, you must use the BeginInvoke method. With this approach, the delegate takes a thread from the pool and executes some code in it.

ThreadPool - implementation of the “object pool” pattern. Its meaning is in effective flow management :: creating, deleting, assigning to it some kind of work. Returning to the construction analogy, ThreadPool is a foreman who controls the number of builders at a construction site and assigns each of them a task.

Thread synchronization tools

The C # language provides tools for synchronizing threads. These tools are presented in the form of lock and Monitor. They are used to ensure that the execution of a block of code is not carried out simultaneously by several threads. But there is one nuance. Using these tools can lead to deadlock (deadlock threads). It happens like this: thread A waits for thread B to return control, and thread B, in turn, waits for thread A to execute the blocked code. Therefore, multithreading and synchronization of threads must be used with caution.

Problems of the built-in mechanisms of a multithreading in Unity

The main problem we face when developing single-threaded applications is UI-friezes, caused by performing complex operations in the main thread. Unity has a task parallelization mechanism, presented in the form of coroutine (coroutine), but it works in one thread, and if you run something “heavy” in coroutine - hello, freeze. If we are satisfied with the parallel execution of functions in the main thread, then we can use korutiny. There is nothing difficult in this, in the Unity documentation this topic is very well covered. However, I would like to remind you that corutins are iterators, which work as follows in Unity:

• the first thing is registration korutina,
• Further, after each call to Update and before the call to LateUpdate , Unity polls all registered corutines and processes the code that is described inside the method that has a return type of IEnumerator .

In addition to the benefits, Korutin also have disadvantages:

1. Unable to get return value
   
   

   private IEnumerator LoadGoogle() { var www = new WWW("http://google.com"); yield return www; //  www.text    . } 

2. Error processing
   
   

    private IEnumerator LoadGoogle() { try { var www = new WWW("http://google.com"); yield return www; } catch { yield return null; } } 

3. Crutches with callbacks
   
   

    private IEnumerator LoadGoogle(Action<string> callback) { var www = new WWW("http://google.com"); yield return www; if (callback != null) { callback(www.text); } } 

4. Do not handle heavyweight methods in corintry
   
   

     void Start() { Debug.Log(string.Format("Thread id in start method = {0}", Thread.CurrentThread.ManagedThreadId)); StartCoroutine(this.HardMethod()); } private IEnumerator HardMethod() { while (true) { Thread.Sleep(1001); Debug.Log(string.Format("Thread id in HardMethod method = {0}", Thread.CurrentThread.ManagedThreadId)); yield return new WaitForEndOfFrame(); } } //Output: //Thread id in start method = 1 //Thread id in HardMethod method = 1 //Thread id in HardMethod method = 1 //Thread id in HardMethod method = 1 

As mentioned earlier, the cortinos are run in the main thread. For this reason, we get friezes by running heavy methods in them.

A number of these shortcomings are easily eliminated with the help of reactive extensions, which in the future will bring many more different features and facilitate development.

What is reactive expansion?

Reactive extensions - a set of libraries that allow you to work with events and asynchronous calls in the style of Linq. The task of such extensions is to simplify the writing of code, in which asynchronous interaction appears. Unity uses the UniRx library, which provides the basic functionality of reactive extensions. UniRx - implementation of reactive extensions for Unity based on .NET Reactive Extensions. Why not use this native implementation? Because standard RX in Unity do not work. The library is cross-platform and is supported on PC / Mac / Android / iOS / WP8 / WindowsStore platforms.

What provides us UniRx?

• Multithreading
• LINQ-like methods
• Simplified syntax for asynchronous communication
• Cross platform

How it works?

The basis of reactive extensions are the IObserver and IObservable . They provide a generic mechanism for push notifications, also known as the “Observer” design pattern.

• The IObservable interface represents the class that sends notifications (provider).
  The IObserver interface represents the class that receives them (the supervisor).
  T represents a class that provides information for notifications.
  
  The IObserver implementation prepares to receive notifications from the provider (the IObservable implementation) by passing its copy to the IObservable.Subscribe provider IObservable.Subscribe . This method returns an IDisposable object that can be used to unsubscribe an observer before the provider completes sending notifications.
  
  The IObserver interface defines the following three methods that the observer must implement:
  
  
  • The OnNext method, which is usually called by the provider to provide the observer with new data or status information.
  • The OnError method, which is usually called by the provider to indicate that the data is inaccessible, corrupted, or the provider has other errors.
  • The OnCompleted method, which is usually called by the supplier to confirm that notifications have been sent to the observer.
  
  Also in UniRx Scheduler is implemented - the main component with which multithreading is implemented. Basic time operations (Interval, Timer) in UniRx are implemented using MainThread. This means that most operations (except Observable.Start ) work in the main thread and thread safety, in this case, can be neglected. Observable.Start uses the ThreadPool Scheduler by default, this means that a thread will be created.
  
  We have familiarized with the basic concepts and theoretical knowledge, now we will consider examples of using the UniRx library.
  
  

  An example of creating an observer

  
  In this example, we will try to get data from an Internet resource using the UniRx library. To download data using reactive extensions, we need to create an observer and use the ObservableWWW class, which is a wrapper over the standard WWW Unity class. Get
method Get
uses coroutines and returns IObservable, to which we will sign an observer. This approach avoids the crutches described in the “Problems of embedded threading mechanisms in Unity” section.
  
  

   private void Start() { var observer = Observer.Create<string>( x => { Debug.Log("OnNext: " + x); }, ex => Debug.Log("OnError: " + ex.Message), () => Debug.Log("OnCompleted")); ObservableWWW.Get("http://qweqweqwe.qwer.qwer/").Subscribe(observer); } //Output: //OnError: Exception of type 'UniRx.WWWErrorException' was thrown. 

  
  If we change the link to an adequate one, say, at http://www.nixsolutions.com/ , we get the following result:
  
  

   //Output: //OnNext: ”html  ” //OnCompleted 

  
  

  Example of creating a sequence subject

  
  Here we subscribe to two Debug.Logs, the first one is always executed when the OnNext method is OnNext , and the second one is triggered only under the condition.
  
  

   void Start() { this.subject = new Subject<int>(); this.subject.Subscribe(x => Debug.Log(x)); this.subject.Where(x => x % 2 == 0).Subscribe(x => Debug.Log(string.Format("Hello from {0}", x))); } // Update is called once per frame void Update() { this.sub.OnNext(this.i++); } //Output: //0 //Hello from 0 //1 //2 //Hello from 2 

  
  

  EveryUpdate usage example

  
  An important feature in these extensions is the EveryUpdate method. It allows you to remove the code from the Update methods and MonoBehaviour successor MonoBehaviour . Here we check the mouse clicks and display some text.
  
  

   Observable.EveryUpdate() .Where(x => Input.GetMouseButton(buttonIndex)) .Subscribe(x => Debug.Log(outputString)); //Output: //Left button pressed //Right button pressed 

  
  

  An example of working with arrays

  
  Also an interesting feature of these extensions is working with arrays. The code that is presented below, when executed in a single-threaded application, will freeze the display stream.
  
  

   var arr = Enumerable.Range(0, 5); foreach (var i in arr) { Thread.Sleep(1000); Debug.Log(string.Format("Result = {0}, UtcNow = {1}, ThreadId = {2}", i, DateTime.UtcNow, Thread.CurrentThread.ManagedThreadId)); } //Output: //Result = 0, UtcNow = 8/25/2015 1:23:14 PM, ThreadId = 1 //Result = 1, UtcNow = 8/25/2015 1:23:16 PM, ThreadId = 1 //Result = 2, UtcNow = 8/25/2015 1:23:17 PM, ThreadId = 1 //Result = 3, UtcNow = 8/25/2015 1:23:18 PM, ThreadId = 1 //Result = 4, UtcNow = 8/25/2015 1:23:19 PM, ThreadId = 1 

  
  To solve this problem, you can use UniRx, with an explicit indication of the ThreadPool scheduler, which will itself distribute the load between the threads.
  
  

    var arr2 = Enumerable.Range(0, 5).ToObservable(Scheduler.ThreadPool); arr2.Subscribe( x => { Thread.Sleep(1000); Debug.Log(string.Format("Result = {0}, UtcNow = {1}, ThreadId = {2}", x, DateTime.UtcNow, Thread.CurrentThread.ManagedThreadId)); }); //Output: //Result = 0, UtcNow = 8/25/2015 1:23:20 PM, ThreadId = 2 //Result = 1, UtcNow = 8/25/2015 1:23:21 PM, ThreadId = 3 //Result = 2, UtcNow = 8/25/2015 1:23:22 PM, ThreadId = 4 //Result = 3, UtcNow = 8/25/2015 1:23:23 PM, ThreadId = 5 //Result = 4, UtcNow = 8/25/2015 1:23:24 PM, ThreadId = 5 

  
  An interesting feature of this approach is that a free flow will be assigned to each iteration.
  
  

  An example of processing complex methods

  
  In this example, we have a couple of complex methods that, when executed in the main thread, freeze our application. When using Rx, everything will be fine, we will get the return values ​​from the methods and process them.
  
  

   private void Awake() { var heavyMethod = Observable.Start(() => { var timeToSleep = 1000; var returnedValue = 10; Debug.Log(string.Format("Thread = {0} UtcNow = {1}", Thread.CurrentThread.ManagedThreadId, DateTime.UtcNow)); Thread.Sleep(timeToSleep); return returnedValue; }); var heavyMethod2 = Observable.Start(() => { var timeToSleep = 2000; var returnedValue = 20; Debug.Log(string.Format("Thread = {0} UtcNow = {1}", Thread.CurrentThread.ManagedThreadId, DateTime.UtcNow)); Thread.Sleep(timeToSleep); return returnedValue; }); Observable.WhenAll(heavyMethod, heavyMethod2) .ObserveOnMainThread() .Subscribe(result => { Debug.Log(string.Format("Thread = {0}, first result = {1}, second result = {2} UtcNow = {3}", Thread.CurrentThread.ManagedThreadId, result[0], result[1], DateTime.UtcNow)); }); } //Output: //Thread = 5 UtcNow = 8/25/2015 2:06:55 PM //Thread = 3 UtcNow = 8/25/2015 2:06:55 PM //Thread = 1, first result = 10, second result = 20 UtcNow = 8/25/2015 2:06:57 PM 

  
  

  Example use of bindings

  
  Binding is another great mechanism. With it, you can easily implement the MVP pattern. In this example, the model is the class Enemy, in which we describe the reactive properties. The IsDead property depends directly on CurrentHp : when it is less than zero, IsDead becomes = true.
  
  

   public class Enemy { public Enemy(int initialHp) { this.CurrentHp = new ReactiveProperty<long>(initialHp); this.IsDead = this.CurrentHp.Select(x => x <= 0).ToReactiveProperty(); } public ReactiveProperty<long> CurrentHp { get; private set; } public ReactiveProperty<bool> IsDead { get; private set; } } 

  
  Presenter is responsible for the connection of the model and the display, with its help we can tie the reactive properties of the model to the parts of the display. The MvpExample class is a presenter and has a link to both the model (the Enemy class) and the display ( Button and Toggle ). Also, due to reactive extensions, we have the ability to set the behavior of various UI elements using a code. Using the OnClickAsObservable and OnValueChangedAsObservable we described the behavior of Button and Toggle.
  
  

   public class MvpExample : MonoBehaviour { private const int EnemyHp = 1000; [SerializeField] private Button myButton; [SerializeField] private Toggle myToggle; [SerializeField] private Text myText; private void Start() { var enemy = new Enemy(EnemyHp); this.myButton.OnClickAsObservable().Subscribe(x => enemy.CurrentHp.Value -= 99); //      CurrentHp   Enemy this.myToggle.OnValueChangedAsObservable().SubscribeToInteractable(this.myButton); //   Toggle     enemy.CurrentHp.SubscribeToText(this.myText); enemy.IsDead.Where(isDead => isDead) .Subscribe(_ => { this.myToggle.interactable = this.myButton.interactable = false; }); } } 

  
  Next we pribindili reactive properties for UI-elements. When changing CurrentHp in Enemy , we will automatically change the text. When IsDead changes its state to true , then both the button and Toggle will be disabled.
  
  

  findings

  
  Using reactive extensions when developing applications on Unity has many advantages. The main one is to simplify the syntax for building multi-threaded applications. The number of crutches with korutinami significantly reduced, the application becomes more flexible and faster. Also, when building a multi-threaded application using UniRx, it must be remembered that any part of the data must be protected from being modified by many threads.
  
  Useful links:
  
  • http://rsdn.ru/article/dotnet/CSThreading1.xml
  • http://professorweb.ru/my/csharp/thread_and_files/1/1_4.php
  • https://msdn.microsoft.com/ru-ru/library/ms173179.aspx
  • https://github.com/neuecc/UniRx