Turn Your Rockets into Washing Machines

May 20, 2021

There are plenty of ways to build reactive interfaces, but my favorite is to use finite state machines. I've written before on how JavaScript Promises are finite state machines, but I only briefly touched on how finite state machines communicate between each other. With communication comes the potential for race conditions and race conditions bring sadness.

There are different channels for this communication. In the browser we have: event listeners, the DOM, message channels, callbacks (including promises), methods, shared memory, etc.

What's the difference between a rocket and a washing machine? For our purposes, the difference is how long they last. A rocket is created, performs it's task and (hopefully) burns up on reentry. A washing machine is created, and lasts for a long time, performing its task many times over and over again.

To illustrate a race condition we'll implement a search box using various rocket and washing machine styles. Each style is available from this search box where you can switch between them and inspect how they behave. The full code is on GitHub.

A Broken Search Box

Let's look at some faulty code to implement a search box.

const {
	form, results
} = elements();
	form.addEventListener('submit', async (e) => {
	e.preventDefault();
	
	// Clear previous search results and show the spinner
	clear_and_spin();
	
	// Fetch search results
	const args = new URLSearchParams(new FormData(e.target));
	const response = await fetch('https://www.googleapis.com/customsearch/v1?' + args.toString());
	const json = await response.json();
	const { items } = json;
	
	// Display results and hide spinner
	for (const { htmlTitle, link, snippet } of items ?? []) {
		results.insertAdjacentHTML('beforeend', `<li>
			<a href="${link}">${htmlTitle}</a>
			<p>${snippet}</p>
		</li>`);
	}
	hide_spinner();
});

An immediate red flag is the async event listener. While it doesn't mean for certain that a race condition exists, it does indicate that you should slow down to verify that the code works. In this case it doesn't work and can be tripped up with this sequence of events:

The user submits a search for "How do draw?": the results element is cleared and the spinner is shown.
The user edits their query and submits a search for “How to draw?”: the results element is already clear, and the spinner is already shown so nothing changes even though they are run again.
The results for the second query come back and are written to the page. The spinner is hidden. These are the results for the most recent search and are what the user cares about.
Some time later, the results to the first query come back and get appended to those from the second query. The spinner is already hidden so it doesn't change.

This would be confusing to the user because the spinner was hidden while the results for the search they don't care about continued to load. And then, those results randomly append themselves at the end. (It would have been even more broken if we had tried to toggle the spinner. For this reason I avoid toggle based apis.)

Promises act like rockets in that they are self propelling. Once you launch a Promise or rocket, there's no taking it back. The best you can do is to ask the rocket to abort itself — hopefully it is able to listen.

I repeat, there is no way to kill a Promise externally. The abort controller and abort signal are a request to have the Promise terminate itself. This is in contrast to generators in JavaScript or Futures in other languages where the task won't progress unless being actively driven from the outside.

The chance of encountering a weirdly delayed search request is small, especially if you're testing from a computer with good internet, so you might only see this kind of error in production where internet conditions are more diverse and users interact quickly with your interface.

To increase the chances of hitting this race condition, I wrapped the global fetch with a function that randomly delays ~50% of requests.

// delay ~50% of requests by 4 seconds
const original_fetch = fetch;
window.fetch = async function () {
	const chance = Math.random();
	if (chance >= 0.5) {
		console.log("slowing this request: ", chance);
		await new Promise(res => setTimeout(res, 4_000));
	}
	return await original_fetch(...arguments);
};

Next, let's add an abort system to fix our faulty search box implementation.

Launch Abort System

It's not too difficult to wrap our async function with a synchronous function which will provide it with an abort signal.

function single(async_func) {
	let controller;

	return (...args) => {
		if (controller) {
			controller.abort();
		}
		controller = new AbortController();
		const signal = controller.signal;
		async_func(signal, ...args).finally(() => {
			// If our promise settles without being interrupted (the controller's signal is the same as the one we gave to the async_func) then delete the controller because it doesn't need to be cancelled next time around.
			if (signal == controller.signal) controller = false;
		});
	};
}

form.addEventListener('submit', single(async (signal, e) => {
	e.preventDefault();
	
	// Clear previous search results and show the spinner
	clear_and_spin();
	
	// Fetch search results
	const args = new URLSearchParams(new FormData(e.target));
	const response = await fetch(
		'https://www.googleapis.com/customsearch/v1?' + args.toString(),
		{ signal }
	);
	
	const json = await response.json();
	if (signal.aborted) return;
	const { items } = json;
	
	// Display results and hide spinner
	for (const { htmlTitle, link, snippet } of items ?? []) {
		results.insertAdjacentHTML('beforeend', `<li>
			<a href="${link}">${htmlTitle}</a>
			<p>${snippet}</p>
		</li>`);
	}
	hide_spinner();
}));

