Communicating Sequential Processes (CSP) in JavaScript

Max Klymyshyn CTO at CartFresh

CartFresh‣ Grocery Delivery startup

‣ Working in Ukraine as ZAKAZ.UA (Kiev, Dnepropetrovsk, Kharkiv) and as CartFresh (Boston, US)

‣ React.js on front-end

‣ Python on back-end

‣ 12+ years of experience: 7 with Python, 6 with JS

‣ Was part of oDesk (Upwork), Helios, 42cc

‣ Co-organizer of PyCon Ukraine, KyivJS

‣ CTO at CartFresh

Asynchronous I/Oasynchronous I/O leads to callback API’s,

which lead to nested lambdas, which lead to… the pyramid of doom:

range.on("preheat", function() { pot.on("boil", function() { rice.on("cooked", function() { dinner.serve(rice); }); }); });

Pyramid of doom

clusterfuck.

Understanding the code

‣ IBM 1989: 50% of the effort in accomplishing a task for the programmer is towards understanding the system

‣ Bell Labs 1992: 60%-80% of their time understanding code, 20% as the developers gain experience with current codebase

‣ National Research Council 1997 (Canada): over 25% of their time either searching for or looking at code

‣ Microsoft 2006: 50%

‣ Peter Hallam 2006: 70% during personal experiment

‣ Microsoft 2007: 65% (survey)

CSP Communicating sequential processes

CSP‣ initially is a formal language for

describing patterns of interaction in concurrent systems

‣ first described in a 1978 paper by Tony Hoare

‣ influential in the design of the occam, Go, Limbo

‣ core.async in clojure

‣ js-csp for JS

Key points

‣ CSP created for communicating between different components and subsystems

‣ CSP solve problem of coordinating anything asynchronous

‣ CSP alongside others solve problem of easy-to-understand code

Channels

CHANNEL

put

take

Core API

take timeout

‣ blocking semantic

‣ duplex

‣ indirect process communication

‣ easy-to-implement remote channels

Channel Properties

example

import {chan, take, CLOSED, timeout, put} from "js-csp";

var ch = chan();

go(function * () { var val; while((val = yield take(ch)) !== CLOSED) { console.log(val); } });

go(function * () { yield put(ch, 1); yield take(timeout(2000)); yield put(ch, 2); ch.close(); });

Features

merge

Read from multiple channels

CH

AN

NEL

1

CH

AN

NEL

2

put put

Read from multiple channels

var chan1 = chan(), chan2 = chan(), merged = operations.merge([chan1, chan2]);

go(function*(){ var value = yield merged; while (value !== CLOSED) { console.log("Got ", value); value = yield merged; }; });

put

CH

AN

NEL

1

CH

AN

NEL

2take take

Supply value into multiple channels

Supply value into multiple channels

var src = chan(), mult = operations.mult(src), chan1 = chan(), chan2 = chan(0);

operations.mult.tap(mult, chan1); operations.mult.tap(mult, chan2);

go(function * () { yield put(src, 1); });

go(function * () { var value = yield chan1; while (value !== CLOSED) { console.log("Got ", value, " in `chan1`"); value = yield chan1; } });

put

Reduce channel values

CH

AN

NEL

1

CH

AN

NEL

2

RED

UC

Etake

take

Reduce channel values

var ch = chan(), append = (a, b) => a + " " + b;

var reduceCh = operations.reduce(append, "Hello", ch);

go(function * () { yield put(ch, "CSP"); yield put(ch, "World"); console.log(yield reduceCh); });

ch.close();

put

Buffering

CHANNEL 1

BUFFER

take

Features

‣ Channel buffering: fixed size, sliding, dropping

‣ poll / offer values: taking/putting immediately

‣ alts: get value or execute second operation

Common processes communication features

‣ Mixing channels with mute/unmute

‣ Pub/Sub mode

‣ Filtering with predicates and/or transducers

CHANNEL

PROCESS 2

PROCESS 1

put(1)

take

Communication example

FLOW

take

put(2)

import {go, chan, take, timeout, put} from "js-csp"; var ch = chan();

go(function*() { log(1, "line 1"); log(1, " line 2"); log(1, " <- take"); log(1, " took: process 2: ", yield take(ch)); log(1, " line 5"); log(1, " <- take"); log(1, " took: process 2: ", yield take(ch)); log(1, " line 8");

});

go(function*() { log(2, "line 1"); log(2, " -> put"); yield put(ch, 1); log(2, " line 4"); log(2, " -> put"); yield put(ch, 2); log(2, " line 7"); });

sync point

p( 1 ): line 1 p( 1 ): line 2 p( 1 ): <- take p( 2 ): line 1 p( 2 ): -> put p( 2 ): line 4 p( 2 ): -> put p( 1 ): took: process 2: val: 1 p( 1 ): line 5 p( 1 ): <- take p( 2 ): line 7 p( 1 ): took: process 2: val: 2 p( 1 ): line 8

p( 1 ): line 1 p( 1 ): line 2 p( 1 ): <- take p( 2 ): line 1 p( 2 ): -> put p( 2 ): line 4 p( 2 ): -> put p( 1 ): took: process 2: val: 1 p( 1 ): line 5 p( 1 ): <- take p( 2 ): line 7 p( 1 ): took: process 2: val: 2 p( 1 ): line 8

