#  Faabric: Fine-Grained Distribution of Scientific Workloads in the Cloud

Simon Shillaker
Imperial College London  Carlos Segarra
Imperial College London  Eleftheria Mappoura
Imperial College London  Mayeul Fournial
Imperial College London  Lluís Vilanova
Imperial College London  Peter Pietzuch
Imperial College London

###### Abstract

With their high parallelism and resource needs, many scientific applications
benefit from cloud deployments. Today, scientific applications are executed on
dedicated pools of VMs, resulting in resource fragmentation: users pay for
underutilised resources, and providers cannot reallocate unused resources
between applications. While serverless cloud computing could address these
issues, its programming model is incompatible with the use of shared memory
and message passing in scientific applications: serverless functions do not
share memory directly on the same VM or support message passing semantics when
scheduling functions dynamically.

We describe Faabric, a new serverless cloud runtime that transparently
distributes applications with shared memory and message passing across VMs.
Faabric achieves this by scheduling computation in a fine-grained
(thread/process) fashion through a new execution abstraction called
_Granules_. To support shared memory, Granules are isolated using WebAssembly
but share memory directly; to support message passing, Granules offer
asynchronous point-to-point communication. Faabric schedules Granules to meet
an application’s parallelism needs. It also synchronises changes to Granule’s
shared memory, and migrates Granules to improve locality.

##  1 Introduction

Cloud computing offers on-demand access to plentiful resources, making it an
attractive choice for highly-parallel scientific applications in hydrodynamics
[44], genomics [7], and epidemiology [28]. Such applications often employ
parallel programming models that use multi-threading with _shared memory_
(e.g. OpenMP [45]) and distributed processing with _message passing_ (e.g. MPI
[40]).

Cloud providers have introduced platforms, e.g. AWS Batch [3] and Azure Batch
[30], that support such workloads. These platforms typically dedicate a pool
of virtual machines (VMs) to execute a queue of jobs [30, 3]. This allows such
platforms to support shared memory and message passing applications: the
provider schedules jobs on one or more VMs according to the requested
parallelism (e.g. MPI’s world size or OpenMP’s OMP_NUM_THREADS). Shared memory
applications are normally executed on a single VM with many CPU cores and
large memory sizes.

Deploying scientific applications on a fixed set of VMs, however, leads to
resource fragmentation: if a job does not use all available resources or
cannot be bin-packed onto available VMs, users pay for idle or under-utilised
VMs; providers also cannot exploit a user’s idle resources, diminishing the
efficiency of their infrastructure [3, 30].

To achieve higher utilisation, serverless computing [4, 32, 13] distributes
workloads as fine-grained _functions_ , which execute on large VM clusters,
isolated through lightweight mechanisms (e.g. containers [43], micro-VMs [1,
4], or WebAssembly [15, 60]). Serverless computing thus reduces under-
utilisation by distributing workloads at a fine granularity across machines
[4, 32, 13]. By making more fine-grained decisions about how to allocate each
VM’s resources, a provider achieves higher utilisation and thus lower per-
tenant costs.

The programming models used in scientific applications, namely shared memory
and message passing [23], make it challenging to support such applications in
serverless environments: (1)  _shared memory_ applications rely on threads for
parallelism, requiring state to be shared: threads must access the same
address space, which is not possible among serverless functions running in
different containers, potentially on different VMs. In addition, when deployed
in a serverless setting, multi-threaded code is restricted to the parallelism
available within a function’s isolation boundary, e.g. a single container; and
(2)  _message passing_ applications need a fixed-size pool of stateful
processes to support consistent synchronous communication – this is
unavailable in a serverless setting, in which functions tend to be short-
lived, stateless and only communicate through storage [17].

We describe Faabric, a new serverless cloud runtime that executes scientific
applications with shared memory and message passing. For this, Faabric
exploits the new abstraction of _Granules_ , which allow for thread- and
process-granular scheduling. Faabric does not require changes to existing
parallel programming models: it transparently executes applications that use
OpenMP API [45] for shared memory or MPI API [40] for message passing. It
achieves this through the following contributions:

(1) Supporting multi-threading/processing via Granules. Faabric executes
applications as batches of distributed _Granules_ (§3.1) running on shared
VMs. A Granule can share memory with other Granules to offer thread semantics,
or have private memory for process semantics. Granules can be spawned (or
migrated) based on a _snapshot_ taken from a parent Granule. Faabric uses
WebAssembly [16] to isolate Granules and take snapshots of a Granule’s state:
its shared/private memory, message queues, address information and execution
state, e.g. stack pointers and function tables.

(2) Transparent distribution of Granules. Faabric allows the provider to
distribute computation using Granules. A scheduler in Faabric can choose to
spawn or migrate Granules across VM. It makes scheduling decisions when
Granules reach _control points_ , which are triggered by system calls and
calls to parallel APIs. At each control point, Faabric may spawn new Granules
to add a logical thread or process to the application, increasing parallelism;
or migrate existing Granules, e.g. to increase locality of execution.

(3) Distributed synchronisation of address spaces. To support shared memory
programming, Faabric must offer sequential consistency within each Granule,
provide distributed synchronisation primitives, e.g. mutexes and barriers, and
synchronise distributed writes to shared address spaces (§4). Granules
synchronise writes to the address space by building lists of _byte-wise
diffs_. Each Granule maintains a record of writes to shared memory pages,
performs byte-wise comparisons against its parent snapshot and propagates
changes back to a main Granule via byte-wise diffs. To support updates to
shared variables across Granules, byte-wise diff specify a _merge_ operation,
e.g. summation over shared variables.

(4) Asynchronous messaging through Granule groups. To support message passing
between Granules, Faabric organises them into _Granule groups_. Each Granule
is assigned an index within the group and can send/receive messages to/from
Granules in the group. Faabric maintains a set of queues for each Granule to
buffer incoming messages, thus sending/receiving messages asynchronously
without the need for Granules to have been scheduled. This prevents message
loss during Granule migration. Faabric implements common collective
communication operations, such as all-reduce [41]. Its implementation uses
fast in-memory message exchange between co-located Granules on the same VM.

In our evaluation, we use Faabric to execute scientific applications
implemented using OpenMP and MPI (LAMMPS [55], and the ParRes kernels [48])
and compare them to native OpenMP and OpenMPI. When executing a queue of 100
applications on a 32 VM cluster, Faabric can reduce makespan by up to 23%
thanks to its granular scheduling of threads and processes.

##  2 Scientific Applications in Cloud

Domain | Name | Language | SM | MP
---|---|---|---|---
Molecular dynamics |  LAMMPS [54] | C++ | ✗ | ✓
MDAnalysis [29] | Python | ✓ | ✗
Bio-informatics |  BioPython [9] | Python | ✓ | ✗
gatk [10] | Java | ✓ | ✗
Fluid dynamics |  OpenFOAM [44] | C++ | ✗ | ✓
SU2 [62] | C++ | ✓ | ✗
Deep learning |  OpenCV [42] | C++ | ✗ | ✓
Tensorflow [63] | Python | ✓ | ✗
Table 1: Github’s most-starred projects in scientific domains use shared
memory (SM) or message passing (MP)

Next, we outline the benefits and challenges associated with using cloud
models for scientific applications (§2.1). We then analyse what support
representative APIs for shared memory and message passing require from their
execution environment (§2.2). Based on this, we develop a list of features
necessary to deploy such applications in serverless cloud (§2.3).

###  2.1 Cloud models for scientific applications

Cloud platforms for scientific applications such as AWS Batch [3] and Azure
Batch [30] allocate a dedicated pool of VMs to execute jobs. Each VM has the
same size, as determined by the number of vCPUs and memory in GB. For general
purpose VMs, memory increases linearly with vCPUs and price per hour [33].
Jobs in a queue are assigned VMs [38] based on their requested parallelism.
For example, MPI jobs specify the number of processes through the command
line: mpirun -np <num_processes> and are assigned enough VMs such that the sum
of vCPUs is greater or equal to the requested processes.

This introduces efficiency challenges due to fragmentation: if a job does not
use all of its assigned resources, these resources are wasted. For example,
the number of MPI process requested may not be a multiple of the number of
vCPUs per VM. In addition, different jobs cannot execute concurrently on the
same VM [34]. In all of these cases, users pay for under-utilised or idle
resources, and providers cannot allocate these resources to other jobs in the
queue.

