Libprocess User Guide

Bazel

Follows a "repos/deps" pattern (in order to help with recursive dependencies). To use:

1. Copy

   bazel/repos.bzl

   into your repository at

   3rdparty/libprocess/repos.bzl

   and add an empty

   BUILD

   (or

   BUILD.bazel

   ) to

   3rdparty/libprocess

   as well.

2. Copy all of the directories from

   3rdparty

   that you don't already have in your repository's

   3rdparty

   directory.

3. Either ... add the following to your

   WORKSPACE

   (or

   WORKSPACE.bazel

   ):

load("//3rdparty/libprocess:repos.bzl", libprocess_repos="repos")
libprocess_repos()

load("@com_github_3rdparty_libprocess//bazel:deps.bzl", libprocess_deps="deps")
libprocess_deps()

Or ... to simplify others depending on your repository, add the following to your

repos.bzl

load("//3rdparty/libprocess:repos.bzl", libprocess="repos")

def repos():
    libprocess()

And the following to your

deps.bzl

load("@com_github_3rdparty_libprocess//bazel:deps.bzl", libprocess="deps")

def deps():
    libprocess()

1. You can then use

   @com_github_3rdparty_libprocess//:process

   in your target's

   deps

   .

2. Repeat the steps starting at (1) at the desired version of this repository that you want to use.

libprocess provides general primitives and abstractions for asynchronous programming with futures/promises, HTTP, and actors.

Inspired by Erlang, libprocess gets it's name from calling an "actor" a "process" (not to be confused by an operating system process).

Table of Contents

• Presentations
• Overview
• Futures and Promises
• HTTP
• Processes (aka Actors)
• Clock Management and Timeouts
• Miscellaneous Primitives
• Optimized Run Queue and Event Queue

Presentations

The following talks are recommended to get an overview of libprocess:

• libprocess, a concurrent and asynchronous programming library (San Francisco C++ Meetup)

Overview

This user guide is meant to help understand the constructs within the libprocess library. The main constructs are:

1. Futures and Promises which are used to build ...
2. HTTP abstractions, which make the foundation for ...
3. Processes (aka Actors).

For most people processes (aka actors) are the most foreign of the concepts, but they are arguably the most critical part of the library (they library is named after them!). Nevertheless, we organized this guide to walk through futures/promises and HTTP before processes because the former two are prerequisites for the latter.

Futures and Promises

The

Future

Promise

A

Future

Promise

Looking for a specific topic?

• Basics
• States
• Disarding a Future (aka Cancellation)
• Abandoned Futures
• Composition:

  Future::then()

  ,

  Future::repair()

  , and

  Future::recover()

• Discarding and Composition
• Callback Semantics

• CHECK()

  Overloads

Basics

A

Promise

Promise

Promise

You can get a

Future

Promise

Promise::future()

Promise

Future

As of this time, the templated type of the future must be the exact same as the promise: you cannot create a covariant or contravariant future.

Here is a simple example of using

Promise

Future

using process::Future;
using process::Promise;

int main(int argc, char** argv)
{
  Promise promise;
  Future future = promise.future();
  // You can copy a future.
  Future future2 = future;
  // You can also assign a future (NOTE: this future will never
  // complete because the Promise goes out of scope, but the
  // Future is still valid and can be used normally.)
  future = Promise().future();
  return 0;
}

States

A promise starts in the

PENDING

READY

FAILED

DISCARDED

Future::isPending()

Future::isReady()

Future::isFailed()

Future::isDiscarded()

We typically refer to transitioning to

READY

as completing the promise/future.

You can also add a callback to be invoked when (or if) a transition occurs (or has occcured) by using the

Future::onReady()

Future::onFailed()

Future::onDiscarded()

Future::onAny()

READY

FAILED

DISCARDED

The following table is meant to capture these transitions:

| Transition |

Promise::*()

Future::is*()

Future::on*()

READY

Promise::set(T)

Future::isReady()

Future::onReady(F&&)

FAILED

Promise::fail(const std::string&)

Future::isFailed()

Future::onFailed(F&&)

DISCARDED

Promise::discard()

Future::isDiscarded()

Future::onDiscarded(F&&)

Code Style: prefer composition using

Future::then()

and

Future::recover()

over

Future::onReady()

,

Future::onFailed()

,

Future::onDiscarded()

, and

