JavaScript’s Event Loop

Let’s start with an example. We’ll set a couple of timers, one that fires after one millisecond, and the other after two:

setTimeout(() => console.info('A'), 1)
setTimeout(() => console.info('B'), 2)
console.info('C')

Now, what will get logged to the console? Is it A, B, C?

If you’re a JavaScript programmer, you know intuitively the answer is no—the actual firing order is C, A, then B. If you haven’t worked with JavaScript or TypeScript before, this behavior might seem mysterious and unintuitive. In reality, it’s pretty straightforward; it just doesn’t follow the same concurrency model as a sleep would in C, or scheduling work in another thread would in Java.

At a high level, the JavaScript VM simulates concurrency like this (see Figure 8-1):

• The main JavaScript thread calls into native asynchronous APIs like XMLHTTPRequest (for AJAX requests), setTimeout (for sleeping), readFile (for reading a file from disk), and so on. These APIs are provided by the JavaScript platform—you can’t create them yourself.¹

• Once you call into a native asynchronous API, control returns to the main thread and execution continues as if the API was never called.

• Once the asynchronous operation is done, the platform puts a task in its event queue. Each thread has its own queue, used for relaying the results of asynchronous operations back to the main thread. A task includes some metainformation about the call, and a reference to a callback function from the main thread.

• Whenever the main thread’s call stack is emptied, the platform will check its event queue for pending tasks. If there’s a task waiting, the platform runs it; that triggers a function call, and control returns to that main thread function. When the call stack resulting from that function call is once again empty, the platform again checks the event queue for tasks that are ready to go. This loop repeats until both the call stack and the event queue are empty, and all asynchronous native API calls have completed.

Figure 8-1. JavaScript’s event loop: what happens when you call an asynchronous API

Armed with this information, it’s time to go back to our setTimeout example. Here’s what happens:

1. We call setTimeout, which calls a native timeout API with a reference to the callback we passed in and the argument 1.

2. We call setTimeout again, which calls the native timeout API again with a reference to the second callback we passed in and the argument 2.

3. We log C to the console.

4. In the background, after at least one millisecond, our JavaScript platform adds a task to its event queue indicating that the timeout for the first setTimeout has elapsed, and that its callback is now ready to be called.

5. After another millisecond, the platform adds a second task to the event queue for the second setTimeout’s callback.

6. Since the call stack is empty, after step 3 is done the platform looks at its event queue to see if there are any tasks in it. If steps 4 and/or 5 are done, then it will find some tasks. For each task, it will call the corresponding callback function.

7. Once both timers have elapsed and the event queue and call stack are empty, the program exits.

That’s why we logged C, A, B, and not A, B, C. With this baseline out of the way, we can start talking about how to type asynchronous code safely.

Working with Callbacks

The basic unit of the asynchronous JavaScript program is the callback. A callback is a plain old function that you pass as an argument to another function. As in a synchronous program, that other function invokes your function when it’s done doing whatever it does (making a network request, etc.). Callbacks invoked by asynchronous code are just functions, and there’s no giveaway in their type signatures that they are invoked asynchronously.

For NodeJS native APIs like fs.readFile (used to asynchronously read the contents of a file from disk) and dns.resolveCname (used to asynchronously resolve CNAME records), the convention for callbacks is that the first parameter is an error or null, and the second parameter is a result or null.

Here’s what readFile’s type signature looks like:

function readFile(
  path: string,
  options: {encoding: string, flag?: string},
  callback: (err: Error | null, data: string | null) => void
): void

Notice that there’s nothing special about either readFile’s type or callback’s type: both are regular JavaScript functions. Looking at the signature, there’s no indication that readFile is asynchronous and that control will be passed to the next line right after readFile is called (not waiting for its result).

npm install @types/node --save-dev

For example, let’s write a NodeJS program that reads and writes to your Apache access log:

import * as fs from 'fs'

// Read data from an Apache server's access log
fs.readFile(
  '/var/log/apache2/access_log',
  {encoding: 'utf8'},
  (error, data) => {
    if (error) {
      console.error('error reading!', error)
      return
    }
    console.info('success reading!', data)
  }
)

