.gitignore

+.obj/
+
+lockserver/client-main
+lockserver/server-main
+lockserver/lockserver-repl
+store/benchmark/benchClient
+store/benchmark/retwisClient
+store/benchmark/terminalClient
+store/strongstore/server
+store/tapirstore/server
+store/weakstore/server
+timeserver/timeserver
+
+lib/tests/configuration-test
+lib/tests/simtransport-test
+lockserver/tests/lockserver-test
+replication/ir/tests/ir-test
+replication/vr/tests/vr-test
+store/common/backend/tests/kvstore-test
+store/common/backend/tests/lockserver-test
+store/common/backend/tests/versionstore-test
+
+################################################################################
+# vim
+################################################################################
+# Swap
+[._]*.s[a-v][a-z]
+[._]*.sw[a-p]
+[._]s[a-v][a-z]
+[._]sw[a-p]
+
+# Session
+Session.vim
+
+# Temporary
+.netrwhist
+*~
+# Auto-generated tag files
+tags
+
+################################################################################
+# c++
+################################################################################
+# Prerequisites
+*.d
+
+# Compiled Object files
+*.slo
+*.lo
+*.o
+*.obj
+
+# Precompiled Headers
+*.gch
+*.pch
+
+# Compiled Dynamic libraries
+*.so
+*.dylib
+*.dll
+
+# Fortran module files
+*.mod
+*.smod
+
+# Compiled Static libraries
+*.lai
+*.la
+*.a
+*.lib
+
+# Executables
+*.exe
+*.out
+*.app

LICENSE

+The MIT License (MIT)
+
+Copyright (c) 2019 Irene Zhang, Dan Ports, Naveen Kr. Sharma
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

 #
-# Top-level makefile for OR-2PC
+# Top-level makefile for IR and TAPIR
 #
 
 CC = gcc
@@ -7,13 +7,13 @@ CXX = g++
 LD = g++
 EXPAND = lib/tmpl/expand
 
-CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized -levent_pthreads -O2 -DNASSERT
-#CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized -levent_pthreads
-CXXFLAGS := -g -std=c++0x -levent_pthreads
+CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized -O2 -DNASSERT
+#CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized 
+CXXFLAGS := -g -std=c++0x
 LDFLAGS := -levent_pthreads
 ## Debian package: check
-# CHECK_CFLAGS := $(shell pkg-config --cflags check)
-# CHECK_LDFLAGS := $(shell pkg-config --cflags --libs check)
+#CHECK_CFLAGS := $(shell pkg-config --cflags check)
+#CHECK_LDFLAGS := $(shell pkg-config --cflags --libs check)
 # Debian package: libprotobuf-dev
 PROTOBUF_CFLAGS := $(shell pkg-config --cflags protobuf)
 PROTOBUF_LDFLAGS := $(shell pkg-config --cflags --libs protobuf)
@@ -26,8 +26,8 @@ LIBEVENT_LDFLAGS := $(shell pkg-config --libs libevent)
 CFLAGS += $(LIBEVENT_CFLAGS)
 LDFLAGS += $(LIBEVENT_LDFLAGS)
 # Debian package: libssl-dev
-LIBSSL_CFLAGS := $(shell pkg-config --cflags libssl)
-LIBSSL_LDFLAGS := $(shell pkg-config --libs libssl)
+LIBSSL_CFLAGS := $(shell pkg-config --cflags openssl)
+LIBSSL_LDFLAGS := $(shell pkg-config --libs openssl)
 CFLAGS += $(LIBSSL_CFLAGS)
 LDFLAGS += $(LIBSSL_LDFLAGS)
 
@@ -122,7 +122,17 @@ $(foreach bin,$(1),$(eval LDFLAGS-$(bin) += $(2)))
 endef
 
 include lib/Rules.mk
-
+include replication/common/Rules.mk
+include replication/vr/Rules.mk
+include replication/ir/Rules.mk
+include store/common/Rules.mk
+include store/tapirstore/Rules.mk
+include store/strongstore/Rules.mk
+include store/weakstore/Rules.mk
+include store/benchmark/Rules.mk
+include lockserver/Rules.mk
+include timeserver/Rules.mk
+include libtapir/Rules.mk
 ##################################################################
 # General rules
 #
@@ -227,6 +237,12 @@ clean:
 	$(call trace,RM,binaries,rm -f $(BINS) $(TEST_BINS))
 	$(call trace,RM,objects,rm -rf .obj)
 
+#
+# Debugging
+#
+print-%:
+	@echo '$*=$($*)'
+
 ##################################################################
 # Targets
 #

README.md

