15. Intro to Schedulers

Written by Florent Pillet

Until now, you’ve managed to work with schedulers while avoiding any explanation about what they actually are and how they handle threading or concurrency. In earlier chapters, you used methods which implicitly used some sort of concurrency/threading level, such as the buffer, delaySubscription or interval operators.

You might feel like schedulers have some sort of magic under the hood, but before you understand schedulers, you’ll also need to understand what that observeOn operator is all about.

This chapter is going to cover the beauty behind schedulers, where you’ll learn why the RxSwift abstraction is so powerful and why working with asynchronous programming is far less painful than using locks or queues.

Note: Creating custom schedulers is beyond of the scope of this book. Keep in mind that the schedulers and initializers provided by RxSwift, RxCocoa and RxBlocking generally cover 99% of cases. Always try to use the built-in schedulers.

What is a scheduler?

Before getting your hands dirty with schedulers, it’s important to understand what they are — and what they are not. To summarize, a scheduler is a context where a process takes place. This context can be a thread, a dispatch queue or similar entities, or even an Operation used inside the OperationQueueScheduler.

Here’s a good example as to how schedulers can be used:

In this diagram, you have the concept of a cache operator. An observable makes a request to a server and retrieves some data. This data is processed by a custom operator named cache, which stores the data somewhere. After this, the data is passed to all subscribers on a different scheduler, most likely the MainScheduler which sits on top of the main thread, making the update of the UI possible.

Demystifying the scheduler

One common misconception about schedulers is that they are equally related to threads. And that might seem logical at first — after all, schedulers do work similarly to GCD‘s dispatch queues.

But this isn’t the case at all. If you were writing a custom scheduler, which again is not a recommended approach, you could create multiple schedulers using the very same thread, or a single scheduler on top of multiple threads. That would be weird — but it would work!

The important thing to remember is that schedulers are not threads, and they don’t have a one-to-one relationship with threads. Always check the context in which the scheduler is performing an operation — not the thread. Later in this chapter, you’ll encounter some good examples to help you understand this.

Setting up the project

Time to write some code! In this project, you are going to create a simple command-line tool for macOS. Why a command-line tool? Since you are playing with threads and concurrency, plain-text output will be easier to understand than any visual elements you could create in an app.

Install the CocoaPods dependencies for this chapter‘s starter project, as described in Chapter 1, “Hello RxSwift.” (By now you definitely know how to do it by heart, but one never knows how many chapters you skipped through.) Once finished, open the workspace, build and run, and the debugger console should show the following:

===== Schedulers =====

00s | [E] [dog] emitted on Main Thread
00s | [S] [dog] received on Main Thread

Before proceeding, open Utils.swift and take a look at the implementation of dump() and dumpingSubscription().

The first method dumps the element and the current thread information inside a do(onNext:) operator using the [E] prefix (for “Emitted”). The second dumps similar information using the [S] prefix (for ”Subscription”). It subscribes to the observable, showing on which thread it receives the elements. Both functions show the elapsed time, so the 00s above stand for “0 seconds elapsed”.

These functions highlight two different ways of printing info to the console:

• Using do(onNext:) which lets you inject side effects in the operator chain (perform operations “on the side” that do not alter the observable sequence).
• Subscribing to the observable sequence and print from there.

Now that you have a mean to check out which threads you‘re on at any given time, you are ready to learn how easy it is for a chain of observables to switch between schedulers.

Switching schedulers

One of the most important things in RxSwift is the ability to switch schedulers at any time, without any restrictions except for ones imposed by the inner process generating events. There are good reasons why you want to be able to control which scheduler an operator receives elements on:

• To perform expensive work on background schedulers.
• To control whether expensive works occurs serially or in parallel.
• To guarantee delivery on the main thread for user interface updates.

Note: When using operators that let you switch schedulers, make sure that the elements the sequence transports are thread-safe. RwSwift itself acts like Apple‘s Dispatch framework: it lets you switch schedulers / threads regardless of your data‘s thread-safety.

To understand how schedulers behave, you’ll create a simple observable to play with that provides some fruit.

Add the following code to the bottom of main.swift:

let fruit = Observable<String>.create { observer in
  observer.onNext("[apple]")
  sleep(2)
  observer.onNext("[pineapple]")
  sleep(2)
  observer.onNext("[strawberry]")
  return Disposables.create()
}