// Concurrently, write data to the same access log
fs.appendFile(
  '/var/log/apache2/access_log',
  'New access log entry',
  error => {
    if (error) {
      console.error('error writing!', error)
    }
  })

Unless you’re a TypeScript or JavaScript engineer and are familiar with how NodeJS’s built-in APIs work, and know that they’re asynchronous and you can’t rely on the order in which API calls appear in your code to dictate in which order filesystem operations actually happen, you wouldn’t know that we just introduced a subtle bug where the first readFile call may or may not return the access log with our new line appended, depending on how busy the filesystem is at the time this code runs.

You might know that readFile is asynchronous from experience, or because you saw it in NodeJS’s documentation, or because you know that NodeJS generally sticks to the convention that if a function’s last argument is a function that takes two arguments—an Error | null and a T | null, in that order—then the function is usually asynchronous, or because you ran across the hall to your neighbor for a cup of sugar and ended up staying for a while to chit-chat, then you somehow got on the topic of asynchronous programming in NodeJS and they told you about that time they had a similar issue a couple of months ago and how they fixed it.

Whatever it was, the types certainly didn’t help you get there.

Besides the fact that you can’t use types to help guide your intuition about the nature of a function’s synchronicity, callbacks are also difficult to sequence—which can lead to what some people call “callback pyramids”:

async1((err1, res1) => {
  if (res1) {
    async2(res1, (err2, res2) => {
      if (res2) {
        async3(res2, (err3, res3) => {
          // ...
        })
      }
    })
  }
})

When sequencing operations, you usually want to continue down the chain when an operation succeeds, bailing out as soon as you hit an error. With callbacks, you have to do this manually; when you start accounting for synchronous errors too (e.g., the NodeJS convention is to throw when you give it a badly typed argument, rather than calling your provided callback with an Error object), properly sequencing callbacks can get error-prone.

And sequencing is just one kind of operation you might want to run over asynchronous tasks—you might also want to run functions in parallel to know when they’re all done, race them to get the result of the first one that finishes, and so on.

This is a limitation of plain old callbacks. Without more sophisticated abstractions for operating on asynchronous tasks, working with multiple callbacks that depend on each other in some way can get messy fast.

To recap:

• Use callbacks to do simple asynchronous tasks.

• While callbacks are great for modeling simple tasks, they quickly get hairy as you try to do things with lots of asynchronous tasks.

Regaining Sanity with Promises

Luckily, we’re not the first programmers to run into these limitations. In this section we’ll develop the concept of promises, which are a way to abstract over asynchronous work so that we can compose it, sequence it, and so on. Even if you’ve worked with promises or futures before, this will be a helpful exercise to understand how they work.

We’ll start with an example of how we want to use Promise to first append to a file, then read back the result:

function appendAndReadPromise(path: string, data: string): Promise<string> {
  return appendPromise(path, data)
    .then(() => readPromise(path))
    .catch(error => console.error(error))
}

Notice how there’s no callback pyramid here—we’ve effectively linearized what we want to do into a single, easy-to-understand chain of asynchronous tasks. When one succeeds, the next one runs; if it fails, we skip to the catch clause. With a callback-based API, this might have looked more like:

function appendAndRead(
  path: string,
  data: string
  cb: (error: Error | null, result: string | null) => void
) {
  appendFile(path, data, error => {
    if (error) {
      return cb(error, null)
    }
    readFile(path, (error, result) => {
      if (error) {
        return cb(error, null)
      }
      cb(null, result)
    })
  })
}

Let’s design a Promise API that lets us do this.

Promise starts from humble beginnings:

class Promise {
}

A new Promise takes a function we call an executor, which the Promise implementation will call with two arguments, a resolve function and a reject function:

type Executor = (
  resolve: Function,
  reject: Function
) => void

class Promise {
  constructor(f: Executor) {}
}

How do resolve and reject work? Let’s demonstrate it by thinking about how we would manually wrap a callback-based NodeJS API like fs.readFile in a Promise-based API. We use NodeJS’s built-in fs.readFile API like this:

