6477a9fe385314d1db4825952cf71b1beef24d5a to master · syslab / tapir

Some changes are not shown.

For a faster browsing experience, only 47 of 330 files are shown. Download one of the files below to see all changes.

.gitignore

0 → 100644

+75 −0

Original line number	Diff line number	Diff line
		.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

.gitmodules

0 → 100644

+0 −0

Original line number	Diff line number	Diff line

LICENSE

0 → 100644

+21 −0

Original line number	Diff line number	Diff line
		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

+18 −9

Original line number	Diff line number	Diff line
		#
		# Top-level makefile for OR-2PC
		# Top-level makefile for IR and TAPIR
		#

		CC = clang
		CXX = clang++
		LD = clang++
		CC = gcc
		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
		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
		@@ -126,10 +126,13 @@ include replication/common/Rules.mk
		include replication/vr/Rules.mk
		include replication/ir/Rules.mk
		include store/common/Rules.mk
		include store/txnstore/Rules.mk
		include store/qwstore/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
		#
		@@ -234,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

+45 −10

Original line number	Diff line number	Diff line
		# TAPIR

		This repository includes code implementing several replicated and
		transactional key-value stores.
		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:

		@@ -10,20 +35,30 @@ The repo is structured as follows:
		simulate network conditions on a local machine, including packet
		delays and reorderings.

		- /replication
		- /replication - replication library for the distributed stores
		- /vr - implementation of viewstamped replication protocol
		- /ir - implementation of inconsistent replication protocol

		- /store
		- /tapir - implementation of TAPIR designed to work with IR
		- /occstore - implementation of an OCC-based 2PC transactional
		- /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
		- /lockstore - implementation of a strict two-phase locking 2PC
		transactional storage system, designed to work with VR
		- /qwstore - implementation of an eventually consistent storage
		- /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 and libevent, so you will need those development libraries installed on your machine. On Linux, this can be done through apt.
		No newline at end of file
		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

+9 −1

Original line number	Diff line number	Diff line
		@@ -3,7 +3,8 @@ d := $(dir $(lastword $(MAKEFILE_LIST)))
		SRCS += $(addprefix $(d), \
		lookup3.cc message.cc memory.cc \
		latency.cc configuration.cc transport.cc \
		udptransport.cc simtransport.cc)
		udptransport.cc tcptransport.cc simtransport.cc repltransport.cc \
		persistent_register.cc)

		PROTOS += $(addprefix $(d), \
		latency-format.proto)
		@@ -24,6 +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

+1 −1

Original line number	Diff line number	Diff line
		@@ -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

+60 −35

Original line number	Diff line number	Diff line
		@@ -33,10 +33,9 @@
		#include "lib/configuration.h"
		#include "lib/message.h"

		#include <iostream>
		#include <fstream>
		#include <string>
		#include <string.h>
		#include <cstring>
		#include <stdexcept>
		#include <tuple>

		namespace transport {

		@@ -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)
		@@ -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());
		}
		}

		@@ -213,4 +219,23 @@ Configuration::operator==(const Configuration &other) const
		return true;
		}

		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

+24 −4

Original line number	Diff line number	Diff line
		@@ -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);
		}
		};


		@@ -72,6 +82,16 @@ public:
		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
		@@ -82,7 +102,8 @@ private:
		bool hasMulticast;
		};

		} // namespace replication
		} // namespace transport


		namespace std {
		template <> struct hash<transport::ReplicaAddress>
		@@ -110,5 +131,4 @@ template <> struct hash<transport::Configuration>
		};
		}


		#endif /* _LIB_CONFIGURATION_H_ */

lib/hash.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -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

+2 −0

Original line number	Diff line number	Diff line
		syntax = "proto2";

		package transport.latency.format;

		message LatencyDist

lib/latency.cc

+1 −1

Original line number	Diff line number	Diff line
		@@ -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
		*

lib/latency.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -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
		*

lib/memory.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -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

+10 −0

Original line number	Diff line number	Diff line
		@@ -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

0 → 100644

+126 −0

Original line number	Diff line number	Diff line
		// -- 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

0 → 100644

+91 −0

Original line number	Diff line number	Diff line
		// -- 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

0 → 100644

+272 −0

Original line number	Diff line number	Diff line
		// -- 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

0 → 100644

+278 −0

Original line number	Diff line number	Diff line
		// -- 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

+14 −14

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		* simulated message-passing interface for testing use
		*
		* 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/simtransport.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		* simulated message-passing interface for testing use
		*
		* 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/tcptransport.cc

0 → 100644

+681 −0

File added.

Preview size limit exceeded, changes collapsed.

lib/tcptransport.h

0 → 100644

+143 −0