This observable features a sleep function. While this is not something you’d usually see in real applications, in this case it will help you understand how subscriptions and observations work.

Add the following code to subscribe to the observable you created:

fruit
  .dump()
  .dumpingSubscription()
  .disposed(by: bag)

Build and run, and check out the logging in the console:

===== Schedulers =====

00s | [E] [dog] emitted on Main Thread
00s | [S] [dog] received on Main Thread
00s | [E] [apple] emitted on Main Thread
00s | [S] [apple] received on Main Thread
02s | [E] [pineapple] emitted on Main Thread
02s | [S] [pineapple] received on Main Thread
04s | [E] [strawberry] emitted on Main Thread
04s | [S] [strawberry] received on Main Thread

Here you have the original subject, followed by a fruit every two seconds after that.

The fruit is generated on the main thread, but it would be nice to move it to a background thread. To create the fruit in a background thread, you’ll have to use subscribeOn.

Note: if the application doesn‘t compile, showing random errors related to missing derived data, clean it using Clean Build Folder under the Product menu and build again.

Using subscribeOn

In some cases you might want to change on which scheduler the observable computation code runs — not the code in any of the subscription operators, but the code that is actually emitting the observable events.

Note: For the custom observable that you have created, the code that emits events is the one you supply as the trailing closure for Observable.create { ... }.

The way to set the scheduler for that computation code is to use subscribeOn. It might sound like a counterintuitive name at first glance, but after thinking about it for a while, it starts to make sense.

When you want to actually observe an observable, you must first subscribe to it. This determines where the original processing will happen. If subscribeOn is not called, RxSwift automatically uses the current thread:

This process is creating events on the main thread using the main scheduler. The MainScheduler sits on top of the main thread. All the tasks you want to perform on the main thread have to use this scheduler, which is why you used it in previous examples when working with the UI. To switch schedulers, you’ll use subscribeOn.

In main.swift, there’s a predefined scheduler named globalScheduler that uses a background queue. This scheduler is created using the global dispatch queue, which is a concurrent queue:

let globalScheduler = ConcurrentDispatchQueueScheduler(queue: DispatchQueue.global())

So, as the name of the class suggests, all tasks to be computed by this scheduler will be dispatched and handled by the global dispatch queue.

To use this scheduler, replace the previous subscription to fruit you created with this new one:

fruit
  .subscribeOn(globalScheduler)
  .dump()
  .dumpingSubscription()
  .disposed(by: bag)

Now add the following line to the end of the file:

RunLoop.main.run(until: Date(timeIntervalSinceNow: 13))

This is, admittedly, a hack; it prevents Terminal from terminating once all operations have completed on the main thread, which would kill your global scheduler and observable. In this case, Terminal will remain alive for 13 seconds.

Note: Thirteen seconds might be overkill for this example, but as you move through the chapter, your app will need this length of time to finish. So feel free to stop the application once all the observables have completed.

Now that your new scheduler is in place, build and run and check the result:

00s | [E] [dog] emitted on Main Thread
00s | [S] [dog] received on Main Thread
00s | [E] [apple] emitted on Anonymous Thread
00s | [S] [apple] received on Anonymous Thread
02s | [E] [pineapple] emitted on Anonymous Thread
02s | [S] [pineapple] received on Anonymous Thread
04s | [E] [strawberry] emitted on Anonymous Thread
04s | [S] [strawberry] received on Anonymous Thread

The global queue uses a thread that doesn’t have a name, so in this case Anonymous Thread is one of the threads of the global, concurrent dispatch queue.

Now, both the emitter and the subscriber are processing data in the same thread.

That‘s cool, but what can you do if you want to change where the observer performs the code of your operators? You have to use observeOn.

Using observeOn

Observing is one of the three fundamental concepts of Rx. It involves an entity producing events, and an observer for those events. In this case, and in opposition to subscribeOn, the observeOn operator changes the scheduler where the observation happens.

So once an event is pushed by an Observable, this operator ensures that subscribers receive the event on the specified scheduler. This also includes all the operators you added after observeOn!

To switch from the current global scheduler to the main thread, you need to call observeOn before subscribing. One more time, replace your fruits subscription code:

fruit
  .subscribeOn(globalScheduler)
  .dump()
  .observeOn(MainScheduler.instance)
  .dumpingSubscription()
  .disposed(by: bag)

Build and run, and check the console output once more (you will need to wait a few seconds until the program stops printing in the console):

00s | [E] [dog] emitted on Main Thread
00s | [S] [dog] received on Main Thread
00s | [E] [apple] emitted on Anonymous Thread
00s | [S] [apple] received on Main Thread
02s | [E] [pineapple] emitted on Anonymous Thread
02s | [S] [pineapple] received on Main Thread
04s | [E] [strawberry] emitted on Anonymous Thread
04s | [S] [strawberry] received on Main Thread

You’ve achieved the result you wanted: All the events are now processed on the correct thread. The main observable is processing and generating events on the background thread, and the subscriber is receiving them on the main thread.

This is a very common pattern. You’ve used a background scheduler to retrieve data from a server and process the data received, only switching to the MainScheduler to process the final event and display the data in the user interface.

Pitfalls

The ability to switch schedulers and threads looks amazing, but it comes with some pitfalls. To see why, you’ll push some events to the subject using a new thread. Since you need to track on which thread the computation takes place, a good solution is to use an OS Thread.

Right after the fruit observable, add the following code to generate some animals:

let animalsThread = Thread() {
  sleep(3)
  animal.onNext("[cat]")
  sleep(3)
  animal.onNext("[tiger]")
  sleep(3)
  animal.onNext("[fox]")
  sleep(3)
  animal.onNext("[leopard]")
}

Next, name the thread so you will be able to recognize it, and start it up:

animalsThread.name = "Animals Thread"
animalsThread.start()

Build and run; you should see your new thread in action:

...
03s | [E] [cat] emitted on Animals Thread
03s | [S] [cat] received on Animals Thread
04s | [E] [strawberry] emitted on Anonymous Thread
04s | [S] [strawberry] received on Main Thread
06s | [E] [tiger] emitted on Animals Thread
06s | [S] [tiger] received on Animals Thread
09s | [E] [fox] emitted on Animals Thread
09s | [S] [fox] received on Animals Thread
12s | [E] [leopard] emitted on Animals Thread
12s | [S] [leopard] received on Animals Thread

Perfect — you have animals created on the dedicated thread. Next - process the result on the global thread.

Note: It might seem repetitive to keep adding code and then replacing it with something else, but the goal here is to compare the differences between the various schedulers.

Replace the original subscription to the animal subject with the following code:

animal
  .dump()
  .observeOn(globalScheduler)
  .dumpingSubscription()
  .disposed(by: bag)

Build and run, and the new result is as follows:

...
03s | [E] [cat] emitted on Animals Thread
03s | [S] [cat] received on Anonymous Thread
04s | [E] [strawberry] emitted on Anonymous Thread
04s | [S] [strawberry] received on Main Thread
06s | [E] [tiger] emitted on Animals Thread
06s | [S] [tiger] received on Anonymous Thread
09s | [E] [fox] emitted on Animals Thread
09s | [S] [fox] received on Anonymous Thread
12s | [E] [leopard] emitted on Animals Thread
12s | [S] [leopard] received on Anonymous Thread

Now you’re switching threads and nearly running into that 13-second limit!

What if you want to generate animals on the global queue, but receive them on the Main Thread? For the first case, the observeOn is already correct, but for the second it’s necessary to use subscribeOn.

Replace the animal subscription, this time with the following:

animal
  .subscribeOn(MainScheduler.instance)
  .dump()
  .observeOn(globalScheduler)
  .dumpingSubscription()
  .disposed(by: bag)

Build and run, and you’ll get the following result:

03s | [E] [cat] emitted on Animals Thread
03s | [S] [cat] received on Anonymous Thread
04s | [E] [strawberry] emitted on Anonymous Thread
04s | [S] [strawberry] received on Main Thread
06s | [E] [tiger] emitted on Animals Thread
06s | [S] [tiger] received on Anonymous Thread
09s | [E] [fox] emitted on Animals Thread
09s | [S] [fox] received on Anonymous Thread
12s | [E] [leopard] emitted on Animals Thread
12s | [S] [leopard] received on Anonymous Thread

Wait?! What? Why isn’t the computation happening on the correct scheduler? This is a common and dangerous pitfall that comes from thinking of RxSwift as asynchronous or multi-threaded by default — which isn’t the case.