import {readFile} from 'fs'

readFile(path, (error, result) => {
  // ...
})

Wrapping that API in our Promise implementation, it now looks like this:

import {readFile} from 'fs'

function readFilePromise(path: string): Promise<string> {
  return new Promise((resolve, reject) => {
    readFile(path, (error, result) => {
      if (error) {
        reject(error)
      } else {
        resolve(result)
      }
    })
  })
}

So, the type of resolve’s parameter depends on which specific API we’re using (in this case, its parameter’s type would be whatever result’s type is), and the type of reject’s parameter is always some type of Error. Back to our implementation, let’s update our code by replacing our unsafe Function types with more specific types:

type Executor<T, E extends Error> = (
  resolve: (result: T) => void,
  reject: (error: E) => void
) => void
// ...

Because we want to be able to get a sense for what type a Promise will resolve to just by looking at the Promise (for example, Promise<number> represents an asynchronous task that results in a number), we’ll make Promise generic, and pass its type parameters down to the Executor type in its constructor:

// ...
class Promise<T, E extends Error> {
  constructor(f: Executor<T, E>) {}
}

So far, so good. We defined Promise’s constructor API and understand what the types at play are. Now, let’s think about chaining—what are the operations we want to expose to run a sequence of Promises, propagate their results, and catch their exceptions? If you look back to the initial code example at the start of this section, that’s what then and catch are for. Let’s add them to our Promise type:

// ...
class Promise<T, E extends Error> {
  constructor(f: Executor<T, E>) {}
  then<U, F extends Error>(g: (result: T) => Promise<U, F>): Promise<U, F>
  catch<U, F extends Error>(g: (error: E) => Promise<U, F>): Promise<U, F>
}

then and catch are two ways to sequence Promises: then maps a successful result of a Promise to a new Promise,² and catch recovers from a rejection by mapping an error to a new Promise.

Using then looks like this:

let a: () => Promise<string, TypeError> = // ...
let b: (s: string) => Promise<number, never> = // ...
let c: () => Promise<boolean, RangeError> = // ...

a()
  .then(b)
  .catch(e => c()) // b won't error, so this is if a errors
  .then(result => console.info('Done', result))
  .catch(e => console.error('Error', e))

Because the type of b’s second type argument is never (meaning b will never throw an error), the first catch clause will only get called if a errors. But notice that when we use a Promise, we don’t have to care about the fact that a might throw but b won’t—if a succeeds then we map the Promise to b, and otherwise we jump to the first catch clause and map the Promise to c. If c succeeds then we log Done, and if it rejects then we catch again. This mimics how regular old try/catch statements work, and does for asynchronous tasks what try/catch does for synchronous ones (see Figure 8-2).

Figure 8-2. The Promise state machine

We also have to handle the case of Promises that throw actual exceptions (as in, throw Error('foo')). When we implement then and catch, we’ll do this by wrapping code in try/catches and rejecting in the catch clause. This does have a few implications, though. It means that:

1. Every Promise has the potential to reject, and we can’t statically check for this (because TypeScript doesn’t support indicating in a function’s signature which exceptions the function might throw).

2. A Promise won’t always be rejected with an Error. Because TypeScript has no choice but to inherit JavaScript’s behavior, and in JavaScript when you throw you can throw anything—a string, a function, an array, a Promise, and not necessarily an Error—we can’t assume that a rejection will be a subtype of Error. It’s unfortunate, but this is a sacrifice we’ll make in the name of not having to force consumers to try/catch every promise chain (which might be spread across multiple files or modules!).

Taking that into account, let’s loosen our Promise type a bit by not typing errors:

type Executor<T> = (
  resolve: (result: T) => void,
  reject: (error: unknown) => void
) => void

class Promise<T> {
  constructor(f: Executor<T>) {}
  then<U>(g: (result: T) => Promise<U>): Promise<U> {
    // ...
  }
  catch<U>(g: (error: unknown) => Promise<U>): Promise<U> {
    // ...
  }
}

We now have a fully baked Promise interface.