what the fuck?

JavaScript, baby

p( 1 ): line 1 p( 1 ): line 2 p( 1 ): <- take p( 2 ): line 1 p( 2 ): -> offer p( 1 ): took: process 2: val: 1 p( 1 ): line 5 p( 1 ): <- take p( 2 ): line 4 p( 2 ): -> offer p( 2 ): line 7 p( 1 ): took: process 2: val: 2 p( 1 ): line 8

import {go, chan, put, buffers, offer} from "js-csp";

var ch = chan();

go(function*() { log(1, "line 1"); log(1, " line 2"); log(1, " <- take"); log(1, " took: process 2: ", yield take(ch)); log(1, " line 5"); log(1, " <- take"); log(1, " took: process 2: ", yield take(ch)); log(1, " line 8");

});

go(function*() { log(2, "line 1"); log(2, " -> offer"); yield offer(ch, 1); yield timeout(0); log(2, " line 4"); log(2, " -> offer"); yield offer(ch, 2); log(2, " line 7"); });

sync point

CHANNEL

PROCESS 2

PROCESS 1

put

take

wait

put

take

wait

take

put

Channel

FLOW

CHANNEL

PROCESS 2

PROCESS 1

put

put

FLOW

PROCESS 3

take

take

take

put

take

put

real-world

export class Suggest extends React.Component { constructor(props) { super(props); this.state = {active: -1, suggests: props.suggests}} componentWillReceiveProps(nextProps) { switch(nextProps.code) { case 38: this.setState({active: Math.max(-1, this.state.active - 1)}); break; case 40: this.setState({ active: Math.min(this.props.suggests.length - 1, this.state.active + 1)}); break; case 13: search(this.props.suggests[this.state.active]); this.setState({suggests: []}); break; default: this.setState({suggests: nextProps.suggests}); } } render() { return (<ul className="dropdown dropdown-menu"> {this.state.suggests.map((e, n) => <li key={e} className={...}><a href="#">{e}</a></li>)} </ul>); }}

function listen(el, type) { var ch = chan(); el.addEventListener(type, e => { putAsync(ch, [e.keyCode, el.value]); e.preventDefault(); }); return ch; }

export function suggest(elem) { var el = elem[0];

go(function * () { var keydown = listen(el, 'keydown');

while(true) { var [code, value] = yield take(keydown); var response = yield take(Storage.sync([ ["store.suggest", {query: value}, {id: "suggest"}]]));

ReactDOM.render( <Suggest suggests={response.suggests || []} code={code} value={value} />, document.getElementById("suggest")); } }); }

sync code

Tools to write in sync style?

‣ Promises, Promises with generators

‣ Generators

‣ Async/await

‣ Using actors, RxJS, js-csp

sync code with js-csp

Tools

‣ Events

‣ XMLHttpRequest/HTTP

‣ WebSockets

‣ Timers

‣ Web workers

Runtime for low-level async operations: get URL

import {buffers, go, chan, putAsync, operations} from "js-csp";

export function listen(el, type, options={}) { /** * Translate events into CSP channel until channel is not closed. */ var {channel, prevent} = options;

var ch = channel || chan();

var listener = (e) => { if (ch.closed === true) el.removeEventListener(type, listener); else putAsync(ch, e); if (prevent === true) e.preventDefault(); }

el.addEventListener(type, listener);

return ch; }

import {go, take, timeout, CLOSED, close, chan, buffers} from "js-csp"; import {listen} from "./runtime.js";

var mousemove = listen(document, "mousemove", true, {channel: chan(buffers. dropping(1))}); var target = document.getElementById("coords");

go(function * () { var coords; while((coords = yield take(mousemove)) !== CLOSED) { target.innerHTML = `X=${coords.clientX} Y=${coords.clientY}`; } });

go(function * () { yield timeout(3000); yield mousemove.close(); target.innerHTML = 'interrupted.'; });

import {buffers, go, chan, putAsync, take,} from "js-csp";

export function json(options) { var ch = chan(); go(function * () { var value = yield take(request(options)); if(!(value instanceof Error)) { value = JSON.parse(value); } else { console.error("Can't get " + options.url, value); } putAsync(ch, value); });

return ch; }

isoroutesPure, isomorphic and framework-agnostic

js-csp—based router

Pure router

‣ Should be pure

‣ State based only on input

‣ Framework-agnostic

‣ No promises, sync-style code

Example

import {render, router, navigate} from "isoroutes";

exports var routes = router([ ["/", render.react(Home)], ["/about/", render.react(About)], ["/contacts/", render.react(Contact)], ["/articles/:id.html", render.react(Article)], ["/dashboard/", render.react(Dashboard)], ["/dashboard/signin/", render.react(Auth)], ["/dashboard/add/", render.react(ArticleForm)], ["/dashboard/:menu/", render.react(Dashboard)], ]);

Gather stateexport class Home extends React.Component { // state will be executed within CSP `go` routine static state(state, channel, n=0) { // we could use CSP channels here return go(function * () { yield put(channel, [ "talks", yield take(json({url: "/api/upcoming_talks.json"})) ]); channel.close() }) } render() { // ... } }

@maxmaxmaxmaxQuestions?