Reducing the VM size reduces fragmentation but impacts performance: message
passing jobs become less co-located, and shared memory jobs have less
available memory. Allocating a mix of different VM sizes in the pool only
partially alleviates these problems, because the resource requirements of jobs
are unknown ahead of time, making it challenging for providers to provide the
right distribution of VM sizes.

In response, cloud platforms have increasingly adopted fine-grained
distribution to reduce costs for users, and increase infrastructure
utilisation [6, 19]. This has culminated in today’s serverless cloud
offerings, such as AWS Lambda [4] and Azure Functions [32], in which providers
take full control of the parallelism and distribution of applications, billing
users to a millisecond granularity [17].

Serverless applications are divided into thousands of small stateless tasks,
which can be parallelised and distributed [4, 32, 13]. This gives providers
the flexibility to allocate resources, and execute functions on those
resources, according to bespoke policies. The finer-grained the serverless
functions become, the higher the packing density that the provider can achieve
and the more control a provider has over the resource allocation to each
application.

###  2.2 Shared memory and message passing APIs

Although scientific applications cover diverse domains, many employ the same
two programming models: shared memory and message passing. Shared memory is
used for parallelism within a single machine, e.g. using multi-threading
libraries such as OpenMP [45]; and message passing is used for parallelism
across machines, e.g. using MPI [40] and multi-processing. Tab. 1 shows the
most starred open-source repositories in several scientific application
domains on Github, which all use either or both of these programming models.

OpenMP and MPI, along with other parallel programming models, offer high-level
declarative APIs that impose certain features on the underlying execution
environment. If we want to deploy such applications with fine-grained
distribution in a serverless cloud environment, the cloud platform must offer
support to parallelise the computation, and partition, distribute and
synchronise private and shared data.

⬇

1int[] weights = initWeights();

2

3for (int i = 0; i < numSteps; i++) {

4#pragma omp parallel shared(weights) {

5 int threadNum = omp_get_thread_num();

6 int nThreads = omp_get_num_threads();

7 updateWeights(weights, threadNum, nThreads);

8#pragma omp single

9 applyWeights(weights);

10}

Listing 1: Pseudocode for machine learning training using OpenMP’s parallel
abstraction (Within the parallel block, the OpenMP runtime controls the degree
of parallelism, and ensures access to and synchronisation of the shared
variable weights.)

LABEL:lst:sm shows a sample OpenMP implementation of stochastic gradient
descent (SGD) [53], a core algorithm in machine learning training. The omp
parallel construct requests that a for loop be executed in parallel with
access to a single shared variable, the weights vector. The OpenMP runtime has
control over the underlying threads and data partitioning between them. It
must ensure read-only access to the shared address space, except for the
weights vector, which receives synchronised writes from multiple threads.

The environment used to execute this code must spawn and execute parallel
threads, each with access to a shared address space, and provide
synchronisation primitives for accessing shared variables. This is trivial on
a single host but becomes difficult if we distribute the computation. In a
distributed environment, the shared address space must be made available and
synchronised across VMs. Any coordination primitives, such as locks, must also
operate in a distributed manner.

As another example, LABEL:lst:mp shows an SGD implementation based on MPI’s
MPI_Allreduce() operation: concurrent processes execute the same function to
transform and aggregate results on multiple VMs. The MPI runtime manages the
processes, data transfers and messaging topology: it sums the weights and
broadcasts the result to all processes.

⬇

1int worldSize, rank;

2MPI_Comm_size(MPI_COMM_WORLD, &worldSize);

3MPI_Comm_rank(MPI_COMM_WORLD, &rank);

4int[] weights = initWeights();

5

6for (int i = 0; i < numSteps; i++) {

7 updateWeights(weights, rank, worldSize);

8 MPI_Allreduce(MPI_IN_PLACE, weights, nWeights,

9 MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);

10 if(rank == 0) applyWeights(weights);

11}

Listing 2: Pseudocode for machine learning traning using MPI’s MPI_Allreduce
function (The MPI runtime controls the degree of parallelism, data
partitioning and messaging topology.)

The execution environment for this code must provide a fixed-size pool of
processes, each of which maintains in-memory state, and can send/receive
messages to/from the others in the group. The processes count cannot be
changed throughout the application’s lifetime due to the complexity of
maintaining consistent in-memory process state, and preserving the rank-based
addressing scheme between processes. This makes it challenging to vary the
assigned resources.

###  2.3 Requirements

Platform |  Threads/ processes |  Fixed parallelism |  Shared addr. space |  Direct comms.
---|---|---|---|---
AWS Batch [3] | ✓ | ✓ | ✗ | ✓
Azure Batch [30] | ✓ | ✓ | ✗ | ✓
Azure Dur. Funcs. [35] | ✗ | ✓ | ✗ | ✗
AWS Step Functions [5] | ✗ | ✓ | ✗ | ✗
AWS Lambda [4] | ✗ | ✗ | ✗ | ✗
Azure Functions [32] | ✗ | ✗ | ✗ | ✗
Crucial [8] | ✓ | ✗ | ✗ | ✗
Faasm [60] | ✗ | ✗ | ✓ | ✗
Faastlane [20] | ✗ | ✗ | ✓ | ✗
Kappa[65] | ✓ | ✗ | ✗ | ✓
SAND [2] | ✗ | ✗ | ✗ | ✓
Faabric | ✓ | ✓ | ✓ | ✓
Table 2: Cloud support for scientific workloads (We differentiate between
batch-compute, serverless-compute and academic.)

To perform fine-grained distribution of the code in LABEL:lst:sm and
LABEL:lst:mp, a cloud platform must support the following features:
thread/process semantics, fixed parallelism, shared address space and direct
communication. Tab. 2 summarises the support for these features in today’s
scientific cloud platforms, and contrasts them with serverless platforms.

Thread/process semantics. When an application uses threads or processes for
parallel computation, the execution environment must (i) fork and join child
processes, which duplicate the parent’s process state; and (ii) spawn and join
threads, which share the parent’s address space.

Existing scientific and serverless cloud platforms support thread/process
semantics, but only _within_ a given VM or serverless function. Although each
VM/function can fork a process/thread, the degree of parallelism is limited by
the resources allocated to the host. For example, on AWS Lambda [4], it is
possible to spawn new threads but they run inside the same function, competing
for the same resources. This limits the flexibility afforded to the provider,
preventing them from arbitrarily distributing each thread and process on any
VM, e.g. to achieve optimal bin-packing of multiple tenants.

Control of parallelism. Typically, parallel applications use a known number of
threads/processes and employ synchronisation primitives (e.g. barriers,
mutexes and locks) to ensure correct execution. A known parallelism degree
allows applications to partition data and computation appropriately.

Existing scientific cloud platforms [3, 30] provide a fixed level of
parallelism to each job. They only allocate each VM to one job, resulting in
unused resources if the job does not exploit a VM’s full parallelism.
Serverless platforms, on the other hand, do not guarantee a fixed level of
parallelism, instead allocating available resources from the shared
infrastructure. While this means that serverless platforms cannot execute
scientific workloads, they can perform fine-grained distribution according to
their own scheduling policies. In an ideal scientific computing environment,
the platform would both guarantee a fixed level of parallelism, while
retaining control over its own fine-grained scheduling decisions.

Shared address space. As shown in LABEL:lst:sm, multi-threaded applications
assume that threads share the same virtual address space. To avoid concurrency
issues, access to the address space must be coordinated. Standard libraries
[26, 52] and OSs provide implementations of synchronisation primitives such as
mutexes, semaphores and barriers.

As with thread/process semantics, today’s cloud-based scientific and
serverless platforms both support shared memory within a single VM/function.
This means that shared memory parallelism is limited to the scale of a single
VM/function, preventing the provider from arbitrarily distributing threads.
The ideal scientific computing environment would be able to distribute
threads, while maintaining a distributed shared address space across them, as
well as providing distributed coordination primitives.

Figure 1: Key abstractions in Faabric (Granules are restored from snapshots,
and each Granule is a member of a Granule group).

Direct communication. Each process in a message-passing application must be
able to transfer data to other processes (see LABEL:lst:mp). In MPI, this is
done based on an address represented as an integer (rank).