I’ll leave it as an exercise for you to hook it all together with implementations for then and catch. The implementation for Promise is notoriously tricky to write correctly—if you’re ambitious and have a couple of hours free, head over to the ES2015 specification for a walkthrough of how Promise’s state machine should work under the hood.

async and await

Promises are a really powerful abstraction for working with asynchronous code. They’re such a popular pattern that they even have their own JavaScript (and therefore, TypeScript) syntax: async and await. This syntax lets you interact with asynchronous operations the same way you do with synchronous ones.

Let’s say you have the following promise (we didn’t cover finally in the previous section, but it behaves the way you think it would, firing after both then and catch have a chance to fire):

function getUser() {
  getUserID(18)
    .then(user => getLocation(user))
    .then(location => console.info('got location', location))
    .catch(error => console.error(error))
    .finally(() => console.info('done getting location'))
}

To convert this code to async and await, first put it in an async function, then await the promise’s result:

async function getUser() {
  try {
    let user = await getUserID(18)
    let location = await getLocation(user)
    console.info('got location', user)
  } catch(error) {
    console.error(error)
  } finally {
    console.info('done getting location')
  }
}

Since async and await are JavaScript features, we won’t go into them in depth here—suffice it to say that TypeScript has full support for them, and they are completely typesafe. Use them whenever you work with promises, to make it easier to reason about chained operations and avoid lots of thens. To learn more about async and await, head over to their documentation on MDN.

Async Streams

While promises are fantastic for modeling, sequencing, and composing future values, what if you have multiple values, which will become available at multiple points in the future? This is less exotic than it sounds—think bits of a file being read from the filesystem, pixels of a video streaming over the internet from the Netflix server to your laptop, a bunch of keystrokes as you fill out a form, some friends coming over to your house for a dinner party, or votes being deposited into a ballot box throughout the course of Super Tuesday. While these things may sound pretty different on the surface, you can look at them all as asynchronous streams; they are all lists of things where each thing comes in at some point in the future.

There are a few ways to model this, the most common being with an event emitter (like NodeJS’s EventEmitter) or with a reactive programming library like RxJS.³ The difference between the two is like the difference between callbacks and promises: events are quick and lightweight, while reactive programming libraries are more powerful, and give you the ability to compose and sequence streams of events.

We’ll go over event emitters in the following section. To learn more about reactive programming, head over to the documentation for your favorite reactive programming library—for example, RxJS, MostJS, or xstream.

interface Emitter {

  // Send an event
  emit(channel: string, value: unknown): void

  // Do something when an event is sent
  on(channel: string, f: (value: unknown) => void): void

}

import Redis from 'redis'

// Create a new instance of a Redis client
let client = redis.createClient()

// Listen for a few events emitted by the client
client.on('ready', () => console.info('Client is ready'))
client.on('error', e => console.error('An error occurred!', e))
client.on('reconnecting', params => console.info('Reconnecting...', params))

type RedisClient = {
  on(event: 'ready', f: () => void): void
  on(event: 'error', f: (e: Error) => void): void
  on(event: 'reconnecting',
     f: (params: {attempt: number, delay: number}) => void): void
}

type Events = { 
  ready: void
  error: Error
  reconnecting: {attempt: number, delay: number}
}

type RedisClient = { 
  on<E extends keyof Events>(
    event: E,
    f: (arg: Events[E]) => void
  ): void
}

// ...
type RedisClient = {
  on<E extends keyof Events>(
    event: E,
    f: (arg: Events[E]) => void
  ): void
  emit<E extends keyof Events>(
    event: E,
    arg: Events[E]
  ): void
}

Typesafe Multithreading

So far, we’ve been talking about asynchronous programs that you might run on a single CPU thread, a class of programs that most JavaScript and TypeScript programs you’ll write will likely fall into. But sometimes, when doing CPU-intensive tasks, you might opt for true parallelism: the ability to split out work across multiple threads, in order to do it faster or to keep your main thread idle and responsive. In this section, we’ll explore a few patterns for writing safe, parallel programs in the browser and on the server.

{
  "compilerOptions": {
    "lib": ["dom", "es2015"]
  }
}

