Storm was Twitter's first stream processing system. Unfortunately, it wasn't good enough for a number of reasons. Heron is a Storm API compliant rewrite of Storm.

Motivation. In Storm, computation is expressed as a directed graph, or topology, where vertexes are computations and edges transport tuples. Storm topologies are run on a cluster of workers overseen by a central Nimbus. Each worker runs multiple worker processes which run a JVM which run multiple executors which run multiple tasks. The executors run multiple threads, and each worker also has a multi-threaded supervisor.

This worker architecture was far too complex. Multiple components were making scheduling decisions (e.g. OS schedules processes, JVM schedules executors; executors schedule tasks) which made it hard to predict when certain tasks would be run. Moreover, putting different types of tasks on the same executor complicated logs, exception handling, garbage collection, etc. The Storm scheduler was also not good at scheduling tasks with different resource requirements. The fact that workers were very multi-threaded meant that messages were traversing a lot of thread boundaries.

The Nimbus was a complicated piece of code that did too much stuff. It also often became a bottleneck and was a single point of failure. It's scheduling was so poor, that Twitter used to reserve nodes to exclusively run a single topology. The Nimbus also communicated with workers through ZooKeeper which became a bottleneck.

Storm also did not implement backpressure; when bolts became overloaded; packets were just dropped.

Design Alternatives. Twitter considered extending and modifying Storm to fix its problems, but its flaws were deeply entrenched in its design, so a rewrite would be difficult. They considered using other existing stream processing systems, but didn't want to break the Storm API and have to rewrite a bunch of applications. In the end, they felt like a rewrite was the best bet.

Data Model and API. Heron follows the exact same API as Storm. Computation is expressed as a directed graph where vertexes are spouts (sources of tuples) or bolts (tuple processors) and edges transfer tuples between vertexes. Users provide logical plans which are expanded to physical plans in order to exploit data parallelism. Heron provides at least once and at most once semantics.

Architecture Overview. Users submit Heron topologies to Aurora, though Heron is able to run on top of Mesos, YARN, ECS, etc. Each topology is run as a set of containers. One container runs the Topology Master (TM). The rest run a Stream Manager (SM), a Metrics Manager (MM), and multiple Heron Instances (HI). Topology state is kept in ZooKeeper, and the TM can have a standby. All communication is done via protobufs.

Topology Master. The TM is responsible for overseeing the execution of a topology and reporting its status. The TM holds an ephemeral node in ZooKeeper to ensure there is only ever one TM and so that other things can discover it.

Stream Manager. Stream Managers are responsible for routing tuples. There are k Stream Managers that form a clique. Though, O(k^2) connections is a lot, the number of Heron Instances can scale independently of k. Stream Managers communicate via TCP, short-circuiting if delivering within a container.

Heron, unlike Storm, implements backpressure. Here are three kinds of backpressure implementations:

 - TCP Backpressure. Heron Instances communicate with Stream Managers via TCP. This provides a natural form of backpressure. If a TCP consumer is too slow, the TCP producer will slow down. This form of backpressure is easy to implement. However, multiple logical edges are mapped over a single SM to SM TCP connection which means that sometimes nodes are inadvertently slowed down when they shouldn't be.
 - Spout Backpressure. When a bolt is overloaded, the SM in charge of the bolt can tell the spout that is feeding into it to slow down. This is somewhat inefficient in that slowing an intermediate node may be sufficient.
 - Stage-by-Stage Backpressure. Backpressure can be propagated backwards from vertex to vertex.

Heron implements TCP and Spout Backpressure. Each socket is associated with a queue whose size has a high and low watermark. If the size exceeds the high watermark, backpressure is applied until it drops below the low watermark.

Heron Instances. Each Heron instance runs a single JVM which runs a single task which makes debugging significantly easier. Heron instances cannot be single-threaded because slow user code could prevent things like metrics from being reported in a timely manner. So, Heron implements Heron Instances with two threads: a Gateway Thread and a Task Execution Thread. The Gateway Thread communicates with the Task Execution Thread and also communicates with the SM and MM. The Task Execution Thread runs user code and gathers metrics. The Gateway Thread communicates with the Task Execution Thread using a set of one-directional queues. The sizes of these queues can be adjusted to avoid bad GC.

