DEV Community: Zak Henry

Parallel computation in the browser with observable webworkers

Zak Henry — Sun, 18 Aug 2019 22:07:12 +0000

In the previous article we dove deep into the use of webworkers to improve the responsiveness of the main thread by offloading computation into a web worker.

To facilitate the message passing, we used the library observable-webworker to give us an easy to use RxJS stream-based API.

This strategy worked very well for that use case, however we aren't quite maximising the compute resources we have available to us.

Web browsers can spin up multiple parallel web workers that will each run on different threads of the operating system. This means that any computation we do in one worker will not slow down the computation of another worker. There is a limit to this however, as not only does the computer itself have a parallelisation limit (logical processor core count) browsers will self-limit the amount of concurrency web pages are allowed to use.

The amount of parallel threads that can be run can be determined by navigator.hardwareConcurrency. This number varies depending on both underlying hardware and browser implementation. For example most desktop browser give the core count, some mobile ones will limit this value, and some browsers (Safari) won't tell you at all.

To work out how many workers we can run in parallel, simply do const workerCount = navigator.hardwareConcurrency - 1. The - 1 here is to keep a processor core spared for the main ui thread.

Put simply, with the worker pool strategy, we spin up as many workers as we are allowed, and construct a pool of work. Each worker picks up a task from the pool, executes the computation in on it's own core, then returns the results to the main thread, grabbing the next task or shutting down if there is no more work to be done.

To better explain this concept and how to implement it, we will build a simple application that will take a list of files (using an <input type="file" multiple /> element) and return the MD5 hash sum of those files to the main thread. This basic concept can be expanded to many different types of application, and is in itself a fairly useful application as you might want to use this strategy for hashing multiple files for upload to AWS S3 for instance.

Let's start off by outlining exactly what we're trying to achieve. I'll use a little diagram syntax that is loosely based on the RxJS marble testing syntax.

- idle/waiting time
^ transferring file to worker thread
* processing file
| hash computed, returning results

Take the following diagram:

file-1.txt  ----^*******|

With the key above, we can see that we're showing that for file-1.txt there was some idle time, then the file was picked up, processing ran for some time then the result was returned.

We can use this syntax to show what would happen if we were to compute the hashes for the files one at a time:

file-1.txt  ^*******|
file-2.txt  ---------^***|
file-3.txt  --------------^**********|
file-4.txt  --------------------------^********|
file-5.txt  ------------------------------------^*********|
file-6.txt  -----------------------------------------------^****|

results     --------1----2------------3--------4----------5-----6

I've simulated the files being different sizes/taking different time to hash by extending the processing time (*) randomly.
The last line shows the results as they come back to the main thread, with the number just being the file number.

So from this diagram we can see that the total time to get the hash results of all 6 files is the individual durations of each file processing all added up. With a worker pool we can of course take advantage of parallel computing and run a number of these processes in parallel:

file-1.txt  ^*******|               # worker-1
file-2.txt  ^***|                   # worker-2
file-3.txt  ^**********|            # worker-3
file-4.txt  -----^********|         # worker-2
file-5.txt  ---------^*********|    # worker-1
file-6.txt  ------------^****|      # worker-3

results     ----2---1---3-4--6-5

In this diagram I've taken the exact same processing durations from above, but shifted the processing times assuming a parallelism count of three. You can see the dramatic effect this has on the overall processing time as once the first file completes (file-2.txt) the next file (file-4.txt) is immediately picked up for processing. You can also see that even while file-3.txt is still occupying one thread, the other two threads are able to complete their unit of work and pick up a new job. One thing to note for later is that the results are returned as soon as possible, which in this case means out of order. We will need to correlate each unit of work with it's result to marry up the data correctly.

Another way to look at the above scenario is with the same diagram syntax, but considering each worker rather than each file:

worker-1    ^*******|^*********|    # file-1.txt, file-5.txt
worker-2    ^***|^********|         # file-2.txt, file-4.txt
worker-3    ^**********|^****|      # file-3.txt, file-6.txt

results     ----2---1--3--4-6-5

The important thing to note here is that the workers are never idle - as soon as they're finished processing work they pick up another file for processing.

Now in our example we're only using three workers, but let's say the client machine has 8 logical cores and the browser lets you use them, the diagram would look like the following

file-1.txt  ^*******|               # worker-1
file-2.txt  ^***|                   # worker-2
file-3.txt  ^**********|            # worker-3
file-4.txt  ^********|              # worker-4
file-5.txt  ^*********|             # worker-5
file-6.txt  ^****|                  # worker-6

results     ----26--1453

Not so interesting, but we can see that the overall time took only as long as the longest file (file-3.txt) and also that we only needed six of the seven available workers.

Okay, we've talked through the theory, let's get practical!

I'm going to use Angular here as it's what I'm most familiar with, but the two libraries that we'll be using (observable-webworker and js-md5) are both framework agnostic, so if you're following along feel free to use the frontend framework you like best.

Let's start with a quick component that can load some files from the file system and print them to the console.

Template: files-hash.component.html

<input type="file" multiple (change)="hashFiles($event)" />

Component: files-hash.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'app-files-hash',
  templateUrl: './files-hash.component.html',
  styleUrls: ['./files-hash.component.scss'],
})
export class FilesHashComponent {
  public hashFiles($event): void {
    const files: File[] = Array.from($event.target.files);
    console.log(files);
  }
}

Running this we will see a single file select and when we select a few files we see them logged in the console in an array. Easy.

Next let's create a webworker that will take a file as argument and return the hash of that file. We will not just return the hash string alone as we will need to correlate which hash corresponds to which file (remember from earlier we will get the results out of order).

First though, we define an interface to manage passing back this correlation of filename and hash:

file-hasher.interface.ts

export interface FileHashPayload {
  filename: string;
  hash: string;
}

Nothing exciting here, just an interface to hold the filename and the hash.

Now for the worker. We will need both observable-webworker and js-md5 to handle the workers and the hashing respectively.

yarn add -E observable-webworker js-md5

Worker: file-hasher.worker.ts