Future::onAny()

. A good rule of thumb is if you find yourself creating your own instance of a

Promise

to compose an asynchronous operation you should use composition instead!

We use the macros

CHECK_PENDING()

CHECK_READY()

CHECK_FAILED()

CHECK_DISCARDED()

CHECK()

Discarding a Future (aka Cancellation)

You can "cancel" the result of some asynchronous operation by discarding a future. Unlike doing a discard on a promise, discarding a future is a request that may or may not be be satisfiable. You discard a future using

Future::discard()

Future::hasDiscard()

Future::onDiscard()

using process::Future;
using process::Promise;

int main(int argc, char** argv)
{
  Promise promise;
  Future future = promise.future();
  CHECK_PENDING(future);
  future.discard();
  CHECK(promise.future().hasDiscard());
  CHECK_PENDING(future); // THE FUTURE IS STILL PENDING!
  return 0;
}

The provider of the future will often use

Future::onDiscard()

using process::Future;
using process::Promise;

int main(int argc, char** argv)
{
  Promise promise;
  // Set up a callback to discard the future if
  // requested (this is not always possible!).
  promise.future().onDiscard(& {
    promise.discard();
  });
  Future future = promise.future();
  CHECK_PENDING(future);
  future.discard();
  CHECK_DISCARDED(future); // NO LONGER PENDING!
  return 0;
}

Abandoned Futures

An instance of

Promise

PENDING

PENDING

You can check if a future has been abandoned by doing

Future::isAbandoned()

Future::onAbandoned()

using process::Future;
using process::Promise;

int main(int argc, char** argv)
{
  Promise* promise = new Promise();
  Future future = promise->future();
  CHECK(!future.isAbandoned());
  delete promise; // ABANDONMENT!
  CHECK_ABANDONED(future);
  CHECK_PENDING(future); // ALSO STILL PENDING!
  return 0;
}

Composition:

Future::then()

,

Future::repair()

, and

Future::recover()

You can compose together asynchronous function calls using

Future::then()

Future::repair()

Future::recover()

using process::Future;
using process::Promise;

// Returns an instance of Person for the specified name.
Future find(const std::string& name);
// Returns the mother (an instance of Person) of the specified name.
Future mother(const std::string& name)
{
  // First find the person.
  Future person = find(name);
  // Now create a Promise that we can use to compose the two asynchronous calls.
  Promise* promise = new Promise();
  Future mother = promise->future();
  // Here is the boiler plate that can be replaced by Future::then()!
  person.onAny([](const Future& person) {
    if (person.isFailed()) {
      promise->fail(person.failure());
    } else if (person.isDiscarded()) {
      promise->discard();
    } else {
      CHECK_READY(person);
      promise->set(find(person->mother));
    }
    delete promise;
  });
  return mother;
}

Using

Future::then()

using process::Future;

// Returns an instance of Person for the specified name.
Future find(const std::string& name);
// Returns the mother (an instance of Person) of the specified name.
Future mother(const std::string& name)
{
  return find(name)
    .then([](const Person& person) {
      return find(person.mother);
    });
}

Each of

Future::then()

Future::repair()

Future::recover()

| Transition |

Future::*()

READY

Future::then(F&&)

FAILED

Future::repair(F&&)

Future::recover(F&&)

DISCARDED

Future::recover(F&&)

PENDING

Future::isAbandoned()

Future::recover(F&&)

Future::then()

Future

Future::repair()

Future::recover()

Future

Future::recover()

using process::Future;

// Returns an instance of Person for the specified name.
Future find(const std::string& name);
// Returns a parent (an instance of Person) of the specified name.
Future parent(const std::string& name)
{
  return find(name)
    .then([](const Person& person) {
      // Try to find the mother and if that fails try the father!
      return find(person.mother)
        .recover([=](const Future&) {
          return find(person.father);
        });
    });
}

Be careful what you capture in your callbacks! Depending on the state of the future the callback may be executed from a different scope and what ever you captured may no longer be valid; see Callback Semantics for more details.

Discarding and Composition

Doing a

Future::discard()

Future::then()

Future::recover()

1.

Future::then()

enforces discards

The future returned by

Future::then()

READY

Future::then()

DISCARDED

These semantics are surprising to many, and, admittedly, the library may at one point in the future change the semantics and introduce a

discardable()