RxSwift and the general abstraction is free-threaded; there’s no magic thread switching taking place when processing data. The computation always happen on the original thread if you don‘t specify otherwise.

Note: Any thread switching happens after an explicit request by the programmer using the operators subscribeOn and observeOn.

Thinking RxSwift does some thread handling by default is a common trap to fall into. What’s happening above is a misuse of the Subject. The original computation is happening on a specific thread, and those events are pushed in that thread using Thread() { ... }. Due to the nature of Subject, RxSwift has no ability to switch the original computation scheduler and move to another thread, since there’s no direct control over where the subject is pushed.

Why does this work with the fruit thread though? That’s because using Observable.create(_:) puts RxSwift in control of what happens inside the Thread block so you can more finely customize thread handling.

This unexpected outcome is commonly known as the “Hot and Cold” observables problem.

In the case above, you are dealing with a hot observable. The observable doesn‘t have any side-effect during subscription, but it does have its own context in which events are generated and RxSwift can’t control it (namely, it sports its own Thread).

A cold observable in contrast doesn’t produce any elements before any observers subscribe to it. That effectively means it doesn‘t have its own context until, upon subscription, it creates some context and starts producing elements.

Hot vs. cold

The section above touched on the topic of hot and cold observables. The topic of hot and cold observables is quite opinionated and generates a lot of debate, so let‘s briefly look into it here. The concept can be reduced to a very simple question:

Some examples of side effects are:

• Fire a request to the server.
• Edit the local database.
• Write to the file system.
• Launch a rocket.

The world of side effects is endless, so you need to determine whether your Observable instance is performing side effects upon subscription. If you can’t be certain about that, then perform more analysis or dig further into the source code. Launching a rocket on every subscription might not be what you’re looking to achieve…

Another common way to describe this is to ask whether or not the Observable shares side-effects. If you’re performing side effects upon subscription, it means that the side effect is not shared. Otherwise, the side effects are shared with all subscribers.

This is a fairly general rule, and applies to any ObservableType object like a subject and related subtypes.

As you might have noticed, we haven’t spoken much about hot and cold observables so far in the book. It’s a common topic in reactive programming, but in RxSwift you‘ll encounter the concept only in specific cases like the Thread example above or when you need greater control, such as when you run tests.

Keep this section as a point of reference, so in case you need to approach a problem in terms of hot or cold observables, you can quickly open the book to this point and refresh yourself on the concept.

Best practices and built-in schedulers

Schedulers are a non-trivial topic, so they come with some best practices for the most common use cases. In this section, you’ll get a quick introduction to serial and concurrent schedulers, learn how they process the data and see which type works better for a particular context.

Serial vs concurrent schedulers

Considering that a scheduler is simply a context, which could be anything (dispatch queue, thread, custom context), and that all operators transforming sequences need to preserve the implicit guarantees, you need to be sure you’re using the right scheduler.

• If you’re using a serial scheduler, RxSwift will do computations serially. For a serial dispatch queue, schedulers will also be able to perform their own optimizations underneath.

• In a concurrent scheduler, RxSwift will try running code simultaneously, but observeOn and subscribeOn will preserve the sequence in which tasks need to be executed, and ensure that your subscription code ends up on the correct scheduler.

MainScheduler

MainScheduler sits on top of the main thread. This scheduler is used to process changes on the user interface and perform other high-priority tasks. As a general practice when developing applications on iOS, tvOS or macOS, long-running tasks should not be performed using this scheduler, so avoid things like server requests or other heavy tasks.

Additionally, if you perform side effects that update the UI, you must switch to the MainScheduler to guarantee those updates make it to the screen.

The MainScheduler is also used for all observations when using most RxCocoa Traits, and more specifically, Driver and Signal. As discussed in an earlier chapter, these traits ensures the observation is always performed in the MainScheduler to give you the ability to bind data directly to the user interface of your application.

SerialDispatchQueueScheduler

SerialDispatchQueueScheduler manages to abstract the work on a serial DispatchQueue. This scheduler has the great advantage of several optimizations when using observeOn.

You can use this scheduler to process background jobs which are better scheduled in a serial manner. For example, if you have an application talking with a single endpoint of a server (as in a Firebase or GraphQL application), you might want to avoid dispatching multiple, simultaneous requests, which would put too much pressure on the receiving end. This scheduler is definitely the one you would want for any jobs that should advance much like a serial task queue.