{
  "compilerOptions": {
    "lib": ["webworker", "es2015"]
  }
}

// MainThread.ts
let worker = new Worker('WorkerScript.js')

// MainThread.ts
let worker = new Worker('WorkerScript.js')

worker.postMessage('some data')

// WorkerScript.ts
onmessage = e => {
  console.log(e.data) // Logs out 'some data'
}

// MainThread.ts
let worker = new Worker('WorkerScript.js')
worker.onmessage = e => {
  console.log(e.data) // Logs out 'Ack: "some data"'
}
worker.postMessage('some data')

// WorkerScript.ts
onmessage = e => {
  console.log(e.data) // Logs out 'some data'
  postMessage(Ack: "${e.data}")
}

// MainThread.ts
type Message = string
type ThreadID = number
type UserID = number
type Participants = UserID[]

type Commands = {
  sendMessageToThread: [ThreadID, Message]
  createThread: [Participants]
  addUserToThread: [ThreadID, UserID]
  removeUserFromThread: [ThreadID, UserID]
}

type Events = {
  receivedMessage: [ThreadID, UserID, Message]
  createdThread: [ThreadID, Participants]
  addedUserToThread: [ThreadID, UserID]
  removedUserFromThread: [ThreadID, UserID]
}

// WorkerScript.ts
type Command = 
  | {type: 'sendMessageToThread', data: [ThreadID, Message]} 
  | {type: 'createThread', data: [Participants]}
  | {type: 'addUserToThread', data: [ThreadID, UserID]}
  | {type: 'removeUserFromThread', data: [ThreadID, UserID]}

onmessage = e => 
  processCommandFromMainThread(e.data)

function processCommandFromMainThread( 
  command: Command
) {
  switch (command.type) { 
    case 'sendMessageToThread':
      let [threadID, message] = command.data
      console.log(message)
    // ...
  }
}

import EventEmitter from 'events'

class SafeEmitter<
  Events extends Record<PropertyKey, unknown[]>  
> {
  private emitter = new EventEmitter  
  emit<K extends keyof Events>(  
    channel: K,
    ...data: Events[K]
  ) {
    return this.emitter.emit(channel, ...data)
  }
  on<K extends keyof Events>(  
    channel: K,
    listener: (...data: Events[K]) => void
  ) {
    return this.emitter.on(channel, listener)
  }
}

// WorkerScript.ts
type Commands = {
  sendMessageToThread: [ThreadID, Message]
  createThread: [Participants]
  addUserToThread: [ThreadID, UserID]
  removeUserFromThread: [ThreadID, UserID]
}

type Events = {
  receivedMessage: [ThreadID, UserID, Message]
  createdThread: [ThreadID, Participants]
  addedUserToThread: [ThreadID, UserID]
  removedUserFromThread: [ThreadID, UserID]
}

// Listen for events coming from the main thread
let commandEmitter = new SafeEmitter	<Commands>()

// Emit events back to the main thread
let eventEmitter = new SafeEmitter	<Events>()

// Wrap incoming commands from the main thread
// using our typesafe event emitter
onmessage = command =>
  commandEmitter.emit(
    command.data.type,
    ...command.data.data
  )

// Listen for events issued by the worker, and send them to the main thread
eventEmitter.on('receivedMessage', data =>
  postMessage({type: 'receivedMessage', data})
)
eventEmitter.on('createdThread', data =>
  postMessage({type: 'createdThread', data})
)
// etc.

// Respond to a sendMessageToThread command from the main thread
commandEmitter.on('sendMessageToThread', (threadID, message) =>
  console.log(OK, I will send a message to threadID ${threadID})
)

// Send an event back to the main thread
eventEmitter.emit('createdThread', 123, [456, 789])

// MainThread.ts
type Commands = {
  sendMessageToThread: [ThreadID, Message]
  createThread: [Participants]
  addUserToThread: [ThreadID, UserID]
  removeUserFromThread: [ThreadID, UserID]
}

type Events = {
  receivedMessage: [ThreadID, UserID, Message]
  createdThread: [ThreadID, Participants]
  addedUserToThread: [ThreadID, UserID]
  removedUserFromThread: [ThreadID, UserID]
}

