// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
/***********************************************************************
*
* ir-test.cc:
* test cases for Inconsistent Replication Protocol
*
* Copyright 2013 Dan R. K. Ports <drkp@cs.washington.edu>
* Copyright 2015 Irene Zhang Ports <iyzhang@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/configuration.h"
#include "lib/message.h"
#include "lib/transport.h"
#include "lib/simtransport.h"
#include "replication/common/client.h"
#include "replication/common/replica.h"
#include "replication/ir/client.h"
#include "replication/ir/replica.h"
#include <cstdio>
#include <stdlib.h>
#include <stdio.h>
#include <gtest/gtest.h>
#include <vector>
#include <set>
#include <sstream>
#include <memory>
using google::protobuf::Message;
using namespace replication;
using namespace replication::ir;
using namespace replication::ir::proto;
class IRApp : public IRAppReplica {
std::vector<string> *iOps;
std::vector<string> *cOps;
std::vector<string> *unloggedOps;
public:
IRApp(std::vector<string> *i, std::vector<string> *c,
std::vector<string> *u)
: iOps(i), cOps(c), unloggedOps(u) {}
void ExecInconsistentUpcall(const string &req) {
iOps->push_back(req);
}
void ExecConsensusUpcall(const string &req, string &reply) {
cOps->push_back(req);
reply = "1";
}
void UnloggedUpcall(const string &req, string &reply) {
unloggedOps->push_back(req);
reply = "unlreply: " + req;
}
};
class IRTest : public ::testing::Test
{
protected:
std::vector<transport::ReplicaAddress> replicaAddrs;
std::unique_ptr<transport::Configuration> config;
SimulatedTransport transport;
std::vector<std::unique_ptr<IRApp>> apps;
std::vector<std::unique_ptr<IRReplica>> replicas;
std::unique_ptr<IRClient> client;
std::vector<std::vector<string> > iOps;
std::vector<std::vector<string> > cOps;
std::vector<std::vector<string> > unloggedOps;
int requestNum;
IRTest() : requestNum(-1) {
replicaAddrs = {{"localhost", "12345"},
{"localhost", "12346"},
{"localhost", "12347"}};
config = std::unique_ptr<transport::Configuration>(
new transport::Configuration(3, 1, replicaAddrs));
iOps.resize(config->n);
cOps.resize(config->n);
unloggedOps.resize(config->n);
for (int i = 0; i < config->n; i++) {
auto ir_app = std::unique_ptr<IRApp>(
new IRApp(&iOps[i], &cOps[i], &unloggedOps[i]));
auto p = std::unique_ptr<IRReplica>(
new IRReplica(*config, i, &transport, ir_app.get()));
apps.push_back(std::move(ir_app));
replicas.push_back(std::move(p));
}
client = std::unique_ptr<IRClient>(new IRClient(*config, &transport));
}
virtual string RequestOp(int n) {
std::ostringstream stream;
stream << "test: " << n;
return stream.str();
}
virtual string LastRequestOp() {
return RequestOp(requestNum);
}
virtual void ClientSendNextInconsistent(Client::continuation_t upcall) {
requestNum++;
client->InvokeInconsistent(LastRequestOp(), upcall);
}
virtual void ClientSendNextConsensus(Client::continuation_t upcall,
IRClient::decide_t decide) {
requestNum++;
client->InvokeConsensus(LastRequestOp(), decide, upcall);
}
virtual void ClientSendNextUnlogged(
int idx, Client::continuation_t upcall,
Client::error_continuation_t error_continuation = nullptr,
uint32_t timeout = Client::DEFAULT_UNLOGGED_OP_TIMEOUT) {
requestNum++;
client->InvokeUnlogged(idx, LastRequestOp(), upcall,
error_continuation, timeout);
}
virtual void TearDown() {
// Replicas store their view information in the following files:
// - localhost:12345_0.bin
// - localhost:12346_1.bin
// - localhost:12347_2.bin
// We have to make sure to delete them after every test. Otherwise,
// replicas run in recovery mode.
for (std::size_t i = 0; i < replicaAddrs.size(); ++i) {
const transport::ReplicaAddress &addr = replicaAddrs[i];
const std::string filename =
addr.host + ":" + addr.port + "_" + std::to_string(i) + ".bin";
int success = std::remove(filename.c_str());
ASSERT(success == 0);
}
}
};
TEST_F(IRTest, OneInconsistentOp)
{
auto upcall = [this](const string &req, const string &reply) {
EXPECT_EQ(req, LastRequestOp());
// Inconsistent ops do not return a value
EXPECT_EQ(reply, "");
transport.CancelAllTimers();
};
ClientSendNextInconsistent(upcall);
transport.Run();
// By now, they all should have executed the last request.
for (int i = 0; i < config->n; i++) {
EXPECT_EQ(iOps[i].size(), 1);
EXPECT_EQ(iOps[i].back(), LastRequestOp());
}
}
TEST_F(IRTest, OneConsensusOp)
{
auto upcall = [this](const string &req, const string &reply) {
EXPECT_EQ(req, LastRequestOp());
EXPECT_EQ(reply, "1");
transport.CancelAllTimers();
};
auto decide = [this](const std::map<string, std::size_t> &results) {
// shouldn't ever get called
EXPECT_FALSE(true);
return "";
};
ClientSendNextConsensus(upcall, decide);
transport.Run();
// By now, they all should have executed the last request.
for (int i = 0; i < config->n; i++) {
EXPECT_EQ(cOps[i].size(), 1);
EXPECT_EQ(cOps[i].back(), LastRequestOp());
}
}
TEST_F(IRTest, Unlogged)
{
auto upcall = [this](const string &req, const string &reply) {
EXPECT_EQ(req, LastRequestOp());
EXPECT_EQ(reply, "unlreply: "+LastRequestOp());
EXPECT_EQ(unloggedOps[1].back(), req);
transport.CancelAllTimers();
};
int timeouts = 0;
auto timeout = [&](const string &req, ErrorCode) {
timeouts++;
};
ClientSendNextUnlogged(1, upcall, timeout);
transport.Run();
for (unsigned int i = 0; i < iOps.size(); i++) {
EXPECT_EQ(0, iOps[i].size());
EXPECT_EQ((i == 1 ? 1 : 0), unloggedOps[i].size());
}
EXPECT_EQ(0, timeouts);
}
TEST_F(IRTest, UnloggedTimeout)
{
auto upcall = [this](const string &req, const string &reply) {
FAIL();
transport.CancelAllTimers();
};
int timeouts = 0;
auto timeout = [&](const string &req, ErrorCode) {
timeouts++;
};
// Drop messages to or from replica 1
transport.AddFilter(10, [](TransportReceiver *src, int srcIdx,
TransportReceiver *dst, int dstIdx,
Message &m, uint64_t &delay) {
if ((srcIdx == 1) || (dstIdx == 1)) {
return false;
}
return true;
});
// Run for 10 seconds
transport.Timer(10000, [&]() {
transport.CancelAllTimers();
});
ClientSendNextUnlogged(1, upcall, timeout);
transport.Run();
for (unsigned int i = 0; i < iOps.size(); i++) {
EXPECT_EQ(0, iOps[i].size());
EXPECT_EQ(0, unloggedOps[i].size());
}
EXPECT_EQ(1, timeouts);
}
// TEST_F(IRTest, ManyOps)
// {
// Client::continuation_t upcall = [&](const string &req, const string &reply) {
// EXPECT_EQ(req, LastRequestOp());
// EXPECT_EQ(reply, "reply: "+LastRequestOp());
// // Not guaranteed that any replicas except the leader have
// // executed this request.
// EXPECT_EQ(ops[0].back(), req);
// if (requestNum < 9) {
// ClientSendNext(upcall);
// } else {
// transport.CancelAllTimers();
// }
// };
// ClientSendNext(upcall);
// transport.Run();
// // By now, they all should have executed the last request.
// for (int i = 0; i < config->n; i++) {
// EXPECT_EQ(10, ops[i].size());
// for (int j = 0; j < 10; j++) {
// EXPECT_EQ(RequestOp(j), ops[i][j]);
// }
// }
// }