a090e8dd95e5d66d7d1e2c71ca63fdae1c9cf381 to master · syslab / tapir

Some changes are not shown.

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

.gitignore

0 → 100644

+75 −0

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

LICENSE

0 → 100644

+21 −0

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

+9 −2

Original line number	Original line	Diff line number	Diff line
	@@ -7,8 +7,8 @@ CXX = g++
	LD = g++		LD = g++
	EXPAND = lib/tmpl/expand		EXPAND = lib/tmpl/expand

	#CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized -O2 -DNASSERT		CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized -O2 -DNASSERT
	CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized		#CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized
	CXXFLAGS := -g -std=c++0x		CXXFLAGS := -g -std=c++0x
	LDFLAGS := -levent_pthreads		LDFLAGS := -levent_pthreads
	## Debian package: check		## Debian package: check
	@@ -131,6 +131,7 @@ include store/strongstore/Rules.mk
	include store/weakstore/Rules.mk		include store/weakstore/Rules.mk
	include store/benchmark/Rules.mk		include store/benchmark/Rules.mk
	include lockserver/Rules.mk		include lockserver/Rules.mk
			include timeserver/Rules.mk
	include libtapir/Rules.mk		include libtapir/Rules.mk
	##################################################################		##################################################################
	# General rules		# General rules
	@@ -236,6 +237,12 @@ clean:
	$(call trace,RM,binaries,rm -f $(BINS) $(TEST_BINS))		$(call trace,RM,binaries,rm -f $(BINS) $(TEST_BINS))
	$(call trace,RM,objects,rm -rf .obj)		$(call trace,RM,objects,rm -rf .obj)

			#
			# Debugging
			#
			print-%:
			@echo '$=$($)'

	##################################################################		##################################################################
	# Targets		# Targets
	#		#

README.md

+7 −1

Original line number	Original line	Diff line number	Diff line
	@@ -52,7 +52,13 @@ The repo is structured as follows:
	## Compiling & Running		## Compiling & Running
	You can compile all of the TAPIR executables by running make in the root directory		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.		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		## Contact and Questions
	Please email Irene at iyzhang@cs.washington.edu, Dan at drkp@cs.washington.edu and Naveen at naveenks@cs.washington.edu		Please email Irene at iyzhang@cs.washington.edu, Dan at drkp@cs.washington.edu and Naveen at naveenks@cs.washington.edu

lib/Rules.mk

+7 −1

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

	PROTOS += $(addprefix $(d), \		PROTOS += $(addprefix $(d), \
	latency-format.proto)		latency-format.proto)
	@@ -24,8 +25,13 @@ LIB-transport := $(o)transport.o $(LIB-message) $(LIB-configuration)

	LIB-simtransport := $(o)simtransport.o $(LIB-transport)		LIB-simtransport := $(o)simtransport.o $(LIB-transport)

			LIB-repltransport := $(o)repltransport.o $(LIB-transport)

	LIB-udptransport := $(o)udptransport.o $(LIB-transport)		LIB-udptransport := $(o)udptransport.o $(LIB-transport)

	LIB-tcptransport := $(o)tcptransport.o $(LIB-transport)		LIB-tcptransport := $(o)tcptransport.o $(LIB-transport)

			LIB-persistent_register := $(o)persistent_register.o $(LIB-message)

	include $(d)tests/Rules.mk		include $(d)tests/Rules.mk

lib/configuration.cc