We need to make sure that we supply our signal to every Promise that we await or that we check if we were aborted after we await a promise that doesn't accept an abort signal. Since reading the response as JSON doesn't take the signal we need to check for abortion after reading it. Alternatively, you can make a promise abortable using a wrapper function.

This isn't a perfect solution. If you click the search button twice without changing your query this solution throws out the previous (and still valid) request and starts over. We should check if the SearchParams are the same as the previous request and only abort if they are different. Our final implementations will do this.

Locks

If you squint real hard, you might see that our single function looks a bit like a mutex or other lock. Its purpose is to ensure that only one search task is in flight at a time.

JavaScript is ~single threaded~ so in this case, the critical sections that our lock protects is anything in between an await keyword. Since we had two await keywords, we needed to check if the signal had been aborted twice. Once was handled by fetch automatically because it throws an error on abort. The other we used a pair of if and return statements. Looking at the critical sections, it's clear why our faulty example could have the results for query 1 followed by the results for query 2 or the results for query 2 followed by the results for query 1. What's not possible would be a few of 1 and a few of 2 and then a few of 1 again. Because JS is ~single threaded~ the results will never be interleaved because there is no await keyword inside the for loop.

JavaScript has other locks we can use. For example, we also could have disabled the search button so that the user couldn't click it until we'd finished getting the search results. That's a lock.

Being ~single threaded~ means that a lock can be as simple as a single boolean variable.

Aside: sync ⇄ async interfaces

But JavaScript isn't always single threaded now that we have workers. And while we are not allowed to block the main thread, we are allowed to block worker threads. Sure, the JavaScript runtime might block the main thread if the event queue is empty and it has no work to do, but that means that the JavaScript stack will be empty. But with SharedArrayBuffer and Atomics, we can block a JavaScript thread without emptying the JavaScript stack. Think of it like turning any normal JavaScript function into a generator and using yield.

One key use for this is implementing a synchronous WASM interface over an asynchronous JavaScript API. For example, the seek and write methods on the FileSystemWritableFileStream return promises, while Rust's Write and Seek traits are both synchronous. If we put our Rust code in a worker thread with shared memory, then we could transfer the promise we get from calling stream_handle.write() back out to the main thread and use Atomics.wait() to block the worker. When the promise resolves, we can then Atomics.notify() the worker to have it continue. This is one way to deal with this difficult situation.

It's better and more versatile to turn the synchronous interface into an async one since you wouldn't need a worker. If you're using Emscripten, then you can use Asyncify to do exactly that. The worker trick is just for situations where you can't change the sync-ness of the interface you're implementing.

Now, back to inter-machine communication.

Queues

Imagine a webpage that lets users upload multiple files. They select a file, then initiate uploading. While it uploads, they select another file and click upload. Our server can only handle a user uploading a single file at a time, so our web page should queue the second file to be uploaded later. Writing a queue can be pretty easy. Here's one:

export function queue() {
	let items = [];
	let waiting = [];
	return {
		append(item) {
			items.push(item);
			const w = waiting;
			waiting = [];
			w.forEach(c => c());
		},
		async *[Symbol.asyncIterator]() {
			while (true) {
				if (items.length) {
					yield items.shift();
				} else {
					await this;
				}
			}
		},
		then(callback) {
			waiting.push(callback);
		}
	};
}

And using that queue to implement our upload example might look like this:

const upload_queue = queue();

// File drop:
drop_zone.ondrop = function (e) {
	e.preventDefault();  upload_queue.append(...(e.dataTransfer.items ?? []));
};

// Uploader:
(async () => {
	for await (const item of upload_queue) {
		if (item.kind === 'file') {
			const file = item.getAsFile();
			await fetch(/* Upload the file */);
		}
	}
})();

Notice that the uploader is a self-propelling promise, but it is instantiated during initial script execution and it performs its task more than once which makes it a washing machine. You could imagine having two upload servers in which case it would be very easy to just instantiate two of these uploaders instead of one.

It doesn't make sense here, but you could conceivably do pipeline analysis on this code. In circuit design, pipelines are about balancing throughput and speed, but in software, pipelines can be about managing finite resources. Having a queue that fills when a system is overwhelmed can result in a more predictable user experience than having a system that tries to handle every request immediately, but ends up slowing down all requests and becoming ineffective.