Existing scientific cloud computing platforms support low-latency point-to-
point communication for applications, as long as the available parallelism is
sufficient to execute a fixed-size pool of threads. These platforms, however,
cannot redistribute resources within a running application, as they cannot
migrate processes between VMs. Serverless platforms isolate functions even if
they belong to the same application, preventing them from obtaining stable
identities for communication. An ideal scientific cloud environment would
enable long-lived stateful processes with direct communication, yet allow the
resources allocated for these processes to be migrated and shared between
applications.

##  3 Executing Threads and Processes

Faabric is a serverless cloud runtime that supports multi-threading and multi-
processing, yet offers fine-grained distribution. It introduces a new parallel
computing primitive, the _Granule_ , with thread and process semantics (§3.1).
Granules allow Faabric to control an application’s parallelism and
distribution via _control points_ (§3.2). At control points, the Faabric
runtime interrupts the application to add, remove or migrate Granules (§3.3).
Faabric thus efficiently executes parallel applications and gives the cloud
provider control over scheduling (§3.4).

###  3.1 Granule abstraction

Figure 2: Thread/process semantics with Granules

To support multi-threading and multi-processing, Faabric uses Granules.
Granules can be snapshotted and restored across VMs to support the
parent/child semantics of threads and processes. They also share a single
distributed address space for shared memory programming (§4). Finally, they
support the direct exchange of messages within Granule groups for message
passing (§5).

Fig. 1 gives an overview of the key abstractions. Each Granule executes
application code compiled to WebAssembly [16], a binary platform-independent
execution format. The use of WebAssembly enforces lightweight memory safety:
its isolation mechanism allows Granules to execute side-by-side in a single
instance of the Faabric runtime. It also allows for an efficient snapshotting
mechanism because the complete execution state of a Granule is captured in the
single linear memory array of a WebAssembly module.

Faabric creates a Granule by restoring it from a _snapshot_ , which has a copy
of a Granule’s execution state: its linear memory, mutable global variables, a
function table and stack pointers. To restore a Granule, Faabric copies the
stack pointer, function table and globals from the snapshot into the Granule,
and creates a copy-on-write mapping of the Granule memory onto the snapshot’s
linear memory.

Fig. 2 shows how Faabric uses Granules and snapshots to replicate process and
thread semantics across VMs. Each application has a _base snapshot_ , whose
linear memory contains the static data of the application. Granules restored
from point-in-time snapshots of their parent Granules; Granules with _thread_
semantics share a single linear memory mapping with other Granules on that VM.

###  3.2 Intercepting execution with control points

Figure 3: Control points (Control points are triggered when application code
calls functions from supported APIs, before the Faabric implementation of the
given function is executed.)

To execute and distribute scientific applications in a cloud environment,
Faabric must interrupt the application execution periodically: it must (i)
provision new Granules when the degree of parallelism used by the application
changes; (ii) migrate Granules as dictated by the scheduler to improve
locality and utilisation; (iii) synchronise shared memory between VMs; and
(iv) deliver messages between Granules.

As shown in Fig. 3, Faabric triggers _control points_ when an application
invokes certain system calls and parallel programming APIs. Granules execute
application code compiled to WebAssembly, and WebAssembly can pass control to
the execution runtime for arbitrary listed functions. Faabric uses this
approach to transfer control to the runtime on system calls related to thread
and process operations, e.g. pthread_create() and fork(), as well as functions
from OpenMP and MPI APIs, e.g. MPI_Allreduce().

When making such function calls, control passes to the Faabric runtime at a
control point. Before the runtime executes an API or system call
implementation, it may perform one or more of the following actions: (a) spawn
Granules to execute new logical threads/processes, e.g. on fork(); (b) await
the completion of Granules to replicate joining a thread or awaiting process
completion, e.g. on a call to pthread_join(); (c) merge changes to a shared
address space using byte-wise diffs (§4), e.g. when completing OpenMP parallel
sections; (d) send/receive messages between the Granules (§5.1), e.g. due to
MPI_Send(); and (e) migrate a Granule (§3.3) to another host, e.g. on a call
to MPI_Barrier().

###  3.3 Migrating Granules across VMs

Figure 4: Granule migration at barrier control points

Cloud providers must retain control over the scheduling of Granules on VMs,
e.g. to increase host utilisation or to co-locate Granules belonging to the
same tenant. Faabric achieves this control despite the long-lived execution of
Granules because Granules can be _migrated_ between VMs. Migration decisions
are determined by a _scheduling policy_ , e.g. bin-packing to the fewest VMs,
load-balancing across VMs, or exploiting locality for Granules of a single
application.

To simplify the migration process, Granule migration may only be carried out
at _barrier control points_. These are control points that block all Granules
of an application, e.g. as triggered by calls to OpenMP’s barrier directive,
MPI’s MPI_Barrier() or MPI_Allreduce() functions.

Fig. 4 illustrates Granule migration. When Granules reach a barrier control
point, they wait for a notification from the application’s main VM. When the
main Granule reaches the barrier control point on the main VM, it queries the
_scheduler_ for migration decisions. The scheduler, periodically and in the
background, applies its scheduling policy and decides on function migrations
if the current function execution deviates from the desired allocation. It
then sends messages to all Faabric runtimes on the VMs involved in the
migrations.

To migrate a Granule, the involved Faabric runtimes reserve the necessary
resources for the Granules. If the resources have become unavailable, the
migration is aborted. After that, the Granules to be migrated perform the
migration and notify the main VM. Once the main VM has received notifications
from all migrated Granules, it allows the Granules to exit the barrier control
point.

The actual migration of a Granule is performed via the same mechanism that
Faabric uses to create child processes and threads. The migrating Granule
takes a snapshot, and sends the snapshot as part of a migration request to the
target VM. The target VM creates a new Granule with the required semantics,
i.e. with a new private mapping of the snapshot’s linear memory for a process,
or sharing a linear memory mapping with existing Granules for a thread.

###  3.4 Faabric architecture

Figure 5: Faabric architecture (Faabric runtime instances act as either the
_main_ VM or _worker_ VM for each application. They add, remove and migrate
Granules, synchronise replicas of snapshots via byte-wise diffs, and
asynchronously pass messages between Granules.)

Fig. 5 shows Faabric’s architecture. A Faabric runtime executes on each VM,
controlling a variable-sized pool of Granules, snapshots and message queues.
Each Granule runs a single thread/process from an application.

Faabric uses a distributed shared state _scheduler_ : the Faabric runtime on
each VM has a _local_ scheduler, which communicates with the other schedulers
in the cluster. The local scheduler allocates up to one Granule per CPU core.
If the thread or process executing in a given Granule requests more
parallelism, the local scheduler creates new Granules on that VM. If that VM
has exhausted its CPU cores, the local scheduler chooses another VM,
preferably one that already executes Granules for that application, as it then
holds the application code and Granule snapshots in memory. If no such VM is
available, the scheduler selects the VM with the most available resources. It
then transfers the required snapshot to the new VM, and requests that it
create and execute the new Granule.

Similar to most MPI implementations, Faabric currently does not offer fault-
tolerance features—when one or more Granules fail, the whole application
fails. Fault tolerance could be added to Faabric by exploiting Granule
snapshots as checkpoints. If a Granule fails, the most recent snapshot can be
restored, either on the same or on a different VM. Incoming messages to a
Granule can be persisted by the local Faabric runtime until the next snapshot
is reached, and replayed after Granule failure.

##  4 Shared Memory Programming

To execute multi-threaded applications correctly with shared memory, Faabric
must provide consistency guarantees across Granules and VMs, alongside
suitable synchronisation primitives. Faabric makes shared memory consistent by
sending _byte-wise diffs_ between Granules and VMs that communicate updates to
the shared address space (§4.1).

Programming models such as OpenMP use _reductions_ to aggregate parallel
updates to shared variables without the coordination overhead of mutexes.
Faabric supports reductions using _merge operations_ , which allow it to
combine multiple byte-wise diffs to the same memory region using arithmetic
operations (§4.2). Finally, Faabric provides custom implementations of
coordination primitives including mutexes, barriers and latches (§4.3).

###  4.1 Synchronising changes to shared memory

