Which thread does each stage of a pipeline run on, and what do subscribe(on:) and receive(on:) each move?

11.9.1 · L3 · Interview

Predict first

Which thread does each stage run on?

import Combine
import Foundation

var bag = Set<AnyCancellable>()
func tag() -> String { Thread.isMainThread ? "main" : "background" }
let bgq = DispatchQueue(label: "bg")
Just(1)
    .handleEvents(receiveSubscription: { _ in print("subscription on", tag()) })
    .map { v -> Int in print("first map on", tag()); return v }
    .subscribe(on: bgq)
    .map { v -> Int in print("second map on", tag()); return v }
    .receive(on: DispatchQueue.main)
    .sink { _ in print("sink on", tag()) }
    .store(in: &bag)
RunLoop.main.run(until: Date().addingTimeInterval(0.3))
  • subscription on background / first map on background / second map on background / sink on main
  • subscription on main / first map on main / second map on background / sink on main
  • subscription on background / first map on background / second map on main / sink on main

Mental model. receive(on:) is a sign at a junction that redirects all traffic after that point. subscribe(on:) decides which road the vehicle starts its journey on.

By default a stage runs on whatever thread the value arrives on: the publisher decides where it emits. receive(on:) is a downstream switch: from that point on, values (and completion) are delivered on the given scheduler, so the final sink runs on the main queue in the check. subscribe(on:) moves the subscribe, request and cancel operations onto a scheduler, which in practice means upstream work that happens when subscribing runs there. In the check, subscription, the first map and the second map all run on “background”, because the Just emits synchronously on the thread that subscribes, which is the background queue after subscribe(on:); only the receive(on: DispatchQueue.main) at the end brings the sink back to the main thread.

Practical rules: put receive(on: DispatchQueue.main) (or RunLoop.main) immediately before UI-touching work, once, near the end; do not sprinkle it between every stage (each hop costs a dispatch and can reorder timing); use subscribe(on:) only when the subscription itself does blocking work and the publisher does not already do its own async work (URLSession publishers already run off the main thread, so it adds nothing). To find out where a stage runs, log Thread.isMainThread in handleEvents or a map, as here. With Swift concurrency, @MainActor on the type and the sink closure is the compiler-checked alternative.

subscribe(on:) makes the subscription and the synchronous emission happen on the queue; receive(on:) moves only what follows it.

Common traps

  • Assuming a stage runs on the main thread because the pipeline was created there. Emission thread, not creation thread, decides.
  • Adding subscribe(on:) to network publishers “for performance”. They already run off the main thread.

They might follow up with

  • Does receive(on:) preserve ordering? Yes, on a serial queue; each value is delivered in order, but later than the emission.
  • What happens if I forget it before a UI update? The update runs on a background thread, which UIKit and SwiftUI forbid (Q218).

Related questions

  • 11.9.2 What is the difference between receive(on: ImmediateScheduler) and receive(on: DispatchQueue.main) for the…
  • 11.7.5 How do you see what a Combine pipeline is doing: handleEvents, print, breakpoint and the order of…
  • 11.4.1 What do receive(on:) and subscribe(on:) do, and how do you know which thread a stage runs on?
  • 6.3.3 What is the main-thread contract, and why does DispatchQueue.main.sync from the main thread crash?

What is the difference between receive(on: ImmediateScheduler) and… →