ConcurrentDispatchQueueScheduler

ConcurrentDispatchQueueScheduler, similar to SerialDispatchQueueScheduler, manages to abstract work on a DispatchQueue. The main difference here is that instead of a serial queue, the scheduler uses a concurrent one.

This kind of scheduler isn’t optimized when using observeOn, so remember to account for that when deciding which kind of scheduler to use.

A concurrent scheduler might be a good option for multiple, long-running tasks that need to end simultaneously. Combining multiple observables with a blocking operator, so all results are combined together when ready, can prevent serial schedulers from performing at their best. Instead, a concurrent scheduler could perform multiple concurrent tasks and optimize the gathering of the results.

OperationQueueScheduler

OperationQueueScheduler is similar to ConcurrentDispatchQueueScheduler, but instead of abstracting the work over a DispatchQueue, it performs the job over an OperationQueue. Sometimes you need more control over the concurrent jobs you are running, which you can’t do with a concurrent DispatchQueue.

If you need to fine-tune the maximum number of concurrent jobs, this is the scheduler for the job. You can set maxConcurrentOperationCount to cap the number of concurrent operations to suit your application’s needs.

TestScheduler

TestScheduler is a special kind of beast. It’s meant only to be used in testing, so you should not use it in production code. This special scheduler simplifies operator testing; it’s part of the RxTest library. You will have a look into using this scheduler in the dedicated chapter about testing, but let‘s have a quick look since you‘re doing the grand tour of RxSwift schedulers.

A good use case for this scheduler is provided by the test suite of RxSwift. Open the following link: https://bit.ly/2E0xVUq. You‘ll find the dedicated file for testing the delaySubscription operator Observable+DelaySubscriptionTests.swift, and specifically, the single test case named testDelaySubscription_TimeSpan_Simple. Inside this test case, you have the initialization of the scheduler:

let scheduler = TestScheduler(initialClock: 0)

Following this initialization, you have the definition of the observable to test:

let xs = scheduler.createColdObservable([
  next(50, 42),
  next(60, 43),
  completed(70)
])

And just before the definition of the expectations, you have the declaration of how to get the results:

let res = scheduler.start {
  xs.delaySubscription(30, scheduler: scheduler)
}

res will be created by the scheduler using the previously defined xs observable. This result contains all the information about the events sent as well as the time tracked by the test scheduler.

With this, you could write a test case like so:

XCTAssertEqual(res.events, [
  next(280, 42),
  next(290, 43),
  completed(300)
])

Wondering why the event happens at 280, and not at 80 (considering the original 50, plus 30 for the delay)? This is due to the nature of testScheduler, which starts all subscriptions to ColdObservable after 200. This trick ensures that a cold observable won’t start at an unpredictable time — which would make testing a nightmare!

The same thing doesn’t apply to a HotObservable, so a HotObservable will start pushing events right away.

As you’re testing a delaySubscription operator, just the information about the events sent and their time won’t be enough to work with. You’ll need extra information about the time of the subscription to ensure everything is working as expected.

With xs.subscriptions, you can get the list of the subscriptions to make the final part of the test:

XCTAssertEqual(xs.subscriptions, [
  Subscription(230, 300)
])

The first number defines the starting time of the first subscription. The second one defines when the subscription will be disposed. In this case, the second number matches the completed event because completion will dispose of all subscriptions.

Where to go from here?

Schedulers are a non-trivial topic in the RxSwift space; they’re responsible for computing and performing all tasks in RxSwift. The golden rule of a Scheduler is that it can be anything. Keep this in mind, and you’ll get along just fine when working with observables and using and changing schedulers.

As you read earlier, a scheduler can sit on top of a DispatchQueue, a OperationQueue, a Thread or even perform the task immediately on the current thread. There’s no hard rule about this, so make sure you know what scheduler you’re using for the task at hand. Sometimes, using the wrong scheduler can have a negative impact on performance, while a well-chosen scheduler can have great performance returns.

Before proceeding, invest some time in playing around with the current example and test some schedulers to see what impact they have on the final result. Understanding schedulers will make life easier with RxSwift, and will improve your confidence when using subscribeOn and observeOn.