By default, multi-threaded applications assume only weak consistency
guarantees on the memory shared between threads; stronger consistency is
requested explicitly through synchronisation primitives. Assuming code is free
from data races, Faabric must correctly execute multi-threaded applications:
it must ensures that writes to shared memory from a child thread are visible
to the parent thread when it joins that child. Changes must be visible to all
threads when entering a critical section, or exiting an explicit or implicit
barrier [45].

When Faabric needs to execute a child thread, it creates a new Granule from a
snapshot of the main Granule. This snapshot is maintained until all child
threads have finished execution, and acts as the _main snapshot_ for the
shared address space. Faabric synchronises all subsequent changes across
Granules via this main snapshot: it receives updates to it from other Granules
and VMs, and uses it to calculate updates to send to remote VMs.

A Granule maps its linear memory from the main snapshot if executing on the
main VM, or a replica of the main snapshot if executing on a worker VM. The
Granule then tracks the changes made to the shared address space by
application code. Faabric write-protects all memory pages of the Granule’s
linear memory using mprotect() [25] and handles the page faults caused by
application code by marking the page dirty and resetting its read/write
permissions.

To send these changes back to the main VM when the Granule completes or
reaches a barrier, the Granule performs a byte-wise comparison of the modified
pages with its local copy of the main snapshot. This results in a list of
_byte-wise diffs_ that specify the offset at which the changes occurred and
the modified bytes. The main VM receives these byte-wise diffs from worker VMs
and uses them to update the main snapshot.

The Faabric runtime must update the main snapshot replicas on remote VMs, e.g.
when exiting a barrier. It transmits only the byte-wise diffs required to
update the remote replicas, and not the whole snapshot. To enable this, the
Faabric runtime on the main VM keeps track of which bytes have been updated by
incoming byte-wise diffs, then sends a new set of byte-wise diffs with these
changes to the worker VMs.

Figure 6: Synchronisation of a shared address space (Distributed Granules
execute an OpenMP parallel section with a barrier.)

Fig. 6 gives an example of shared memory synchronisation, which shows how
Faabric executes an implementation of the SGD example (LABEL:lst:sm) using 3
Granules across 2 VMs. 1 When the main Granule enters an OpenMP parallel
section, it triggers a control point at which Faabric creates the main
snapshot of the shared address space from the main Granule. 2 Faabric then
creates 2 more Granules from a replica of this main snapshot on the worker VM,
and each Granule executes the body of the parallel section. 3 When all
Granules have reached the barrier, each creates a list of byte-wise diffs that
the Faabric runtime on the main VM uses to update the main snapshot. 4 On
exiting the barrier, the main VM’s Faabric runtime sends another list of byte-
wise diffs to update the snapshot replica on the worker VM. All Granules then
remap their own linear memory to the local copy of the snapshot and continue
execution. 5 At the end of the parallel section, the parent Granule joins the
child Granules, and again Faabric uses the byte-wise diffs from each Granule
to update the main snapshot. Finally, the main Granule remaps its memory from
the snapshot.

###  4.2 Supporting reductions on shared variables

Figure 7: Reductions in OpenMP (Granules on 2 VMs make non-conflicting updates
in a parallel section, and perform a reduction.)

Declarative multi-threading frameworks such as OpenMP allow multiple threads
to aggregate changes to shared variables using _reductions_ [46]. A reduction
spawns multiple threads that update one or more shared variables in parallel,
then aggregates those updates once all threads have completed execution, e.g.
via a summation. By deferring the aggregation of concurrent updates to shared
variables until threads have completed, we can avoid the overhead that would
otherwise be incurred from synchronising those updates using a mutex. Faabric
distributes multi-threaded applications, hence reducing synchronisation
between threads via reductions reduces cross-VM coordination overheads.

To distribute reductions in Faabric, the runtime on each VM performs the
reduction operation locally for the Granules executing on that VM, then
transmits its updates back to the main VM as a byte-wise diff. For each byte-
wise diff sent back to the main VM, Faabric specifies a _merge operation_ ,
with the arithmetic operation that should be used to apply that byte-wise diff
to the main copy of the shared variable.

Fig. 7 shows OpenMP code for a parallel section that performs disjoint updates
to a shared vector and a reduction section to update a shared variable via a
summation. Faabric spawns 3 child Granules when the main Granule reaches the
parallel section, creating the main snapshot on the main VM and a replica on
the worker VM. Each Granule maps its linear memory from its local copy of the
snapshot.

In the first parallel section, each Granule updates its value in the w vector.
The resulting byte-wise diffs can be written directly to the main snapshot
without a merge operation. In the reduction section, each thread updates their
local copy of the variable x, generating a byte-wise diff on the same memory
region. Since the reduction specifies a summation over x, Faabric combines
these byte-wise diffs in the main snapshot using a sum.

Tab. 3 lists the merge operations supported by Faabric. They include simple
arithmetic operations of summation, subtraction, multiplication and division,
as commonly found in parallel reductions. The operations involve four values:
$A_{0}$, the starting value in the main snapshot; $B_{0}$, the value held in
the copy of the snapshot on the remote VM; $B_{1}$, the updated value after
the thread has executed on the remote VM; and $A_{1}$, the value written to
the main snapshot by the operation.

Merge operation | Formula | Data types
---|---|---
sum | $A_{1}=A_{0}+(B_{1}-B_{0})$ | All numeric
subtract | $A_{1}=A_{0}-(B_{0}-B_{1})$ | All numeric
multiply | $A_{1}=A_{0}\,*\,(B_{1}\,/\,B_{0})$ | All numeric
divide | $A_{1}=A_{0}\,/\,(B_{0}\,/\,B_{1})$ | All numeric
overwrite | $A_{1}=B_{1}$ | Arbitrary bytes
Table 3: Merge operations supported by Faabric. (Faabric overwrites the
original value $A_{0}$ in the main snapshot with value $A_{1}$. $B_{0}$ is the
value seen in the snapshot on the remote VM before the Granule executed, and
$B_{1}$ is the value after execution.)

###  4.3 Synchronisation primitives

In addition to providing shared memory and reduction operations, Faabric must
support the synchronisation primitives in multi-threaded code that control
concurrent access to shared data. Faabric offers the following primitives:

Mutexes. A mutex guarantees that only one Granule can access data at a given
time. In Faabric, application code that acquires a mutex triggers a control
point, and the associated Granule requests a lock on the mutex from the
Faabric runtime. When locking the mutex, the Faabric runtime returns the byte-
wise diffs to update that Granule’s local copy of the shared memory snapshot.
This way, the Granule holding the mutex is guaranteed to observe the updates
of other Granules that have also held it; when releasing the mutex, the
Granule returns its own set of byte-wise diffs to the Faabric runtime.

Atomic operations. Atomic arithmetic operations do not guarantee consistency,
only atomicity. To perform such operations, each Granule acquires a VM-local
mutex to avoid data races on the local copy of the shared memory. Faabric then
uses a merge operation corresponding to the arithmetic operation to merge the
resulting byte-wise diffs.

Barriers. A barrier is either implicit or explicit: an implicit barrier is
introduced by a parallel section; an explicit barrier is added manually.
Barriers require that all Granules block until they have completed the
barrier. Afterwards, all Granules must observe a consistent view of the shared
memory. On entering a barrier, Granules send their byte-wise diffs to the main
VM and block. After all threads have completed, the main VM sends the
aggregated byte-wise diffs to all Granules, which unblock.

Latches. A latch allows Granules to decrement a counter and/or wait for it to
reach zero. Latches are used implicitly in nowait OpenMP operations: the main
Granule blocks until all child Granules have reached the latch. Granules can
make non-blocking requests to the main VM to decrement a latch, or blocking
requests to wait for the latch to reach zero.

##  5 Message Passing

To support multi-process applications with message passing, Faabric must
associate Granules with long-lived identities for communication. Faabric
organises Granules into _Granule groups_ , in which each Granule is assigned
an index. It can then asynchronously send and receive messages to and from
other Granules in the group by referring to that index (§5.1).

When migrating Granules between VMs, each Faabric runtime instance updates its
local Granule group metadata to ensure consistent message delivery independent
of each Granule’s placement (§5.2). For efficient collective communication,
Faabric provides Granule group-aware implementations of operations such as
all-reduce and broadcast, maximising fast intra-VM messaging via in-memory
data structures (§5.3).