helper for letting people explicitly decide if/when they want a callback to be discarded. Historically, these semantics were chosen so that people could write infinite loops using

Future::then()

that could be interrupted with

Future::discard()

. The proper way to do infinte loops today is with

loop()

.

Here's an example to clarify this caveat:

using process::Future;
using process::Promise;

int main(int argc, char** argv)
{
  Promise promise;
  Future<:string> future = promise.future()
    .then([](int i) {
      return stringify(i);
    });
  future.discard();
  CHECK_PENDING(future);
  promise.set(42);
  CHECK_DISCARDED(future); // EVEN THOUGH THE PROMISE COMPLETED SUCCESSFULLY!
  return 0;
}
</:string>

2. Sometimes you want something to be

undiscardable()

You may find yourself in a circumstance when you're referencing (or composing) futures but you don't want a disard to propagate to the referenced future! Consider some code that needs to complete some expensive initialization before other functions can be called. We can model that by composing each function with the initialization, for example:

using process::Future;

Future foo()
{
  return initialization
    .then( {
      return ...;
    });
}

In the above example, if someone were to discard the future returned from

foo()

The real intention here is to compose with the initialization but not propagate discarding. This can be accomplished using

undiscardable()

undiscardable()

using process::Future;
using process::undiscardable;

Future foo()
{
  return undiscardable(initialization)
    .then( {
      return ...;
    });
}

Callback Semantics

There are two possible ways in which the callbacks to the

Future::onReady()

Future::then()

1. By the caller of

   Future::onReady()

   ,

   Future::then()

   , etc.
2. By the caller of

   Promise::set()

   ,

   Promise::fail()

   , etc.

The first case occurs if the future is already transitioned to that state when adding the callback, i.e., if the state is already

READY

Future::onReady()

Future::then()

FAILED

Future::onFailed()

Future::recover()

The second case occurs if the future has not yet transitioned to that state. In that case the callback is stored and it is executed by the caller of

Promise::set()

Promise::fail()

This means that it is critical to consider the synchronization that might be necessary for your code given that multiple possible callers could execute the callback!

We call these callback semantics synchronous, as opposed to asynchronous. You can use

defer()

Note that after the callbacks are invoked they are deleted, so any resources that you might be holding on to inside the callback will properly be released after the future transitions and it's callbacks are invoked.

CHECK()

Overloads

CHECK()

assert

CHECK()

CHECK_PENDING()

CHECK_READY()

CHECK_FAILED()

CHECK_DISCARDED()

CHECK_READY(future)

HTTP

libprocess provides facilities for communicating between actors via HTTP messages. With the advent of the HTTP API, HTTP is becoming the preferred mode of communication.

route

route

using namespace process;
using namespace process::http;

class HttpProcess : public Process
{
protected:
  void initialize() override
  {
    route("/testing", None(), [](const Request& request) {
      return testing(request.query);
    });
  }
};
class Http
{
public:
  Http() : process(new HttpProcess())
  {
    spawn(process.get());
  }
  virtual ~Http()
  {
    terminate(process.get());
    wait(process.get());
  }
  Owned process;
};

Now if our program instantiates this class, we can do something like:

$ curl localhost:1234/testing?value=42

Note that the port at which this endpoint can be reached is the port libprocess has bound to, which is determined by the

LIBPROCESS_PORT

--port

get

get

Future

UPID

URL

UPID

upid

Future future = get(upid, "testing");

Or let's assume our serving process has been set up on a remote server and we want to hit its endpoint. We'll construct a

URL

get

URL url = URL("http", "hostname", 1234, "/testing");

Future future = get(url);

post

and

requestDelete

The

post

requestDelete

get

Connection

A

Connection

connect

Future

Connection

Future connect = connect(url);

connect.await();
Connection connection = connect.get();
Request request;
request.method = "GET";
request.url = url;
request.body = "Amazing prose goes here.";
request.keepAlive = true;
Future response = connection.send(request);

It's also worth noting that if multiple requests are sent in succession on a

Connection

Processes (aka Actors)

An actor in libprocess is called a process (not to be confused by an operating system process).

A process receives events that it processes one at a time. Because a process is only processing one event at a time there's no need for synchronization within the process.

There are a few ways to create an event for a process, the most important including:

• You can

  send()

  a process a message.
• You can do a function

  dispatch()

  on a process.
