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 (40)
Hide whitespace changes
Inline Side-by-side
Showing
with 1159 additions and 172 deletions
.gitignore 0 → 100644
View file @ 656d1711
.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
View file @ 656d1711
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
View file @ 656d1711
......@@ -7,8 +7,8 @@ CXX = g++
LD = g++
EXPAND = lib/tmpl/expand
#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 -O2 -DNASSERT
#CFLAGS := -g -Wall -pthread -iquote.obj/gen -Wno-uninitialized
CXXFLAGS := -g -std=c++0x
LDFLAGS := -levent_pthreads
## Debian package: check
......@@ -131,6 +131,7 @@ 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
......@@ -236,6 +237,12 @@ clean:
$(call trace,RM,binaries,rm -f $(BINS) $(TEST_BINS))
$(call trace,RM,objects,rm -rf .obj)
#
# Debugging
#
print-%:
@echo '$*=$($*)'
##################################################################
# Targets
#
......
README.md
View file @ 656d1711
......@@ -52,7 +52,13 @@ The repo is structured as follows:
## 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.
TAPIR depends on protobufs, libevent and openssl, so you will need the following development libraries:
- libprotobuf-dev
- libevent-openssl
- libevent-pthreads
- libevent-dev
- libssl-dev
- protobuf-compiler
## Contact and Questions
Please email Irene at iyzhang@cs.washington.edu, Dan at drkp@cs.washington.edu and Naveen at naveenks@cs.washington.edu
lib/Rules.mk
View file @ 656d1711
......@@ -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 tcptransport.cc simtransport.cc)
udptransport.cc tcptransport.cc simtransport.cc repltransport.cc \
persistent_register.cc)
PROTOS += $(addprefix $(d), \
latency-format.proto)
......@@ -24,8 +25,13 @@ LIB-transport := $(o)transport.o $(LIB-message) $(LIB-configuration)
LIB-simtransport := $(o)simtransport.o $(LIB-transport)
LIB-repltransport := $(o)repltransport.o $(LIB-transport)
LIB-udptransport := $(o)udptransport.o $(LIB-transport)
LIB-tcptransport := $(o)tcptransport.o $(LIB-transport)
LIB-persistent_register := $(o)persistent_register.o $(LIB-message)
include $(d)tests/Rules.mk
lib/configuration.cc
View file @ 656d1711
......@@ -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
lib/configuration.h
View file @ 656d1711
......@@ -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_ */
lib/latency-format.proto
View file @ 656d1711
syntax = "proto2";
package transport.latency.format;
message LatencyDist
......
lib/message.cc
View file @ 656d1711
......@@ -40,6 +40,8 @@
#include <unistd.h>
#include <sys/time.h>
#include <mutex>
#define BACKTRACE_ON_PANIC 1
#if BACKTRACE_ON_PANIC
#include <execinfo.h>
......@@ -48,6 +50,8 @@
#define TIMESTAMP_BASE62 0
#define TIMESTAMP_NUMERIC 1
std::mutex message_mtx;
void __attribute__((weak))
Message_VA(enum Message_Type type,
const char *fname, int line, const char *func,
......@@ -74,6 +78,9 @@ _Message_VA(enum Message_Type type, FILE *fp,
const char *fname, int line, const char *func,
const char *fmt, va_list args)
{
// Lock mutex to make sure the output is not mangled.
message_mtx.lock();
static int haveColor = -1;
struct msg_desc {
const char *prefix;
......@@ -150,6 +157,9 @@ _Message_VA(enum Message_Type type, FILE *fp,
fputs("\033[0m", fp);
fprintf(fp, "\n");
fflush(fp);
// Unlock mutex.
message_mtx.unlock();
}
void _Panic(void)
......
lib/persistent_register.cc 0 → 100644
View file @ 656d1711
// -*- 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
View file @ 656d1711
// -*- 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
View file @ 656d1711
// -*- 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
View file @ 656d1711
// -*- 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
View file @ 656d1711
......@@ -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;
}
......
lib/tcptransport.cc
View file @ 656d1711
// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
/***********************************************************************
*
* udptransport.cc:
* tcptransport.cc:
* message-passing network interface that uses TCP message delivery
* and libasync
*
......@@ -35,11 +35,9 @@
#include "lib/tcptransport.h"
#include <google/protobuf/message.h>
#include <event2/event.h>
#include <event2/buffer.h>
#include <event2/thread.h>
#include <event2/bufferevent.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <fcntl.h>
......@@ -87,6 +85,7 @@ TCPTransport::LookupAddress(const transport::ReplicaAddress &addr)
{
int res;
struct addrinfo hints;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_protocol = 0;
......@@ -151,33 +150,27 @@ BindToPort(int fd, const string &host, const string &port)
}
TCPTransport::TCPTransport(double dropRate, double reorderRate,
int dscp, event_base *evbase)
int dscp, bool handleSignals)
{
lastTimerId = 0;
// Set up libevent
evthread_use_pthreads();
event_set_log_callback(LogCallback);
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();
evthread_make_base_notifiable(libeventBase);
}
libeventBase = event_base_new();
evthread_make_base_notifiable(libeventBase);
// Set up signal handler
signalEvents.push_back(evsignal_new(libeventBase, SIGTERM,
SignalCallback, this));
signalEvents.push_back(evsignal_new(libeventBase, SIGINT,
SignalCallback, this));
for (event *x : signalEvents) {
event_add(x, NULL);
if (handleSignals) {
signalEvents.push_back(evsignal_new(libeventBase, SIGTERM,
SignalCallback, this));
signalEvents.push_back(evsignal_new(libeventBase, SIGINT,
SignalCallback, this));
for (event *x : signalEvents) {
event_add(x, NULL);
}
}
}
......@@ -204,6 +197,13 @@ TCPTransport::ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst)
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();
info->transport = this;
info->acceptFd = 0;
......@@ -230,6 +230,15 @@ TCPTransport::ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst)
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;
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");
}
// 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
// host/port
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");
}
// 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
bev = bufferevent_socket_new(transport->libeventBase, newfd,
BEV_OPT_CLOSE_ON_FREE);
......@@ -553,56 +576,62 @@ TCPTransport::TCPReadableCallback(struct bufferevent *bev, void *arg)
Debug("Readable on bufferevent %p", bev);
uint32_t *magic;
magic = (uint32_t *)evbuffer_pullup(evbuf, sizeof(*magic));
ASSERT(*magic == MAGIC);
size_t *sz;
unsigned char *x = evbuffer_pullup(evbuf, sizeof(*magic) + sizeof(*sz));
sz = (size_t *) (x + sizeof(*magic));
if (x == NULL) {
return;
}
size_t totalSize = *sz;
ASSERT(totalSize < 1073741826);
if (evbuffer_get_length(evbuf) < totalSize) {
Debug("Don't have %ld bytes for a message yet, only %ld",
totalSize, evbuffer_get_length(evbuf));
return;
}
Debug("Receiving %ld byte message", totalSize);
char buf[totalSize];
size_t copied = evbuffer_remove(evbuf, buf, totalSize);
ASSERT(copied == totalSize);
// Parse message
char *ptr = buf + sizeof(*sz) + sizeof(*magic);
size_t typeLen = *((size_t *)ptr);
ptr += sizeof(size_t);
ASSERT((size_t)(ptr-buf) < totalSize);
ASSERT((size_t)(ptr+typeLen-buf) < totalSize);
string msgType(ptr, typeLen);
ptr += typeLen;
size_t msgLen = *((size_t *)ptr);
ptr += sizeof(size_t);
ASSERT((size_t)(ptr-buf) < totalSize);
while (evbuffer_get_length(evbuf) > 0) {
uint32_t *magic;
magic = (uint32_t *)evbuffer_pullup(evbuf, sizeof(*magic));
if (magic == NULL) {
return;
}
ASSERT(*magic == MAGIC);
ASSERT((size_t)(ptr+msgLen-buf) <= totalSize);
string msg(ptr, msgLen);
ptr += msgLen;
auto addr = transport->tcpAddresses.find(bev);
ASSERT(addr != transport->tcpAddresses.end());
// Dispatch
info->receiver->ReceiveMessage(addr->second, msgType, msg);
Debug("Done processing large %s message", msgType.c_str());
size_t *sz;
unsigned char *x = evbuffer_pullup(evbuf, sizeof(*magic) + sizeof(*sz));
sz = (size_t *) (x + sizeof(*magic));
if (x == NULL) {
return;
}
size_t totalSize = *sz;
ASSERT(totalSize < 1073741826);
if (evbuffer_get_length(evbuf) < totalSize) {
Debug("Don't have %ld bytes for a message yet, only %ld",
totalSize, evbuffer_get_length(evbuf));
return;
}
Debug("Receiving %ld byte message", totalSize);
char buf[totalSize];
size_t copied = evbuffer_remove(evbuf, buf, totalSize);
ASSERT(copied == totalSize);
// Parse message
char *ptr = buf + sizeof(*sz) + sizeof(*magic);
size_t typeLen = *((size_t *)ptr);
ptr += sizeof(size_t);
ASSERT((size_t)(ptr-buf) < totalSize);
ASSERT((size_t)(ptr+typeLen-buf) < totalSize);
string msgType(ptr, typeLen);
ptr += typeLen;
size_t msgLen = *((size_t *)ptr);
ptr += sizeof(size_t);
ASSERT((size_t)(ptr-buf) < totalSize);
ASSERT((size_t)(ptr+msgLen-buf) <= totalSize);
string msg(ptr, msgLen);
ptr += msgLen;
auto addr = transport->tcpAddresses.find(bev);
ASSERT(addr != transport->tcpAddresses.end());
// Dispatch
info->receiver->ReceiveMessage(addr->second, msgType, msg);
Debug("Done processing large %s message", msgType.c_str());
}
}
void
......
lib/tcptransport.h
View file @ 656d1711
/ -*- 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
* and libasync
*
......@@ -68,7 +68,7 @@ class TCPTransport : public TransportCommon<TCPTransportAddress>
{
public:
TCPTransport(double dropRate = 0.0, double reogrderRate = 0.0,
int dscp = 0, event_base *evbase = nullptr);
int dscp = 0, bool handleSignals = true);
virtual ~TCPTransport();
void Register(TransportReceiver *receiver,
const transport::Configuration &config,
......
lib/tests/simtransport-testmessage.proto
View file @ 656d1711
syntax = "proto2";
package transport.test;
message TestMessage {
......
lib/udptransport.cc
View file @ 656d1711
......@@ -38,6 +38,7 @@
#include <event2/event.h>
#include <event2/thread.h>
#include <memory>
#include <random>
#include <arpa/inet.h>
......@@ -178,9 +179,8 @@ BindToPort(int fd, const string &host, const string &port)
}
UDPTransport::UDPTransport(double dropRate, double reorderRate,
int dscp, event_base *evbase)
: dropRate(dropRate), reorderRate(reorderRate),
dscp(dscp)
int dscp, bool handleSignals)
: dropRate(dropRate), reorderRate(reorderRate), dscp(dscp)
{
lastTimerId = 0;
......@@ -197,27 +197,23 @@ UDPTransport::UDPTransport(double dropRate, double reorderRate,
}
// Set up libevent
evthread_use_pthreads();
event_set_log_callback(LogCallback);
event_set_fatal_callback(FatalCallback);
// XXX Hack for Naveen: allow the user to specify an existing
// libevent base. This will probably not work exactly correctly
// for error messages or signals, but that doesn't much matter...
if (evbase) {
libeventBase = evbase;
} else {
evthread_use_pthreads();
libeventBase = event_base_new();
evthread_make_base_notifiable(libeventBase);
}
// Set up signal handler
signalEvents.push_back(evsignal_new(libeventBase, SIGTERM,
SignalCallback, this));
signalEvents.push_back(evsignal_new(libeventBase, SIGINT,
SignalCallback, this));
libeventBase = event_base_new();
evthread_make_base_notifiable(libeventBase);
for (event *x : signalEvents) {
event_add(x, NULL);
// Set up signal handler
if (handleSignals) {
signalEvents.push_back(evsignal_new(libeventBase, SIGTERM,
SignalCallback, this));
signalEvents.push_back(evsignal_new(libeventBase, SIGINT,
SignalCallback, this));
for (event *x : signalEvents) {
event_add(x, NULL);
}
}
}
......@@ -385,7 +381,8 @@ UDPTransport::Register(TransportReceiver *receiver,
}
static size_t
SerializeMessage(const ::google::protobuf::Message &m, char **out)
SerializeMessage(const ::google::protobuf::Message &m,
std::unique_ptr<char[]> *out)
{
string data = m.SerializeAsString();
string type = m.GetTypeName();
......@@ -394,8 +391,9 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
ssize_t totalLen = (typeLen + sizeof(typeLen) +
dataLen + sizeof(dataLen));
char *buf = new char[totalLen];
std::unique_ptr<char[]> unique_buf(new char[totalLen]);
char *buf = unique_buf.get();
char *ptr = buf;
*((size_t *) ptr) = typeLen;
ptr += sizeof(size_t);
......@@ -409,8 +407,8 @@ SerializeMessage(const ::google::protobuf::Message &m, char **out)
ASSERT(ptr+dataLen-buf == totalLen);
memcpy(ptr, data.c_str(), dataLen);
ptr += dataLen;
*out = buf;
*out = std::move(unique_buf);
return totalLen;
}
......@@ -423,11 +421,12 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
sockaddr_in sin = dynamic_cast<const UDPTransportAddress &>(dst).addr;
// Serialize message
char *buf;
size_t msgLen = SerializeMessage(m, &buf);
std::unique_ptr<char[]> unique_buf;
size_t msgLen = SerializeMessage(m, &unique_buf);
char *buf = unique_buf.get();
int fd = fds[src];
// XXX All of this assumes that the socket is going to be
// available for writing, which since it's a UDP socket it ought
// to be.
......@@ -435,7 +434,7 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
if (sendto(fd, buf, msgLen, 0,
(sockaddr *)&sin, sizeof(sin)) < 0) {
PWarning("Failed to send message");
goto fail;
return false;
}
} else {
int numFrags = ((msgLen-1) / MAX_UDP_MESSAGE_SIZE) + 1;
......@@ -458,22 +457,17 @@ UDPTransport::SendMessageInternal(TransportReceiver *src,
*((size_t *)ptr) = msgLen;
ptr += sizeof(size_t);
memcpy(ptr, &buf[fragStart], fragLen);
if (sendto(fd, fragBuf, fragLen + fragHeaderLen, 0,
(sockaddr *)&sin, sizeof(sin)) < 0) {
PWarning("Failed to send message fragment %ld",
fragStart);
goto fail;
return false;
}
}
}
}
delete [] buf;
return true;
fail:
delete [] buf;
return false;
}
void
......
lib/udptransport.h
View file @ 656d1711
......@@ -66,7 +66,7 @@ class UDPTransport : public TransportCommon<UDPTransportAddress>
{
public:
UDPTransport(double dropRate = 0.0, double reorderRate = 0.0,
int dscp = 0, event_base *evbase = nullptr);
int dscp = 0, bool handleSignals = true);
virtual ~UDPTransport();
void Register(TransportReceiver *receiver,
const transport::Configuration &config,
......
lockserver/Rules.mk
View file @ 656d1711
d := $(dir $(lastword $(MAKEFILE_LIST)))
SRCS += $(addprefix $(d), server.cc client.cc)
SRCS += $(addprefix $(d), \
server.cc client.cc server-main.cc client-main.cc \
lockserver-repl.cc)
PROTOS += $(addprefix $(d), locks-proto.proto)
$(d)server: $(LIB-udptransport) $(OBJS-ir-replica) $(o)locks-proto.o $(o)server.o
$(d)server-main: $(o)server-main.o \
$(o)locks-proto.o \
$(o)server.o \
$(LIB-udptransport) \
$(OBJS-ir-replica)
$(d)client: $(LIB-udptransport) $(OBJS-ir-client) $(LIB-store-common) \
$(o)locks-proto.o $(o)client.o
$(d)client-main: $(o)client-main.o \
$(o)locks-proto.o \
$(o)client.o \
$(LIB-udptransport) \
$(OBJS-ir-client) \
$(LIB-store-common)
BINS += $(d)server $(d)client
$(d)lockserver-repl: $(o)lockserver-repl.o \
$(o)locks-proto.o \
$(o)server.o \
$(o)client.o \
$(OBJS-ir-replica) \
$(OBJS-ir-client) \
$(LIB-configuration) \
$(LIB-repltransport) \
$(LIB-store-common) \
$(GTEST_MAIN)
BINS += $(d)server-main $(d)client-main $(d)lockserver-repl
include $(d)tests/Rules.mk