###  5.1 Communication in Granule groups

Each Granule that executes a process in a multi-process application may need
to send messages directly to another Granule, e.g. to fulfil API calls in a
message-passing framework such as MPI. To enable such message passing
efficiently, Faabric allows asynchronous messaging between Granules within a
Granule group. The asynchronous nature of Faabric’s message passing avoids
blocking a sender Granule while the receiver Granule is being migrated or
initialised.

By default, all Granules that execute an application are in the same Granule
group. Faabric may create new Granule groups on control points, e.g. as
triggered by MPI_Comm_create(), which allows the application to control
communication groups. A Granule group assigns each Granule an _index_ , which
Faabric uses as an address for that Granule in its implementation of message
passing APIs. Each Faabric runtime that executes a Granule holds a replica of
the Granule group metadata with an _address table_ that maps the indexes of
Granules in the Granule group to the VM on which they have been scheduled. For
each Granule of the group executed on a VM, the Faabric runtime has a set of
queues to buffer messages sent to that index.

When a Granule triggers a control point that requires sending a message, e.g.
MPI_Send(), the Faabric runtime on the VM of the sending Granule looks up the
recipient index in the address table for that group. If the recipient is on
the same VM, Faabric directly enqueues the message on the relevant in-memory
queue. This results in low-latency intra-VM messaging compared to using the
local loopback network interface or inter-process communication (IPC). If the
recipient is on another VM, Faabric sends the message to the runtime on that
VM, where it is enqueued.

###  5.2 Groups when migrating Granules

Figure 8: Preserving a Granule group while migrating (One Granule from the
Granule group is migrated from a worker VM to the main VM, and Faabric updates
the metadata and queues.)

When migrating Granules across VMs (§3.3), the Faabric runtimes must also
update the metadata and queues associated with Granule groups to which the
migrating Granules belong.

Fig. 8 shows how Faabric updates a Granule group during migration. Initially,
the main VM executes one Granule from the group alongside a Granule from
another application; the worker VM executes two other Granules from the group.
When the Granule from the other application completes, it frees up resources
on the main VM; when the Granules reaches a barrier control point, Faabric
migrates one of the Granules from the worker VM to the main VM. Before
completing the migration, each Faabric runtime updates its address table and
creates or deletes queues to accommodate the new or departing Granule,
respectively.

To avoid issues with message delivery arising from migrated Granules while
messages are in-flight or queued, Faabric only inserts barrier control points
on message passing operations where Granules in worker VMs are waiting for one
Granule in the main VM, e.g. MPI’s MPI_Barrier(), MPI_Allreduce() and
MPI_Gather(). This way, Faabric guarantees that there are no messages in-
transit.

###  5.3 Collective communication

Figure 9: MPI collective communication (A VM-leader on each VM sends/receives
intra-VM messages (blue) and sends/receives cross-VM message (red) to/from
other VM-leaders.)

In addition to simple point-to-point messaging, most message passing
applications make use of _collective communication_ operations, such as all-
reduce and broadcast. These operations are widely used in distributed ML
training through specialised libraries [66, 12, 49]. Faabric uses custom
Granule group-aware implementations of these operations, which minimise
latency by exploiting knowledge of Granule placement to maximise intra-VM
messaging.

Fig. 9 shows the message passing performed by Faabric when application code
makes a call to MPI_Allreduce(). 1 When Faabric creates a Granule group, it
selects one Granule on each VM to be the _VM-leader_ for that VM. Any messages
that need to be sent to Granules on other VMs are sent by all Granules to
their VM-leader, which batches the messages into single cross-VM requests.
All-reduce takes place in two steps: 2 an initial reduce in which results from
all Granules on each VM are sent to the main VM via their VM-leader; and 3 a
broadcast of the final result to all Granules, which is delivered via their
VM-leader.

Faabric’s implementation of collective communication operations reduces the
cross-VM messages to one per remote VM involved in each step. It then uses
fast in-memory queues for the Granule to VM-leader communication. This
approach reduces latency (§6.5) and bandwidth usage, and enables pipelining:
after a Granule has asynchronously messaged its local leader, it can continue
execution.

##  6 Evaluation

Our evaluation answers the following questions: (i) what are the benefits of
using Faabric to run scientific applications on shared VMs? (§6.2); (ii) what
is the impact of the cluster size on Faabric? (§6.3); (iii) what is the
performance overhead of executing shared memory applications using Granules?
(§6.4); (iv) what is the performance overhead of executing message passing
applications using Granules? (§6.5); and (v) what is the performance overhead
when migrating Granules at runtime? (§6.6)

###  6.1 Experimental set-up

Implementation. Faabric is written in 24,000 lines of C++20, compiled using
clang-13, and available as open-source at: _removed for anonymity_. Deployed
applications and all transitive dependencies, e.g.  _libc_ , are compiled to
WebAssembly [16] using clang-13 [51], as part of the Faabric CPP toolchain,
also available as open-source at: _removed for anonymity_. The batch scheduler
is written in 1000 lines of Python, and is available at: _removed for
anonymity_.

Test-bed. We deploy Faabric and OpenMPI on a Kubernetes cluster [21] on Azure
[31]. The cluster has 32 _Standard_D8_v5_ VMs [33] with 8 vCPU cores and 32 GB
of memory. Native OpenMP applications execute using the underlying VMs in. We
implement a batch scheduler that monitors cluster resources and executes jobs
as soon as there are enough free resources (in terms of vCPUs).

Workloads. We evaluate Faabric with scientific applications written using
OpenMP [45] for shared memory, and MPI [40] for message passing. We compare
these applications running on Faabric against native implementations running
directly on VMs. All OpenMP code is compiled using clang-13 (OpenMP v4.5)
[27], and we use OpenMPI v4.1 [47]. Faabric is deployed using Kubernetes (K8s)
[21].

###  6.2 Efficiency of running scientific applications

(a) Execution of 100 MPI jobs (Requested parallelism distributed uniformly
from 4–16 vCPUs/MPI processes)

(b) Execution of 100 OpenMP jobs (Requested parallelism distributed uniformly
from 2–8 vCPUs/OpenMP threads)

Figure 10: Execution breakdown of running a trace with 100 jobs (We report the
total time elapsed, i.e. makespan; the CDF of idle vCPUs; and the CDF of job
execution times.)

We explore the efficiency and performance impact of using Faabric to execute
applications on a shared VM cluster. As a workload, we generate a trace of 100
jobs. Each job executes a scientific application with a different level of
parallelism specified by MPI’s world size (as indicated in the mpirun command)
or by the OMP_NUM_THREADS environment variable.

We generate two traces: one with message passing applications (mpi), and one
with shared memory applications (omp). For the message passing applications,
we use LAMMPS [50, 55], a popular molecular dynamics simulator written in C++
using MPI. We pick the Lennard-Jones (LJ) atomic fluid simulation with 4
million atoms, as it is one of the standard benchmarking problems in the
LAMMPS suite [56]. For the shared memory application, we run a dense matrix
multiplication (DGEMM), part of the ParRes kernels [48].

As baselines, we want to compare to different VM sizes, exhaustively exploring
the fragmentation space: larger VM sizes (i.e. higher vCPU counts and larger
memory amounts) yield higher per-job performance but result in more idle
vCPUs; smaller VM sizes lead to lower per-job performance but fewer idle
vCPUs. To avoid managing different VM pools with different VM sizes, we deploy
a single VM pool (32  _Standard_D8_v5_ VMs) and emulate smaller VMs by using
1, 2, 4, or 8 containers, enforcing an even vCPU/memory split using K8s
resource limits [22]. Different jobs execute in different containers: MPI jobs
take up $\lceil\frac{\text{MPI world size}}{\text{vCPUs per ctr}}\rceil$
containers; OpenMP jobs use 1 container, overcommitting vCPUs to threads if
OMP_NUM_THREADS > vCPUs per container.

We configure the batch scheduler to schedule jobs in sequence, as soon as
there are sufficient vCPUs. This means that the scheduling granularity becomes
important: if the cluster is fragmented, jobs will wait for longer in the
queue, increasing makespan; over-fragmenting jobs to pack them tightly will
increase their execution time, also increasing makespan.