• You can send an

  http::Request

  to a process.

That last one is exceptionally powerful; every process is also an HTTP-based service that you can communicate with using the HTTP protocol.

Process Lifecycle

You create a process like any other class in C++ but extending from

Process

Process

Process::self()

Practically you can think of a process as a combination of a thread and an object, except creating/spawning a process is very cheap (no actual thread gets created, and no stack gets allocated).

Here's the simplest process you can create:

using process::Process;

class FooProcess : public Process {};

You start a process using

spawn()

terminate()

wait()

using process::Process;
using process::spawn;
using process::terminate;
using process::wait;

class FooProcess : public Process {};
int main(int argc, char** argv)
{
  FooProcess process;
  spawn(process);
  terminate(process);
  wait(process);
  return 0;
}

Memory Management

A process CAN NOT be deleted until after doing a

wait()

spawn()

true

using process::Process;
using process::spawn;
using process::terminate;
using process::wait;

class FooProcess : public Process {};
int main(int argc, char** argv)
{
  FooProcess* process = new FooProcess();
  spawn(process, true); // 
Process Identity: 
PID
 and 
UPID


A process is uniquely identifiable by it's process id which can be any arbitrary string (but only one process can be spawned at a time with the same id). The 
PID
 and 
UPID
 types encapsulate both the process id as well as the network address for the process, e.g., the IP and port where the process can be reached if libprocess was initialized with an IPv4 or IPv6 network address. You can get the 
PID
 of a process by calling it's 
self()
 method:
using process::PID;
using process::Process;
using process::spawn;

class FooProcess : public Process {};
int main(int argc, char** argv)
{
  FooProcess process;
  spawn(process);
  PID pid = process.self();
  return 0;
}

A 
UPID
 is the "untyped" base class of 
PID
.


If you turn on logging you might see a 
PID
/
UPID
 printed out as 
[email protected]:port
.


Process Event Queue (aka Mailbox)

Each process has a queue of incoming 
Event
's that it processes one at a time. Other actor implementations often call this queue the "mailbox".

There are 5 different kinds of events that can be enqueued for a process:



MessageEvent
: a 
Message
 has been received.

DispatchEvent
: a method on the process has been "dispatched".

HttpEvent
: an 
http::Request
 has been received.

ExitedEvent
: another process which has been linked has terminated.

TerminateEvent
: the process has been requested to terminate.


An event is serviced one at a time by invoking the process' 
serve()
 method which by default invokes the process' 
visit()
 method corresponding to the underlying event type. Most actors don't need to override the implementation of 
serve()
 or 
visit()
 but can rely on higher-level abstractions that simplify serving the event (e.g., 
route()
, which make it easy to set up handlers for an 
HttpEvent
, discussed below in HTTP).

MessageEvent

A 
MessageEvent
 gets enqueued for a process when it gets sent a 
Message
, either locally or remotely. You use 
send()
 to send a message from within a process and 
post()
 to send a message from outside a process. A 
post()
 sends a message without a return address because there is no process to reply to. Here's a classic ping pong example using 
send()
:
using process::MessageEvent;
using process::PID;
using process::Process;
using process::spawn;
using process::terminate;
using process::wait;

class ClientProcess : public Process
{
public:
  ClientProcess(const UPID& server) : server(server) {}
  void initialize() override
  {
    send(server, "ping");
  }
  void visit(const MessageEvent& event) override
  {
    if (event.message.from == server &&
        event.message.name == "pong") {
      terminate(self());
    }
  }
};
class ServerProcess : public Process
{
public:
protected:
  void visit(const MessageEvent& event) override
  {
    if (event.message.name == "ping") {
      send(event.message.from, "pong");
    }
    terminate(self());
  }
};
int main(int argc, char** argv)
{
  PID server = spawn(new ServerProcess(), true);
  PID client = spawn(new ClientProcess(server), true);
  wait(server);
  wait(client);
  return 0;
}

DispatchEvent

TODO.

HttpEvent

TODO.

ExitedEvent

TODO.

TerminateEvent

TODO.


 Processes and the Asynchronous Pimpl Pattern

It's tedious to require everyone to have to explicitly 
spawn()
, 
terminate()
, and 
wait()
 for a process. Having everyone call 
dispatch()
 when they really just want to invoke a function (albeit asynchronously) is unfortnate too! To alleviate these burdenes, a common pattern that is used is to wrap a process within another class that performs the 
spawn()
, 
terminate()
, 
wait()
, and 
dispatch()
's for you. Here's a typical example:
class FooProcess : public Process
{
public:
  void foo(int i);
};

class Foo
{
public:
  Foo()
  {
    process::spawn(process);
  }
  ~Foo()
  {
    process::terminate(process);
    process::wait(process);
  }
  void foo(int i)
  {
    dispatch(process, &FooProcess::foo, i);
  }
private:
  FooProcess process;
};
int main(int argc, char** argv)
{
  Foo foo;
  foo.foo(42);
  return 0;
}

Anyone using 
Foo
 uses it similarly to how they would work with any other synchronous object where they don't know, or need to know, the implementation details under the covers (i.e., that it's implemented using a process). This is similar to the Pimpl pattern, except we need to 