import * as md5 from 'js-md5';
import { DoWorkUnit, ObservableWorker } from 'observable-webworker';
import { Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { FileHashPayload } from './file-hasher.interface';

@ObservableWorker()
export class WorkerPoolHashWorker implements DoWorkUnit<File, FileHashPayload> {
  public workUnit(file: File): Observable<FileHashPayload> {
    return this.readFileAsArrayBuffer(file).pipe(
      map(arrayBuffer => ({ filename: file.name, hash: md5(arrayBuffer) })),
    );
  }

  private readFileAsArrayBuffer(blob: Blob): Observable<ArrayBuffer> {
    return new Observable(observer => {
      if (!(blob instanceof Blob)) {
        observer.error(
          new Error('`blob` must be an instance of File or Blob.'),
        );
        return;
      }

      const reader = new FileReader();

      reader.onerror = err => observer.error(err);
      reader.onload = () => observer.next(reader.result as ArrayBuffer);
      reader.onloadend = () => observer.complete();

      reader.readAsArrayBuffer(blob);

      return () => reader.abort();
    });
  }
}

Ok there's a few things going on here, so let's break them down.

@ObservableWorker()
export class WorkerPoolHashWorker implements DoWorkUnit<File, FileHashPayload> {

In the class header we decorate it with @ObservableWorker() to register the class as a worker so that we can communicate with it from the main thread.

We also implement DoWorkUnit<File, FileHashPayload> which means that we need to implement a public method that takes a file and returns an observable of FileHashPayload.

public workUnit(file: File): Observable<FileHashPayload> {
  return this.readFileAsArrayBuffer(file).pipe(
    map(arrayBuffer => ({ filename: file.name, hash: md5(arrayBuffer) })),
  );
}

This method is pretty straightforward - we read the file as an array buffer (see the private method defined next for the implementation), then map the resulting array buffer to the FileHashPayload we defined earlier with the hash computed synchronously using the js-md5 library.

Note that we're returning a completing observable here. This is critical to the functioning of the observable-webworker library when managing thread pools. It uses the completion notification to determine that the worker has completed that unit of work and is ready for another unit of work. We return observable as it allows us to return multiple events from the worker, which is useful for outputting progress events from long running processes.

Now back to the main thread to actually implement the worker pool.

import { Component } from '@angular/core';
import { fromWorkerPool } from 'observable-webworker';
import { FileHashPayload } from './file-hasher.interface';

@Component({
  selector: 'app-files-hash',
  templateUrl: './files-hash.component.html',
  styleUrls: ['./files-hash.component.scss'],
})
export class FilesHashComponent {
  public hashFiles($event): void {
    const files: File[] = Array.from($event.target.files);
    console.log(`files`, files);
    fromWorkerPool<File, FileHashPayload>(
      () => new Worker(`./file-hasher.worker.ts`, { type: 'module' }),
      files,
    ).subscribe((hashPayload: FileHashPayload) => {
      console.log('Hashed file', hashPayload.filename, hashPayload.hash);
    });
  }
}

This is quite similar to what we were doing in earlier articles with fromWorker() but this time we use fromWorkerPool() from observable-webworker. Like fromWorker the arguments are a factory to create a worker, and the input itself.

In this example we've used an Array<File> as input, however the fromWorkerPool method also takes an Observable<T> or an Iterator<T>.

Iterators can be very useful with this pattern as you can use their lazy evaluation feature to know when the work has been picked up for processing, not just that it has been queued.

Observable can be useful as you're probably using observable streams anyway, and fromWorkerPool buffers the stream when the input rate exceeds the rate the worker pool can handle the work.

If we now run our application defined above, and select some filesm, we will see something like the following:

files (10) [File, File, File, File, File, File, File, File, File, File]
Hashed file 1-monospondylic-Notelaea-140MB.txt 8b3e48ad838ba4b3024d42fa10591c82
Hashed file 2-safekeeper-unheedful-307MB.txt d1bef08d19b30f471f161c8f5fbf9a8a
Hashed file 4-pseudochromesthesia-laryngeal-162MB.txt d77b94eea8ad01c4e4b804fe9ecd26a2
Hashed file 3-gnatcatcher-incudes-361MB.txt 30c6b4c2ea8b1dd9e5222e8ab9f2119d
Hashed file 5-amacrine-Nance-188MB.txt 7be20804dde038994a2c0e6630643046
Hashed file 8-ankaratrite-dermatoheteroplasty-149MB.txt 72cb1a511a27ac023b9960dbc3959c40
Hashed file 7-enjambed-escutcheon-256MB.txt 1f5039e50bd66b290c56684d8550c6c2
Hashed file 6-idiograph-freckly-393MB.txt bcee52113567c0040a0db2c678dfe3c3
Hashed file 9-strandage-barrelage-179MB.txt 9eab43b111f3c7ff67536d031d06f69b
Hashed file 10-refinish-mellowy-378MB.txt 212374c37a433a7b06105090002297d0

To test I've generated a bunch of random files with the file size in their filename.

From the output we can see that we've achieved the expected outcome that we theorised earlier - the files are appearing in the logs, out of order and the files that are particularly large are ordered towards the end, as we would expect. Success!

Don't just take my word for it though - I've created a far more interactive demo at https://cloudnc.github.io/observable-webworker which uses basically the exact same strategy outlined here, just with a bunch more progress messages to build a pretty timeline:

To try it out, jump to the "Multiple Worker Pool" section and select some files. Don't worry - no files are sent out of the browser, feel free to check the source to verify or generate some random files like I did. Once you've selected the files, you'll see a timeline graph of the files being processed with whatever concurrency your machine can handle.

If you inspect the sources tab in your browser devtools you will see multiple workers being created to handle the workload, and they will be shut down as soon as there is no further work in the pool.

Additionally, if you select less files than you have concurrency available, you'll only spin up as many workers as files you've selected.

That's all for this article, I hope you've enjoyed it! Next up in the series we will be digging in to the performance penalties of transferring large objects to webworkers and a strategy to mitigate this cost.

The source for this article & demo is available at https://github.com/zakhenry/blog-posts/tree/master/posts/observable-workers-parallel-computation-in-the-browser

Photo by Max Vertsanov on Unsplash

TC39 Pipeline proposal comparison - RxJS case study

Zak Henry — Thu, 04 Jul 2019 20:20:27 +0000

As you may or may not be aware there is an exciting new operator possible coming to javascript - the |> pipeline operator. I don't want to re-tell what it is and how it is useful, Ben Lesh has already done an excellent job of this so read his article first. Instead, I'm going to investigate the two competing proposals, using RxJS as a case study as this is a use-case where I'm particularly excited about the new operator.

Forewarning, this article is opinionated! If you have a differing opinion I'd love to have my mind changed as I have swung opinions on this topic a couple of times.

Before we dive into RxJS, let's just recap first the two types and their features/shortcomings.

Minimal/Point-free Style

^{try it!}

const map = fn => arr => arr.map(fn);
const filter = fn => arr => arr.filter(fn);
const reduce = (fn, i) => arr => arr.reduce(fn, i);
const increment = v => v + 1;
const add = a => b => a + b;

const result = [1,2,3]
 |> map(a => a * 2)
 |> filter(a => a > 5)
 |> reduce((sum, a) => a+sum, 0)
 |> increment
 |> add(3)

console.log(result); // 10

In this example you can see that we had to do a bit of preparatory work with our functions to make them point-free style, that is they return a function that takes one argument. Once this prep is done, the pipeline becomes a simple composition of operators. It's worth noting at this point that this isn't really a great example as you should probably in future still be using the Array.prototype.* methods anyway. I just picked an example that should feel familiar.

Topic Style (a.k.a Smart Mix)

^{try it! - (pick "Smart" in the pipeline proposal on the left)}

const increment = a => a + 1;
const add = (a, b) => a + b;

const result = [1,2,3]
 |> #.map(a => a * 2 )
 |> #.filter(a => a > 5)
 |> #.reduce((sum, a) => a+sum, 0)
 |> increment
 |> add(#, 3)

console.log(result); // 10

Topic style takes a different approach by essentially considering the bit where you use the function as a "template" for its actual invocation. Because the argument is not known ahead of time, we use # as a placeholder for that value to be used once it is known.

This strategy does have the advantage that the functions used in the pipeline are simple in their declaration, at the cost that the pipeline itself is slightly more verbose.

Additionally in my personal opinion it feels like additional cognitive load that when I do something like add(#, 3) I need to understand that I'm not really invoking the add function at all when this line of code is reached, rather I am defining a template for its execution.

With this dummy example there isn't really a lot differentiating the two proposals, but I don't think it is really very realistic anyway, so let's get real!

Using RxJS we will define a pipeline which takes user input from a textbox, ignores it when character count is less than two, debounces their keystrokes, then queries an api for search results.

RxJS pre-pipeline

const searchResults$ = fromEvent(document.querySelector('input'), 'input').pipe(
  map(event => event.target.value),
  filter(searchText => searchText.length > 2),
  debounce(300),
  distinctUntilChanged(),
  switchMap(searchText => queryApi(searchText).pipe(retry(3))),
  share(),
)

This simply implements the description given above. RxJS uses the .pipe() method in lieu of something better - Javascript pipeline operator:

RxJS Minimal/Point-free Style


const searchResults$ = fromEvent(document.querySelector('input'), 'input')
  |> map(event => event.target.value)
  |> filter(searchText => searchText.length > 2)
  |> debounce(300)
  |> distinctUntilChanged()
  |> switchMap(searchText => queryApi(searchText) |> retry(3))
  |> share()

This is pretty much exactly the same syntax as before, but much tidier and cleaner to read due to the new |> operator. How nice is that inline operator within the switchMap?

For comparison purposes, let's see how the above might look with the topic-style operator implementation:

RxJS Topic Style


const searchResults$ = fromEvent(document.querySelector('input'), 'input')
  |> map(#, event => event.target.value)
  |> filter(#, searchText => searchText.length > 2)
  |> debounce(#, 300)
  |> distinctUntilChanged
  |> switchMap(#, searchText => queryApi(searchText) |> retry(#, 3))
  |> share

Hmmm. I am not a fan. Maybe it is just a lack of familiarity talking, but the requirement that all operators need to be passing in the observable as a topic feels really unclean. Of course this would require rewriting all the RxJS operators to use this new syntax and take an observable as the first argument.

What do you think? Is there something really compelling about the topic style operator that I am missing? I know much of the discussion has been around how to handle the await keyword, but honestly I rarely use async/await now that RxJS is mostly what I use, but that is just my personal experience.

On a different note, the TC39 proposal feels like it has stagnated a bit, as the Github issues are pretty much all comments from last year. Let's kick start the discussion again and see if we can nudge it one step closer to being used!

^{Photo by Rodion Kutsaev on Unsplash}

Observable Web Workers, a deep dive into a realistic use case

Zak Henry — Sun, 30 Jun 2019 21:20:03 +0000

In the previous article we took a look at the library observable-webworker which lets us use familiar observable patterns to construct and manage Web Workers and the communication between the threads. In this article we will develop an application, first without a web worker, then refactor it to use a web worker to demonstrate the power and usefulness of web workers. If you haven't read the previous article, I do recommend you do so first as it goes over all the prerequisites to getting up and running, and the background of what web workers are, which I will skip over here.

This article is a deep dive for advanced developers, so get yourself a beverage and settle in. I'm assuming moderate to advanced knowledge of Typescript and RxJS in this article, so if you're a beginner this may not be for you I'm afraid. Here's a table of contents so you can come back to this article in stints if you like.

Brief
Data Flow Plan
Algorithm
Application
- Common functionality
- Service
- Component
- Main Thread Test
- Worker
- Worker Thread Test
Performance
Wrap Up

Brief

The application we're going to build has a few requirements, I'll lay these out as acceptance criteria that we can refer to later.

As a user I want to be able to search through text of a variety of novels fetched from Project Gutenberg. My search results should contain my search phrase highlighted within the context of the matching paragraph. When I typo or misspell words I should still get expected results.

For fun I'll detail the criteria in the BDD Gherkin syntax



Feature: Find phrases in books by searching for them

    Background:
        Given I'm a user on the home page
        And I have selected the book "Alice in Wonderland"
        And I see a free text field in which I can type my search phrase

    # basic search
    Scenario
        When I type "we’re all mad here"
        Then I should see the first search result containing the text "‘Oh, you can’t help that,’ said the Cat: ‘we’re all mad here. I’m mad. You’re mad.’"

    # multiple result search
    Scenario
        When I type "mad"
        Then I should see search results with "mad" highlighted at least "10" times

    # search with typos
    Scenario
        When I type "were all madd her"
        Then I should see the first search result containing the text "‘Oh, you can’t help that,’ said the Cat: ‘we’re all mad here. I’m mad. You’re mad.’"

Additional to these user journey requirements, we will have the following performance requirements:

the user interface must remain responsive at all times
the user gets their search results live with every keystroke
the user can change book and the search immediate begins returning results for their previous search

The difficult part to take particular note of in the requirements is the ability to handle typos and misspellings. This explodes the problem complexity as it is not as simple as finding a substring of a paragraph; rather we need to score candidate substrings to find the best match.

Data Flow Plan

To start with we will put together a simple typescript file proving out the observable flow, then later build it into the application.



// playground/observable-flow.ts

import { from, Observable, of } from 'rxjs';
import { scan, shareReplay, switchMap } from 'rxjs/operators';

/**
 * First off, we create a quick enum of the books we will use as a demo. Later
 * we will update the urls to be correct
 */
enum BookChoice {
  ALICE_IN_WONDERLAND = 'http://some-url-to-alice-in-wonderland-text',
  SHERLOCK_HOLMES = 'http://some-url-to-sherlock-holmes-text',
}

/**
 * This observable will be something like a regular dropdown, emitted whenever
 * the user changes their selection.
 */
const userBookSelection$ = of(BookChoice.ALICE_IN_WONDERLAND);

/**
 * This observable represents the stream of search terms the user will enter
 * into a text box. We `shareReplay(1)` so that subsequent subscribers will get
 * the latest value
 */
const userSearchTerm$ = of(`we’re all mad here`).pipe(shareReplay(1));

/**
 * This function will be responsible for fetching the content of the book given
 * the enum. We're cheating a little by making the enum value the url to fetch.
 * For now we will just pretend the url is the content of the book.
 * @todo implement
 */
function getBookText(bookChoice: BookChoice): Observable<string> {
  return of(bookChoice);
}

/**
 * This function will be responsible for taking the search term and returning
 * the stream of paragraphs found, as soon as they are found.
 *
 * For the purposes of quick testing, we've hardcoded as if two search results
 * were found
 *
 * We will also likely extend this return type in future to handle highlighting
 * the search phrase, but for now just the paragraph is sufficient for testing
 * @todo implement
 */
function getSearchResults(
  searchTerm: string,
  bookText: string,
): Observable<string> {
  return from([
    searchTerm + ' (this will be the first result)',
    searchTerm + ' (this will be the second result)',
  ]);
}

/**
 * Here we take the user selected book stream and pipe it via switchMap to fetch
 * the content of the book. We use switchMap() because we want to cancel the
 * download of the book if the user switches to a different book to search
 * before the book has finished downloading.
 *
 * Next we again switchMap() the result of the book content to the user search
 * term observable so that if the user has changed books, once it is loaded we
 * will cancel the processing of the current search term.
 *
 * Next we pass that stream of search terms to getSearchResults() which will
 * itself be returning a stream of search results for that search string.
 *
 * Finally, we use a scan() operator to collate the stream of search results
 * into an array so that we can present all results to the user, not just the
 * most recent one
 */
const searchResults$ = userBookSelection$.pipe(
  switchMap(selection => getBookText(selection)),
  switchMap(bookText => {
    return userSearchTerm$.pipe(
      switchMap(searchTerm =>
        getSearchResults(searchTerm, bookText).pipe(
          scan((searchResults: string[], searchResult) => {
            return [...searchResults, searchResult];
          }, []),
        ),
      ),
    );
  }),
);

/**
 * Last but not least, to check our logic, we subscribe to the observable and
 * bind the console so we see output.
 */
searchResults$.subscribe(console.log);

Alrighty, we've got a concept of what we're trying to build, so what do we get when we run it? I like to use ts-node for these kind of quick tests, so just run npx ts-node --skip-project playground/observable-flow.ts (tip: the --skip-project bit is just because I'm currently in a working directory that has a tsconfig.json file that is not compatible with just running a plain nodejs script.)

Our output of the above file is as follows:



[ 'we’re all mad here (this will be the first result)' ]
[
  'we’re all mad here (this will be the first result)',
  'we’re all mad here (this will be the second result)'
]

Okay, so this is expected - we first get one result in an array, then we get both results in the array. Success? Well kinda, but we'd really like to put some realism into this script so that we can see that our observables are having the right behavior.

So, we will now edit the script we just wrote to add a bunch of realistic delays and some more logging to see what is really going on:



// playground/observable-flow-test.ts

import { from, Observable, of } from 'rxjs';
import {
  concatMap,
  delay,
  map,
  pairwise,
  scan,
  shareReplay,
  startWith,
  switchMap,
  timestamp,
} from 'rxjs/operators';

enum BookChoice {
  ALICE_IN_WONDERLAND = 'http://alice.text',
  SHERLOCK_HOLMES = 'http://sherlock.text',
}

/**
 * This is a nice little custom operator that spaces out observables by a certain amount, this is super handy for
 * emulating user events (humans are slooow!)
 */
function separateEmissions<T>(delayTime: number) {
  return (obs$: Observable<T>): Observable<T> => {
    return obs$.pipe(
      concatMap((v, i) => (i === 0 ? of(v) : of(v).pipe(delay(delayTime)))),
    );
  };
}

/**
 * For the book selection, we've piped to separateEmissions() with 4000ms
 * defined, this means when subscribed the observable will immediately emit
 * Alice in Wonderland content, then 4 seconds later emit Sherlock Holmes content.
 */
const userBookSelection$ = from([
  BookChoice.ALICE_IN_WONDERLAND,
  BookChoice.SHERLOCK_HOLMES,
]).pipe(separateEmissions(4000));

/**
 * Slightly different strategy for this one - we're
 * 1. Piping delayed user book selection to vary the search phrase depending on
 * which book is selected
 * 2. creating a streams of individual characters
 * 3. spacing out the emissions by 100ms (this is the inter-keystroke time)
 * 4. using scan to combine the previous characters
 * The result is a pretty good simulation of the user typing the phrase at 10
 * keys per second
 */
const userSearchTerm$ = userBookSelection$.pipe(
  delay(200),
  switchMap(book => {
    const searchPhrase =
      book === BookChoice.ALICE_IN_WONDERLAND
        ? `we’re all mad here`
        : `nothing more deceptive than an obvious fact`;

    return from(searchPhrase).pipe(
      separateEmissions(100),
      scan((out, char) => out + char, ''),
    );
  }),
  shareReplay(1),
);

/**
 * Here, we're guessing it will take about 200ms to download the book. We've
 * also put in a console.log so we can make sure we're not going to try download
 * the book on every keystroke!
 * @param bookChoice
 */
function getBookText(bookChoice: BookChoice): Observable<string> {
  console.log(`getBookText called (${bookChoice})`);
  return of(bookChoice).pipe(delay(200));
}

/**
 * With this function we're saying that the search takes (20 milliseconds * the
 * length of the search string)
 * This is actually totally unrealistic, but the linear variability will help
 * when understanding the logs
 */
function getSearchResults(
  searchTerm: string,
  bookText: string,
): Observable<string> {
  return from([' (first search result)', ' (second search result)']).pipe(
    map(result => `${bookText} : ${searchTerm} : ${result}`),
    delay(20 * searchTerm.length),
    separateEmissions(200),
  );
}

/**
 * This is unchanged from before
 */
const searchResults$ = userBookSelection$.pipe(
  switchMap(selection => getBookText(selection)),
  switchMap(bookText => {
    return userSearchTerm$.pipe(
      switchMap(searchTerm =>
        getSearchResults(searchTerm, bookText).pipe(
          scan((searchResults: string[], searchResult) => {
            return [...searchResults, searchResult];
          }, []),
        ),
      ),
    );
  }),
);

/**
 * Lastly we'd doing a few tricks to make the output express what happened
 * better.
 * The combination of timestamp and pairwise gives us a stream of when the
 * emission happened and bundles it with the previous one so we can compare
 * times to get a time taken value. The startWith(null) just gives us the
 * startup time as a baseline.
 * Lastly we use our old friend map() to output the data in a nice format for
 * the logger.
 */
searchResults$
  .pipe(
    startWith(null),
    timestamp(),
    pairwise(),
    map(([before, tsResult], i) => {
      const timeSinceLast = (tsResult.timestamp - before.timestamp) / 1000;
      return `${i} : Search Result: [${tsResult.value.join(
        ', ',
      )}] (+${timeSinceLast} seconds)`;
    }),
  )
  .subscribe(console.log);



getBookText called (http://alice.text)
0 : Search Result: [http://alice.text : w :  (first search result)] (+0.431 seconds)
1 : Search Result: [http://alice.text : we :  (first search result)] (+0.123 seconds)
2 : Search Result: [http://alice.text : we’ :  (first search result)] (+0.128 seconds)
3 : Search Result: [http://alice.text : we’r :  (first search result)] (+0.121 seconds)
4 : Search Result: [http://alice.text : we’re :  (first search result)] (+0.121 seconds)
5 : Search Result: [http://alice.text : we’re all mad here :  (first search result)] (+1.6 seconds)
6 : Search Result: [http://alice.text : we’re all mad here :  (first search result), http://alice.text : we’re all mad here :  (second search result)] (+0.204 seconds)
getBookText called (http://sherlock.text)
7 : Search Result: [http://sherlock.text : n :  (first search result)] (+1.704 seconds)
8 : Search Result: [http://sherlock.text : no :  (first search result)] (+0.125 seconds)
9 : Search Result: [http://sherlock.text : not :  (first search result)] (+0.124 seconds)
10 : Search Result: [http://sherlock.text : noth :  (first search result)] (+0.125 seconds)
11 : Search Result: [http://sherlock.text : nothi :  (first search result)] (+0.122 seconds)
12 : Search Result: [http://sherlock.text : nothing more deceptive than an obvious fact :  (first search result)] (+4.68 seconds)
13 : Search Result: [http://sherlock.text : nothing more deceptive than an obvious fact :  (first search result), http://sherlock.text : nothing more deceptive than an obvious fact :  (second search result)] (+0.201 seconds)

Alrighty let's dig into this a bit.

We can see straight away that we fetch the Alice in Wonderland book immediately, and never again - this is perfect.

Next as the phrase begins to be typed we first get one result, then the other is appended to the results, good good.

Later on (the line starting with 5) we can see that the search result slowdown has meant that we're getting results less frequently and they are for longer search phrases than just the next character - this is what we expect as it means that the switchMap is unsubscribing from the search processor function as there is different data to be processed. This is a great opportunity to compare the different behaviors of switchMap(), mergeMap() and exhaustMap():

If we had chosen mergeMap(), we would see every single search result for every keystroke, but likely all overlapping with each other and would be pretty confusing. Also the overall time would be longer, assuming the CPU was saturated while processing the search.
If we had chosen exhaustMap(), we would get the exhaustive set of results (hence the name!) in the correct order, however the overall time would be way longer as we had to wait sequentially.

In this case I think switchMap() is the correct behavior as the user is not interested in interim search results before they have finished typing, and we gain efficiency by immediately cancelling computation of irrelevant search results.

Back to the output analysis, in the line marked 7 we see we've switched to fetching Sherlock Holmes with the new search phrase. Success!

Okay, we're in a state now where we're pretty confident in the general data flow, let's cover the algorithm we will use to score paragraphs, then later start building the application.

Algorithm

Given the requirement that our search terms may contain typos and misspellings, the algorithm to find the best paragraph to show the user is non trivial. There are a number of different fuzzy string matching algorithms, each with their own strengths & weaknesses. Given our requirement is about typos and misspellings, not word order mistakes, computing the Levenshtein distance sounds like a great candidate.

Levenshtein distance is the number of changes required to change one string to another. The changes can be either insertions, deletions or substitutions. The distance is simply the minimum count of changes.

As an example the distance between hello and helo is 1 because there is one insertion required. Two identical strings will have a distance of 0.

Now our inputs don't quite suit this algorithm as we're comparing a small search phrase with a large paragraph - our distance scoring will mostly indicate the difference in length of the search term and the paragraph. Instead we will modify the algorithm to ignore the start and end of the search string such that we compute the minimum Levenshtein distance of the search string when placed anywhere in the paragraph.

Finally, we will normalize Levenshtein distance against the length of the search string in order to compute a similarity score. This will be used to rank the paragraphs by best match.

In order to not make this article any longer than it already is, I'm going to skip over the implementation, but you can dig in to it yourself at https://github.com/zakhenry/blog-posts/tree/master/posts/observable-workers-deep-dive/src/app/book-search/common/fuzzy-substring.ts

What is important to know is that there is a function fuzzySubstringSimilarity:



// src/app/book-search/common/fuzzy-substring.ts#L70-L73

export function fuzzySubstringSimilarity(
  needle: string,
  haystack: string,
): FuzzyMatchSimilarity {

which returns FuzzyMatchSimilarity:



// src/app/book-search/common/fuzzy-substring.ts#L1-L9

export interface FuzzyMatch {
  substringDistance: number;
  startIndex: number;
  endIndex: number;
}

export interface FuzzyMatchSimilarity extends FuzzyMatch {
  similarityScore: number;
}

We will be using this return type to both score the paragraphs of the novel and return the start and end index of the match so we can highlight it in the output.

Let's test the algorithm against the user story defined above:



import { fuzzySubstringSimilarity } from '../src/app/book-search/common/fuzzy-substring';

const similarity = fuzzySubstringSimilarity(
  'were all madd her',
  '‘Oh, you can’t help that,’ said the Cat: ‘we’re all mad here. I’m mad. You’re mad.’',
);
console.log(`Similarity: `, similarity);

All we're doing here is defining the needle as the search string, and the haystack as the expected matching sentence. If we run this file we get the following:



Similarity:  {
  substringDistance: 2,
  startIndex: 42,
  endIndex: 59,
  similarityScore: 0.8823529411764706
}

As expected, we see a substringDistance of 2 (to get to the expected substring we need to add a ' and remove a d, two changes, distance is 2). We can also see that the similarityScore is very high (range is 0-1) so we can expect this search to score highly on this paragraph against other paragraphs.

Okay, we now have a algorithm sorted, and a data flow to manage it. Let's build an app!

Application

So far we've just been working with single typescript files, to work through our ideas, but now we're going to jump into using a framework as this is much more realistic to our real world problems we hope to solve by understanding this article.

I'm going to use Angular, but if you prefer any of the other awesome frameworks don't fret as this is going to be fairly framework agnostic anyway. If you've made it this far, it's probably safe to assume you have the basics of framework development down pat so I'll gloss over that.



ng new observable-workers-deep-dive



ng generate component book-search

Ok, we've got a basic framework scaffolded, given we're trying to test the difference between using observables directly, and using observables with web workers, as we build out the application we will try to extract the common functionality into one place. That way as many variables as possible are controlled during our performance tests as the worker and main thread strategies will be using the exact same core algorithm code.

Common functionality

To start with we will build a utility function to handle getting a stream of fuzzy-match scored paragraphs:



// src/app/book-search/common/book-search.utils.ts#L8-L25

interface SearchMatch {
  paragraph: string;
  paragraphNumber: number;
  searchMatch: FuzzyMatchSimilarity;
}

function getSearchResults(
  searchTerm: string,
  paragraphs: string[],
): Observable<SearchMatch> {
  return from(paragraphs).pipe(
    observeOn(asyncScheduler),
    map((paragraph, index) => {
      const searchMatch = fuzzySubstringSimilarity(searchTerm, paragraph);
      return { searchMatch, paragraph, paragraphNumber: index };
    }),
  );
}

Some very important things to note here. We're converting the array of paragraphs to an observable stream of individual paragraphs with from().

After this we use observeOn(asyncScheduler) - this is critical to the responsiveness of the application. Basically what this is doing is rescheduling the emissions of the from() observable from being synchronous to asynchronous. This allows subscribers to our stream to disconnect from the paragraph stream without having to compute the fuzzy substring scores for the entire book. This will allow us to discard partial search results when the search string becomes invalid due to the user typing more characters.

Finally we map() to the computation function, and attach the paragraph index number to the output - we will later use this to work out how far along the computation is as a percentage.

Next up for our common functions is an accumulation function that will

take our stream of search results
collate them into an array
order the array of results by score
take the top ten results
slice up the resulting paragraph into non matching text and matching text



// src/app/book-search/common/book-search.utils.ts#L27-L77

export interface MatchingParagraph {
  before: string;
  match: string;
  after: string;
  score: number;
}

export interface SearchResults {
  paragraphs: MatchingParagraph[];
  searchedParagraphCount: number;
  paragraphCount: number;
}

function accumulateResults(paragraphCount: number) {
  return (obs$: Observable<SearchMatch>): Observable<SearchResults> => {
    return obs$.pipe(
      scan((searchResults: SearchMatch[], searchResult: SearchMatch) => {
        searchResults.push(searchResult);
        return searchResults;
      }, []),
      startWith([]),
      map(
        (searchMatches: SearchMatch[]): SearchResults => {
          const last = searchMatches[searchMatches.length - 1];

          return {
            searchedParagraphCount: last ? last.paragraphNumber : 0,
            paragraphCount,
            paragraphs: searchMatches
              .sort(
                (a, b) =>
                  b.searchMatch.similarityScore - a.searchMatch.similarityScore,
              )
              .slice(0, 10)
              .map(({ searchMatch, paragraph }) => {
                return {
                  score: searchMatch.similarityScore,
                  match: paragraph.substring(
                    searchMatch.startIndex,
                    searchMatch.endIndex,
                  ),
                  before: paragraph.substring(0, searchMatch.startIndex),
                  after: paragraph.substring(searchMatch.endIndex),
                };
              }),
          };
        },
      ),
    );
  };
}

Despite the line count, there is nothing particularly complex in here, some points to note however:

The scan() operator is used so that we get results streamed as they come in, if we wanted to wait for completion of the search we would use reduce() (the rxjs operator, not Array.prototype.reduce)
We use startWith([]) so that a subscriber immediately gets an empty result set on subscription

Lastly we combine these two functions into a single one that accepts a search term and a book, and returns the accumulated stream of search results.



// src/app/book-search/common/book-search.utils.ts#L79-L87

export function getAccumulatedSearchResults(
  searchTerm: string,
  bookText: string,
): Observable<SearchResults> {
  const paragraphs = bookText.split('\n\n');
  return getSearchResults(searchTerm, paragraphs).pipe(
    accumulateResults(paragraphs.length),
  );
}

Service

Ok, basic functions that do the heavy lifting are all sorted, we're pretty much on the home stretch! Let's create a service to wrap this logic that we will later use in our component.



// src/app/book-search/main-thread/book-search.service.ts#L11-L44

@Injectable({
  providedIn: 'root',
})
export class BookSearchService {
  constructor(private http: HttpClient) {}

  public search(
    bookSelection$: Observable<BookChoice>,
    searchTerm$: Observable<string>,
  ): Observable<SearchResults> {
    return this.processSearch(bookSelection$, searchTerm$).pipe(
      auditTime(1000 / 60), // emit results at a maximum of 60fps
      share(),
    );
  }

  protected processSearch(
    url$: Observable<string>,
    search$: Observable<string>,
  ): Observable<SearchResults> {
    const sharedSearchTerm$ = search$.pipe(shareReplay(1));

    return url$.pipe(
      switchMap(url => this.http.get(url, { responseType: 'text' })),
      switchMap(bookText => {
        return sharedSearchTerm$.pipe(
          switchMap(searchTerm => {
            return getAccumulatedSearchResults(searchTerm, bookText);
          }),
        );
      }),
    );
  }
}

With this service, we inject an HttpClient into the constructor; this allows us to fetch the book content. Next is a public method that we will use in our component. It takes two streams:

the first is the BookChoice. Recall from earlier this is a string enum with the values being the URl of the book
the second is the search phrase itself that is used for fuzzy matching against the paragraphs of the book

We call out to a local processSearch method which simply takes the url and pipes it to a switchMap() that fetches the content, then in turn pipes to the search term observable and finally we switch on that observable to call out to getAccumulatedSearchResults that we built earlier.

Back in the search() method, we pipe the results from processSearch() and first use the auditTime() operator. This restricts the data output rate to 60 frames per second otherwise we could potentially get tens of thousands of SearchResults per second, which would completely oversaturate the change detection strategy of the framework when we try to display the results in the DOM.

Lastly we add a share() to the search results as we don't want multiple subscribers to trigger computation of the search results more than once.

Next let's define the component logic to connect this service to user input.

Component

Our component requirements are super simple - it needs to provide a form control for both the book selection and the search term, then output the results of the search.



// src/app/book-search/main-thread/book-search.component.ts#L13-L59

@Component({
  selector: 'app-book-search',
  templateUrl: './book-search.component.html',
  styleUrls: ['./book-search.component.css'],
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class BookSearchComponent {
  public componentName = 'Main thread search';

  public bookChoices = [
    {
      url: BookChoice.ALICE_IN_WONDERLAND,
      name: 'Alice in Wonderland',
    },
    {
      url: BookChoice.SHERLOCK_HOLMES,
      name: 'Sherlock Holmes',
    },
  ];

  public bookSelectionFormControl = new FormControl(null);
  public userBookSelection$: Observable<BookChoice> = this
    .bookSelectionFormControl.valueChanges;

  public searchTermFormControl = new FormControl(null);
  public userSearchTerm$: Observable<string> = this.searchTermFormControl
    .valueChanges;

  private searchResults$: Observable<
    SearchResults
  > = this.bookSearchHandler.search(
    this.userBookSelection$,
    this.userSearchTerm$,
  );

  public searchResultParagraphs$: Observable<
    MatchingParagraph[]
  > = this.searchResults$.pipe(map(result => result.paragraphs));

  public searchResultProgress$: Observable<
    [number, number]
  > = this.searchResults$.pipe(
    map(result => [result.searchedParagraphCount, result.paragraphCount]),
  );

  constructor(private bookSearchHandler: BookSearchService) {}
}

Points of interest in this component:

we construct a new FormControl for the book selection, then immediately set up an observable that observes the valueChanges of that control
we do the same thing for the search terms
next we construct the private searchResults$: Observable<SearchResults> using the previous two observables, it calls out to the service that we just defined earlier
the last two public members observe the searchResults$, the first extracting the matching paragraphs, the second extracting the search progress information

Next let's take a look how this will be displayed in the template:



<!-- src/app/book-search/main-thread/book-search.component.html -->

<h2>{{ componentName }}</h2>

<select [formControl]="bookSelectionFormControl">
  <option *ngFor="let book of bookChoices" [value]="book.url">
    {{ book.name }}
  </option>
</select>

<input type="text" [formControl]="searchTermFormControl" />

<span *ngIf="searchResultProgress$ | async as searchResultProgress">
  Progress: {{ searchResultProgress[0] }} / {{ searchResultProgress[1] }} ({{
    searchResultProgress[0] / searchResultProgress[1] | percent
  }})
</span>

<ng-container *ngFor="let matchingParagraph of searchResultParagraphs$ | async">
  <hr />
  <blockquote>
    <span>{{ matchingParagraph.before }}</span>
    <strong>{{ matchingParagraph.match }}</strong>
    <span>{{ matchingParagraph.after }}</span>
    <footer>Score: {{ matchingParagraph.score }}</footer>
  </blockquote>
</ng-container>

Nice simple template, we've got

a basic <select> to manage the book selection
then an <input> to manage the user search
then a <span> that outputs the search result progress
lastly a repeated <blockquote> to output the results of the search

Note that we're using AsyncPipe to managed all subscriptions - this greatly simplifies the component logic, and allows us to use OnPush change detection strategy as the AsyncPipe manages marking the component to be checked.

At long last, we're there, let us test it.

Main Thread Test

Ok so when we run the app, we're presented with the two controls. So let's select "Alice and Wonderland" and type that typo-ridden string "were all madd her"

Okay! This is good - we've managed to fetch search results, and the first result is the expected match:



were all madd her
 Progress: 900 / 901 (100%)
‘Oh, you can’t help that,’ said the Cat: ‘we’re all mad here. I’m mad. You’re mad.’
Score: 0.8823529411764706

However the responsiveness of the application was awful - that gif is not stuttering, it really did lock up after typing the first character then all the results appeared after a few seconds.
Fortunately we have the incredible tool that is Web Workers up our sleeves, and we can use one to free up this oversaturated main thread to just do UI stuff, and keep our expensive search result computation in a separate thread.

Worker

So, let's create a worker



ng generate web-worker book-search

This will put a book-search.worker.ts file into our application:



/// <reference lib="webworker" />

addEventListener('message', ({ data }) => {
  const response = `worker response to ${data}`;
  postMessage(response);
});

We don't want any of that boilerplate, so we will delete it and implement our own worker version of the processSearch method we created in the service earlier:



// src/app/book-search/worker-thread/book-search.worker.ts

import { DoWork, ObservableWorker } from 'observable-webworker';
import { Observable } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import {
  distinctUntilChanged,
  map,
  shareReplay,
  switchMap,
} from 'rxjs/operators';
import {
  getAccumulatedSearchResults,
  SearchResults,
  WorkerInput,
} from '../common/book-search.utils';

@ObservableWorker()
export class BookSearchWorker implements DoWork<WorkerInput, SearchResults> {
  public work(input$: Observable<WorkerInput>): Observable<SearchResults> {
    const url$ = input$.pipe(
      map(({ url }) => url),
      distinctUntilChanged(),
    );

    const searchTerm$ = input$.pipe(
      map(({ searchTerm }) => searchTerm),
      distinctUntilChanged(),
      shareReplay(1),
    );

    return url$.pipe(
      switchMap(url => ajax({ url, responseType: 'text' })),
      map(result => result.response),
      switchMap(bookText => {
        return searchTerm$.pipe(
          switchMap(searchTerm =>
            getAccumulatedSearchResults(searchTerm, bookText),
          ),
        );
      }),
    );
  }
}

This ought to look pretty familiar, as it is really just the processSearch function within a @ObservableWorker()- decorated class from the observable-webworker package. The only difference is that the input to the work() method is a single stream of Observable<WorkerInput> which we split and observe separately.

From there it is essentially identical, though we do use the rxjs ajax() method rather than HttpClient as the Injector is not available in the worker context.

Now for the main thread class to manage this worker:



// src/app/book-search/worker-thread/book-search-worker.service.ts

import { Injectable } from '@angular/core';
import { fromWorker } from 'observable-webworker';
import { combineLatest, Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { BookSearchService } from '../main-thread/book-search.service';
import { SearchResults, WorkerInput } from '../common/book-search.utils';

@Injectable({
  providedIn: 'root',
})
export class BookSearchWorkerService extends BookSearchService {
  protected processSearch(
    url$: Observable<string>,
    search$: Observable<string>,
  ): Observable<SearchResults> {
    const input$: Observable<WorkerInput> = combineLatest(url$, search$).pipe(
      map(([url, searchTerm]) => ({ searchTerm, url })),
    );

    return fromWorker(
      () => new Worker('./book-search.worker', { type: 'module' }),
      input$,
    );
  }
}

Because we've already implemented the search() logic in BookSearchService we simply extend that class and override the implementation of processSearch() in a new BookSearchWorkerService class.

In a real situation you probably would just replace the processSearch method, but we're going to have two implementations in this article so we can compare them side-by-side later.

Our implementation of processSearch is super simple, we combineLatest the two input streams into the form that the Worker required. Then we use the fromWorker() method from 'observable-webworker' to wrap the construction of a new Worker and pass the input stream as the second argument.

Now, while we're trying to preserve the old behaviour, we need a new component to show off this shiny new worker backed service:



// src/app/book-search/worker-thread/book-search-worker.component.ts

import { ChangeDetectionStrategy, Component } from '@angular/core';
import { BookSearchWorkerService } from './book-search-worker.service';
import { BookSearchComponent } from '../main-thread/book-search.component';
import { BookSearchService } from '../main-thread/book-search.service';

@Component({
  selector: 'app-book-search-worker',
  templateUrl: '../main-thread/book-search.component.html',
  styleUrls: ['../main-thread/book-search.component.css'],
  changeDetection: ChangeDetectionStrategy.OnPush,
  providers: [
    { provide: BookSearchService, useClass: BookSearchWorkerService },
  ],
})
export class BookSearchWorkerComponent extends BookSearchComponent {
  public componentName = 'Worker thread search';
}

Not much going on here! All we're doing is extending the BookSearchComponent, reusing it's template and styles, and providing our new BookSearchWorkerService instead of the default BookSearchService.

Now in our app.component.html template we can insert this new component:



<!-- src/app/app.component.html -->

<app-book-search></app-book-search>
<app-book-search-worker></app-book-search-worker>

So now we have our two implementations of book search - one in the main thread, the other backed by a worker thread service.

Worker Thread Test

Let's test the new worker component!

Wow! How's that for performance?! Now I wouldn't call this the nicest of user experiences, but you can really see the power of web workers in action here.

Also in this demo you can really see the reason we had to auditTime() the search results as the DOM update frequency here is super high. In fact, if you comment out that function, you will find that the web worker implementation performance is significantly worse as it is emitting search results at such a high frequency that it causes a backpressure of change detection cycles for Angular, making the overall performance actually worse. This was a real takeaway lesson for me, and hopefully you too -

Web Workers can make your computation output so fast that displaying the results in the DOM can become a major overall bottleneck

Visual comparisons are all well and good, but let's properly look into how the performance differs between the two strategies

Performance

To test the relative performance I will search the phrase "There is nothing more deceptive than an obvious fact" within the Sherlock Holmes book. I'll do it three times using the regular main thread strategy, clearing the input each time, then the same for the web worker strategy.

From these graphs it is clear the difference using a web worker makes - the main thread (in yellow in this graph) holds at 100% for a full 15 seconds, during which repainting will suffer greatly. With the web worker strategy, the main thread barely peaks at 50% utilisation, and is actually mostly idle for the duration.

It is worth noting at this time that the overall duration is roughly the same, and it would be expected that in some circumstances the worker strategy might be slightly slower overall. This is because the strategy requires spinning up a new thread, and structured copying of the data flowing between threads is not zero-cost.

One other consideration is that we're not fully utilising the power of web workers - the computation job that we're running can easily be broken up into smaller tasks for multiple workers, and processed using a thread pool strategy. This is a topic for another article, and I might just be covering that one next.

Wrap Up

You've reached the end! This was a bit of a marathon I'm afraid but there was an awful lot of content to cover. In summary, we outlined a real world scenario where users want to robustly do full text search over a novel, we then came up with a data flow strategy, then an algorithm, then built it out into an application, then refactored to use an observable web worker strategy, and finally did some performance metrics to prove the utility of web workers. Phew!

The code is all available on Github so please feel free to clone it and have a play. If you spot any errors or room for improvement please do raise an issue or a pull request, I'm learning too!

Next in this series, I will be either demonstrating how to further improve the performance of this application using thread pools, or implementing a new real world use case that does intensive image manipulation which will demonstrate high performance techniques of passing large data between threads. Let me know if you have a preference!

Thanks again for staying with me here, I hope that this article has been useful to you.

Photo by Annie Spratt on Unsplash

Ensuring accuracy of README code snippets

Zak Henry — Thu, 20 Jun 2019 17:33:33 +0000

The problem

It is the first interaction a developer will have with your library. They hear about the library through various means, but they will at some point all end up at the README. It is critical that their first impression is good, and nothing can turn a developer off a library than the example code blocks in the README not actually working.

Markdown itself doesn't do much to help us - it is a simple markup language that knows nothing about the libraries that we're building, and it certainly can't compile the code within the code snippets that we create.

Generally, this leaves us with two options:

We write the code directly in markdown and cross our fingers that it will be okay
We write the code in a separate file, that we can test against the library and we copy-paste into the markdown document

With both options, there is a huge amount of room for mistakes. In the first case, our code might not even be valid! In the second, there is a number of files to remember to update and then remember to copy paste the content.

In both instances, we may later on introduce a breaking change to the library, and completely forget to update the docs and then we're in the worst case scenario - the documentation does not reflect the library.

I've contributed to open source a bit, and have managed to run into all these cases and while they do get picked up eventually, it is fairly embarrassing.

The solution

Becoming fed up with this annoyance, I decided to automate the process of managing updating the README from source code snippets. I've built a little nodejs commandline utility which takes all the hassle out of maintaining a README (and other markdown documents too!).

Assuming you have nodejs installed you can simply run

npx embedme README.md

Now to begin with you're probably not going to get very useful output; you need to annotate your code blocks such that embedme can find the appropriate file to insert

So, let us step through an example:

Say I have the following README.md:

What you can do with embedme is insert a comment in the first line of the code block:

Next, if you now run the following:

You will see output from the library indicating where it found code blocks and what it did with them (i.e. in this case, inserted the content from path/to/example-usage.ts).

Next, to prove the utility of the library, make a change to your source code

import { hello } from 'my-amazing-library';

console.log('An update here!');
hello(); // outputs Hello World!

Then rerun embedme:

You will see your readme file has updated:

So, as you've now seen, we're in a situation where out example code is outside of the README, but with a simple utility we can have it be up to date. Not only that, but we can (and should!) unit test this example code against our library, which will ensure that if we cause breaking changes to our library, we are guaranteed that the documentation will be kept up to date with correct code snippets.

Embedme has a number of other features, including...

extracting specific lines (//path/to/your/file.ts#L10-L20)
--verify flag which doesn't actually write to file, but checks there are not changes (good for CI scripts!)
--stdout flag to output the change to be stdout in order to be redirected to another file
using markdown comment syntax to define code blocks without being required to insert the comment within the code block. ()
and a few other handy features...

I encourage you to view the library for more info - https://github.com/zakhenry/embedme and add it to your automated workflow to get documentation that is both a pleasure to write, and guaranteed to be valid.

Photo by Andrea Sonda on Unsplash

Observable Web Workers with Angular (8) - Introduction

Zak Henry — Sun, 16 Jun 2019 21:28:54 +0000

tl;dr

Web Workers are awesome and Angular CLI now supports them natively.

Unfortunately, the Web Worker API is not Angular-like and therefore, I'm introducing a library observable-webworker

If you already know what web workers are and how to use them in Angular, feel free to skip ahead where I will show you how to refactor the initial worker generated by the CLI into a superpowered worker with observable-webworker.

Context

In this article series we will explore using Web Workers to offload heavy computation to another thread. Keeping the main thread free for interaction with the user is important. We don't typically think of threading very often in the frontend world, as non-blocking APIs and the event loop model of Javascript typically allows us to have execution of user interaction events in-between longer running processes such as waiting for an HTTP request to resolve.

Because of this, for the most part, you don't need threading or web workers for most tasks. However, there are a set of problems that do require heavy computation that would ordinarily block the main thread, and therefore user interaction (or modification of the DOM). This issue can manifest itself in stuttering animations, unresponsive inputs, buttons that appear not to work immediately etc.

Often the answer to this has been to run intensive computations server side, then send the result back to the browser once done. This solution does have a real cost however - you need to manage a server API, monetarily computation itself isn't free (you're paying for the server) and there may be a significant latency issue if you need to interact with the computation frequently.

Fortunately, Web Workers are the solution to all this - they allow you to schedule units of work in browser that run in parallel to the main execution context, then once done can pass their result back to the main thread for rendering in the DOM etc.

Creating your first Web Worker in Angular

The worker API is relatively simple - in Typescript you can create a new worker with just

const myWorker = new Worker('./worker.js');

This is all very well, however we love to use Typescript, and be able to import other files etc. Being stuck with a single javascript file is not very scalable.

Previously working with workers (heh) in Angular was fairly painful as bundling them requires custom Webpack configuration. However, Angular CLI version 8 brings built-in support for properly compiling & bundling web workers.

The docs are at https://angular.io/guide/web-worker but we will step through everything required (and more!) here.

Before we get going, we've set up an empty Angular CLI project with

ng new observable-workers

Next we need to enable web worker support, so we run

ng generate web-worker

We're prompted for a name, so we will call it demo.

Angular CLI will now update tsconfig.app.json and angular.json to enable web worker support, and create us a new demo.worker.ts:

/// <reference lib="webworker" />

addEventListener('message', ({ data }) => {
  const response = `worker response to ${data}`;
  postMessage(response);
});

To get this worker running, lets update our AppComponent:

import { Component } from '@angular/core';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})
export class AppComponent {

  runWorker() {

    const demoWorker = new Worker('./demo.worker', { type: 'module'});

    demoWorker.onmessage = (message) => {
      console.log(`Got message`, message.data);
    };

    demoWorker.postMessage('hello');

  }

}

And the template:

<button (click)="runWorker()">Run Worker</button>

Okay, so now we're all good to go, run the application if you aren't already, and you'll be greeted with a single button in the DOM. Are you excited yet?

Click that sucker, and you will see in the dev console

Got message worker response to hello

Yeehaw, we got a response from a separate thread, and Angular compiled the worker for us.

Also, to prove this is actually a worker, if you go to the Sources tab in Chrome, you will see that there is a worker running there as 1.worker.js.

The fact that we can still see it running is important - it means that we have not destroyed the worker, despite receiving the only message we will get back from it. If we click the button again, we will construct a brand new worker, and the old one will continue to hang around. This is a bad idea! We should always clean up after we're done with a worker.

Before we worry about how to destroy a worker, let's reflect for a bit on the API that we have with workers so far - we need to:

construct the worker
declare a property and assign a function to it in order to get the response back from the worker
addEventListener within the worker itself, we have to
call a global postMessage function to send back information to the main thread

This API doesn't feel very Angular-like does it? We're used to deal with clean hook-based APIs and have fallen in love with RxJS for dealing with streams of data and asynchronous responses, why can't we have this for Web Workers too?

Well, we can. And this happens to be the whole point of this article. I'd like to introduce a new library observable-webworker which seeks to address this clunky API and give us a familiar experience we're used to with Angular.

I should note that this library doesn't actually depend on Angular at all, and will work beautifully with React or any other framework or lack thereof! The only dependency is a peerDependency on RxJS.

To best introduce the concepts of the library, we will refactor our current web worker to use observable-webworker.

Implementing observable-webworker

To start, we will install observable-webworker. I'm using Yarn, but you know how to install packages right?!

 yarn add -E observable-webworker

First of all, we'll update the AppComponent

import { Component } from '@angular/core';
import { fromWorker } from 'observable-webworker';
import { Observable, of } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})
export class AppComponent {

  runWorker() {

    const input$: Observable<string> = of('hello');

    fromWorker<string, string>(() => new Worker('./demo.worker', { type: 'module'}), input$)
      .subscribe(message => {
        console.log(`Got message`, message);
      });

  }

}

Note we've imported fromWorker from observable-webworker. The first argument is a worker factory - we need to have the ability to lazily construct a worker on-demand, so we pass a factory and the library can construct it when needed. Also Webpack needs to find the new Worker('./path/to/file.worker') in the code in order to be able to bundle it for us.
The second argument input$ is a simple stream of messages that will go to the worker. The generics (<string, string>) that we pass to the worker indicate the input and output types. In our case the input is a very simple of('hello').

Now for the worker:

import { DoWork, ObservableWorker } from 'observable-webworker';
import { Observable } from 'rxjs';
import { map } from 'rxjs/operators';

@ObservableWorker()
export class DemoWorker implements DoWork<string, string> {

  public work(input$: Observable<string>): Observable<string> {

    return input$.pipe(
      map(data => `worker response to ${data}`)
    );
  }

}

You can see we have completely restructured the worker to use an API that is far more familiar to Angular developers - we use a decorator @ObservableWorker() which lets the library bootstrap and work its magic, and we implement the DoWork interface, which requires us to create a public work(input$: Observable<string>): Observable<string>; method.

The actual content of the method is super simple. We process the observable stream with a simple map, which is returned from the method. Behind the scenes the library will call postMessage to get the data back to the main thread.

You can now run the application again and it will work exactly as before, with the one exception that the worker will automatically be terminated as no more work needs to be done.

As we're now using observables, the library is able to know when subscribers have finished listening and can clean up the worker appropriately.

This wraps up Part One of this series. In later articles we will discuss a more real-world implementation of workers, how to deal with errors, and how to work with very large object passing between main and worker threads. Possibly I will also do a deep dive into the implementation of observable-webworker as it uses some of the more exotic RxJS operators like materialize() and dematerialize().

All of these features are available with observable-webworker now, so I encourage you to check out the readme, as it goes into more detail about the features than I do here in this article.

This is my first blog post ever! I'd love to hear from you if you have any feedback at all, good or bad!