Original line number	Diff line number	Diff line
		// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --
		/***********************************************************************
		*
		* tcptransport.h:
		* message-passing network interface that uses UDP message delivery
		* and libasync
		*
		* 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_TCPTRANSPORT_H_
		#define _LIB_TCPTRANSPORT_H_

		#include "lib/configuration.h"
		#include "lib/transport.h"
		#include "lib/transportcommon.h"

		#include <event2/event.h>
		#include <event2/buffer.h>
		#include <event2/bufferevent.h>

		#include <map>
		#include <unordered_map>
		#include <list>
		#include <random>
		#include <mutex>
		#include <netinet/in.h>

		class TCPTransportAddress : public TransportAddress
		{
		public:
		TCPTransportAddress * clone() const;
		private:
		TCPTransportAddress(const sockaddr_in &addr);

		sockaddr_in addr;
		friend class TCPTransport;
		friend bool operator==(const TCPTransportAddress &a,
		const TCPTransportAddress &b);
		friend bool operator!=(const TCPTransportAddress &a,
		const TCPTransportAddress &b);
		friend bool operator<(const TCPTransportAddress &a,
		const TCPTransportAddress &b);
		};

		class TCPTransport : public TransportCommon<TCPTransportAddress>
		{
		public:
		TCPTransport(double dropRate = 0.0, double reogrderRate = 0.0,
		int dscp = 0, bool handleSignals = true);
		virtual ~TCPTransport();
		void Register(TransportReceiver *receiver,
		const transport::Configuration &config,
		int replicaIdx);
		void Run();
		void Stop();
		int Timer(uint64_t ms, timer_callback_t cb);
		bool CancelTimer(int id);
		void CancelAllTimers();

		private:
		std::mutex mtx;
		struct TCPTransportTimerInfo
		{
		TCPTransport *transport;
		timer_callback_t cb;
		event *ev;
		int id;
		};
		struct TCPTransportTCPListener
		{
		TCPTransport *transport;
		TransportReceiver *receiver;
		int acceptFd;
		int replicaIdx;
		event *acceptEvent;
		std::list<struct bufferevent *> connectionEvents;
		};
		event_base *libeventBase;
		std::vector<event *> listenerEvents;
		std::vector<event *> signalEvents;
		std::map<int, TransportReceiver*> receivers; // fd -> receiver
		std::map<TransportReceiver*, int> fds; // receiver -> fd
		int lastTimerId;
		std::map<int, TCPTransportTimerInfo *> timers;
		std::list<TCPTransportTCPListener *> tcpListeners;
		std::map<TCPTransportAddress, struct bufferevent *> tcpOutgoing;
		std::map<struct bufferevent *, TCPTransportAddress> tcpAddresses;

		bool SendMessageInternal(TransportReceiver *src,
		const TCPTransportAddress &dst,
		const Message &m, bool multicast = false);

		TCPTransportAddress
		LookupAddress(const transport::ReplicaAddress &addr);
		TCPTransportAddress
		LookupAddress(const transport::Configuration &cfg,
		int replicaIdx);
		const TCPTransportAddress *
		LookupMulticastAddress(const transport::Configuration*config) { return NULL; };

		void ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst);
		void OnTimer(TCPTransportTimerInfo *info);
		static void TimerCallback(evutil_socket_t fd,
		short what, void *arg);
		static void LogCallback(int severity, const char *msg);
		static void FatalCallback(int err);
		static void SignalCallback(evutil_socket_t fd,
		short what, void *arg);
		static void TCPAcceptCallback(evutil_socket_t fd, short what,
		void *arg);
		static void TCPReadableCallback(struct bufferevent *bev,
		void *arg);
		static void TCPEventCallback(struct bufferevent *bev,
		short what, void *arg);
		static void TCPIncomingEventCallback(struct bufferevent *bev,
		short what, void *arg);
		static void TCPOutgoingEventCallback(struct bufferevent *bev,
		short what, void *arg);
		};

		#endif // _LIB_TCPTRANSPORT_H_

lib/tests/simtransport-testmessage.proto

+2 −0

Original line number	Diff line number	Diff line
		syntax = "proto2";

		package transport.test;

		message TestMessage {

lib/timeval.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		* utility functions for manipulating timevals
		*
		* 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/transport.cc

+1 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		* message-passing network interface; common definitions
		*
		* 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/transportcommon.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		* template support for implementing transports
		*
		* 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/udptransport.cc

+31 −37

Original line number	Diff line number	Diff line
		@@ -38,6 +38,7 @@
		#include <event2/event.h>
		#include <event2/thread.h>

		#include <memory>
		#include <random>

		#include <arpa/inet.h>
		@@ -178,9 +179,8 @@ BindToPort(int fd, const string &host, const string &port)
		}

		UDPTransport::UDPTransport(double dropRate, double reorderRate,
		int dscp, event_base *evbase)
		: dropRate(dropRate), reorderRate(reorderRate),
		dscp(dscp)
		int dscp, bool handleSignals)
		: dropRate(dropRate), reorderRate(reorderRate), dscp(dscp)
		{

		lastTimerId = 0;
		@@ -197,20 +197,15 @@ UDPTransport::UDPTransport(double dropRate, double reorderRate,
		}

		// Set up libevent
		evthread_use_pthreads();
		event_set_log_callback(LogCallback);
		event_set_fatal_callback(FatalCallback);
		// XXX Hack for Naveen: allow the user to specify an existing
		// libevent base. This will probably not work exactly correctly
		// for error messages or signals, but that doesn't much matter...
		if (evbase) {
		libeventBase = evbase;
		} else {
		evthread_use_pthreads();

		libeventBase = event_base_new();
		evthread_make_base_notifiable(libeventBase);
		}

		// Set up signal handler
		if (handleSignals) {
		signalEvents.push_back(evsignal_new(libeventBase, SIGTERM,
		SignalCallback, this));
		signalEvents.push_back(evsignal_new(libeventBase, SIGINT,
		@@ -220,6 +215,7 @@ UDPTransport::UDPTransport(double dropRate, double reorderRate,
		event_add(x, NULL);
		}
		}
		}

		UDPTransport::~UDPTransport()
		{
		@@ -385,7 +381,8 @@ UDPTransport::Register(TransportReceiver *receiver,
		}

		static size_t
		SerializeMessage(const ::google::protobuf::Message &m, char **out)
		SerializeMessage(const ::google::protobuf::Message &m,
		std::unique_ptr<char[]> *out)
		{
		string data = m.SerializeAsString();
		string type = m.GetTypeName();
		@@ -394,7 +391,8 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
		ssize_t totalLen = (typeLen + sizeof(typeLen) +
		dataLen + sizeof(dataLen));

		char *buf = new char[totalLen];
		std::unique_ptr<char[]> unique_buf(new char[totalLen]);
		char *buf = unique_buf.get();

		char *ptr = buf;
		((size_t ) ptr) = typeLen;
		@@ -410,7 +408,7 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
		memcpy(ptr, data.c_str(), dataLen);
		ptr += dataLen;

		*out = buf;
		*out = std::move(unique_buf);
		return totalLen;
		}

		@@ -423,8 +421,9 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
		sockaddr_in sin = dynamic_cast<const UDPTransportAddress &>(dst).addr;

		// Serialize message
		char *buf;
		size_t msgLen = SerializeMessage(m, &buf);
		std::unique_ptr<char[]> unique_buf;
		size_t msgLen = SerializeMessage(m, &unique_buf);
		char *buf = unique_buf.get();

		int fd = fds[src];

		@@ -435,7 +434,7 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
		if (sendto(fd, buf, msgLen, 0,
		(sockaddr *)&sin, sizeof(sin)) < 0) {
		PWarning("Failed to send message");
		goto fail;
		return false;
		}
		} else {
		int numFrags = ((msgLen-1) / MAX_UDP_MESSAGE_SIZE) + 1;
		@@ -463,17 +462,12 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
		(sockaddr *)&sin, sizeof(sin)) < 0) {
		PWarning("Failed to send message fragment %ld",
		fragStart);
		goto fail;
		return false;
		}
		}
		}

		delete [] buf;
		return true;

		fail:
		delete [] buf;
		return false;
		}

		void

lib/udptransport.h

+1 −1

Original line number	Diff line number	Diff line
		@@ -66,7 +66,7 @@ class UDPTransport : public TransportCommon<UDPTransportAddress>
		{
		public:
		UDPTransport(double dropRate = 0.0, double reorderRate = 0.0,
		int dscp = 0, event_base *evbase = nullptr);
		int dscp = 0, bool handleSignals = true);
		virtual ~UDPTransport();
		void Register(TransportReceiver *receiver,
		const transport::Configuration &config,

libtapir/Rules.mk

0 → 100644

+6 −0

Original line number	Diff line number	Diff line
		d := $(dir $(lastword $(MAKEFILE_LIST)))

		$(d)libtapir.so: $(patsubst %.o,%-pic.o, $(OBJS-all-clients))
		LDFLAGS-$(d)libtapir.so += -shared

		BINS += $(d)libtapir.so

lockserver/Rules.mk

+28 −5

Original line number	Diff line number	Diff line
		d := $(dir $(lastword $(MAKEFILE_LIST)))

		SRCS += $(addprefix $(d), server.cc client.cc)
		SRCS += $(addprefix $(d), \
		server.cc client.cc server-main.cc client-main.cc \
		lockserver-repl.cc)

		PROTOS += $(addprefix $(d), locks-proto.proto)

		$(d)server: $(LIB-udptransport) $(OBJS-ir-replica) $(o)locks-proto.o $(o)server.o
		$(d)server-main: $(o)server-main.o \
		$(o)locks-proto.o \
		$(o)server.o \
		$(LIB-udptransport) \
		$(OBJS-ir-replica)

		$(d)client: $(LIB-udptransport) $(OBJS-ir-client) $(LIB-store-common) \
		$(o)locks-proto.o $(o)client.o
		$(d)client-main: $(o)client-main.o \
		$(o)locks-proto.o \
		$(o)client.o \
		$(LIB-udptransport) \
		$(OBJS-ir-client) \
		$(LIB-store-common)

		BINS += $(d)server $(d)client
		$(d)lockserver-repl: $(o)lockserver-repl.o \
		$(o)locks-proto.o \
		$(o)server.o \
		$(o)client.o \
		$(OBJS-ir-replica) \
		$(OBJS-ir-client) \
		$(LIB-configuration) \
		$(LIB-repltransport) \
		$(LIB-store-common) \
		$(GTEST_MAIN)

		BINS += $(d)server-main $(d)client-main $(d)lockserver-repl

		include $(d)tests/Rules.mk

lockserver/client-main.cc

0 → 100644

+103 −0

Original line number	Diff line number	Diff line
		#include <thread>

		#include "lockserver/client.h"
		#include "lib/udptransport.h"

		namespace {

		void
		usage()
		{
		printf("Unknown command.. Try again!\n");
		printf("Usage: exit \| q \| lock <key> \| unlock <key>\n");
		}

		} // namespace

		int
		main(int argc, char **argv)
		{
		const char *configPath = NULL;

		// Parse arguments
		int opt;
		while ((opt = getopt(argc, argv, "c:")) != -1) {
		switch (opt) {
		case 'c':
		configPath = optarg;
		break;

		default:
		fprintf(stderr, "Unknown argument %s\n", argv[optind]);
		}
		}

		if (!configPath) {
		fprintf(stderr, "option -c is required\n");
		return EXIT_FAILURE;
		}

		// Load configuration
		std::ifstream configStream(configPath);
		if (configStream.fail()) {
		Panic("Unable to read configuration file: %s\n", configPath);
		}
		transport::Configuration config(configStream);

		// Create lock client.
		UDPTransport transport(0.0, 0.0, 0);
		lockserver::LockClient locker(&transport, config);
		std::thread run_transport([&transport]() { transport.Run(); });

		char c, cmd[2048], *tok;
		int clen, status;
		string key, value;

		while (1) {
		printf(">> ");
		fflush(stdout);

		clen = 0;
		while ((c = getchar()) != '\n')
		cmd[clen++] = c;
		cmd[clen] = '\0';

		tok = strtok(cmd, " ,.-");
		if (tok == NULL) continue;

		if (strcasecmp(tok, "exit") == 0 \|\| strcasecmp(tok, "q") == 0) {
		printf("Exiting..\n");
		break;
		} else if (strcasecmp(tok, "lock") == 0) {
		tok = strtok(NULL, " ,.-");
		if (tok == NULL) {
		usage();
		continue;
		}
		key = string(tok);
		status = locker.lock(key);

		if (status) {
		printf("Lock Successful\n");
		} else {
		printf("Failed to acquire lock..\n");
		}
		} else if (strcasecmp(tok, "unlock") == 0) {
		tok = strtok(NULL, " ,.-");
		if (tok == NULL) {
		usage();
		continue;
		}
		key = string(tok);
		locker.unlock(key);
		printf("Unlock Successful\n");
		} else {
		usage();
		}
		fflush(stdout);
		}

		transport.Stop();
		run_transport.join();
		return EXIT_SUCCESS;
		}

lockserver/client.cc

+57 −97

Original line number	Diff line number	Diff line
		@@ -30,88 +30,14 @@

		#include "lockserver/client.h"

		int
		main(int argc, char **argv)
		{
		const char *configPath = NULL;

		// Parse arguments
		int opt;
		while ((opt = getopt(argc, argv, "c:")) != -1) {
		switch (opt) {
		case 'c':
		configPath = optarg;
		break;

		default:
		fprintf(stderr, "Unknown argument %s\n", argv[optind]);
		}
		}

		if (!configPath) {
		fprintf(stderr, "option -c is required\n");
		return EXIT_FAILURE;
		}

		lockserver::LockClient locker(configPath);

		char c, cmd[2048], *tok;
		int clen, status;
		string key, value;

		while (1) {
		printf(">> ");
		fflush(stdout);

		clen = 0;
		while ((c = getchar()) != '\n')
		cmd[clen++] = c;
		cmd[clen] = '\0';

		if (clen == 0) continue;
		tok = strtok(cmd, " ,.-");

		if (strcasecmp(tok, "exit") == 0 \|\| strcasecmp(tok, "q") == 0) {
		printf("Exiting..\n");
		break;
		} else if (strcasecmp(tok, "lock") == 0) {
		tok = strtok(NULL, " ,.-");
		key = string(tok);
		status = locker.lock(key);

		if (status) {
		printf("Lock Successful\n");
		} else {
		printf("Failed to acquire lock..\n");
		}
		} else if (strcasecmp(tok, "unlock") == 0) {
		tok = strtok(NULL, " ,.-");
		key = string(tok);
		locker.unlock(key);
		printf("Unlock Successful\n");
		} else {
		printf("Unknown command.. Try again!\n");
		}
		fflush(stdout);
		}

		return EXIT_SUCCESS;
		}

		namespace lockserver {

		using namespace std;
		using namespace proto;

		LockClient::LockClient(const string &configPath) : transport(0.0, 0.0, 0)
		{
		// Load configuration
		std::ifstream configStream(configPath);
		if (configStream.fail()) {
		Panic("Unable to read configuration file: %s\n", configPath.c_str());
		}
		transport::Configuration config(configStream);

		LockClient::LockClient(Transport *transport,
		const transport::Configuration &config)
		: transport(transport) {
		client_id = 0;
		while (client_id == 0) {
		random_device rd;
		@@ -120,23 +46,14 @@ LockClient::LockClient(const string &configPath) : transport(0.0, 0.0, 0)
		client_id = dis(gen);
		}

		client = new replication::ir::IRClient(config, &transport, client_id);

		/* Run the transport in a new thread. */
		clientTransport = new thread(&LockClient::run_client, this);
		client = new replication::ir::IRClient(config, transport, client_id);
		}

		LockClient::~LockClient() { }

		void
		LockClient::run_client()
		{
		transport.Run();
		}

		bool
		LockClient::lock(const string &key)
		{
		LockClient::lock_async(const std::string &key) {
		ASSERT(waiting == nullptr);
		Debug("Sending LOCK");

		string request_str;
		@@ -147,19 +64,29 @@ LockClient::lock(const string &key)
		request.SerializeToString(&request_str);

		waiting = new Promise(1000);
		transport.Timer(0, [=]() {
		transport->Timer(0, [=]() {
		client->InvokeConsensus(request_str,
		bind(&LockClient::Decide,
		this,
		placeholders::_1),
		bind(&LockClient::LockCallback,
		this,
		placeholders::_1,
		placeholders::_2),
		bind(&LockClient::ErrorCallback,
		this,
		placeholders::_1,
		placeholders::_2));
		});
		}

		bool
		LockClient::lock_wait() {
		ASSERT(waiting != nullptr);

		int status = waiting->GetReply();
		delete waiting;
		waiting = nullptr;

		if (status == 0) {
		return true;
		@@ -170,8 +97,8 @@ LockClient::lock(const string &key)
		}

		void
		LockClient::unlock(const string &key)
		{
		LockClient::unlock_async(const std::string &key) {
		ASSERT(waiting == nullptr);
		Debug("Sending UNLOCK");

		string request_str;
		@@ -182,31 +109,53 @@ LockClient::unlock(const string &key)
		request.SerializeToString(&request_str);

		waiting = new Promise(1000);
		transport.Timer(0, [=]() {
		transport->Timer(0, [=]() {
		client->InvokeInconsistent(request_str,
		bind(&LockClient::UnlockCallback,
		this,
		placeholders::_1,
		placeholders::_2));
		});
		}

		void
		LockClient::unlock_wait() {
		waiting->GetReply();
		delete waiting;
		waiting = nullptr;
		}

		bool
		LockClient::lock(const string &key)
		{
		lock_async(key);
		return lock_wait();
		}

		void
		LockClient::unlock(const string &key)
		{
		unlock_async(key);
		return unlock_wait();
		}

		string
		LockClient::Decide(const set<string> &results)
		LockClient::Decide(const map<string, std::size_t> &results)
		{
		// If a majority say lock, we say lock.
		int success_count = 0;
		string key;
		for (string s : results) {
		for (const auto& string_and_count : results) {
		const string& s = string_and_count.first;
		const std::size_t count = string_and_count.second;

		Reply reply;
		reply.ParseFromString(s);
		key = reply.key();

		if (reply.status() == 0)
		success_count ++;
		if (reply.status() == 0) {
		success_count += count;
		}
		}

		string final_reply_str;
		@@ -243,4 +192,15 @@ LockClient::UnlockCallback(const std::string &request_str, const std::string &re
		w->Reply(0);
		}

		void
		LockClient::ErrorCallback(const std::string &request_str,
		replication::ErrorCode err)
		{
		Debug("Error Callback: %s %s", request_str.c_str(),
		replication::ErrorCodeToString(err).c_str());
		Promise *w = waiting;
		waiting = NULL;
		w->Reply(-3); // Invalid command.
		}

		} // namespace lockserver

lockserver/client.h

+22 −6

Original line number	Diff line number	Diff line
		@@ -33,49 +33,65 @@

		#include "lib/assert.h"
		#include "lib/message.h"
		#include "lib/udptransport.h"
		#include "lib/transport.h"
		#include "replication/ir/client.h"
		#include "store/common/promise.h"
		#include "lockserver/locks-proto.pb.h"

		#include <map>
		#include <set>
		#include <string>
		#include <thread>
		#include <random>

		namespace lockserver {

		class LockClient
		{
		public:
		LockClient(const std::string &configPath);
		LockClient(Transport* transport, const transport::Configuration &config);
		~LockClient();

		// Synchronously lock and unlock. Calling lock (or unlock) will block until
		// the lock (or unlock) request is fully processed.
		bool lock(const std::string &key);
		void unlock(const std::string &key);

		// Asynchronously lock and unlock. Calling lock_async or unlock_async will
		// not block. Calling lock_wait (or unlock_wait) will block for the
		// previous invocation of lock_async (or unlock_async) to complete.
		//
		// All async calls must be followed by a corresponding wait call. It is an
		// error to issue multiple async requests without waiting. It is also
		// erroneous to wait for a request which was never issued.
		void lock_async(const std::string &key);
		bool lock_wait();
		void unlock_async(const std::string &key);
		void unlock_wait();

		private:
		/* Unique ID for this client. */
		uint64_t client_id;

		/* Transport layer and thread. */
		UDPTransport transport;
		std::thread *clientTransport;
		Transport *transport;

		/* Function to run the transport thread. */
		void run_client();

		/* Decide function for a lock server. */
		string Decide(const std::set<string> &results);
		string Decide(const std::map<string, std::size_t> &results);

		/* IR client proxy. */
		replication::ir::IRClient *client;

		/* Promise to wait for pending operation. */
		Promise *waiting;
		Promise *waiting = nullptr;

		/* Callbacks for hearing back for an operation. */
		void LockCallback(const std::string &, const std::string &);
		void UnlockCallback(const std::string &, const std::string &);
		void ErrorCallback(const std::string &, replication::ErrorCode);
		};

		} // namespace lockserver

lockserver/locks-proto.proto

+2 −0

Original line number	Diff line number	Diff line
		syntax = "proto2";

		package lockserver.proto;

		message Request {

lockserver/lockserver-repl.cc

0 → 100644

+80 −0

Original line number	Diff line number	Diff line
		// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --
		/***********************************************************************
		*
		* lockserver-repl.cc: Step-by-step lock server evaluation.
		*
		* 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 <thread>
		#include <memory>

		#include "lib/configuration.h"
		#include "lib/repltransport.h"
		#include "lockserver/client.h"
		#include "lockserver/server.h"
		#include "replication/ir/replica.h"

		int main() {
		ReplTransport transport;
		std::vector<transport::ReplicaAddress> replica_addrs = {
		{"replica", "0"},
		{"replica", "1"},
		{"replica", "2"},
		{"replica", "3"},
		{"replica", "4"}};
		transport::Configuration config(5 /* n /, 2 / f */, replica_addrs);

		// Clients.
		lockserver::LockClient client_a(&transport, config);
		lockserver::LockClient client_b(&transport, config);
		lockserver::LockClient client_c(&transport, config);
		client_a.lock_async("a");
		client_b.lock_async("b");
		client_c.lock_async("c");

		// Servers.
		std::vector<std::unique_ptr<lockserver::LockServer>> servers;
		std::vector<std::unique_ptr<replication::ir::IRReplica>> replicas;
		for (std::size_t i = 0; i < replica_addrs.size(); ++i) {
		auto server = std::unique_ptr<lockserver::LockServer>(
		new lockserver::LockServer());
		servers.push_back(std::move(server));
		auto replica = std::unique_ptr<replication::ir::IRReplica>(
		new replication::ir::IRReplica(config, i, &transport,
		servers[i].get()));
		replicas.push_back(std::move(replica));
		}

		// Launch REPL.
		transport.Run();

		// Remove persisted files.
		for (std::size_t i = 0; i < replica_addrs.size(); ++i) {
		const transport::ReplicaAddress &addr = replica_addrs[i];
		const std::string filename =
		addr.host + ":" + addr.port + "_" + std::to_string(i) + ".bin";
		int success = std::remove(filename.c_str());
		ASSERT(success == 0);
		}
		}

lockserver/server-main.cc

0 → 100644

+62 −0

Original line number	Diff line number	Diff line
		#include "lockserver/server.h"

		int
		main(int argc, char **argv)
		{
		int index = -1;
		const char *configPath = NULL;

		// Parse arguments
		int opt;
		char *strtolPtr;
		while ((opt = getopt(argc, argv, "c:i:")) != -1) {
		switch (opt) {
		case 'c':
		configPath = optarg;
		break;

		case 'i':
		index = strtol(optarg, &strtolPtr, 10);
		if ((optarg == '\0') \|\| (strtolPtr != '\0') \|\| (index < 0)) {
		fprintf(stderr, "option -i requires a numeric arg\n");
		}
		break;

		default:
		fprintf(stderr, "Unknown argument %s\n", argv[optind]);
		}
		}

		if (!configPath) {
		fprintf(stderr, "option -c is required\n");
		return EXIT_FAILURE;
		}

		if (index == -1) {
		fprintf(stderr, "option -i is required\n");
		return EXIT_FAILURE;
		}

		// Load configuration
		std::ifstream configStream(configPath);
		if (configStream.fail()) {
		fprintf(stderr, "unable to read configuration file: %s\n", configPath);
		return EXIT_FAILURE;
		}
		transport::Configuration config(configStream);

		if (index >= config.n) {
		fprintf(stderr, "replica index %d is out of bounds; "
		"only %d replicas defined\n", index, config.n);
		return EXIT_FAILURE;
		}

		UDPTransport transport(0.0, 0.0, 0);

		lockserver::LockServer server;
		replication::ir::IRReplica replica(config, index, &transport, &server);

		transport.Run();

		return EXIT_SUCCESS;
		}

lockserver/server.cc

+130 −60

Original line number	Diff line number	Diff line
		@@ -30,66 +30,9 @@

		#include "lockserver/server.h"

		int
		main(int argc, char **argv)
		{
		int index = -1;
		const char *configPath = NULL;

		// Parse arguments
		int opt;
		char *strtolPtr;
		while ((opt = getopt(argc, argv, "c:i:")) != -1) {
		switch (opt) {
		case 'c':
		configPath = optarg;
		break;

		case 'i':
		index = strtol(optarg, &strtolPtr, 10);
		if ((optarg == '\0') \|\| (strtolPtr != '\0') \|\| (index < 0)) {
		fprintf(stderr, "option -i requires a numeric arg\n");
		}
		break;

		default:
		fprintf(stderr, "Unknown argument %s\n", argv[optind]);
		}
		}

		if (!configPath) {
		fprintf(stderr, "option -c is required\n");
		return EXIT_FAILURE;
		}

		if (index == -1) {
		fprintf(stderr, "option -i is required\n");
		return EXIT_FAILURE;
		}

		// Load configuration
		std::ifstream configStream(configPath);
		if (configStream.fail()) {
		fprintf(stderr, "unable to read configuration file: %s\n", configPath);
		return EXIT_FAILURE;
		}
		transport::Configuration config(configStream);

		if (index >= config.n) {
		fprintf(stderr, "replica index %d is out of bounds; "
		"only %d replicas defined\n", index, config.n);
		return EXIT_FAILURE;
		}

		UDPTransport transport(0.0, 0.0, 0);

		lockserver::LockServer server;
		replication::ir::IRReplica replica(config, index, &transport, &server);

		transport.Run();

		return EXIT_SUCCESS;
		}
		#include <algorithm>
		#include <iterator>
		#include <unordered_set>

		namespace lockserver {

		@@ -173,4 +116,131 @@ LockServer::UnloggedUpcall(const string &str1, string &str2)
		Debug("Unlogged: %s\n", str1.c_str());
		}

		void
		LockServer::Sync(const std::map<opid_t, RecordEntry>& record) {
		locks.clear();

		struct KeyLockInfo {
		std::unordered_set<uint64_t> locked;
		std::unordered_set<uint64_t> unlocked;
		};
		std::unordered_map<std::string, KeyLockInfo> key_lock_info;

		for (const std::pair<const opid_t, RecordEntry> &p : record) {
		const opid_t &opid = p.first;
		const RecordEntry &entry = p.second;

		Request request;
		request.ParseFromString(entry.request.op());
		Reply reply;
		reply.ParseFromString(entry.result);
		KeyLockInfo &info = key_lock_info[request.key()];

		Debug("Sync opid=(%lu, %lu), clientid=%lu, key=%s, type=%d, status=%d.",
		opid.first, opid.second, request.clientid(),
		request.key().c_str(), request.type(), reply.status());

		if (request.type() && reply.status() == 0) {
		// Lock.
		info.locked.insert(request.clientid());
		} else if (!request.type() && reply.status() == 0) {
		// Unlock.
		info.unlocked.insert(request.clientid());
		}
		}

		for (const std::pair<const std::string, KeyLockInfo> &p : key_lock_info) {
		const std::string &key = p.first;
		const KeyLockInfo &info = p.second;
		std::unordered_set<uint64_t> diff;
		std::set_difference(std::begin(info.locked), std::end(info.locked),
		std::begin(info.unlocked), std::end(info.unlocked),
		std::inserter(diff, diff.begin()));

		ASSERT(diff.size() == 0 \|\| diff.size() == 1);
		if (diff.size() == 1) {
		uint64_t client_id = *std::begin(diff);
		Debug("Assigning lock %lu: %s", client_id, key.c_str());
		locks[key] = client_id;
		}
		}
		}

		std::map<opid_t, std::string>
		LockServer::Merge(const std::map<opid_t, std::vector<RecordEntry>> &d,
		const std::map<opid_t, std::vector<RecordEntry>> &u,
		const std::map<opid_t, std::string> &majority_results_in_d) {
		// First note that d and u only contain consensus operations, and lock
		// requests are the only consensus operations (unlock is an inconsistent
		// operation), so d and u only contain lock requests. To merge, we grant
		// any majority successful lock request in d if it does not conflict with a
		// currently held lock. We do not grant any other lock request.

		std::map<opid_t, std::string> results;

		using EntryVec = std::vector<RecordEntry>;
		for (const std::pair<const opid_t, EntryVec>& p: d) {
		const opid_t &opid = p.first;
		const EntryVec &entries = p.second;

		// Get the request and reply.
		const RecordEntry &entry = *std::begin(entries);

		Request request;
		request.ParseFromString(entry.request.op());

		Reply reply;
		auto iter = majority_results_in_d.find(opid);
		ASSERT(iter != std::end(majority_results_in_d));
		reply.ParseFromString(iter->second);

		// Form the final result.
		const bool operation_successful = reply.status() == 0;
		if (operation_successful) {
		// If the lock was successful, then we acquire the lock so long as
		// it is not already held.
		const std::string &key = reply.key();
		if (locks.count(key) == 0) {
		Debug("Assigning lock %lu: %s", request.clientid(),
		key.c_str());
		locks[key] = request.clientid();
		results[opid] = iter->second;
		} else {
		Debug("Rejecting lock %lu: %s", request.clientid(),
		key.c_str());
		reply.set_status(-1);
		std::string s;
		reply.SerializeToString(&s);
		results[opid] = s;
		}
		} else {
		// If the lock was not successful, then we maintain this as the
		// majority result.
		results[opid] = iter->second;
		}
		}

		// We reject all lock requests in u. TODO: We could acquire a lock if
		// it is free, but it's simplest to just reject them unilaterally.
		for (const std::pair<const opid_t, EntryVec>& p: u) {
		const opid_t &opid = p.first;
		const EntryVec &entries = p.second;

		const RecordEntry &entry = *std::begin(entries);
		Request request;
		request.ParseFromString(entry.request.op());

		Debug("Rejecting lock %lu: %s", request.clientid(),
		request.key().c_str());
		Reply reply;
		reply.set_key(request.key());
		reply.set_status(-1);
		std::string s;
		reply.SerializeToString(&s);
		results[opid] = s;
		}

		return results;
		}

		} // namespace lockserver

lockserver/server.h

+19 −7

Original line number	Diff line number	Diff line
		@@ -31,15 +31,18 @@
		#ifndef _IR_LOCK_SERVER_H_
		#define _IR_LOCK_SERVER_H_

		#include "lib/udptransport.h"
		#include "replication/ir/replica.h"
		#include "lockserver/locks-proto.pb.h"

		#include <string>
		#include <unordered_map>

		#include "lib/transport.h"
		#include "replication/ir/replica.h"
		#include "lockserver/locks-proto.pb.h"

		namespace lockserver {

		using opid_t = replication::ir::opid_t;
		using RecordEntry = replication::ir::RecordEntry;

		class LockServer : public replication::ir::IRAppReplica
		{
		public:
		@@ -47,13 +50,22 @@ public:
		~LockServer();

		// Invoke inconsistent operation, no return value
		void ExecInconsistentUpcall(const string &str1);
		void ExecInconsistentUpcall(const string &str1) override;

		// Invoke consensus operation
		void ExecConsensusUpcall(const string &str1, string &str2);
		void ExecConsensusUpcall(const string &str1, string &str2) override;

		// Invoke unreplicated operation
		void UnloggedUpcall(const string &str1, string &str2);
		void UnloggedUpcall(const string &str1, string &str2) override;

		// Sync
		void Sync(const std::map<opid_t, RecordEntry>& record) override;

		// Merge
		std::map<opid_t, std::string> Merge(
		const std::map<opid_t, std::vector<RecordEntry>> &d,
		const std::map<opid_t, std::vector<RecordEntry>> &u,
		const std::map<opid_t, std::string> &majority_results_in_d) override;

		private:
		std::unordered_map<std::string, uint64_t> locks;

lockserver/tests/Rules.mk

0 → 100644

+16 −0

Original line number	Diff line number	Diff line
		d := $(dir $(lastword $(MAKEFILE_LIST)))

		GTEST_SRCS += $(addprefix $(d), lockserver-test.cc)

		$(d)lockserver-test: $(o)lockserver-test.o \
		$(o)../locks-proto.o \
		$(o)../server.o \
		$(o)../client.o \
		$(OBJS-ir-replica) \
		$(OBJS-ir-client) \
		$(LIB-configuration) \
		$(LIB-repltransport) \
		$(LIB-store-common) \
		$(GTEST_MAIN)

		TEST_BINS += $(d)lockserver-test

lockserver/tests/lockserver-test.cc

0 → 100644

+343 −0

Original line number	Diff line number	Diff line
		// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --
		/***********************************************************************
		*
		* lockserver_test.cc:
		* test cases for lock server
		*
		* 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 <fstream>
		#include <memory>
		#include <thread>

		#include <gtest/gtest.h>

		#include "lib/configuration.h"
		#include "lib/repltransport.h"
		#include "lockserver/client.h"
		#include "lockserver/server.h"
		#include "replication/ir/replica.h"

		class LockServerTest : public testing::Test {
		protected:
		std::vector<transport::ReplicaAddress> replica_addrs_;
		std::unique_ptr<transport::Configuration> config_;
		ReplTransport transport_;
		std::vector<std::unique_ptr<lockserver::LockClient>> clients_;
		std::vector<std::unique_ptr<lockserver::LockServer>> servers_;
		std::vector<std::unique_ptr<replication::ir::IRReplica>> replicas_;

		LockServerTest() {
		replica_addrs_ = {{"replica", "0"},
		{"replica", "1"},
		{"replica", "2"},
		{"replica", "3"},
		{"replica", "4"}};
		config_ = std::unique_ptr<transport::Configuration>(
		new transport::Configuration(5, 2, replica_addrs_));
		RemovePersistedFiles();

		for (std::size_t i = 0; i < 3; ++i) {
		auto client = std::unique_ptr<lockserver::LockClient>(
		new lockserver::LockClient(&transport_, *config_));
		client->lock_async(std::to_string(i));
		clients_.push_back(std::move(client));
		}

		for (std::size_t i = 0; i < replica_addrs_.size(); ++i) {
		auto server = std::unique_ptr<lockserver::LockServer>(
		new lockserver::LockServer());
		servers_.push_back(std::move(server));
		auto replica = std::unique_ptr<replication::ir::IRReplica>(
		new replication::ir::IRReplica(*config_, i, &transport_,
		servers_[i].get()));
		replicas_.push_back(std::move(replica));
		}
		}

		virtual void TearDown() {
		RemovePersistedFiles();
		}

		virtual void RemovePersistedFiles() {
		for (std::size_t i = 0; i < replica_addrs_.size(); ++i) {
		const transport::ReplicaAddress &addr = replica_addrs_[i];
		const std::string filename =
		addr.host + ":" + addr.port + "_" + std::to_string(i) + ".bin";
		std::ifstream f(filename);
		if (f.good()) {
		int success = std::remove(filename.c_str());
		ASSERT(success == 0);
		}
		}
		}
		};

		// Note that these tests are all white box smoke tests. They depend on the
		// low-level details of knowing exactly which timeouts are registered and which
		// messages are sent. If an implementation detail is changed to make some of
		// these tests fail, you should cal transport_.Run() and walk through the
		// execution to trigger the desired behavior. Also, they only check to make
		// sure that nothing crashes, though you can read through the Debug prints to
		// make sure everything looks right.
		//
		// TODO: Use a ReplTransport for tests like the ones in ir-test.cc to assert
		// that the correct messages are being sent.

		TEST_F(LockServerTest, SuccessfulFastPathLock) {
		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Deliver lock reply to client.
		for (std::size_t i = 0; i < replica_addrs_.size(); ++i) {
		transport_.DeliverMessage({"client", "0"}, i);
		}

		// Deliver finalize to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 1);
		}

		// Deliver confirm to client.
		int j = replica_addrs_.size();
		for (std::size_t i = j; i < j + replica_addrs_.size(); ++i) {
		transport_.DeliverMessage({"client", "0"}, i);
		}
		}

		TEST_F(LockServerTest, SuccessfulSlowPathLock) {
		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Transition to slow path.
		transport_.TriggerTimer(clients_.size() + replica_addrs_.size() + 1);

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Deliver lock reply to client.
		for (std::size_t i = 0; i < replica_addrs_.size(); ++i) {
		transport_.DeliverMessage({"client", "0"}, i);
		}

		// Deliver finalize to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 1);
		}

		// Deliver confirm to client.
		int j = replica_addrs_.size();
		for (std::size_t i = j; i < j + replica_addrs_.size(); ++i) {
		transport_.DeliverMessage({"client", "0"}, i);
		}
		}

		TEST_F(LockServerTest, SuccessfulViewChange) {
		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Initiate view changes on all replicas.
		const std::size_t nclients = clients_.size();
		const std::size_t nreplicas = replica_addrs_.size();
		for (std::size_t i = nclients + 1; i < nclients + nreplicas + 1; ++i) {
		transport_.TriggerTimer(i);
		}

		// Deliver DoViewChangeMessages to new primary.
		const transport::ReplicaAddress& primary = replica_addrs_[1];
		for (std::size_t i = 1; i < 1 + nreplicas - 1; ++i) {
		transport_.DeliverMessage({primary.host, primary.port}, i);
		}

		// Deliver StartViewMessage to all replicas.
		for (std::size_t i = 0; i < nreplicas; ++i) {
		if (i == 1) {
		continue;
		}
		const transport::ReplicaAddress& addr = replica_addrs_[i];
		transport_.DeliverMessage({addr.host, addr.port}, nreplicas);
		}
		}

		TEST_F(LockServerTest, SuccessfulViewChangeNonemptyRdu) {
		const std::size_t nclients = clients_.size();
		const std::size_t nreplicas = replica_addrs_.size();
		ASSERT_GE(nclients, 3);
		ASSERT_GE(nreplicas, 3);

		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Deliver lock reply to client.
		for (std::size_t i = 0; i < replica_addrs_.size(); ++i) {
		transport_.DeliverMessage({"client", "0"}, i);
		}

		// Deliver finalize to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 1);
		}

		// Send client 1's lock request.
		transport_.TriggerTimer(2);

		// Deliver lock request to first three replicas.
		for (std::size_t i = 0; i < 3; ++i) {
		const transport::ReplicaAddress &addr = replica_addrs_[i];
		transport_.DeliverMessage({addr.host, addr.port}, 2);
		}

		// Send client 2's lock request.
		transport_.TriggerTimer(3);

		// Deliver lock request to first replica.
		const transport::ReplicaAddress &addr = replica_addrs_[0];
		transport_.DeliverMessage({addr.host, addr.port}, 3);

		// View change first three replicas.
		for (std::size_t i = nclients + 1; i < nclients + 1 + 3; ++i) {
		transport_.TriggerTimer(i);
		}

		// Deliver DoViewChangeMessages to new primary.
		const transport::ReplicaAddress& primary = replica_addrs_[1];
		for (std::size_t i = 4; i < 4 + 2; ++i) {
		transport_.DeliverMessage({primary.host, primary.port}, i);
		}

		// Deliver StartViewMessage to replica 0 and 2.
		const transport::ReplicaAddress& addr0 = replica_addrs_[0];
		const transport::ReplicaAddress& addr2 = replica_addrs_[2];
		transport_.DeliverMessage({addr0.host, addr0.port}, 6);
		transport_.DeliverMessage({addr2.host, addr2.port}, 6);
		}

		TEST_F(LockServerTest, FinalizeConsensusReply) {
		const std::size_t nclients = clients_.size();
		const std::size_t nreplicas = replica_addrs_.size();

		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Trigger view change.
		for (std::size_t i = nclients + 1; i < nclients + 1 + nreplicas; ++i) {
		transport_.TriggerTimer(i);
		}

		// Deliver DoViewChangeMessages to new primary.
		const transport::ReplicaAddress& primary = replica_addrs_[1];
		for (std::size_t i = 1; i < 1 + nreplicas - 1; ++i) {
		transport_.DeliverMessage({primary.host, primary.port}, i);
		}

		// Deliver StartViewMessage to all replicas.
		for (std::size_t i = 0; i < nreplicas; ++i) {
		if (i == 1) {
		continue;
		}
		const transport::ReplicaAddress& addr = replica_addrs_[i];
		transport_.DeliverMessage({addr.host, addr.port}, nreplicas);
		}

		// Deliver lock request to replicas.
		for (const auto &addr : replica_addrs_) {
		transport_.DeliverMessage({addr.host, addr.port}, 0);
		}

		// Deliver finalized reply to client.
		transport_.DeliverMessage({"client", "0"}, nreplicas);
		}

		TEST_F(LockServerTest, MismatchedConsensus) {
		const std::size_t nclients = clients_.size();
		const std::size_t nreplicas = replica_addrs_.size();

		// Send client 0's lock request.
		transport_.TriggerTimer(1);

		// Transition to slow path.
		transport_.TriggerTimer(nclients + nreplicas + 1);

Source

Target

Files

Some changes are not shown.