You can also use pipelines to represent the finite resources of a real world system. If a task requires an employee's action, then it makes sense to limit the number of machines to the number of employees you have.

In any case, whether you're instantiating a state machine per task or instantiating a few long running state machines, the purpose of using locks, channels, or queues, is the same: to constrain and coordinate your system's critical sections. The fewer state machines and critical sections that you have to worry about, the easier that will be. That's why I recommend using a few long lived washing machines over many short lived rockets.

Let's get back to the search box. I'll show you two examples of how to implement a search box using a single state machine. The first is a single instantiated async function — like our uploader example — and the second uses an experimental state machine library I made. Both examples share the same state diagram:

Search Box Washing Machine

This implementation is a bit longer than the singleton version, but I think it's fairly readable.

import { elements, clear_and_spin, hide_spinner } from './base.mjs';
const {
	form, results
} = elements();function get_args(old_args) {
	return new Promise(res => {
		function handler(e) {
			e.preventDefault();
			
			const new_args = new URLSearchParams(new FormData(e.target));
			if (!old_args || new_args.get('q') !== old_args.get('q')) {
				form.removeEventListener('submit', handler);
				res(new_args);
			}
		}
		form.addEventListener('submit', handler);
	});
}

// Search Box:
(async () => {
	while (true) {
		// STATE: wait_for_search; TRANSITIONS: query
		let args = await get_args();
		
		clear_and_spin();
		
		let search_results;
		let controller;
		while (true) {
			if (controller) controller.abort();
			controller = new AbortController();
			
			let response;
			const t_response = fetch(
				'https://www.googleapis.com/customsearch/v1?' + args.toString(),
				{ signal: controller.signal }
			).then(res => response = res);
			
			const t_query = get_args(args).then(new_args => args = new_args);
			
			// STATE: searching.1; TRANSITIONS: query, response
			await Promise.race([t_query, t_response]);
			
			if (!response) continue;
			
			let items;
			const t_json = response.json().then(json => items = json?.items ?? []);
			
			// STATE: searching.2; TRANSITIONS: query, json
			await Promise.race([t_query, t_json]);      if (items) {
				search_results = items;
				break;
			}
		}
		for (const { htmlTitle, link, snippet } of search_results ?? []) {
			results.insertAdjacentHTML('beforeend', `<li>
								<a href="${link}">${htmlTitle}</a>
								<p>${snippet}</p>
						</li>`);
		}    hide_spinner();
	}
})();

One of the problems with building state machines this way is how difficult it is to specify transitions and states. Each state with more than one possible transition ends up being a Promise.race() of a few promises immediately followed by conditionals to check which transition was taken. We use the .then() to start the transition and extract the data which identifies which transition ran.

The next solution is just for fun. I wanted to see what a syntax for describing state machines might look like.

Re-Entrant FSM Library

The library uses exceptions to “suspend” the function at calls to state().

import { machine, state, transition } from './machine.mjs';
import { elements, clear_and_spin, hide_spinner } from './base.mjs';
const {
	form, results
} = elements();machine(function () {
	form.addEventListener('submit',
		transition('query', e => {
			e.preventDefault();

			const new_args = new URLSearchParams(new FormData(e.target));
			if (!this.args || new_args.get('q') !== this.args.get('q')) {
				this.args = new_args;
				if (this.controller) {
					this.controller.abort();
				}
			}
		}),
		{ once: true }
	);

	// Get the search query
	if (!this.args) {
		state('waiting_for_search');
	}  if (!this.controller || this.controller.signal.aborted) {
		// Send the request
		this.controller = new AbortController();
		fetch(
			'https://www.googleapis.com/customsearch/v1?' + this.args.toString()
		).then(
			transition('response', response => this.response = response)
		);
	} else if (this.response) {
		// Turn the request into json and display the results
		this.response.json().then(
			transition('json', ({ items }) => {
				for (const { htmlTitle, link, snippet } of items ?? []) {
					results.insertAdjacentHTML('beforeend', `<li>
						<a href="${link}">${htmlTitle}</a>
						<p>${snippet}</p>
					</li>`);
				}
				this.args = undefined;
				this.controller = undefined;
				this.response = undefined;
			})
		);
	}
	state('searching', clear_and_spin, hide_spinner);
});

Well, what do you think? Would you consider writing your state machines this way, or do you prefer using a library like xstate where machines are defined via objects?