+60 −35

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

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

	namespace transport {		namespace transport {

	@@ -52,6 +51,12 @@ ReplicaAddress::operator==(const ReplicaAddress &other) const {
	(port == other.port));		(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)		Configuration::Configuration(const Configuration &c)
	: n(c.n), f(c.f), replicas(c.replicas), hasMulticast(c.hasMulticast)		: n(c.n), f(c.f), replicas(c.replicas), hasMulticast(c.hasMulticast)
	@@ -94,52 +99,53 @@ Configuration::Configuration(std::ifstream &file)
	}		}

	// Get the command		// Get the command
	// This is pretty horrible, but C++ does promise that &line[0]		unsigned int t1 = line.find_first_of(" \t");
	// is going to be a mutable contiguous buffer...		string cmd = line.substr(0, t1);
	char *cmd = strtok(&line[0], " \t");

	if (strcasecmp(cmd, "f") == 0) {		if (strcasecmp(cmd.c_str(), "f") == 0) {
	char *arg = strtok(NULL, " \t");		unsigned int t2 = line.find_first_not_of(" \t", t1);
	if (!arg) {		if (t2 == string::npos) {
	Panic ("'f' configuration line requires an argument");		Panic ("'f' configuration line requires an argument");
	}		}
	char *strtolPtr;
	f = strtoul(arg, &strtolPtr, 0);		try {
	if ((arg == '\0') \|\| (strtolPtr != '\0')) {		f = stoul(line.substr(t2, string::npos));
			} catch (std::invalid_argument& ia) {
	Panic("Invalid argument to 'f' configuration line");		Panic("Invalid argument to 'f' configuration line");
	}		}
	} else if (strcasecmp(cmd, "replica") == 0) {		} else if (strcasecmp(cmd.c_str(), "replica") == 0) {
	char *arg = strtok(NULL, " \t");		unsigned int t2 = line.find_first_not_of(" \t", t1);
	if (!arg) {		if (t2 == string::npos) {
	Panic ("'replica' configuration line requires an argument");		Panic ("'replica' configuration line requires an argument");
	}		}

	char *host = strtok(arg, ":");		unsigned int t3 = line.find_first_of(":", t2);
	char *port = strtok(NULL, "");		if (t3 == string::npos) {

	if (!host \|\| !port) {
	Panic("Configuration line format: 'replica host:port'");		Panic("Configuration line format: 'replica host:port'");
	}		}

	replicas.push_back(ReplicaAddress(string(host), string(port)));		string host = line.substr(t2, t3-t2);
	} else if (strcasecmp(cmd, "multicast") == 0) {		string port = line.substr(t3+1, string::npos);
	char *arg = strtok(NULL, " \t");
	if (!arg) {		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");		Panic ("'multicast' configuration line requires an argument");
	}		}

	char *host = strtok(arg, ":");		unsigned int t3 = line.find_first_of(":", t2);
	char *port = strtok(NULL, "");		if (t3 == string::npos) {
			Panic("Configuration line format: 'replica host:port'");
	if (!host \|\| !port) {
	Panic("Configuration line format: 'multicast host:port'");
	}		}

	multicastAddress = new ReplicaAddress(string(host),		string host = line.substr(t2, t3-t2);
	string(port));		string port = line.substr(t3+1, string::npos);

			multicastAddress = new ReplicaAddress(host, port);
	hasMulticast = true;		hasMulticast = true;
	} else {		} 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;		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		} // namespace transport

lib/configuration.h

+24 −4