spawn()
 and 
terminate()/wait()
 and rather than synchronously invoking the underlying object we're asynchronously invoking the underlying object using 
dispatch()
.


 Clock Management and Timeouts

Asynchronous programs often use timeouts, e.g., because a process that initiates
an asynchronous operation wants to take action if the operation hasn't completed
within a certain time bound. To facilitate this, libprocess provides a set of
abstractions that simplify writing timeout logic. Importantly, test code has the
ability to manipulate the clock, in order to ensure that timeout logic is
exercised (without needing to block the test program until the appropriate
amount of system time has elapsed).

To invoke a function after a certain amount of time has elapsed, use 
delay
:
using namespace process;

class DelayedProcess : public Process
{
public:
  void action(const string& name)
  {
    LOG(INFO) << "hello, " << name;
promise.set(Nothing());
  }
  Promise promise;
};
int main()
{
  DelayedProcess process;
  spawn(process);
  LOG(INFO) << "Starting to wait";
  delay(Seconds(5), process.self(), &DelayedProcess::action, "Neil");
  AWAIT_READY(process.promise.future());
  LOG(INFO) << "Done waiting";
  terminate(process);
  wait(process);
  return 0;
}

This invokes the 
action
 function after (at least) five seconds of time
have elapsed. When writing unit tests for this code, blocking the test for five
seconds is undesirable. To avoid this, we can use 
Clock::advance
:
int main()
{
  DelayedProcess process;

  spawn(process);
  LOG(INFO) << "Starting to wait";
  Clock::pause();
  delay(Seconds(5), process.self(), &DelayedProcess::action, "Neil");
  Clock::advance(Seconds(5));
  AWAIT_READY(process.promise.future());
  LOG(INFO) << "Done waiting";
  terminate(process);
  wait(process);
  Clock::resume();
  return 0;
}


 Miscellaneous Primitives

async

Async defines a function template for asynchronously executing
function closures. It provides their results as
futures.


 Optimized Run Queue and Event Queue

There are a handful of compile-time optimizations that can be
configured to improve the run queue and event queue performance. These
are currently not enabled by default as they are considered
alpha. These optimizations include:




--enable-lock-free-run-queue
 (autotools) or
-DENABLE_LOCK_FREE_RUN_QUEUE
 (cmake) which enables the lock-free
run queue implementation.


--enable-lock-free-event-queue
 (autotools) or
-DENABLE_LOCK_FREE_EVENT_QUEUE
 (cmake) which enables the lock-free
event queue implementation.


--enable-last-in-first-out-fixed-size-semaphore
 (autotools) or
-DENABLE_LAST_IN_FIRST_OUT_FIXED_SIZE_SEMAPHORE
 (cmake) which
enables an optimized semaphore implementation.


Details

Both the lock-free run queue implementation and the lock-free event
queue implementation use 
moodycamel::ConcurrentQueue
 which can be
found here.

For the run queue we use a semaphore to block threads when there are
not any processes to run. On Linux we found that using a semaphore
from glibc (i.e., 
sem_create
, 
sem_wait
, 
sem_post
, etc) had some
performance issues. We discuss those performance issues and how our
optimized semaphore overcomes them in more detail in
semaphore.hpp.

Benchmark

The benchmark that we've used to drive the run queue and event queue
performance improvements can be found in
benchmarks.cpp. You
can run the benchmark yourself by invoking 
./benchmarks
--gtest_filter=ProcessTest.*ThroughputPerformance
.