Fig. 10 shows the results. For each job trace (mpi in 10(a) and omp in 10(b)),
we report the total time to execute the 100 jobs (i.e. makespan), the CDF of
the percentage of idle vCPUs in the cluster, and the CDF of job execution
times.

In terms of makespan, MPI jobs in Faabric have a 13%–23% lower makespan
compared to all fixed-sized baselines except for 8-ctr-per-vm. 8-ctr-per-vm is
equivalent to running each MPI process in a separate container. Given the
MPI’s message passing nature, this approach is acceptable and performs on-par
with Faabric. For OpenMP jobs, however, using 1 container per vCPU reduces
performance: Faabric’s makespan is 38% lower than 8-ctr-per-vm’s. However,
Faabric has between 8% and 25% higher makespan than the other baselines. This
is due to the overhead of Faabric’s shared memory implementation (see §6.4).
In summary, Faabric achieves lower makespan than the native baselines,
reducing both user costs and provider under-utilisation.

To understand why Faabric reduces the makespan, we analyse the CDF of the
percentage of idle vCPUs, and the CDF of the job execution time. For MPI jobs,
half of time, Faabric has at most 5% of idle vCPUs, whereas the native
baselines leave 10%–30% of vCPUs idle (with the exception of 8-ctr-per-vm).
This shows that Faabric better utilised the available vCPUs by packing and
distributing jobs at a finer granularity using Granules. For OpenMP, 100 jobs
are not enough to saturate the cluster, specially for baselines that
overcommit. Per-job execution time in Faabric is on-par with all baselines for
the MPI jobs, with the exception of the last 15% of jobs – these jobs take
longer to execute because they are over-fragmented. For OpenMP jobs, Faabric’s
execution time becomes worse than the baselines due to the overhead of shared
memory executions. In summary, Faabric achieves lower makespan by allocating
resources at finer-grained granularity, paying the price of over-fragmenting
some jobs, making them run for longer.

###  6.3 Scalability

(a) Makespan

(b) Job execution time

Figure 11: Scaling the VM count (We report the makespan to execute a batch of
jobs, which increases linearly with cluster size, and the distribution of
execution times.)

We explore Faabric’s scalability with respect to the number of VMs in the
cluster. As a workload, we use traces of 50, 100, 200 and 400 MPI jobs in a
cluster with 16, 32, 64 and 128 _Standard_D8_v5_ VMs, respectively. We measure
the makespan and the average job execution time. By increasing the number of
tasks with the cluster size, we expect the makespan to stay constant. We use
the same batch scheduler configuration and baselines from §6.2.

In Fig. 11, we report, for each cluster size, the makespan (11(a)) and the
distribution of execution times with a box plot that includes the median,
boxes for the 25th and 75th percentiles, and whiskers extending 1.5$\times$
the inter-quartile range (11(b)).

Faabric achieves a 7%–16%, 13%–23% and 10%–20% lower makespan for 16, 32 and
64 VMs respectively, because it manages to utilise the cluster resources more
efficiently. Its makespan is on par with native-8, as explained in §6.2. Note
that the makespan values for 16–64 VMs vary within 5%–10% of each other, which
is caused by the different job sizes in each trace. For 128 VMs, however, the
performance of all deployments degrades due to the implementation of our
centralised batch scheduler, which becomes a bottleneck. Faabric’s makespan
also is 5% higher than the baselines because Faabric’s current implementation
centrally manages all registered VMs and their resources.

To explore this performance degradation further, 11(b) shows the distribution
of execution times for each baseline and cluster size using a boxplot. Each
job in the trace has a different level of parallelism, and longer traces have
more jobs, introducing randomness. In spite of this randomness, 11(b) shows
that the 25th, median, and 75th percentiles are very similar across baselines,
and cluster sizes. The whiskers are more variable, as they account for the
tails of the distribution. These results confirm our hypothesis that the
performance degradation in 11(a) is due to limitations of Faabric’s cluster
management and batch scheduler implementation, rather than Granules. These
limitations can be overcome with additional engineering work or by deploying
Faabric across multiple 64 VM clusters, and load-balancing requests across
them.

###  6.4 Shared memory performance

Figure 12: Speed-up executing shared memory applications (When the number of
threads spans multiple VMs, we compare Faabric’s time with OpenMP’s execution
time with 8 threads.)

Next we investigate the performance overhead introduced by Faabric when
executing shared memory applications using OpenMP. We use the same OpenMP
application from §6.2: a dense parallel matrix multiplication (DGEMM) from
ParResKernels [48]. DGEMM is computationally-intensive, with some light use of
Faabric’s shared memory synchronisation and coordination primitives.

Fig. 12 shows the speed-up achieved by Faabric, computed as the ratio between
native OpenMP and Faabric’s execution time. We execute both baselines with
different numbers of OpenMP threads: native OpenMP cannot be scaled out,
limiting its parallelism to that that of 1 VM (i.e. 8 vCPU cores, with each
thread pinned to one core); Faabric can scale out shared memory applications
using Granules. For distributed execution (i.e. with thread counts greater
than 8; faded bars in Fig. 12), we measure the speed-up as the native
execution time with 8 OpenMP threads divided by Faabric’s execution with the
higher thread count.

For DGEMM, Faabric is 20%–30% slower than native OpenMP in a single VM, due to
the overhead of performing floating-point arithmetic in WebAssembly [18]. When
scaling out to another VM, Faabric achieves the optimal native performance in
one VM with 50% more threads. Faabric achieves a 25% speed-up (over native
OpenMP with 8 threads) when executing with twice as many threads. In summary,
Faabric distributes shared memory applications with Granules, surpassing the
performance of a single VM deployment.

###  6.5 Message passing performance

(a) LAMMPS

(b) ParRes Kernels

Figure 13: Speed-up executing message passing applications

This experiment explores Faabric’s performance overhead when executing message
passing applications using MPI. We run the same MPI application as in §6.2:
the LAMMPS simulator running the Lennard-Jones (LJ) benchmark with 4 million
atoms. To stress Faabric’s communication layer, we update LAMMPS’ _controller_
example [57] and increase the synchronisation steps, resulting in three orders
of magnitude more cross-VM messaging. We refer to the LJ benchmark as
_compute-bound_ , and the modified _controller_ one as _network-bound_. We
also run a subset of the ParRes kernels [48] to evaluate specific parts of
Faabric’s implementation.

Fig. 13 shows the speed-up that Faabric achieves, computed as the ratio
between OpenMPI’s and Faabric’s execution times. We execute the two LAMMPS
simulations (13(a)) and the ParRes kernels (13(b)) with different levels of
parallelism.

For LAMMPS’ compute-bound benchmark, Faabric achieves a 5%–10% speed-up over
OpenMPI. Faabric slightly outperforms OpenMPI due to the faster intra-VM
messaging using shared memory and the locality-aware collective communication
implementation. For the network-bound benchmark, Faabric’s execution time is
up to 15% higher than OpenMPI’s. This slow-down is due to an additional level
of indirection in Faabric’s transport layer to support concurrent
applications, which becomes a bottleneck with higher message throughputs.

To analyse the performance of different parts of Faabric’s message passing
implementation further, we execute ParRes kernels [48] for distributed
computation. For the point-to-point messaging kernel (p2p), Faabric achieves a
50%–70% speed-up over OpenMPI because most messages are intra-VM, and Faabric
can use in-memory queues. The nstream kernel updates an array in streaming
fashion and synchronises with a barrier, and Faabric matches the OpenMPI
performance. For the reduction kernel (reduce), Faabric is between 25% faster
and 25% slower in a single VM, and 25%–50% slower when distributing MPI
processes over 2 VMs. Faabric performs worse than OpenMPI when there are more
cross-VM messages, as previously discussed. Finally, for the stencil kernel,
Faabric is up to 30% slower because cross-VM messages dominate execution.

In summary, Faabric’s message passing performance is comparable to that of
OpenMPI. Faabric performs best for intra-VM messages, as it can use its in-
memory queues, and worst for cross-VM messages, which add extra overhead.

###  6.6 Migration of Granules

(a) All-to-all

(b) LAMMPS

Figure 14: Speed-up when migrating Granules (We deploy 8 MPI processes
fragmented across two VMs and migrate 4 at runtime. We report the speed-up
compared to not migrating.)