Original line number	Original line	Diff line number	Diff line
	@@ -52,6 +52,16 @@ struct ReplicaAddress
	inline bool operator!=(const ReplicaAddress &other) const {		inline bool operator!=(const ReplicaAddress &other) const {
	return !(*this == other);		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 {		inline bool operator!=(const Configuration &other) const {
	return !(*this == other);		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:		public:
	int n; // number of replicas		int n; // number of replicas
	@@ -82,7 +102,8 @@ private:
	bool hasMulticast;		bool hasMulticast;
	};		};

	} // namespace replication		} // namespace transport


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


	#endif /* _LIB_CONFIGURATION_H_ */		#endif /* _LIB_CONFIGURATION_H_ */

lib/latency-format.proto

+2 −0

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

	package transport.latency.format;		package transport.latency.format;

	message LatencyDist		message LatencyDist

lib/message.cc

+10 −0

Original line number	Original line	Diff line number	Diff line
	@@ -40,6 +40,8 @@
	#include <unistd.h>		#include <unistd.h>
	#include <sys/time.h>		#include <sys/time.h>

			#include <mutex>

	#define BACKTRACE_ON_PANIC 1		#define BACKTRACE_ON_PANIC 1
	#if BACKTRACE_ON_PANIC		#if BACKTRACE_ON_PANIC
	#include <execinfo.h>		#include <execinfo.h>
	@@ -48,6 +50,8 @@
	#define TIMESTAMP_BASE62 0		#define TIMESTAMP_BASE62 0
	#define TIMESTAMP_NUMERIC 1		#define TIMESTAMP_NUMERIC 1

			std::mutex message_mtx;

	void __attribute__((weak))		void __attribute__((weak))
	Message_VA(enum Message_Type type,		Message_VA(enum Message_Type type,
	const char fname, int line, const char func,		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 fname, int line, const char func,
	const char *fmt, va_list args)		const char *fmt, va_list args)
	{		{
			// Lock mutex to make sure the output is not mangled.
			message_mtx.lock();

	static int haveColor = -1;		static int haveColor = -1;
	struct msg_desc {		struct msg_desc {
	const char *prefix;		const char *prefix;
	@@ -150,6 +157,9 @@ _Message_VA(enum Message_Type type, FILE *fp,
	fputs("\033[0m", fp);		fputs("\033[0m", fp);
	fprintf(fp, "\n");		fprintf(fp, "\n");
	fflush(fp);		fflush(fp);

			// Unlock mutex.
			message_mtx.unlock();
	}		}

	void _Panic(void)		void _Panic(void)

lib/persistent_register.cc

0 → 100644

+126 −0

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

+13 −13

File changed.

Contains only whitespace changes.

lib/tcptransport.cc

+99 −70

Original line number	Original line	Diff line number	Diff line
	// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --		// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --
	/***********************************************************************		/***********************************************************************
	*		*
	* udptransport.cc:		* tcptransport.cc:
	* message-passing network interface that uses TCP message delivery		* message-passing network interface that uses TCP message delivery
	* and libasync		* and libasync
	*		*
	@@ -35,11 +35,9 @@
	#include "lib/tcptransport.h"		#include "lib/tcptransport.h"

	#include <google/protobuf/message.h>		#include <google/protobuf/message.h>
	#include <event2/event.h>
	#include <event2/buffer.h>
	#include <event2/thread.h>		#include <event2/thread.h>
	#include <event2/bufferevent.h>

			#include <netinet/tcp.h>
	#include <arpa/inet.h>		#include <arpa/inet.h>
	#include <netinet/in.h>		#include <netinet/in.h>
	#include <fcntl.h>		#include <fcntl.h>
	@@ -87,6 +85,7 @@ TCPTransport::LookupAddress(const transport::ReplicaAddress &addr)
	{		{
	int res;		int res;
	struct addrinfo hints;		struct addrinfo hints;
			memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;		hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_STREAM;		hints.ai_socktype = SOCK_STREAM;
	hints.ai_protocol = 0;		hints.ai_protocol = 0;
	@@ -151,27 +150,20 @@ BindToPort(int fd, const string &host, const string &port)
	}		}

	TCPTransport::TCPTransport(double dropRate, double reorderRate,		TCPTransport::TCPTransport(double dropRate, double reorderRate,
	int dscp, event_base *evbase)		int dscp, bool handleSignals)
	{		{
	lastTimerId = 0;		lastTimerId = 0;

	// Set up libevent		// Set up libevent
			evthread_use_pthreads();
	event_set_log_callback(LogCallback);		event_set_log_callback(LogCallback);
	event_set_fatal_callback(FatalCallback);		event_set_fatal_callback(FatalCallback);
	libeventBase = event_base_new();

	// 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();		libeventBase = event_base_new();
	evthread_make_base_notifiable(libeventBase);		evthread_make_base_notifiable(libeventBase);
	}

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

	TCPTransport::~TCPTransport()		TCPTransport::~TCPTransport()
	{		{
	@@ -204,6 +197,13 @@ TCPTransport::ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst)
	PWarning("Failed to set O_NONBLOCK on outgoing TCP socket");		PWarning("Failed to set O_NONBLOCK on outgoing TCP socket");
	}		}

			// Set TCP_NODELAY
			int n = 1;
			if (setsockopt(fd, IPPROTO_TCP,
			TCP_NODELAY, (char *)&n, sizeof(n)) < 0) {
			PWarning("Failed to set TCP_NODELAY on TCP listening socket");
			}

	TCPTransportTCPListener *info = new TCPTransportTCPListener();		TCPTransportTCPListener *info = new TCPTransportTCPListener();
	info->transport = this;		info->transport = this;
	info->acceptFd = 0;		info->acceptFd = 0;
	@@ -230,6 +230,15 @@ TCPTransport::ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst)
	Panic("Failed to enable bufferevent");		Panic("Failed to enable bufferevent");
	}		}

			// Tell the receiver its address
			struct sockaddr_in sin;
			socklen_t sinsize = sizeof(sin);
			if (getsockname(fd, (sockaddr *) &sin, &sinsize) < 0) {
			PPanic("Failed to get socket name");
			}
			TCPTransportAddress *addr = new TCPTransportAddress(sin);
			src->SetAddress(addr);

	tcpOutgoing[dst] = bev;		tcpOutgoing[dst] = bev;
	tcpAddresses.insert(pair<struct bufferevent*, TCPTransportAddress>(bev,dst));		tcpAddresses.insert(pair<struct bufferevent*, TCPTransportAddress>(bev,dst));

	@@ -271,6 +280,13 @@ TCPTransport::Register(TransportReceiver *receiver,
	PWarning("Failed to set SO_REUSEADDR on TCP listening socket");		PWarning("Failed to set SO_REUSEADDR on TCP listening socket");
	}		}

			// Set TCP_NODELAY
			n = 1;
			if (setsockopt(fd, IPPROTO_TCP,
			TCP_NODELAY, (char *)&n, sizeof(n)) < 0) {
			PWarning("Failed to set TCP_NODELAY on TCP listening socket");
			}

	// Registering a replica. Bind socket to the designated		// Registering a replica. Bind socket to the designated
	// host/port		// host/port
	const string &host = config.replica(replicaIdx).host;		const string &host = config.replica(replicaIdx).host;
	@@ -526,6 +542,13 @@ TCPTransport::TCPAcceptCallback(evutil_socket_t fd, short what, void *arg)
	PWarning("Failed to set O_NONBLOCK");		PWarning("Failed to set O_NONBLOCK");
	}		}

			// Set TCP_NODELAY
			int n = 1;
			if (setsockopt(newfd, IPPROTO_TCP,
			TCP_NODELAY, (char *)&n, sizeof(n)) < 0) {
			PWarning("Failed to set TCP_NODELAY on TCP listening socket");
			}

	// Create a buffered event		// Create a buffered event
	bev = bufferevent_socket_new(transport->libeventBase, newfd,		bev = bufferevent_socket_new(transport->libeventBase, newfd,
	BEV_OPT_CLOSE_ON_FREE);		BEV_OPT_CLOSE_ON_FREE);
	@@ -553,8 +576,13 @@ TCPTransport::TCPReadableCallback(struct bufferevent bev, void arg)

	Debug("Readable on bufferevent %p", bev);		Debug("Readable on bufferevent %p", bev);


			while (evbuffer_get_length(evbuf) > 0) {
	uint32_t *magic;		uint32_t *magic;
	magic = (uint32_t )evbuffer_pullup(evbuf, sizeof(magic));		magic = (uint32_t )evbuffer_pullup(evbuf, sizeof(magic));
			if (magic == NULL) {
			return;
			}
	ASSERT(*magic == MAGIC);		ASSERT(*magic == MAGIC);

	size_t *sz;		size_t *sz;
	@@ -604,6 +632,7 @@ TCPTransport::TCPReadableCallback(struct bufferevent bev, void arg)
	info->receiver->ReceiveMessage(addr->second, msgType, msg);		info->receiver->ReceiveMessage(addr->second, msgType, msg);
	Debug("Done processing large %s message", msgType.c_str());		Debug("Done processing large %s message", msgType.c_str());
	}		}
			}

	void		void
	TCPTransport::TCPIncomingEventCallback(struct bufferevent *bev,		TCPTransport::TCPIncomingEventCallback(struct bufferevent *bev,

lib/tcptransport.h

+3 −3

Original line number	Original line	Diff line number	Diff line
	/ -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --		// -- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 --
	/***********************************************************************		/***********************************************************************
	*		*
	* udptransport.h:		* tcptransport.h:
	* message-passing network interface that uses UDP message delivery		* message-passing network interface that uses UDP message delivery
	* and libasync		* and libasync
	*		*
	@@ -68,7 +68,7 @@ class TCPTransport : public TransportCommon<TCPTransportAddress>
	{		{
	public:		public:
	TCPTransport(double dropRate = 0.0, double reogrderRate = 0.0,		TCPTransport(double dropRate = 0.0, double reogrderRate = 0.0,
	int dscp = 0, event_base *evbase = nullptr);		int dscp = 0, bool handleSignals = true);
	virtual ~TCPTransport();		virtual ~TCPTransport();
	void Register(TransportReceiver *receiver,		void Register(TransportReceiver *receiver,
	const transport::Configuration &config,		const transport::Configuration &config,

lib/tests/simtransport-testmessage.proto

+2 −0

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

	package transport.test;		package transport.test;

	message TestMessage {		message TestMessage {

lib/udptransport.cc

+31 −37

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

			#include <memory>
	#include <random>		#include <random>

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

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

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

	// Set up libevent		// Set up libevent
			evthread_use_pthreads();
	event_set_log_callback(LogCallback);		event_set_log_callback(LogCallback);
	event_set_fatal_callback(FatalCallback);		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();		libeventBase = event_base_new();
	evthread_make_base_notifiable(libeventBase);		evthread_make_base_notifiable(libeventBase);
	}

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

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

	static size_t		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 data = m.SerializeAsString();
	string type = m.GetTypeName();		string type = m.GetTypeName();
	@@ -394,7 +391,8 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
	ssize_t totalLen = (typeLen + sizeof(typeLen) +		ssize_t totalLen = (typeLen + sizeof(typeLen) +
	dataLen + sizeof(dataLen));		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;		char *ptr = buf;
	((size_t ) ptr) = typeLen;		((size_t ) ptr) = typeLen;
	@@ -410,7 +408,7 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
	memcpy(ptr, data.c_str(), dataLen);		memcpy(ptr, data.c_str(), dataLen);
	ptr += dataLen;		ptr += dataLen;

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

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

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

	int fd = fds[src];		int fd = fds[src];

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

	delete [] buf;
	return true;		return true;

	fail:
	delete [] buf;
	return false;
	}		}

	void		void

lib/udptransport.h

+1 −1

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

lockserver/Rules.mk

+28 −5

Original line number	Original line	Diff line number	Diff line
	d := $(dir $(lastword $(MAKEFILE_LIST)))		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)		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) \		$(d)client-main: $(o)client-main.o \
	$(o)locks-proto.o $(o)client.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	Original line	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	Original line	Diff line number	Diff line
	@@ -30,88 +30,14 @@

	#include "lockserver/client.h"		#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 {		namespace lockserver {

	using namespace std;		using namespace std;
	using namespace proto;		using namespace proto;

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

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

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

	/* Run the transport in a new thread. */
	clientTransport = new thread(&LockClient::run_client, this);
	}		}

	LockClient::~LockClient() { }		LockClient::~LockClient() { }

	void		void
	LockClient::run_client()		LockClient::lock_async(const std::string &key) {
	{		ASSERT(waiting == nullptr);
	transport.Run();
	}

	bool
	LockClient::lock(const string &key)
	{
	Debug("Sending LOCK");		Debug("Sending LOCK");

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

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

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

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

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

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

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

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

			void
			LockClient::unlock_wait() {
	waiting->GetReply();		waiting->GetReply();
	delete waiting;		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		string
	LockClient::Decide(const set<string> &results)		LockClient::Decide(const map<string, std::size_t> &results)
	{		{
	// If a majority say lock, we say lock.		// If a majority say lock, we say lock.
	int success_count = 0;		int success_count = 0;
	string key;		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 reply;
	reply.ParseFromString(s);		reply.ParseFromString(s);
	key = reply.key();		key = reply.key();

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

	string final_reply_str;		string final_reply_str;
	@@ -243,4 +192,15 @@ LockClient::UnlockCallback(const std::string &request_str, const std::string &re
	w->Reply(0);		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		} // namespace lockserver

lockserver/client.h

+22 −6

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

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

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

	namespace lockserver {		namespace lockserver {

	class LockClient		class LockClient
	{		{
	public:		public:
	LockClient(const std::string &configPath);		LockClient(Transport* transport, const transport::Configuration &config);
	~LockClient();		~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);		bool lock(const std::string &key);
	void unlock(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:		private:
	/* Unique ID for this client. */		/* Unique ID for this client. */
	uint64_t client_id;		uint64_t client_id;

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

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

	/* Decide function for a lock server. */		/* 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. */		/* IR client proxy. */
	replication::ir::IRClient *client;		replication::ir::IRClient *client;

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

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

	} // namespace lockserver		} // namespace lockserver

lockserver/locks-proto.proto

+2 −0

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

	package lockserver.proto;		package lockserver.proto;

	message Request {		message Request {

lockserver/lockserver-repl.cc

0 → 100644

+80 −0

Original line number	Original line	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	Original line	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	Original line	Diff line number	Diff line
	@@ -30,66 +30,9 @@

	#include "lockserver/server.h"		#include "lockserver/server.h"

	int		#include <algorithm>
	main(int argc, char **argv)		#include <iterator>
	{		#include <unordered_set>
	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;
	}

	namespace lockserver {		namespace lockserver {

	@@ -173,4 +116,131 @@ LockServer::UnloggedUpcall(const string &str1, string &str2)
	Debug("Unlogged: %s\n", str1.c_str());		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		} // namespace lockserver

lockserver/server.h

+19 −7

Original line number	Original line	Diff line number	Diff line
	@@ -31,15 +31,18 @@
	#ifndef _IR_LOCK_SERVER_H_		#ifndef _IR_LOCK_SERVER_H_
	#define _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 <string>
	#include <unordered_map>		#include <unordered_map>

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

	namespace lockserver {		namespace lockserver {

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

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

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

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

	// Invoke unreplicated operation		// 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:		private:
	std::unordered_map<std::string, uint64_t> locks;		std::unordered_map<std::string, uint64_t> locks;

lockserver/tests/Rules.mk

0 → 100644

+16 −0

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

Compare revisions

Source

Target

Files

Some changes are not shown.

.gitignore

LICENSE

Makefile

README.md

lib/Rules.mk

lib/configuration.cc

lib/configuration.h

lib/latency-format.proto

lib/message.cc

lib/persistent_register.cc

lib/persistent_register.h

lib/repltransport.cc

lib/repltransport.h

lib/simtransport.cc

lib/tcptransport.cc

lib/tcptransport.h

lib/tests/simtransport-testmessage.proto

lib/udptransport.cc

lib/udptransport.h

lockserver/Rules.mk

lockserver/client-main.cc

lockserver/client.cc

lockserver/client.h

lockserver/locks-proto.proto

lockserver/lockserver-repl.cc

lockserver/server-main.cc

lockserver/server.cc

lockserver/server.h

lockserver/tests/Rules.mk

lockserver/tests/lockserver-test.cc