+# TAPIR
+
+This repository includes code implementing TAPIR -- the Transaction
+Application Protocol for Inconsistent Replication. This code was used
+for the SOSP 2015 paper, ["Building Consistent Transactions with
+Inconsistent Replication."](http://dl.acm.org/authorize?N93281)
+
+TAPIR is a new protocol for linearizable distributed transactions
+built using replication with no consistency guarantees. By enforcing
+consistency only at the transaction layer, TAPIR eliminates
+coordination at the replication layer, enabling TAPIR to provide the
+same transaction model and consistency guarantees as existing systems,
+like Spanner, with better latency and throughput.
+
+In addition to TAPIR, this repo includes several other useful
+implementations of distributed systems, including:
+
+1. An implementation of a lock server designed to work with
+   inconsistent replication (IR), our high-performance, unordered
+   replication protocol.
+
+2. An implementation of Viewstamped Replication (VR), detailed in this
+   [older paper](http://dl.acm.org/citation.cfm?id=62549) and this
+   [more recent paper](http://18.7.29.232/handle/1721.1/71763).
+
+3. An implementation of a scalable, distributed storage system
+   designed to work with VR that uses two-phase commit to support
+   distributed transactions and supports both optimistic concurrency
+   control and strict two-phase locking.
+
+The repo is structured as follows:
+
+- /lib - the transport library for communication between nodes. This
+  includes UDP based network communcation as well as the ability to
+  simulate network conditions on a local machine, including packet
+  delays and reorderings.
+
+- /replication - replication library for the distributed stores
+  - /vr - implementation of viewstamped replication protocol
+  - /ir - implementation of inconsistent replication protocol
+
+- /store - partitioned/sharded distributed store
+  - /common - common data structures, backing stores and interfaces for all of stores
+  - /tapirstore - implementation of TAPIR designed to work with IR
+  - /strongstore - implementation of both an OCC-based and locking-based 2PC transactional
+  storage system, designed to work with VR
+  - /weakstore - implementation of an eventually consistent storage
+    system, using quorum writes for replication
+
+- /lockserver - a lock server designed to be used with IR
+
+## Compiling & Running
+You can compile all of the TAPIR executables by running make in the root directory
+
+TAPIR depends on protobufs, libevent and openssl, so you will need the following development libraries:
+- libprotobuf-dev
+- libevent-openssl
+- libevent-pthreads
+- libevent-dev
+- libssl-dev
+- protobuf-compiler
+
+## Contact and Questions
+Please email Irene at iyzhang@cs.washington.edu, Dan at drkp@cs.washington.edu and Naveen at naveenks@cs.washington.edu

lib/Rules.mk

@@ -2,7 +2,9 @@ d := $(dir $(lastword $(MAKEFILE_LIST)))
 
 SRCS += $(addprefix $(d), \
 	lookup3.cc message.cc memory.cc \
-	latency.cc configuration.cc transport.cc udptransport.cc simtransport.cc)
+	latency.cc configuration.cc transport.cc \
+	udptransport.cc tcptransport.cc simtransport.cc repltransport.cc \
+	persistent_register.cc)
 
 PROTOS += $(addprefix $(d), \
           latency-format.proto)
@@ -23,7 +25,13 @@ LIB-transport := $(o)transport.o $(LIB-message) $(LIB-configuration)
 
 LIB-simtransport := $(o)simtransport.o $(LIB-transport)
 
+LIB-repltransport := $(o)repltransport.o $(LIB-transport)
+
 LIB-udptransport := $(o)udptransport.o $(LIB-transport)
 
+LIB-tcptransport := $(o)tcptransport.o $(LIB-transport)
+
+LIB-persistent_register := $(o)persistent_register.o $(LIB-message)
 
 include $(d)tests/Rules.mk
+

lib/assert.h

@@ -5,7 +5,7 @@
  *   assertion macros that integrate with the logging framework
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  *
  * Permission is hereby granted, free of charge, to any person

lib/configuration.cc

@@ -29,16 +29,15 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/assert.h"
-#include "paxos-lib/lib/configuration.h"
-#include "paxos-lib/lib/message.h"
+#include "lib/assert.h"
+#include "lib/configuration.h"
+#include "lib/message.h"
 
-#include <iostream>
-#include <fstream>
-#include <string>
-#include <string.h>
+#include <cstring>
+#include <stdexcept>
+#include <tuple>
 
-namespace specpaxos {
+namespace transport {
 
 ReplicaAddress::ReplicaAddress(const string &host, const string &port)
     : host(host), port(port)
@@ -52,6 +51,12 @@ ReplicaAddress::operator==(const ReplicaAddress &other) const {
             (port == other.port));
 }
 
+bool
+ReplicaAddress::operator<(const ReplicaAddress &other) const {
+    auto this_t = std::forward_as_tuple(host, port);
+    auto other_t = std::forward_as_tuple(other.host, other.port);
+    return this_t < other_t;
+}
 
 Configuration::Configuration(const Configuration &c)
     : n(c.n), f(c.f), replicas(c.replicas), hasMulticast(c.hasMulticast)
@@ -61,7 +66,7 @@ Configuration::Configuration(const Configuration &c)
         multicastAddress = new ReplicaAddress(*c.multicastAddress);
     }
 }
-    
+
 Configuration::Configuration(int n, int f,
                              std::vector<ReplicaAddress> replicas,
                              ReplicaAddress *multicastAddress)
@@ -82,7 +87,7 @@ Configuration::Configuration(std::ifstream &file)
     f = -1;
     hasMulticast = false;
     multicastAddress = NULL;
-    
+
     while (!file.eof()) {
         // Read a line
         string line;
@@ -94,52 +99,53 @@ Configuration::Configuration(std::ifstream &file)
         }
 
         // Get the command
-        // This is pretty horrible, but C++ does promise that &line[0]
-        // is going to be a mutable contiguous buffer...
-        char *cmd = strtok(&line[0], " \t");
+        unsigned int t1 = line.find_first_of(" \t");
+        string cmd = line.substr(0, t1);
 
-        if (strcasecmp(cmd, "f") == 0) {
-            char *arg = strtok(NULL, " \t");
-            if (!arg) {
+        if (strcasecmp(cmd.c_str(), "f") == 0) {
+            unsigned int t2 = line.find_first_not_of(" \t", t1);
+            if (t2 == string::npos) {
                 Panic ("'f' configuration line requires an argument");
             }
-            char *strtolPtr;
-            f = strtoul(arg, &strtolPtr, 0);
-            if ((*arg == '\0') || (*strtolPtr != '\0')) {
+
+            try {
+                f = stoul(line.substr(t2, string::npos));
+            } catch (std::invalid_argument& ia) {
                 Panic("Invalid argument to 'f' configuration line");
             }
-        } else if (strcasecmp(cmd, "replica") == 0) {
-            char *arg = strtok(NULL, " \t");
-            if (!arg) {
+        } else if (strcasecmp(cmd.c_str(), "replica") == 0) {
+            unsigned int t2 = line.find_first_not_of(" \t", t1);
+            if (t2 == string::npos) {
                 Panic ("'replica' configuration line requires an argument");
             }
 
-            char *host = strtok(arg, ":");
-            char *port = strtok(NULL, "");
-            
-            if (!host || !port) {
+            unsigned int t3 = line.find_first_of(":", t2);
+            if (t3 == string::npos) {
                 Panic("Configuration line format: 'replica host:port'");
             }
 
-            replicas.push_back(ReplicaAddress(string(host), string(port)));
-        } else if (strcasecmp(cmd, "multicast") == 0) {
-            char *arg = strtok(NULL, " \t");
-            if (!arg) {
+            string host = line.substr(t2, t3-t2);
+            string port = line.substr(t3+1, string::npos);
+
+            replicas.push_back(ReplicaAddress(host, port));
+        } else if (strcasecmp(cmd.c_str(), "multicast") == 0) {
+            unsigned int t2 = line.find_first_not_of(" \t", t1);
+            if (t2 == string::npos) {
                 Panic ("'multicast' configuration line requires an argument");
             }
 
-            char *host = strtok(arg, ":");
-            char *port = strtok(NULL, "");
-            
-            if (!host || !port) {
-                Panic("Configuration line format: 'multicast host:port'");
+            unsigned int t3 = line.find_first_of(":", t2);
+            if (t3 == string::npos) {
+                Panic("Configuration line format: 'replica host:port'");
             }
 
-            multicastAddress = new ReplicaAddress(string(host),
-                                                  string(port));
+            string host = line.substr(t2, t3-t2);
+            string port = line.substr(t3+1, string::npos);
+
+            multicastAddress = new ReplicaAddress(host, port);
             hasMulticast = true;
         } else {
-            Panic("Unknown configuration directive: %s", cmd);
+            Panic("Unknown configuration directive: %s", cmd.c_str());
         }
     }
 
@@ -209,8 +215,27 @@ Configuration::operator==(const Configuration &other) const
             return false;
         }
     }
-    
+
     return true;
 }
 
-} // namespace specpaxos
+bool
+Configuration::operator<(const Configuration &other) const {
+    auto this_t = std::forward_as_tuple(n, f, replicas, hasMulticast);
+    auto other_t = std::forward_as_tuple(other.n, other.f, other.replicas,
+                                         other.hasMulticast);
+    if (this_t < other_t) {
+        return true;
+    } else if (this_t == other_t) {
+        if (hasMulticast) {
+            return *multicastAddress < *other.multicastAddress;
+        } else {
+            return false;
+        }
+    } else {
+        // this_t > other_t
+        return false;
+    }
+}
+
+} // namespace transport

lib/configuration.h

@@ -32,7 +32,7 @@
 #ifndef _LIB_CONFIGURATION_H_
 #define _LIB_CONFIGURATION_H_
 
-#include "paxos-lib/lib/viewstamp.h"
+#include "replication/common/viewstamp.h"
 
 #include <fstream>
 #include <stdbool.h>
@@ -41,7 +41,7 @@
 
 using std::string;
 
-namespace specpaxos {
+namespace transport {
 
 struct ReplicaAddress
 {
@@ -52,6 +52,16 @@ struct ReplicaAddress
     inline bool operator!=(const ReplicaAddress &other) const {
         return !(*this == other);
     }
+    bool operator<(const ReplicaAddress &other) const;
+    bool operator<=(const ReplicaAddress &other) const {
+        return *this < other || *this == other;
+    }
+    bool operator>(const ReplicaAddress &other) const {
+        return !(*this <= other);
+    }
+    bool operator>=(const ReplicaAddress &other) const {
+        return !(*this < other);
+    }
 };
 
 
@@ -69,10 +79,20 @@ public:
     int QuorumSize() const;
     int FastQuorumSize() const;
     bool operator==(const Configuration &other) const;
-    inline bool operator!= (const Configuration &other) const {
+    inline bool operator!=(const Configuration &other) const {
         return !(*this == other);
     }
-    
+    bool operator<(const Configuration &other) const;
+    bool operator<=(const Configuration &other) const {
+        return *this < other || *this == other;
+    }
+    bool operator>(const Configuration &other) const {
+        return !(*this <= other);
+    }
+    bool operator>=(const Configuration &other) const {
+        return !(*this < other);
+    }
+
 public:
     int n;                      // number of replicas
     int f;                      // number of failures tolerated
@@ -82,12 +102,13 @@ private:
     bool hasMulticast;
 };
 
-}      // namespace specpaxos
+}      // namespace transport
+
 
 namespace std {
-template <> struct hash<specpaxos::ReplicaAddress>
+template <> struct hash<transport::ReplicaAddress>
 {
-    size_t operator()(const specpaxos::ReplicaAddress & x) const
+    size_t operator()(const transport::ReplicaAddress & x) const
         {
             return hash<string>()(x.host) * 37 + hash<string>()(x.port);
         }
@@ -95,20 +116,19 @@ template <> struct hash<specpaxos::ReplicaAddress>
 }
 
 namespace std {
-template <> struct hash<specpaxos::Configuration>
+template <> struct hash<transport::Configuration>
 {
-    size_t operator()(const specpaxos::Configuration & x) const
+    size_t operator()(const transport::Configuration & x) const
         {
             size_t out = 0;
             out = x.n * 37 + x.f;
             for (int i = 0; i < x.n; i++) {
                 out *= 37;
-                out += hash<specpaxos::ReplicaAddress>()(x.replica(i));
+                out += hash<transport::ReplicaAddress>()(x.replica(i));
             }
             return out;
         }
 };
 }
 
-
 #endif  /* _LIB_CONFIGURATION_H_ */

lib/hash.h

@@ -5,7 +5,7 @@
  *   header defining hash functions
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  * Copyright 2009-2012 Massachusetts Institute of Technology
  *

lib/latency-format.proto

-package specpaxos.latency.format;
+syntax = "proto2";
+
+package transport.latency.format;
 
 message LatencyDist
 {

lib/latency.cc

@@ -4,7 +4,9 @@
  * latency.cc:
  *   latency profiling functions
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
+ *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  * Copyright 2009-2012 Massachusetts Institute of Technology
  *
  * Permission is hereby granted, free of charge, to any person
@@ -40,8 +42,8 @@
 #include <iostream>
 #include <fstream>
 
-#include "paxos-lib/lib/message.h"
-#include "paxos-lib/lib/latency-format.pb.h"
+#include "lib/message.h"
+#include "lib/latency-format.pb.h"
 
 static struct Latency_t *latencyHead;
 
@@ -326,10 +328,10 @@ Latency_FlushTo(const char *fname)
 {
     std::ofstream outfile(fname);
     Latency_t *l = latencyHead;
-    ::specpaxos::latency::format::LatencyFile out;
+    ::transport::latency::format::LatencyFile out;
     
     for (; l; l = l->next) {
-        ::specpaxos::latency::format::Latency lout;
+        ::transport::latency::format::Latency lout;
         Latency_Put(l, lout);
         *(out.add_latencies()) = lout;
     }
@@ -354,14 +356,14 @@ Latency_Flush(void)
 }
 
 void
-Latency_Put(Latency_t *l, ::specpaxos::latency::format::Latency &out)
+Latency_Put(Latency_t *l, ::transport::latency::format::Latency &out)
 {
     out.Clear();
     out.set_name(l->name);
     
     for (int i = 0; i < l->distPoolNext; ++i) {
         Latency_Dist_t *d = &l->distPool[i];
-        ::specpaxos::latency::format::LatencyDist *outd = out.add_dists();
+        ::transport::latency::format::LatencyDist *outd = out.add_dists();
         outd->set_type(d->type);
         outd->set_min(d->min);
         outd->set_max(d->max);
@@ -375,12 +377,12 @@ Latency_Put(Latency_t *l, ::specpaxos::latency::format::Latency &out)
 }
 
 bool
-Latency_TryGet(const ::specpaxos::latency::format::Latency &in, Latency_t *l)
+Latency_TryGet(const ::transport::latency::format::Latency &in, Latency_t *l)
 {
     LatencyInit(l, strdup(in.name().c_str())); // XXX Memory leak
     l->distPoolNext = in.dists_size();
     for (int i = 0; i < l->distPoolNext; ++i) {
-        const ::specpaxos::latency::format::LatencyDist &ind =
+        const ::transport::latency::format::LatencyDist &ind =
             in.dists(i);
         Latency_Dist_t *d = &l->distPool[i];
         d->type = ind.type();

lib/latency.h

@@ -5,7 +5,7 @@
  *   latency profiling functions
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  * Copyright 2009-2012 Massachusetts Institute of Technology
  *
@@ -108,8 +108,8 @@ void Latency_FlushTo(const char *fname);
 void Latency_Flush(void);
 
 void Latency_Put(Latency_t *l,
-                 ::specpaxos::latency::format::Latency &out);
-bool Latency_TryGet(const ::specpaxos::latency::format::Latency &in,
+                 ::transport::latency::format::Latency &out);
+bool Latency_TryGet(const ::transport::latency::format::Latency &in,
                     Latency_t *l);
 
 static inline void

lib/memory.h

@@ -5,7 +5,7 @@
  *   parsing and pretty-printing of memory sizes
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  * Copyright 2009-2012 Massachusetts Institute of Technology
  *

lib/message.cc

@@ -40,6 +40,8 @@
 #include <unistd.h>
 #include <sys/time.h>
 
+#include <mutex>
+
 #define BACKTRACE_ON_PANIC 1
 #if BACKTRACE_ON_PANIC
 #include <execinfo.h>
@@ -48,6 +50,8 @@
 #define TIMESTAMP_BASE62 0
 #define TIMESTAMP_NUMERIC 1
 
+std::mutex message_mtx;
+
 void __attribute__((weak))
 Message_VA(enum Message_Type type,
            const char *fname, int line, const char *func,
@@ -74,6 +78,9 @@ _Message_VA(enum Message_Type type, FILE *fp,
             const char *fname, int line, const char *func,
             const char *fmt, va_list args)
 {
+    // Lock mutex to make sure the output is not mangled.
+    message_mtx.lock();
+
     static int haveColor = -1;
     struct msg_desc {
         const char *prefix;
@@ -150,6 +157,9 @@ _Message_VA(enum Message_Type type, FILE *fp,
         fputs("\033[0m", fp);
     fprintf(fp, "\n");
     fflush(fp);
+
+    // Unlock mutex.
+    message_mtx.unlock();
 }
 
 void _Panic(void)

lib/persistent_register.cc

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * persistent_register.cc: A disk-backed persistent register.
+ *
+ * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+#include "lib/persistent_register.h"
+
+#include <cstdio>
+#include <cstring>
+#include <unistd.h>
+
+#include <fstream>
+#include <memory>
+
+#include "lib/message.h"
+
+bool PersistentRegister::Initialized() const
+{
+    // Check to see if the file exists. If it doesn't, then we default to
+    // returning an empty string. Refer to [1] for some ways to check if a file
+    // exists in C++.
+    //
+    // [1]: https://stackoverflow.com/a/12774387/3187068
+    std::ifstream f(filename_.c_str());
+    return f.good();
+}
+
+std::string PersistentRegister::Read() const
+{
+    if (!Initialized()) {
+        return "";
+    }
+
+    std::FILE *file = OpenFile(filename_, "rb");
+
+    // Seek to the end of the file and get it's size.
+    int success = std::fseek(file, 0, SEEK_END);
+    if (success != 0) {
+        Panic("Unable to fseek file %s", filename_.c_str());
+    }
+    long length = ftell(file);
+    if (length == -1) {
+        Panic("%s", std::strerror(errno));
+    }
+
+    // Seek back to the beginning of the file and read its contents. Now that
+    // we know the size, we can allocate an appropriately sized buffer.
+    success = std::fseek(file, 0, SEEK_SET);
+    if (success != 0) {
+        Panic("Unable to fseek file %s", filename_.c_str());
+    }
+    std::unique_ptr<char[]> buffer(new char[length]);
+    std::size_t num_read = std::fread(buffer.get(), length, 1, file);
+    if (num_read != 1) {
+        Panic("Unable to read file %s", filename_.c_str());
+    }
+
+    CloseFile(file);
+    return std::string(buffer.get(), length);
+}
+
+void PersistentRegister::Write(const std::string &s)
+{
+    // Perform the write.
+    std::FILE *file = OpenFile(filename_, "wb");
+    std::size_t num_written =
+        std::fwrite(s.c_str(), sizeof(char), s.size(), file);
+    if (num_written != s.size()) {
+        Panic("Unable to write to file %s", filename_.c_str());
+    }
+
+    // Persist the write.
+    int fd = fileno(file);
+    if (fd == -1) {
+        Panic("%s", std::strerror(errno));
+    }
+    int success = fsync(fd);
+    if (success != 0) {
+        Panic("%s", std::strerror(errno));
+    }
+
+    CloseFile(file);
+}
+
+std::string PersistentRegister::Filename() { return filename_; }
+
+std::FILE *PersistentRegister::OpenFile(const std::string &filename,
+                                        const std::string &mode)
+{
+    std::FILE *file = std::fopen(filename.c_str(), mode.c_str());
+    if (file == nullptr) {
+        Panic("%s", std::strerror(errno));
+    }
+    return file;
+}
+
+void PersistentRegister::CloseFile(std::FILE *file)
+{
+    int success = std::fclose(file);
+    if (success != 0) {
+        Panic("Unable to close file.");
+    }
+}

lib/persistent_register.h

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * persistent_register.h: A disk-backed persistent register.
+ *
+ * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+#ifndef _LIB_PERSISTENT_REGISTER_H_
+#define _LIB_PERSISTENT_REGISTER_H_
+
+#include <cstdio>
+
+#include <string>
+
+// A PersistentRegister is used to read and write a string that is persisted to
+// disk. It's like a database for a single string value. Here's how you might
+// use it.
+//
+//     // Persist x to the file "x.bin".
+//     PersistentRegister x("x.bin");
+//
+//     if (!x.Initialized()) {
+//         // If x has not yet been written, write "Hello, World!".
+//         x.Write("Hello, World!");
+//     } else {
+//         // If x has been written, read and print the value of x.
+//         std::cout << x.Read() << std::endl;
+//     }
+//
+// The first time this program is called, it will detect that x has not been
+// written and will write "Hello, World!". The second time it's called, it will
+// read and print "Hello, World!".
+class PersistentRegister {
+public:
+    PersistentRegister(const std::string &filename) : filename_(filename) {}
+
+    // Returns whether a PersistentRegister is initalized (i.e. the file into
+    // which the register is persisted exists).
+    bool Initialized() const;
+
+    // Read a value from the register, or return an empty string if the
+    // register is not initalized. Read panics on error.
+    std::string Read() const;
+
+    // Write a value to the register. Write panics on error.
+    void Write(const std::string &s);
+
+    // Return the filename in which the register is persisted.
+    std::string Filename();
+
+private:
+    // Note that using C++ file IO, there is not really a way to ensure that
+    // data has been forced to disk [1]. Thus, our implementation of
+    // PersistentRegister uses C file IO so that it can use primitives like
+    // fsync.
+    //
+    // [1]: https://stackoverflow.com/q/676787/3187068
+
+    // `OpenFile(f, m)` calls `std::fopen(f, m)` but calls `Panic` on error.
+    static std::FILE *OpenFile(const std::string &filename,
+                               const std::string &mode);
+
+    // `CloseFile(f, m)` calls `std::fclose(f)` but calls `Panic` on error.
+    static void CloseFile(std::FILE *file);
+
+    // The filename of the file that contains the persisted data.
+    const std::string filename_;
+};
+
+#endif  // _LIB_PERSISTENT_REGISTER_H_

lib/repltransport.cc

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * repltransport.cc: REPL-driven step-by-step simulated transport.
+ *
+ * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
+ *                     Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+#include "lib/repltransport.h"
+
+#include <iostream>
+#include <iterator>
+#include <sstream>
+#include <string>
+#include <tuple>
+#include <utility>
+
+namespace {
+
+// https://stackoverflow.com/a/236803/3187068
+template <typename Out>
+void split(const std::string &s, char delim, Out result) {
+    std::stringstream ss(s);
+    std::string item;
+    while (std::getline(ss, item, delim)) {
+        *(result++) = item;
+    }
+}
+
+// https://stackoverflow.com/a/236803/3187068
+std::vector<std::string> split(const std::string &s, char delim) {
+    std::vector<std::string> elems;
+    split(s, delim, std::back_inserter(elems));
+    return elems;
+}
+
+// https://stackoverflow.com/a/4654718/3187068
+bool is_number(const std::string &s) {
+    std::string::const_iterator it = s.begin();
+    while (it != s.end() && std::isdigit(*it)) ++it;
+    return !s.empty() && it == s.end();
+}
+
+// https://stackoverflow.com/a/1494435/3187068
+void string_replace(std::string *str, const std::string &oldStr,
+                    const std::string &newStr) {
+    std::string::size_type pos = 0u;
+    while ((pos = str->find(oldStr, pos)) != std::string::npos) {
+        str->replace(pos, oldStr.length(), newStr);
+        pos += newStr.length();
+    }
+}
+
+}  // namespace
+
+void ReplTransport::Register(TransportReceiver *receiver,
+                             const transport::Configuration &config,
+                             int replicaIdx) {
+    // If replicaIdx is -1, then the registering receiver is a client.
+    // Otherwise, replicaIdx is in the range [0, config.n), and the registering
+    // receiver is a replica.
+    bool is_client = replicaIdx == -1;
+
+    if (is_client) {
+        // Create the client's address.
+        std::string port = std::to_string(client_id_);
+        auto repl_addr = new ReplTransportAddress("client", std::move(port));
+        receiver->SetAddress(repl_addr);
+        client_id_++;
+
+        // Register receiver.
+        receivers_[*repl_addr].receiver = receiver;
+    } else {
+        // Set the receiver's address.
+        transport::ReplicaAddress addr = config.replica(replicaIdx);
+        auto repl_addr = new ReplTransportAddress(addr.host, addr.port);
+        receiver->SetAddress(repl_addr);
+
+        // Register receiver.
+        receivers_[*repl_addr].receiver = receiver;
+    }
+
+    // Register with superclass.
+    RegisterConfiguration(receiver, config, replicaIdx);
+}
+
+int ReplTransport::Timer(uint64_t ms, timer_callback_t cb) {
+    timer_id_++;
+    ASSERT(timers_.count(timer_id_) == 0);
+    timers_[timer_id_] = cb;
+    return timer_id_;
+}
+
+bool ReplTransport::CancelTimer(int id) {
+    if (timers_.count(id) == 0) {
+        return false;
+    } else {
+        timers_.erase(id);
+        return true;
+    }
+}
+
+void ReplTransport::CancelAllTimers() {
+    timers_.clear();
+}
+
+bool ReplTransport::DeliverMessage(const ReplTransportAddress &addr,
+                                   int index) {
+    history_.push_back("transport.DeliverMessage({\"" + addr.Host() + "\", \"" +
+                       addr.Port() + "\"}, " + std::to_string(index) + ");");
+    ASSERT(receivers_.count(addr) != 0);
+    TransportReceiverState &state = receivers_[addr];
+
+    // If the recipient of this address hasn't yet been registered, then
+    // state.receiver is null.
+    if (state.receiver == nullptr) {
+        return false;
+    }
+
+    // Deliver the message.
+    const QueuedMessage &m = state.msgs.at(index);
+    string data;
+    m.msg->SerializeToString(&data);
+    state.receiver->ReceiveMessage(m.src, m.msg->GetTypeName(), data);
+    return true;
+}
+
+void ReplTransport::TriggerTimer(int timer_id) {
+    history_.push_back("transport.TriggerTimer(" + std::to_string(timer_id) +
+                       ");");
+    ASSERT(timers_.count(timer_id) != 0);
+    timers_[timer_id]();
+}
+
+void ReplTransport::PrintState() const {
+    // Show the history.
+    std::cout << "- History" << std::endl;
+    for (const std::string &command : history_) {
+        std::cout << "    " << command << std::endl;
+    }
+
+    // Show the timers.
+    std::cout << "- Timers" << std::endl;
+    for (const std::pair<const int, timer_callback_t> &p : timers_) {
+        std::cout << "  - [" << p.first << "]" << std::endl;
+    }
+
+    // Show the message buffers.
+    for (const std::pair<const ReplTransportAddress, TransportReceiverState>
+             &p : receivers_) {
+        const ReplTransportAddress &addr = p.first;
+        const TransportReceiverState &state = p.second;
+
+        std::cout << "- " << addr;
+        if (state.receiver == nullptr) {
+            std::cout << " [not registered]";
+        }
+        std::cout << std::endl;
+
+        for (std::size_t i = 0; i < state.msgs.size(); ++i) {
+            const Message *msg = state.msgs[i].msg.get();
+            std::string debug = msg->DebugString();
+            string_replace(&debug, "\n", "\n        ");
+            std::cout << "  - [" << i << "] " << msg->GetTypeName() << std::endl
+                      << "        " << debug << std::endl;
+        }
+    }
+}
+
+bool ReplTransport::RunOne() {
+    // Parse response.
+    while (true) {
+        // Prompt user and read response.
+        std::cout << "> " << std::flush;
+        std::string line;
+        std::getline(std::cin, line);
+        if (std::cin.fail() || std::cin.eof()) {
+            return true;
+        }
+        std::vector<std::string> words = split(line, ' ');
+
+        const std::string usage =
+            "Usage: quit | show | <timer_id> | <host> <port> <index>";
+        if (words.size() == 1) {
+            if (words[0] == "quit") {
+                return true;
+            }
+            if (words[0] == "show") {
+                PrintState();
+                return false;
+            }
+
+            if (is_number(words[0])) {
+                int timer_id = std::stoi(words[0]);
+                TriggerTimer(timer_id);
+                return false;
+            } else {
+                std::cout << usage << std::endl;
+            }
+        } else if (words.size() == 3) {
+            if (!is_number(words[2])) {
+                std::cout << usage << std::endl;
+            } else {
+                ReplTransportAddress addr(words[0], words[1]);
+                int index = std::stoi(words[2]);
+                if (receivers_.count(addr) == 0) {
+                    std::cout << "Receiver not found." << std::endl;
+                } else {
+                    DeliverMessage(addr, index);
+                    return false;
+                }
+            }
+        } else {
+            std::cout << usage << std::endl;
+        }
+    }
+}
+
+void ReplTransport::Run() {
+    bool done = false;
+    while (!done) {
+        done = RunOne();
+    }
+}
+
+bool ReplTransport::SendMessageInternal(TransportReceiver *src,
+                                        const ReplTransportAddress &dst,
+                                        const Message &m,
+                                        bool multicast) {
+    // Multicast is not supported.
+    ASSERT(!multicast);
+
+    const ReplTransportAddress &repl_addr =
+        dynamic_cast<const ReplTransportAddress &>(src->GetAddress());
+    std::unique_ptr<Message> msg(m.New());
+    msg->CheckTypeAndMergeFrom(m);
+    receivers_[dst].msgs.push_back(QueuedMessage(repl_addr, std::move(msg)));
+    return true;
+}
+
+ReplTransportAddress ReplTransport::LookupAddress(
+    const transport::Configuration &cfg, int replicaIdx) {
+    transport::ReplicaAddress addr = cfg.replica(replicaIdx);
+    return ReplTransportAddress(addr.host, addr.port);
+}
+
+const ReplTransportAddress *ReplTransport::LookupMulticastAddress(
+    const transport::Configuration *cfg) {
+    return nullptr;
+}

lib/repltransport.h

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * repltransport.h: REPL-driven step-by-step simulated transport.
+ *
+ * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
+ *                     Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+
+// Distributed algorithms have to handle arbitrary message delays, message
+// loss, message reordering, node failure, network partitions, etc. However,
+// these failure scenarios are rare, which can make it difficult to flesh out
+// all the corner cases of a distributed algorithm.
+//
+// Take IR for example. If we want to trigger the IR-MERGE-RECORDS function to
+// run with a non-empty d and a non-empty u, we have to
+//   1. propose two separate messages,
+//   2. deliver one to a supermajority,
+//   3. deliver the other to a simple majority,
+//   4. prevent both messages from being finalized, and
+//   5. trigger a view change.
+//
+// ReplTransport is a simulated transport, like SimulatedTransport, that lets
+// users manipulate every aspect of the execution of a distributed system. When
+// run, a ReplTransport opens up a REPL with which users can use to trigger an
+// arbitrary timeout or deliver an arbitrary message to a receiver.
+//
+// For example, imagine a simple distributed system with two nodes: ping
+// (localhost:8000) and pong (localhost:9000). Initially, ping sends a message
+// to pong, pong replies with a message, ping replies with another message, and
+// so on. If a node hasn't heard from the other after some timeout, it resends
+// its message. An interaction with a ReplTransport would look something like
+// this (comments inline):
+//
+//     $ ./ping_pong
+//     > show                  # show the state
+//     - History               # A history of all commands (empty at first)
+//     - Timers                # A list of all timer ids
+//       - [1]                 # ping's timeout
+//       - [2]                 # pong's timeout
+//     - localhost:8000        # ping (no pending messages)
+//     - localhost:9000        # pong
+//       - [0] PingMessage     # pong's pending message from ping
+//
+//     > localhost 9000 0      # Deliver the 0th message to pong
+//     > show
+//     - History               # A history of all executed commands
+//         transport.DeliverMessage({"localhost", "9000"}, 0);
+//     - Timers
+//       - [1]
+//       - [2]
+//     - localhost:8000
+//       - [0] PongMessage     # pings's pending message from pong
+//     - localhost:9000
+//       - [0] PingMessage     # Notice that this message wasn't removed. We
+//                             # can deliver the same message multiple times.
+//
+//     > localhost 8000 0      # Deliver the 0th message to ping
+//     > show
+//     - History
+//         transport.DeliverMessage({"localhost", "9000"}, 0);
+//         transport.DeliverMessage({"localhost", "8000"}, 0);
+//     - Timers
+//       - [1]
+//       - [2]
+//     - localhost:8000
+//       - [0] PongMessage
+//     - localhost:9000
+//       - [0] PingMessage
+//       - [1] PingMessage
+//
+//     > 1                     # Trigger ping's timeout
+//     > show
+//     - History
+//         transport.DeliverMessage({"localhost", "9000"}, 0);
+//         transport.DeliverMessage({"localhost", "8000"}, 0);
+//         transport.TriggerTimer(1);
+//     - Timers
+//       - [1]
+//       - [2]
+//     - localhost:8000
+//       - [0] PongMessage
+//     - localhost:9000
+//       - [0] PingMessage
+//       - [1] PingMessage
+//       - [2] PingMessage     # ping resent its message to pong
+//
+//     > quit
+//
+// Also notice that the ReplTransport prints out a history of the executed
+// commands. You can copy and paste these commands into your code to replay
+// your interaction with the REPL.
+
+#ifndef _LIB_REPLTRANSPORT_H_
+#define _LIB_REPLTRANSPORT_H_
+
+#include <functional>
+#include <map>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <tuple>
+
+#include "lib/configuration.h"
+#include "lib/transport.h"
+#include "lib/transportcommon.h"
+
+class ReplTransportAddress : public TransportAddress {
+public:
+    // Constructors.
+    ReplTransportAddress() {}
+
+    ReplTransportAddress(std::string host, std::string port)
+        : host_(std::move(host)), port_(std::move(port)) {}
+
+    ReplTransportAddress(const ReplTransportAddress &other)
+        : ReplTransportAddress(other.host_, other.port_) {}
+
+    ReplTransportAddress(ReplTransportAddress &&other)
+        : ReplTransportAddress() {
+        swap(*this, other);
+    }
+
+    ReplTransportAddress &operator=(ReplTransportAddress other) {
+        swap(*this, other);
+        return *this;
+    }
+
+    friend void swap(ReplTransportAddress &x, ReplTransportAddress &y) {
+        std::swap(x.host_, y.host_);
+        std::swap(x.port_, y.port_);
+    }
+
+    // Comparators.
+    bool operator==(const ReplTransportAddress &other) const {
+        return Key() == other.Key();
+    }
+    bool operator!=(const ReplTransportAddress &other) const {
+        return Key() != other.Key();
+    }
+    bool operator<(const ReplTransportAddress &other) const {
+        return Key() < other.Key();
+    }
+    bool operator<=(const ReplTransportAddress &other) const {
+        return Key() <= other.Key();
+    }
+    bool operator>(const ReplTransportAddress &other) const {
+        return Key() > other.Key();
+    }
+    bool operator>=(const ReplTransportAddress &other) const {
+        return Key() >= other.Key();
+    }
+
+    // Getters.
+    const std::string& Host() const {
+        return host_;
+    }
+
+    const std::string& Port() const {
+        return port_;
+    }
+
+    ReplTransportAddress *clone() const override {
+        return new ReplTransportAddress(host_, port_);
+    }
+
+    friend std::ostream &operator<<(std::ostream &out,
+                                    const ReplTransportAddress &addr) {
+        out << addr.host_ << ":" << addr.port_;
+        return out;
+    }
+
+private:
+    std::tuple<const std::string&, const std::string&> Key() const {
+        return std::forward_as_tuple(host_, port_);
+    }
+
+    std::string host_;
+    std::string port_;
+};
+
+class ReplTransport : public TransportCommon<ReplTransportAddress> {
+public:
+    void Register(TransportReceiver *receiver,
+                  const transport::Configuration &config,
+                  int replicaIdx) override;
+    int Timer(uint64_t ms, timer_callback_t cb) override;
+    bool CancelTimer(int id) override;
+    void CancelAllTimers() override;
+
+    // DeliverMessage(addr, i) delivers the ith queued inbound message to the
+    // receiver with address addr. It's possible to send a message to the
+    // address of a receiver that hasn't yet registered. In this case,
+    // DeliverMessage returns false. Otherwise, it returns true.
+    bool DeliverMessage(const ReplTransportAddress& addr, int index);
+
+    // Run timer with id timer_id.
+    void TriggerTimer(int timer_id);
+
+    // Launch the REPL.
+    void Run();
+
+protected:
+    bool SendMessageInternal(TransportReceiver *src,
+                             const ReplTransportAddress &dst, const Message &m,
+                             bool multicast = false) override;
+    ReplTransportAddress LookupAddress(const transport::Configuration &cfg,
+                                       int replicaIdx) override;
+    const ReplTransportAddress *LookupMulticastAddress(
+        const transport::Configuration *cfg) override;
+
+private:
+    // Prompt the user for input and either (1) trigger a timer, (2) deliver a
+    // message, or (3) quit. RunOne returns true if the user decides to quit.
+    bool RunOne();
+
+    // Pretty print the current state of the system. For example, PrintState
+    // prints the queued messages for every node in the system.
+    void PrintState() const;
+
+    struct QueuedMessage {
+        ReplTransportAddress src;
+        std::unique_ptr<Message> msg;
+
+        QueuedMessage(ReplTransportAddress src, std::unique_ptr<Message> msg)
+            : src(std::move(src)), msg(std::move(msg)) {}
+    };
+
+    struct TransportReceiverState {
+        // receiver can be null if it has queued messages but hasn't yet been
+        // registered with a ReplTransport.
+        TransportReceiver *receiver;
+
+        // Queued inbound messages.
+        std::vector<QueuedMessage> msgs;
+    };
+
+    // receivers_ maps a receiver r's address to r and r's queued messages.
+    std::map<ReplTransportAddress, TransportReceiverState> receivers_;
+
+    // timer_id_ is an incrementing counter used to assign timer ids.
+    int timer_id_ = 0;
+
+    // timers_ maps timer ids to timers.
+    std::map<int, timer_callback_t> timers_;
+
+    // client_id_ is an incrementing counter used to assign addresses to
+    // clients. The first client gets address client:0, the next client gets
+    // address client:1, etc.
+    int client_id_ = 0;
+
+    // A history of all the command issued to this ReplTransport.
+    std::vector<std::string> history_;
+};
+
+#endif // _LIB_REPLTRANSPORT_H_

lib/simtransport.cc

@@ -4,7 +4,9 @@
  * simtransport.cc:
  *   simulated message-passing interface for testing use
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
+ *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  *
  * Permission is hereby granted, free of charge, to any person
  * obtaining a copy of this software and associated documentation
@@ -28,15 +30,15 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/assert.h"
-#include "paxos-lib/lib/message.h"
-#include "paxos-lib/lib/simtransport.h"
+#include "lib/assert.h"
+#include "lib/message.h"
+#include "lib/simtransport.h"
 #include <google/protobuf/message.h>
 
 SimulatedTransportAddress::SimulatedTransportAddress(int addr)
     : addr(addr)
 {
-    
+
 }
 
 int
@@ -68,12 +70,12 @@ SimulatedTransport::SimulatedTransport()
 
 SimulatedTransport::~SimulatedTransport()
 {
-    
+
 }
 
 void
 SimulatedTransport::Register(TransportReceiver *receiver,
-                             const specpaxos::Configuration &config,
+                             const transport::Configuration &config,
                              int replicaIdx)
 {
     // Allocate an endpoint
@@ -97,12 +99,12 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
                                         bool multicast)
 {
     ASSERT(!multicast);
-    
+
     int dst = dstAddr.addr;
-    
+
     Message *msg = m.New();
     msg->CheckTypeAndMergeFrom(m);
-    
+
     int srcAddr =
         dynamic_cast<const SimulatedTransportAddress &>(src->GetAddress()).addr;
 
@@ -117,11 +119,11 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
             return true;
         }
     }
-                      
+
     string msgData;
     msg->SerializeToString(&msgData);
     delete msg;
-    
+
     QueuedMessage q(dst, srcAddr, m.GetTypeName(), msgData);
 
     if (delay == 0) {
@@ -131,18 +133,18 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
                 queue.push_back(q);
             });
     }
-    return true;    
+    return true;
 }
 
 SimulatedTransportAddress
-SimulatedTransport::LookupAddress(const specpaxos::Configuration &cfg,
+SimulatedTransport::LookupAddress(const transport::Configuration &cfg,
                                   int idx)
 {
     // Check every registered replica to see if its configuration and
     // idx match. This is the least efficient possible way to do this,
     // but that's why this is the simulated transport not the real
     // one... (And we only have to do this once at runtime.)
-    
+
 
     for (auto & kv : configurations) {
         if (*(kv.second) == cfg) {
@@ -155,12 +157,12 @@ SimulatedTransport::LookupAddress(const specpaxos::Configuration &cfg,
             }
         }
     }
-    
+
     Panic("No replica %d was registered", idx);
 }
 
 const SimulatedTransportAddress *
-SimulatedTransport::LookupMulticastAddress(const specpaxos::Configuration *cfg)
+SimulatedTransport::LookupMulticastAddress(const transport::Configuration *cfg)
 {
     return NULL;
 }
@@ -169,7 +171,7 @@ void
 SimulatedTransport::Run()
 {
     LookupAddresses();
-    
+
     do {
         // Process queue
         while (!queue.empty()) {
@@ -188,7 +190,7 @@ SimulatedTransport::Run()
             timers.erase(iter);
             cb();
         }
-        
+
         // ...then retry to see if there are more queued messages to
         // deliver first
     } while (!queue.empty() || (processTimers && !timers.empty()));
@@ -232,7 +234,7 @@ SimulatedTransport::CancelTimer(int id)
             iter++;
         }
     }
-    
+
     return found;
 }