let commandEmitter = new SafeEmitter	<Commands>()
let eventEmitter = new SafeEmitter	<Events>()

let worker = new Worker('WorkerScript.js')

// Listen for events coming from our worker,
// and re-emit them using our typesafe event emitter
worker.onmessage = event =>
  eventEmitter.emit(
    event.data.type,
    ...event.data.data
  )

// Listen for commands issues by this thread, and send them to our worker
commandEmitter.on('sendMessageToThread', data =>
  worker.postMessage({type: 'sendMessageToThread', data})
)
commandEmitter.on('createThread', data =>
  worker.postMessage({type: 'createThread', data})
)
// etc.

// Do something when the worker tells us a new thread was created
eventEmitter.on('createdThread', (threadID, participants) =>
  console.log('Created a new chat thread!', threadID, participants)
)

// Send a command to our worker
commandEmitter.emit('createThread', [123, 456])

type Matrix = number[][]

type MatrixProtocol = {
  determinant: {
    in: [Matrix]
    out: number
  }
  'dot-product': {
    in: [Matrix, Matrix]
    out: Matrix
  }
  invert: {
    in: [Matrix]
    out: Matrix
  }
}

type Protocol = { 
  [command: string]: {
    in: unknown[]
    out: unknown
  }
}

function createProtocol<P extends Protocol>(script: string) { 
 return <K extends keyof P>(command: K) => 
   (...args: P[K]['in']) => 
    new Promise<P[K]['out']>((resolve, reject) => { 
      let worker = new Worker(script)
      worker.onerror = reject
      worker.onmessage = event => resolve(event.data.data)
      worker.postMessage({command, args})
    })
}

let runWithMatrixProtocol = createProtocol<MatrixProtocol>(
  'MatrixWorkerScript.js'
)
let parallelDeterminant = runWithMatrixProtocol('determinant')

parallelDeterminant([[1, 2], [3, 4]])
  .then(determinant =>
    console.log(determinant) // -2
  )

npm install @types/node --save-dev

// MainThread.ts
import {fork} from 'child_process'

let child = fork('./ChildThread.js') 

child.on('message', data => 
  console.info('Child process sent a message', data)
)

child.send({type: 'syn', data: [3]}) 

// ChildThread.ts
process.on('message', data => 
  console.info('Parent process sent a message', data)
)

process.send({type: 'ack', data: [3]}) 

Summary

In this chapter we started with the basics of JavaScript’s event loop, and continued on to a discussion of the building blocks of asynchronous code in JavaScript and how to safely express them in TypeScript: callbacks, promises, async/await, and event emitters. We then covered multithreading, exploring passing messages between threads (in the browser and on the server) and building full protocols for communicating between threads.

As with Chapter 7, which technique you use is up to you:

• For simple asynchronous tasks, callbacks are as straightforward as it gets.

• For more complex tasks that need to be sequenced and parallelized, promises and async/await are your friend.

• When a promise doesn’t cut it (e.g., if you’re firing an event multiple times), reach for event emitters or a reactive streams library like RxJS.

• To extend these techniques to multiple threads, use event emitters, typesafe protocols, or typesafe APIs (see “Typesafe APIs”).

Exercises

1. Implement a general-purpose promisify function, which takes any function that takes exactly one argument and a callback and wraps it in a function that returns a promise. When you’re done, you should be able to use promisify like this (install type declarations for NodeJS first, with npm install @types/node --save-dev):

   import {readFile} from 'fs'
   
   let readFilePromise = promisify(readFile)
   readFilePromise('./myfile.ts')
     .then(result => console.log('success reading file', result.toString()))
     .catch(error => console.error('error reading file', error))

2. In the section on “Typesafe protocols” we derived one half of a protocol for typesafe matrix math. Given this half of the protocol that runs in the main thread, implement the other half that runs in a Web Worker thread.

3. Use a mapped type (as in “In the Browser: With Web Workers”) to implement a typesafe message-passing protocol for NodeJS’s child_process.