Metrics Manager. The metrics manager, well, manages metrics. It reports metrics to the TM and to a Twitter monitoring system.

Google has spent the last decade developing three container management systems. Borg is Google's main cluster management system that manages long running production services and non-production batch jobs on the same set of machines to maximize cluster utilization. Omega is a clean-slate rewrite of Borg using more principled architecture. In Omega, all system state lives in a consistent Paxos-based storage system that is accessed by a multitude of components which act as peers. Kubernetes is the latest open source container manager that draws on lessons from both previous systems.

All three systems use containers for security and performance isolation. Container technology has evolved greatly since the inception of Borg from chroot to jails to cgroups. Of course containers cannot prevent all forms of performance isolation. Today, containers also contain program images.

Containers allow the cloud to shift from a machine-oriented design to an application oriented-design and tout a number of advantages.

- The gap between where an application is developed and where it is deployed is shrunk.
 - Application writes don't have to worry about the details of the operating system on which the application will run.
 - Infrastructure operators can upgrade hardware without worrying about breaking a lot of applications.
 - Telemetry is tied to applications rather than machines which improves introspection and debugging.

Container management systems typically also provide a host of other niceties including:

 - naming services,
 - autoscaling,
 - load balancing,
 - rollout tools, and
 - monitoring tools.

In borg, these features were integrated over time in ad-hoc ways. Kubernetes organizes these features under a unified and flexible API.

Google's experience has led a number of things to avoid:

 - Container management systems shouldn't manage ports. Kubernetes gives each job a unique IP address allowing it to use any port it wants.
 - Containers should have labels, not just numbers. Borg gave each task an index within its job. Kubernetes allows jobs to be labeled with key-value pairs and be grouped based on these labels.
 - In Borg, every task belongs to a single job. Kubernetes makes task management more flexible by allowing a task to belong to multiple groups.
 - Omega exposed the raw state of the system to its components. Kubernetes avoids this by arbitrating access to state through an API.

Despite the decade of experience, there are still open problems yet to be solved:

 - Configuration. Configuration languages begin simple but slowly evolve into complicated and poorly designed Turing complete programming languages. It's ideal to have configuration files be simple data files and let real programming languages manipulate them.
 - Dependency management. Programs have lots of dependencies but don't manually state them. This makes automated dependency management very tough.

Strong consistency increases latency; weak consistency is hard to reason about. Compromising between the two, a number of databases allow each operation to run with either strong or weak consistency. Ideally, users choose the minimal set of strongly consistent operations needed to enforce some application specific invariant. However, deciding which operations to run with strong consistency and which to run with weak consistency can be very challenging. This paper

 - introduces a formal hybrid consistency model which subsumes many existing consistency models,
 - introduces a modular proof rule which can determine whether a given consistency model enforces an invariant, and
 - implements a prototype using a standard SMT solver.

Consistency Model, Informally. Consider

 - a set of states $s, s_{init} \in$ State,
 - a set of operations Op = {o, ...},
 - a set of values $\bot$ in Val, and
 - a set of replicas $r_1, r_2,$ ....

The denotation of an operation is denoted F_o where

 - F_o: State -> (Val x (State -> State)),
 - F_o^val(s) = F_o(s)[0] (the value returned by the operation), and
 - F_o^eff(s) = F_o(s)[1] (the effect of the operation).

For example, a banking operation may let states range over natural numbers where

 - F_deposit_a(s) = (\bot, fun s' -> s' + a)
 - F_interest(s) = (\bot, fun s' -> s' + 0.05 * s)
 - F_query(s) = (s, fun s' -> s')

If all operations commute (i.e. forall o1, o2, s1, s2. F_o1(s1) o F_o2(s2) = F_o2(s2) o F_o1(s1)), then all replicas are guaranteed to converge. However, convergence does not guarantee all application invariants are maintained. For example, an invariant I = {s | s >= 0} could be violated by merging concurrent withdrawals. To enforce invariants, we introduce a token system which can be used to order certain operations.

