Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • syslab/tapir
  • aaasz/tapir
  • ashmrtnz/tapir
3 results
Show changes
Commits on Source (73)
.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
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.
#
# 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,11 +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/tapir/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
#
......@@ -235,6 +237,12 @@ clean:
$(call trace,RM,binaries,rm -f $(BINS) $(TEST_BINS))
$(call trace,RM,objects,rm -rf .obj)
#
# Debugging
#
print-%:
@echo '$*=$($*)'
##################################################################
# Targets
#
......
# 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
......@@ -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
......@@ -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
......
......@@ -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)
......@@ -61,7 +66,7 @@ Configuration::Configuration(const Configuration &c)
multicastAddress = new ReplicaAddress(*c.multicastAddress);
}
}
Configuration::Configuration(int n, int f,
std::vector<ReplicaAddress> replicas,
ReplicaAddress *multicastAddress)
......@@ -82,7 +87,7 @@ Configuration::Configuration(std::ifstream &file)
f = -1;
hasMulticast = false;
multicastAddress = NULL;
while (!file.eof()) {
// Read a line
string line;
......@@ -94,52 +99,53 @@ Configuration::Configuration(std::ifstream &file)
}
// Get the command
// This is pretty horrible, but C++ does promise that &line[0]
// is going to be a mutable contiguous buffer...
char *cmd = strtok(&line[0], " \t");
unsigned int t1 = line.find_first_of(" \t");
string cmd = line.substr(0, t1);
if (strcasecmp(cmd, "f") == 0) {
char *arg = strtok(NULL, " \t");
if (!arg) {
if (strcasecmp(cmd.c_str(), "f") == 0) {
unsigned int t2 = line.find_first_not_of(" \t", t1);
if (t2 == string::npos) {
Panic ("'f' configuration line requires an argument");
}
char *strtolPtr;
f = strtoul(arg, &strtolPtr, 0);
if ((*arg == '\0') || (*strtolPtr != '\0')) {
try {
f = stoul(line.substr(t2, string::npos));
} catch (std::invalid_argument& ia) {
Panic("Invalid argument to 'f' configuration line");
}
} else if (strcasecmp(cmd, "replica") == 0) {
char *arg = strtok(NULL, " \t");
if (!arg) {
} else if (strcasecmp(cmd.c_str(), "replica") == 0) {
unsigned int t2 = line.find_first_not_of(" \t", t1);
if (t2 == string::npos) {
Panic ("'replica' configuration line requires an argument");
}
char *host = strtok(arg, ":");
char *port = strtok(NULL, "");
if (!host || !port) {
unsigned int t3 = line.find_first_of(":", t2);
if (t3 == string::npos) {
Panic("Configuration line format: 'replica host:port'");
}
replicas.push_back(ReplicaAddress(string(host), string(port)));
} else if (strcasecmp(cmd, "multicast") == 0) {
char *arg = strtok(NULL, " \t");
if (!arg) {
string host = line.substr(t2, t3-t2);
string port = line.substr(t3+1, string::npos);
replicas.push_back(ReplicaAddress(host, port));
} else if (strcasecmp(cmd.c_str(), "multicast") == 0) {
unsigned int t2 = line.find_first_not_of(" \t", t1);
if (t2 == string::npos) {
Panic ("'multicast' configuration line requires an argument");
}
char *host = strtok(arg, ":");
char *port = strtok(NULL, "");
if (!host || !port) {
Panic("Configuration line format: 'multicast host:port'");
unsigned int t3 = line.find_first_of(":", t2);
if (t3 == string::npos) {
Panic("Configuration line format: 'replica host:port'");
}
multicastAddress = new ReplicaAddress(string(host),
string(port));
string host = line.substr(t2, t3-t2);
string port = line.substr(t3+1, string::npos);
multicastAddress = new ReplicaAddress(host, port);
hasMulticast = true;
} else {
Panic("Unknown configuration directive: %s", cmd);
Panic("Unknown configuration directive: %s", cmd.c_str());
}
}
......@@ -209,8 +215,27 @@ Configuration::operator==(const Configuration &other) const
return false;
}
}
return true;
}
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
......@@ -52,6 +52,16 @@ struct ReplicaAddress
inline bool operator!=(const ReplicaAddress &other) const {
return !(*this == other);
}
bool operator<(const ReplicaAddress &other) const;
bool operator<=(const ReplicaAddress &other) const {
return *this < other || *this == other;
}
bool operator>(const ReplicaAddress &other) const {
return !(*this <= other);
}
bool operator>=(const ReplicaAddress &other) const {
return !(*this < other);
}
};
......@@ -69,10 +79,20 @@ public:
int QuorumSize() const;
int FastQuorumSize() const;
bool operator==(const Configuration &other) const;
inline bool operator!= (const Configuration &other) const {
inline bool operator!=(const Configuration &other) const {
return !(*this == other);
}
bool operator<(const Configuration &other) const;
bool operator<=(const Configuration &other) const {
return *this < other || *this == other;
}
bool operator>(const Configuration &other) const {
return !(*this <= other);
}
bool operator>=(const Configuration &other) const {
return !(*this < other);
}
public:
int n; // number of replicas
int f; // number of failures tolerated
......@@ -82,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_ */
......@@ -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
*
......
syntax = "proto2";
package transport.latency.format;
message LatencyDist
......
......@@ -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
*
......
......@@ -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
*
......
......@@ -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
*
......
......@@ -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)
......
// -*- 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.");
}
}
// -*- 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_
// -*- 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;
}
// -*- 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_
......@@ -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
......@@ -38,7 +38,7 @@
SimulatedTransportAddress::SimulatedTransportAddress(int addr)
: addr(addr)
{
}
int
......@@ -70,7 +70,7 @@ SimulatedTransport::SimulatedTransport()
SimulatedTransport::~SimulatedTransport()
{
}
void
......@@ -99,12 +99,12 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
bool multicast)
{
ASSERT(!multicast);
int dst = dstAddr.addr;
Message *msg = m.New();
msg->CheckTypeAndMergeFrom(m);
int srcAddr =
dynamic_cast<const SimulatedTransportAddress &>(src->GetAddress()).addr;
......@@ -119,11 +119,11 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
return true;
}
}
string msgData;
msg->SerializeToString(&msgData);
delete msg;
QueuedMessage q(dst, srcAddr, m.GetTypeName(), msgData);
if (delay == 0) {
......@@ -133,7 +133,7 @@ SimulatedTransport::SendMessageInternal(TransportReceiver *src,
queue.push_back(q);
});
}
return true;
return true;
}
SimulatedTransportAddress
......@@ -144,7 +144,7 @@ SimulatedTransport::LookupAddress(const transport::Configuration &cfg,
// idx match. This is the least efficient possible way to do this,
// but that's why this is the simulated transport not the real
// one... (And we only have to do this once at runtime.)
for (auto & kv : configurations) {
if (*(kv.second) == cfg) {
......@@ -157,7 +157,7 @@ SimulatedTransport::LookupAddress(const transport::Configuration &cfg,
}
}
}
Panic("No replica %d was registered", idx);
}
......@@ -171,7 +171,7 @@ void
SimulatedTransport::Run()
{
LookupAddresses();
do {
// Process queue
while (!queue.empty()) {
......@@ -190,7 +190,7 @@ SimulatedTransport::Run()
timers.erase(iter);
cb();
}
// ...then retry to see if there are more queued messages to
// deliver first
} while (!queue.empty() || (processTimers && !timers.empty()));
......@@ -234,7 +234,7 @@ SimulatedTransport::CancelTimer(int id)
iter++;
}
}
return found;
}
......