This experiment measures the benefit of migrating Granules at runtime. As
baselines, we run a compute-bound LAMMPS simulation, and a network-bound _all-
to-all_ kernel that performs synchronisation over a vector in a loop. To
migrate a function, Faabric must guarantee that there are no messages in-
flight, and it uses calls to barrier synchronisation points to check for
migration opportunities.

Fig. 14 shows the speed-up achieved when migrating. We force the scheduler to
over-fragment the jobs, and then migrate at 20%, 40%, 60%, or 80% of
execution. For reference, we also include the speed-up for a co-located
deployment (1 VM).

For a network-bound kernel, over-fragmenting has a high cost: the speed-up for
1 VM is 7.5$\times$. By migrating after 20%, 40%, 60%, and 80%, of execution,
we achieve speed-ups of 3.5$\times$, 2.7$\times$, 1.7$\times$, and
1.2$\times$, respectively. We conclude that it is always worth to migrate
Granules at runtime for network-bound applications.

For a compute-bound kernel, over-fragmenting has a lower cost: the speed-up
for 1 VM is 1.2$\times$. This is because the fragmentation splits 4 processes
in 1 VM, and 4 processes into another, which means that there is substantial
intra-VM messaging. By migrating after 20%, 40%, and 60% of execution, we
achieve speed-ups of 10%, 8%, and 5%, respectively. When migrating after 80%
of execution, the costs of migrating outweigh its benefits, achieving a slow-
down of 5%. LAMMPS has large code and data sections, which leads to larger
Granule snapshot, increasing the cost of migration.

##  7 Related Work

Scientific applications in the cloud. Nowadays, all major cloud providers
offer targeted solutions to support scientific applications in the cloud [7,
37, 36]. To schedule and execute these applications, providers deploy batch
scheduling solutions [30, 3] inspired by HPC batch schedulers, and there is
related work focuses on optimising scheduling decisions [24]. Instead, Faabric
focuses on utilising scheduled resources more efficiently—an orthogonal
problem to better scheduling decisions. Recent work on scheduling for deep
learning training on shared GPU clusters [64] uses traits of the scheduled
resources to improve scheduling decisions, and we plan on exploring this in
the future.

Shared memory and message passing in the cloud. Hoplite [66] uses well-known
collective communication algorithms for building fault-tolerant task-based
distributed systems. Faabric adopts a similar approach: it focuses on dynamic
group membership without considerations of fault-tolerance; Ray [39] is a
distributed system that unifies task-parallel and actor-based computations in
a single interface. It offers transparent state and message passing
irrespective of the distribution, together with transparent unlimited scaling.
Faabric focuses on sharing resources among multiple users more efficiently.

Fine-grained distribution in serverless. Faabric borrows techniques from
research on serverless runtimes to allow fine-grained scheduling and
distribution of scientific applications. Faasm [60] and Cloudburst [61] add
state to serverless functions, but do not provide generic shared memory
required for multi-threading, nor do they support message passing; PLASMA [58]
supports annotations to specify elasticity constraints but only within an
actor-based programming model; Crucial [8] uses Java concurrency abstractions
to execute serverless functions, but it lacks general multi-threading
capability. The lack of communication primitives in serverless has been
recognised as a limitation: SAND [2] includes a message bus, but does not
provide an associated programming model to support collective communication.

Checkpointing and Migration. CloudScale [59] automates fine-grained elastic
resource scaling in a shared (multi-tenant) cluster. CloudScale also uses
migration to correct scheduling or scaling issues. Faabric could benefit from
being used together with CloudScale, as it allows for finer-grained resource
management, allowing CloudScale to operate at a thread/process level rather
than at a VM one. CRIU [11] is a software tool for checkpointing and restoring
processes in userspace. Faabric’s use of WebAssembly means that snapshot can
be obtained more easily without using CRIU or other similar tools such as
DMTCP [14].

##  8 Conclusions

Cloud computing offers on-demand parallelism that is well-suited to scientific
workloads. Today’s cloud services for scientific applications execute
workloads on dedicated VMs, which reduces providers’ abilities to re-allocate
underused resources. Serverless cloud computing promises to overcome these
issues through the fine-grained allocation of tasks to resources.

We have described Faabric, a new cloud runtime that transparently distributes
scientific workloads at a fine granularity while remaining compatible with the
popular OpenMP and MPI APIs. Faabric relies on Granules, which permit the
arbitrary distribution of threads and processes. Its scheduler allocates
Granules in a flexible fashion, allowing Granules to exchange messages
asynchronously and supporting a distributed shared memory implementation.