A token system TS = (Token, #) is a set of tokens Token and a symmetric relation # over Token. We say two sets of tokens T1 # T2 if exists t1 in T1, t2 in T2. t1 # t2. We also update our definition of operations to acquire tokens:

 - F_o: State -> (Val x (State -> State) x P(Token)),
 - F_o^val(s) = F_o(s)[0] (the value returned by the operation),
 - F_o^eff(s) = F_o(s)[1] (the effect of the operation), and
 - F_o^tok(s) = F_o(s)[2] (the tokens acquired by the operation).

Our consistency model will ensure that two operations that acquire conflicting operations will be ordered.

Formal Semantics. Recall a strict partial order is a partial order that is transitive and irreflexive (e.g. sets ordered by strict subset). Given a partial order R, we say (a, b) \in R or a -R-> b. Consider a countably infinite set Event of events ranged over by e, f, g. If operations are like transactions, an event is like applying a transaction at a replica.

- Definition 1. Given token system TS = (Token, #), an execution is a tuple X = (E, oper, rval, tok, hb) where
   - E \subset Event is a finite subset of events,
   - oper: E -> Op designates the operation of each event,
   - rval: E -> Val designates the return value of each event,
   - tok: E -> P(Token) designates the tokens acquired by each event, and
   - hb \subset Event x Event is a happens before strict partial order where forall e, f. tok(e) # tok(f) => (e-hb->f or f-hb->e).

An execution formalizes operations executing at various replicas concurrently, and the happens before relation captures how these operations are propagated between replicas. The transitivity of the happens before relation ensures at least causal consistency.

Let

 - Exec(TS) be the set of all executions over token system TS, and
 - ctxt(e, X) = (E, X.oper|E, X.rval|E, X.tok|E, X.hb|E) be the context of e where E = X.hb^-1(e). Intuitively, ctxt(e, X) is the subexection of X that only includes operations causally preceding e.

Executions are directed graphs of operations, but without a semantics, they are rather meaningless. Here, we define a relation evald_F \subset Exec(TS) x P(State) where evald_F(Y) is the set of all final states Y can be in after all operations are propagated to all replicas. We'll see shortly that if all non-token-conflicting operations commute, then evald_F is a function.

 - evald_F(Y) = {} if Y.E = {}, and
 - evald_F(Y) = {F_e^eff(s)(s') | e \in max(Y), s \in evald_F(ctxt(e, Y)), s' \in evalfd_F(Y|Y.E - {e})} otherwise.

Now

 - Definition 2. An execution X \in Exec(TS) is consistent with TS and F denoted X |= TS, F if forall e \in X.e. exists s \in evald_F(ctxt(e, x)). X.val(e) = F_X.oper(e)^val(s) and X.tok(e = F_X.oper(e)^tok(s)).

We let Exec(TS, F) = {X | X |= TS, F}. Consistent operations are closed under context. Furthermore, evald_F is a function when restricted to consistent executions where non-token-conflicting operations commute. We call this function eval_F.

This model can model a number of consistency models:

 - Causal consistency. Let Token = {}.
 - Sequential consistency. Let Token = {t}, t # t, and F_o^tok(s) = {t} for all o.
 - RedBlue Consistency. Let Token = {t}, t # t, and F_o^tok(s) = {t} for all red o and F_o^tok(s) = {} for all blue o.

State Based Proof Rule. We want to construct a proof rule to establish the fact that Exec(TS, F) \subset eval_F^-1(I). That is, every execution results in a state that satisfies the invariant. Since executions are closed under context, this also means that all operations execute on a state that satisfies the invariant.

Our proof rule involves a guarantee relation G(t) over states which describes all possible state changes that can occur while holding token t. Similarly, G_0 describes the state transitions that can occur without holding any tokens.

Here is the proof rule.

 - S1: s_init \in I.
 - S2: G_0(I) \subset I and forall t. G(t)(I) \subset I.
 - S3: forall o, s, s'.
    - s \in I and
    - (s, s') \in (G_0 \cup G(F_o^tok(s)^\bot))* =>
    - (s', F_o^eff(s)(s')) \in G_0 \cup G(F_o^tok(s)).


In English,

 - S1: s_init satisfies the invariant.
 - S2: G and G_0 preserve the invariant.
 - S3: If we start in any state s that satisfies the invariant and can transition in any finite number of steps to any state s' without acquiring any tokens conflicting with o, then we can transition from s' to F_o^eff(s)(s') in a single step using the tokens acquired by o.

Event Based Proof Rule and Soundness. Instead of looking at states, we can instead look at executions. That is, if we let invariants I \subset Exec(TS), then we want to write a proof rule to ensure Exec(TS, F) \subset I. That is, all consistent executions satisfy the invariant. Again, we use a guarantee G \subset Exec(TS) x Exec(TS).

 - E1: X_init \in I.
 - E2: G(I) \in I.
 - E3: forall X, X', X''. forall e in X''.E.
    - X'' |= TS, F and
    - X' = X''|X''.E - {e} and
    - e \in max(X'') and
    - X = ctxt(e, X'') and
    - X \in I and
    - (X, X') \in G* =>
    - (X', X'') \in G.

This proof rule is proven sound. The event based rule and its soundness is derived from this.

Examples and Automation. The authors have built a banking, auction, and courseware application in this style. They have also built a prototype that you give TS, F, and I and it determines if Exec(T, f) \subset eval_F^-1(I). Their prototype modifies the state-based proof rule eliminating the transitive closure and introducing intermediate predicates.

Decibel is like git for relational data.

Often, teams need to simultaneously query, analyze, clean, or curate a collection of data without clobbering each other's work. Currently, the best solution available involves each team member making copies of the entire database. This is undesirable for a number of reasons:

 - Data is stored very redundantly, wasting a huge amount of storage.
 - It is difficult to share or merge changes made to local snapshots of the data.
 - There is no systematic way to say which version of the data was used for a given experiment.

Version control solves these problems, but existing version control systems (e.g. git) have a couple of shortcomings when applied naively to large datasets:

 - Using distributed version control systems like git, clients clone an entire copy of a repository's contents. This is infeasible when working with large datasets.
 - Version control systems like git operate on arbitrary data and do not understand the structure of relational data. This makes certain operations like generic diffing a bad fit.
 - Systems like git do not support high-level data management or query APIs.

Decibel manages datasets which are collections of relations, and each relation's schema includes an immutable primary key which is used to track the version of each row. Beginning with an initial snapshot of a dataset, users can check out, branch, and commit changes to the data set in a style very similar to git. When data is merged, non-conflicting changes to separate columns of the same row are both applied. If conflicting changes to the same column of the same row occur, one branch's changes take priority. Diffing two tables across two branches results in a table of insertions and deletions. Finally, data can be queried across versions using VQuel: a query language which is notably not SQL.

This paper describes three physical realizations of Decibel's logical data model.

1. In a tuple-first representation, tables contain every version of every row, and each row is annotated with a bitmap indicating the set of versions in which the row is live. In a tuple-oriented approach, each of N tuples comes with a B-bit bitmap for B branches. In a branch-oriented approach, there are B N-bit bitmaps.
2. In a version-first representation, all the changes made to a table in a branch are stored together in the same file, and these branched files contain pointers to their ancestors forming a directed acyclic graph.
3. In a hybrid representation, data is stored similarly to the version-first approach, but each of the branched files (called segments) includes a segment index: a bitmap, like in the tuple-first representation, that tracks the liveness of each row for all descendent branches. Moreover, there is a single branch-segment bitmap which for each branch, records the set of segments with a tuple live in that branch.

The tuple-first representation is good for multi-version queries, the version-first representation is good for single-version queries, and the hybrid approach is good at both.

This paper also presents a versioned database benchmarking framework in the form of a set of branching strategies and characteristic queries to analyze Decibel and any future versioned databases.

Commentary. A git-like versioned relational database seems like a great idea! This paper focuses on how to implement and analyze such a database efficiently; it focuses less on the higher-level semantics of data versioning. Notably, two unanswered questions stuck out to me that seem like interesting areas for future research.

1. The current merging strategy seems inconvenient, if not inconsistent. Imagine a table R(key, a, b) with the invariant that a + b < 10. Imagine branch b1 updates a tuple (0, 0, 0) to (0, 9, 0) and branch b2 updates it to (0, 0, 9). If these tuples are merged, the tuple becomes (0, 9, 9) violating the invariant. In systems like git, users can manually inspect and verify the correctness of the merge, but if a large dataset is being merged, this becomes infeasible. I think more research is needed to determine if this strategy is good enough and won't cause problems in practice. Or if it is insufficient, what merging strategies can be applied on large datasets. Perhaps ideas could be borrowed from CRDT literature; if all columns were semilattices, the columns could be merged in a sane way.
2. How useful is diffing when dealing with large datasets? When working with git, even a modestly sized diff can become confusing. Would a non-trivial diff of a large dataset be incomprehensible?

In the face of latency, partitions, and failures, application sometimes turn to weak consistency for improved performance. However, strong consistency cannot always be avoided, so some data stores (e.g. Cassandra, Riak) allow applications to operate on some data with weak consistency and on other data with strong consistency. However, it is difficult to ensure that data read with weak consistency does not leak into the strongly consistent data.

For example, imagine a ticketing application that displays the number of remaining tickets for an upcoming movie. To improve performance, an application may choose to read the data using weak consistency; it's okay if the displayed number is slightly erroneous. However, now imagine that we want to increase the price of the last 10 tickets. If we used the same weakly consistent value to determine the price of a ticket, we may make less money that we expect. Weak and strong consistency cannot be carelessly mixed.

The Inconsistent, Performance-bound, Approximate (IPA) storage system uses a type system of consistency types to ensure consistency safety: the guarantee that weakly consistent data does not leak into strongly consistent data. They also introduce error-bounded consistency which allows a runtime to dynamically choose the strongest consistency that satisfies certain constraints (e.g. latency or accuracy).

Programming Model. The IPA programming model involves three components: abstract data types, consistency policies, and consistency types. Users annotate ADTs with consistency policies which determines the consistency types returned by the ADT's methods.

1. ADTs. IPA is bundled with a set of common ADTs (e.g. sets, maps, lists) that can be implemented with any number of backing stores (e.g. Cassandra, Riak, Redis). The collection of ADTs is also extensible; users can add their own ADTs.
2. Consistency Policies. ADT instances (or their methods) are annotated with consistency policies which designate the level of consistency the user desires. Static policies, like strong consistency, are fixed throughout the lifetime of the ADT. Dynamic policies allow the runtime to dynamically choose the consistency of the system to meet some constraint. For example, a user can set a policy LatencyBound(x), and the system will choose the strongest consistency that can be served and still satisfy the latency bound. Or, a user can set a ErrorTolerance(x) allowing the system to return approximate results with weaker consistency.
3. Consistency Types. Consistency types are parameterized on a type T and are partially ordered

```
          Consistent[T]
          /     |     \
Interval[T] Rushed[T] ...
          \     |     /
         Inconsistent[T]
```

Users can explicitly endorse, or upcast, weak types to stronger types. Rushed types are produced by ADTs with LatencyBound policies and are a sum of other consistency types. Interval types are a numeric interval which are guaranteed to contain the correct value.

Enforcing Consistency Policies. Static consistency policies are easy to enforce. The ADT operations simply set flags that are used by the underlying store. Dynamic policies are trickier to implement.

- Latency bounds. To service a read with a latency bound of x milliseconds, for example, a read is issued at every single consistency level. The strongest to return in less that x milliseconds is used. This naive approach can put unwanted load on a system, so IPA can also monitor and predict the strongest consistency model for a given latency bound in order to issue only a couple of reads.

- Error bounds. Error bounds are enforced using a reservation scheme similar to the one used by bounded CRDTs. For example, imagine a counter with value 10 should be concurrently decremented but never drop below 0. We can allocate 10 decrement tokens and distribute them to nodes. A node can decrement so long as it has enough tokens. IPA involves reservation servers which are assigned tokens. A node can return an approximate read by knowing the number of outstanding tokens. For example, given a value of 100 and knowing there are 10 outstanding increment tokens, the true value is somewhere in the range [100, 110]. The more outstanding tokens there are, the weaker the error bounds. To avoid unnecessarily granting tokens, IPA uses an allocation table which maps reservation servers to tokens. Tokens are only allocated when necessary.

Implementaton. IPA is implemented in Scala on top of Cassandra with some middleware for the reservation servers. Type checking is done by leveraging Scala's type system.