## References

  * [1] Alexandru Agache, Marc Brooker, Andreea Florescu, Alexandra Iordache, Anthony Liguori, Rolf Neugebauer, Phil Piwonka, and Diana-Maria Popa.  Firecracker: Lightweight virtualization for serverless applications.  In Proceedings of the 17th Usenix Conference on Networked Systems Design and Implementation, NSDI’20, page 419–434, USA, 2020. USENIX Association.
  * [2] Istemi Ekin Akkus, Ruichuan Chen, Ivica Rimac, Manuel Stein, Klaus Satzke, Andre Beck, Paarijaat Aditya, and Volker Hilt.  SAND: Towards High-Performance serverless computing.  In 2018 USENIX Annual Technical Conference (USENIX ATC 18). USENIX Association, 2018.
  * [3] Amazon Web Services.  AWS Batch.  https://aws.amazon.com/batch/, 2021.
  * [4] Amazon Web Services.  AWS Lambda.  https://aws.amazon.com/lambda/, 2021.
  * [5] Amazon Web Services.  AWS Step Functions.  https://aws.amazon.com/step-functions, 2022.
  * [6] Michael Armbrust, Armando Fox, Rean Griffith, Anthony D Joseph, Randy H Katz, Andrew Konwinski, Gunho Lee, David A Patterson, Ariel Rabkin, Ion Stoica, et al.  Above the clouds: A berkeley view of cloud computing.  2009\.
  * [7] AWS.  Genomics in the cloud.  https://aws.amazon.com/health/genomics/, 2022.
  * [8] Daniel Barcelona-Pons, Marc Sánchez-Artigas, Gerard París, Pierre Sutra, and Pedro García-López.  On the FaaS Track: Building Stateful Distributed Applications with Serverless Architectures.  In ACM/IFIP Middleware Conference, 2019.
  * [9] BioPython.  Github - BioPython.  https://github.com/biopython/biopython, 2022.
  * [10] Broad Institute.  Github - gatk.  https://github.com/broadinstitute/gatk, 2022.
  * [11] CRIU.  Checkpoint-Restore in Userspace.  https://www.criu.org/Main_Page, 2021.
  * [12] Facebook Incubator.  Gloo: Collective Communication Library with Various Primitives for Multi-Machine Training.  https://github.com/facebookincubator/gloo, 2020.
  * [13] Google.  Google Cloud Functions.  https://cloud.google.com/functions, 2021.
  * [14] Google.  Github - DeepVariant.  https://github.com/google/deepvariant, 2022.
  * [15] Andreas Haas, Andreas Rossberg, Derek L. Schuff, Ben L. Titzer, Michael Holman, Dan Gohman, Luke Wagner, Alon Zakai, and JF Bastien.  Bringing the Web up to Speed with WebAssembly.  ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI), 2017.
  * [16] Andreas Haas, Andreas Rossberg, Derek L. Schuff, Ben L. Titzer, Michael Holman, Dan Gohman, Luke Wagner, Alon Zakai, and JF Bastien.  Bringing the Web up to Speed with WebAssembly.  ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI), 2017.
  * [17] Joseph M. Hellerstein, Jose M. Faleiro, Joseph Gonzalez, Johann Schleier-Smith, Vikram Sreekanti, Alexey Tumanov, and Chenggang Wu.  Serverless computing: One step forward, two steps back.  ArXiv, 2019.
  * [18] Abhinav Jangda, Bobby Powers, Emery Berger, and Arjun Guha.  Not So Fast: Analyzing the Performance of WebAssembly vs. Native Code.  In USENIX Annual Technical Conference (USENIX ATC), 2019.
  * [19] Eric Jonas, Johann Schleier-Smith, Vikram Sreekanti, Chia-Che Tsai, Anurag Khandelwal, Qifan Pu, Vaishaal Shankar, Joao Carreira, Karl Krauth, Neeraja Yadwadkar, et al.  Cloud programming simplified: A berkeley view on serverless computing.  arXiv preprint arXiv:1902.03383, 2019.
  * [20] Swaroop Kotni, Ajay Nayak, Vinod Ganapathy, and Arkaprava Basu.  Faastlane: Accelerating Function-as-a-Service workflows.  In 2021 USENIX Annual Technical Conference (USENIX ATC 21). USENIX Association, 2021.
  * [21] Kubernetes.  Kubernetes- Production-Grade Container Orchestration.  https://kubernetes.io/, 2022.
  * [22] Kubernetes.  Resource management for pods and containers.  https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/, 2022\.
  * [23] Hugh C. Lauer and Roger M. Needham.  On the duality of operating system structures.  SIGOPS Oper. Syst. Rev., 1979.
  * [24] Bing Lin, Wenzhong Guo, Xianghan Zheng, Hong Zhang, Chunming Rong, and Guolong Chen.  Optimization scheduling for scientific applications with different priorities across multiple clouds.  In 2014 IEEE 6th International Conference on Cloud Computing Technology and Science, 2014.
  * [25] Linux Manual Page.  mprotect.  https://man7.org/linux/man-pages/man2/mprotect.2.html, 2022.
  * [26] Linux Manual Page.  pthreads.  https://man7.org/linux/man-pages/man7/pthreads.7.html, 2022.
  * [27] LLVM Project.  LLVM/OpenMP documentation.  https://openmp.llvm.org/, 2022.
  * [28] MRC-IDE Imperial College London.  CovidSim: COVID-19 Simulator.  https://github.com/mrc-ide/covid-sim, 2022.
  * [29] MDAnalysis.  Github - mdanalysis.  https://github.com/MDAnalysis/mdanalysis, 2022.
  * [30] Microsoft.  Azure Batch.  https://azure.microsoft.com/en-us/services/batch/, 2021.
  * [31] Microsoft.  Azure: Cloud Computing Services.  https://azure.microsoft.com/en-us/, 2021.
  * [32] Microsoft.  Azure Functions.  https://docs.microsoft.com/en-us/azure/azure-functions/, 2021.
  * [33] Microsoft.  Azure Virtual Machines.  https://docs.microsoft.com/en-us/azure/virtual-machines/dv2-dsv2-series, 2021\.
  * [34] Microsoft.  Azure Batch - Requirements for multi-instance tasks.  https://learn.microsoft.com/en-us/azure/batch/batch-mpi#requirements-for-multi-instance-tasks, 2022\.
  * [35] Microsoft.  Azure Durable Functions.  https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-overview, 2022\.
  * [36] Microsoft.  Azure Genomics.  https://azure.microsoft.com/en-us/services/genomics/, 2022.
  * [37] Microsoft.  Azure HPC.  https://azure.microsoft.com/en-gb/solutions/high-performance-computing, 2022\.
  * [38] Microsoft.  Use multi-instance tasks to run Message Passing Interface (MPI) applications in Batch.  https://learn.microsoft.com/en-us/azure/batch/batch-mpi, 2022.
  * [39] Philipp Moritz, Robert Nishihara, Stephanie Wang, Alexey Tumanov, Richard Liaw, Eric Liang, Melih Elibol, Zongheng Yang, William Paul, Michael I. Jordan, and Ion Stoica.  Ray: A Distributed Framework for Emerging AI Applications.  In USENIX Symposium on Operating Systems Design and Implementation (OSDI), 2017.
  * [40] MPI.  MPI Forum.  https://www.mpi-forum.org/, 2022.
  * [41] MPI Forum.  All-Reduce.  https://www.mpi-forum.org/docs/mpi-2.2/mpi22-report/node109.htm, 2022.
  * [42] OpenCV.  Github - OpenCV.  https://github.com/opencv/opencv, 2022.
  * [43] OpenFaaS.  Serverless Functions Made Simple.  https://www.openfaas.com/, 2022.
  * [44] OpenFOAM.  Github - OpenFOAM.  https://github.com/OpenFOAM/OpenFOAM-dev, 2022.
  * [45] OpenMP.  The OpenMP API specification for parallel programming.  https://www.openmp.org/specifications/, 2021.
  * [46] OpenMP.  Reduction Clauses and Directives.  https://www.openmp.org/spec-html/5.0/openmpsu107.html, 2022.
  * [47] OpenMPI.  OpenMPI: Open Source High Performance Computing.  https://www.open-mpi.org/, 2021.
  * [48] ParResKernels Team.  Parallel Research Kernels.  https://github.com/ParRes/Kernels, 2021.
  * [49] Pitch Patarasuk and Xin Yuan.  Bandwidth optimal all-reduce algorithms for clusters of workstations.  Journal of Parallel and Distributed Computing, (2), 2009.
  * [50] S. Plimpton.  Fast parallel algorithms for short-range molecular dynamics.  Journal of Computational Physics, 1993.
  * [51] LLVM Project.  Clang: a C language family frontend for LLVM.  https://clang.llvm.org/, 2022.
  * [52] CPP Reference.  Thread Support Library.  https://en.cppreference.com/w/cpp/thread, 2022.
  * [53] Herbert E. Robbins.  A stochastic approximation method.  Annals of Mathematical Statistics, 2007.
  * [54] Sandia National Laboratories.  Github - LAMMPS.  https://github.com/lammps/lammps, 2020.
  * [55] Sandia National Laboratories.  LAMMPS Molecular Dynamics Simulator.  https://lammps.sandia.gov/index.html, 2020.
  * [56] Sandia National Laboratories.  Benchmarks - LAMMPS Documentation.  https://docs.lammps.org/Speed_bench.html, 2022.
  * [57] Sandia National Laboratories.  Examples - LAMMPS Documentation.  https://docs.lammps.org/Examples.html, 2022.
  * [58] Bo Sang, Pierre-Louis Roman, Patrick Eugster, Hui Lu, Srivatsan Ravi, and Gustavo Petri.  Plasma: Programmable elasticity for stateful cloud computing applications.  In Proceedings of the Fifteenth European Conference on Computer Systems, EuroSys ’20. Association for Computing Machinery, 2020.
  * [59] Zhiming Shen, Sethuraman Subbiah, Xiaohui Gu, and John Wilkes.  Cloudscale: Elastic resource scaling for multi-tenant cloud systems.  In Proceedings of the 2nd ACM Symposium on Cloud Computing, SOCC ’11. Association for Computing Machinery, 2011.
  * [60] Simon Shillaker and Peter Pietzuch.  Faasm: Lightweight isolation for efficient stateful serverless computing.  In USENIX Annual Technical Conference (USENIX ATC). USENIX Association, 2020.
  * [61] Vikram Sreekanti, Chenggang Wu, Xiayue Charles Lin, Johann Schleier-Smith, Joseph E. Gonzalez, Joseph M. Hellerstein, and Alexey Tumanov.  Cloudburst: Stateful functions-as-a-service.  Proc. VLDB Endow., (12), 2020.
  * [62] su2code.  Github - SU2.  https://github.com/su2code/SU2, 2022.
  * [63] TensorFlow.  Github - TensorFlow.  https://github.com/tensorflow/tensorflow, 2022.
  * [64] Wencong Xiao, Shiru Ren, Yong Li, Yang Zhang, Pengyang Hou, Zhi Li, Yihui Feng, Wei Lin, and Yangqing Jia.  AntMan: Dynamic scaling on GPU clusters for deep learning.  In 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20). USENIX Association, 2020.
  * [65] Wen Zhang, Vivian Fang, Aurojit Panda, and Scott Shenker.  Kappa: A programming framework for serverless computing.  In Proceedings of the 11th ACM Symposium on Cloud Computing, SoCC ’20. Association for Computing Machinery, 2020.
  * [66] Siyuan Zhuang, Zhuohan Li, Danyang Zhuo, Stephanie Wang, Eric Liang, Robert Nishihara, Philipp Moritz, and Ion Stoica.  Hoplite: Efficient and fault-tolerant collective communication for task-based distributed systems.  In Proceedings of the 2021 ACM SIGCOMM 2021 Conference, SIGCOMM ’21. Association for Computing Machinery, 2021.