lib/simtransport.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

lib/tcptransport.h

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * tcptransport.h:
+ *   message-passing network interface that uses UDP message delivery
+ *   and libasync
+ *
+ * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+
+#ifndef _LIB_TCPTRANSPORT_H_
+#define _LIB_TCPTRANSPORT_H_
+
+#include "lib/configuration.h"
+#include "lib/transport.h"
+#include "lib/transportcommon.h"
+
+#include <event2/event.h>
+#include <event2/buffer.h>
+#include <event2/bufferevent.h>
+
+#include <map>
+#include <unordered_map>
+#include <list>
+#include <random>
+#include <mutex>
+#include <netinet/in.h>
+
+class TCPTransportAddress : public TransportAddress
+{
+public:
+    TCPTransportAddress * clone() const;
+private:
+    TCPTransportAddress(const sockaddr_in &addr);
+    
+    sockaddr_in addr;
+    friend class TCPTransport;
+    friend bool operator==(const TCPTransportAddress &a,
+                           const TCPTransportAddress &b);
+    friend bool operator!=(const TCPTransportAddress &a,
+                           const TCPTransportAddress &b);
+    friend bool operator<(const TCPTransportAddress &a,
+                          const TCPTransportAddress &b);
+};
+
+class TCPTransport : public TransportCommon<TCPTransportAddress>
+{
+public:
+    TCPTransport(double dropRate = 0.0, double reogrderRate = 0.0,
+                    int dscp = 0, bool handleSignals = true);
+    virtual ~TCPTransport();
+    void Register(TransportReceiver *receiver,
+                  const transport::Configuration &config,
+                  int replicaIdx);
+    void Run();
+    void Stop();
+    int Timer(uint64_t ms, timer_callback_t cb);
+    bool CancelTimer(int id);
+    void CancelAllTimers();
+    
+private:
+    std::mutex mtx;
+    struct TCPTransportTimerInfo
+    {
+        TCPTransport *transport;
+        timer_callback_t cb;
+        event *ev;
+        int id;
+    };
+    struct TCPTransportTCPListener
+    {
+        TCPTransport *transport;
+        TransportReceiver *receiver;
+        int acceptFd;
+        int replicaIdx;
+        event *acceptEvent;
+        std::list<struct bufferevent *> connectionEvents;
+    };
+    event_base *libeventBase;
+    std::vector<event *> listenerEvents;
+    std::vector<event *> signalEvents;
+    std::map<int, TransportReceiver*> receivers; // fd -> receiver
+    std::map<TransportReceiver*, int> fds; // receiver -> fd
+    int lastTimerId;
+    std::map<int, TCPTransportTimerInfo *> timers;
+    std::list<TCPTransportTCPListener *> tcpListeners;
+    std::map<TCPTransportAddress, struct bufferevent *> tcpOutgoing;
+    std::map<struct bufferevent *, TCPTransportAddress> tcpAddresses;
+    
+    bool SendMessageInternal(TransportReceiver *src,
+                             const TCPTransportAddress &dst,
+                             const Message &m, bool multicast = false);
+
+    TCPTransportAddress
+    LookupAddress(const transport::ReplicaAddress &addr);
+    TCPTransportAddress
+    LookupAddress(const transport::Configuration &cfg,
+                  int replicaIdx);
+    const TCPTransportAddress *
+    LookupMulticastAddress(const transport::Configuration*config) { return NULL; };
+
+    void ConnectTCP(TransportReceiver *src, const TCPTransportAddress &dst);
+    void OnTimer(TCPTransportTimerInfo *info);
+    static void TimerCallback(evutil_socket_t fd,
+                              short what, void *arg);
+    static void LogCallback(int severity, const char *msg);
+    static void FatalCallback(int err);
+    static void SignalCallback(evutil_socket_t fd,
+                               short what, void *arg);
+    static void TCPAcceptCallback(evutil_socket_t fd, short what,
+                                  void *arg);
+    static void TCPReadableCallback(struct bufferevent *bev,
+                                    void *arg);
+    static void TCPEventCallback(struct bufferevent *bev,
+                                 short what, void *arg);
+    static void TCPIncomingEventCallback(struct bufferevent *bev,
+                                         short what, void *arg);
+    static void TCPOutgoingEventCallback(struct bufferevent *bev,
+                                         short what, void *arg);
+};
+
+#endif  // _LIB_TCPTRANSPORT_H_

lib/tests/simtransport-test.cc

@@ -67,11 +67,9 @@ TestReceiver::ReceiveMessage(const TransportAddress &src,
 class SimTransportTest : public testing::Test
 {
 protected:
-    std::vector<transport::ReplicaAddress> replicaAddrs =
-    { { "localhost", "12345" },
-      { "localhost", "12346" },
-      { "localhost", "12347" }};
-    transport::Configuration config{3, 1, replicaAddrs};
+    std::vector<transport::ReplicaAddress> replicaAddrs;
+    
+    transport::Configuration *config;
 
     TestReceiver *receiver0;
     TestReceiver *receiver1;
@@ -80,14 +78,19 @@ protected:
     SimulatedTransport *transport;
 
     virtual void SetUp() {
+	replicaAddrs.push_back(*(new transport::ReplicaAddress("localhost", "12345")));
+	replicaAddrs.push_back(*(new transport::ReplicaAddress("localhost", "12346")));
+	replicaAddrs.push_back(*(new transport::ReplicaAddress("localhost", "12347")));
+
         receiver0 = new TestReceiver();
         receiver1 = new TestReceiver();
         receiver2  = new TestReceiver();
 
+	config = new transport::Configuration(3, 1, replicaAddrs);
         transport = new SimulatedTransport();
-        transport->Register(receiver0, config, 0);
-        transport->Register(receiver1, config, 1);
-        transport->Register(receiver2, config, 2);
+        transport->Register(receiver0, *config, 0);
+        transport->Register(receiver1, *config, 1);
+        transport->Register(receiver2, *config, 2);
     }
     
     virtual void TearDown() {

lib/tests/simtransport-testmessage.proto

+syntax = "proto2";
+
 package transport.test;
 
 message TestMessage {

lib/timeval.h

@@ -5,7 +5,7 @@
  *   utility functions for manipulating timevals
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  * Copyright 2009-2012 Massachusetts Institute of Technology
  *

lib/transport.cc

@@ -5,7 +5,7 @@
  *   message-passing network interface; common definitions
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  *
  * Permission is hereby granted, free of charge, to any person

lib/transportcommon.h

@@ -5,7 +5,7 @@
  *   template support for implementing transports
  *
  * Copyright 2013-2015 Irene Zhang <iyzhang@cs.washington.edu>
- *                     Naveen Kr. Sharma <nksharma@cs.washington.edu>
+ *                     Naveen Kr. Sharma <naveenks@cs.washington.edu>
  *                     Dan R. K. Ports  <drkp@cs.washington.edu>
  *
  * Permission is hereby granted, free of charge, to any person

lib/udptransport.cc

@@ -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

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

libtapir/Rules.mk

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

lockserver/Rules.mk

+d := $(dir $(lastword $(MAKEFILE_LIST)))
+
+SRCS += $(addprefix $(d), \
+			server.cc client.cc server-main.cc client-main.cc \
+			lockserver-repl.cc)
+
+PROTOS += $(addprefix $(d), locks-proto.proto)
+
+$(d)server-main: $(o)server-main.o \
+	$(o)locks-proto.o \
+	$(o)server.o \
+	$(LIB-udptransport) \
+	$(OBJS-ir-replica)
+
+$(d)client-main: $(o)client-main.o \
+	$(o)locks-proto.o \
+	$(o)client.o \
+	$(LIB-udptransport) \
+   	$(OBJS-ir-client) \
+   	$(LIB-store-common)
+
+$(d)lockserver-repl: $(o)lockserver-repl.o \
+	$(o)locks-proto.o \
+	$(o)server.o \
+	$(o)client.o \
+	$(OBJS-ir-replica) \
+	$(OBJS-ir-client) \
+	$(LIB-configuration) \
+	$(LIB-repltransport) \
+   	$(LIB-store-common) \
+	$(GTEST_MAIN)
+
+BINS += $(d)server-main $(d)client-main $(d)lockserver-repl
+
+include $(d)tests/Rules.mk

lockserver/client-main.cc

+#include <thread>
+
+#include "lockserver/client.h"
+#include "lib/udptransport.h"
+
+namespace {
+
+void
+usage()
+{
+    printf("Unknown command.. Try again!\n");
+    printf("Usage: exit | q | lock <key> | unlock <key>\n");
+}
+
+} // namespace
+
+int
+main(int argc, char **argv)
+{
+    const char *configPath = NULL;
+
+    // Parse arguments
+    int opt;
+    while ((opt = getopt(argc, argv, "c:")) != -1) {
+        switch (opt) {
+        case 'c':
+            configPath = optarg;
+            break;
+
+        default:
+            fprintf(stderr, "Unknown argument %s\n", argv[optind]);
+        }
+    }
+
+    if (!configPath) {
+        fprintf(stderr, "option -c is required\n");
+        return EXIT_FAILURE;
+    }
+
+    // Load configuration
+    std::ifstream configStream(configPath);
+    if (configStream.fail()) {
+        Panic("Unable to read configuration file: %s\n", configPath);
+    }
+    transport::Configuration config(configStream);
+
+    // Create lock client.
+    UDPTransport transport(0.0, 0.0, 0);
+    lockserver::LockClient locker(&transport, config);
+    std::thread run_transport([&transport]() { transport.Run(); });
+
+    char c, cmd[2048], *tok;
+    int clen, status;
+    string key, value;
+
+    while (1) {
+        printf(">> ");
+        fflush(stdout);
+
+        clen = 0;
+        while ((c = getchar()) != '\n')
+            cmd[clen++] = c;
+        cmd[clen] = '\0';
+
+        tok = strtok(cmd, " ,.-");
+        if (tok == NULL) continue;
+
+        if (strcasecmp(tok, "exit") == 0 || strcasecmp(tok, "q") == 0) {
+            printf("Exiting..\n");
+            break;
+        } else if (strcasecmp(tok, "lock") == 0) {
+            tok = strtok(NULL, " ,.-");
+            if (tok == NULL) {
+                usage();
+                continue;
+            }
+            key = string(tok);
+            status = locker.lock(key);
+
+            if (status) {
+                printf("Lock Successful\n");
+            } else {
+                printf("Failed to acquire lock..\n");
+            }
+        } else if (strcasecmp(tok, "unlock") == 0) {
+            tok = strtok(NULL, " ,.-");
+            if (tok == NULL) {
+                usage();
+                continue;
+            }
+            key = string(tok);
+            locker.unlock(key);
+            printf("Unlock Successful\n");
+        } else {
+            usage();
+        }
+        fflush(stdout);
+    }
+
+    transport.Stop();
+    run_transport.join();
+    return EXIT_SUCCESS;
+}

lockserver/client.cc

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * lockserver/client.cc:
+ *   A single lockserver client with commandline interface.
+ *
+ * Copyright 2015 Naveen Kr. Sharma <naveenks@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 "lockserver/client.h"
+
+namespace lockserver {
+
+using namespace std;
+using namespace proto;
+
+LockClient::LockClient(Transport *transport,
+                       const transport::Configuration &config)
+    : transport(transport) {
+    client_id = 0;
+    while (client_id == 0) {
+        random_device rd;
+        mt19937_64 gen(rd());
+        uniform_int_distribution<uint64_t> dis;
+        client_id = dis(gen);
+    }
+
+    client = new replication::ir::IRClient(config, transport, client_id);
+}
+
+LockClient::~LockClient() { }
+
+void
+LockClient::lock_async(const std::string &key) {
+    ASSERT(waiting == nullptr);
+    Debug("Sending LOCK");
+
+    string request_str;
+    Request request;
+    request.set_clientid(client_id);
+    request.set_key(key);
+    request.set_type(true);
+    request.SerializeToString(&request_str);
+
+    waiting = new Promise(1000);
+    transport->Timer(0, [=]() {
+            client->InvokeConsensus(request_str,
+                bind(&LockClient::Decide,
+                    this,
+                    placeholders::_1),
+                bind(&LockClient::LockCallback,
+                    this,
+                    placeholders::_1,
+                    placeholders::_2),
+                bind(&LockClient::ErrorCallback,
+                    this,
+                    placeholders::_1,
+                    placeholders::_2));
+            });
+}
+
+bool
+LockClient::lock_wait() {
+    ASSERT(waiting != nullptr);
+
+    int status = waiting->GetReply();
+    delete waiting;
+    waiting = nullptr;
+
+    if (status == 0) {
+        return true;
+    } else if (status == -1) {
+        Debug("Lock held by someone else.");
+    }
+    return false;
+}
+
+void
+LockClient::unlock_async(const std::string &key) {
+    ASSERT(waiting == nullptr);
+    Debug("Sending UNLOCK");
+
+    string request_str;
+    Request request;
+    request.set_clientid(client_id);
+    request.set_key(key);
+    request.set_type(false);
+    request.SerializeToString(&request_str);
+
+    waiting = new Promise(1000);
+    transport->Timer(0, [=]() {
+            client->InvokeInconsistent(request_str,
+                bind(&LockClient::UnlockCallback,
+                    this,
+                    placeholders::_1,
+                    placeholders::_2));
+            });
+}
+
+void
+LockClient::unlock_wait() {
+    waiting->GetReply();
+    delete waiting;
+    waiting = nullptr;
+}
+
+bool
+LockClient::lock(const string &key)
+{
+    lock_async(key);
+    return lock_wait();
+}
+
+void
+LockClient::unlock(const string &key)
+{
+    unlock_async(key);
+    return unlock_wait();
+}
+
+string
+LockClient::Decide(const map<string, std::size_t> &results)
+{
+    // If a majority say lock, we say lock.
+    int success_count = 0;
+    string key;
+    for (const auto& string_and_count : results) {
+        const string& s = string_and_count.first;
+        const std::size_t count = string_and_count.second;
+
+        Reply reply;
+        reply.ParseFromString(s);
+        key = reply.key();
+
+        if (reply.status() == 0) {
+            success_count += count;
+        }
+    }
+
+    string final_reply_str;
+    Reply final_reply;
+    final_reply.set_key(key);
+    if (success_count >= 2) {
+        final_reply.set_status(0);
+    } else {
+        final_reply.set_status(-1);
+    }
+    final_reply.SerializeToString(&final_reply_str);
+    return final_reply_str;
+}
+
+void
+LockClient::LockCallback(const std::string &request_str, const std::string &reply_str)
+{
+    Debug("Lock Callback: %s %s", request_str.c_str(), reply_str.c_str());
+    Reply reply;
+    reply.ParseFromString(reply_str);
+
+    Promise *w = waiting;
+    waiting = NULL;
+    w->Reply(reply.status());
+}
+
+void
+LockClient::UnlockCallback(const std::string &request_str, const std::string &reply_str)
+{
+    Debug("Lock Callback: %s %s", request_str.c_str(), reply_str.c_str());
+
+    Promise *w = waiting;
+    waiting = NULL;
+    w->Reply(0);
+}
+
+void
+LockClient::ErrorCallback(const std::string &request_str,
+                          replication::ErrorCode err)
+{
+    Debug("Error Callback: %s %s", request_str.c_str(),
+          replication::ErrorCodeToString(err).c_str());
+    Promise *w = waiting;
+    waiting = NULL;
+    w->Reply(-3);  // Invalid command.
+}
+
+} // namespace lockserver

lockserver/client.h

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * lockserver/client.h:
+ *   A lockserver client interface.
+ *
+ * Copyright 2015 Naveen Kr. Sharma <naveenks@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 _IR_LOCK_CLIENT_H_
+#define _IR_LOCK_CLIENT_H_
+
+#include "lib/assert.h"
+#include "lib/message.h"
+#include "lib/transport.h"
+#include "replication/ir/client.h"
+#include "store/common/promise.h"
+#include "lockserver/locks-proto.pb.h"
+
+#include <map>
+#include <set>
+#include <string>
+#include <thread>
+#include <random>
+
+namespace lockserver {
+
+class LockClient
+{
+public:
+    LockClient(Transport* transport, const transport::Configuration &config);
+    ~LockClient();
+
+    // Synchronously lock and unlock. Calling lock (or unlock) will block until
+    // the lock (or unlock) request is fully processed.
+    bool lock(const std::string &key);
+    void unlock(const std::string &key);
+
+    // Asynchronously lock and unlock. Calling lock_async or unlock_async will
+    // not block. Calling lock_wait (or unlock_wait) will block for the
+    // previous invocation of lock_async (or unlock_async) to complete.
+    //
+    // All async calls must be followed by a corresponding wait call. It is an
+    // error to issue multiple async requests without waiting. It is also
+    // erroneous to wait for a request which was never issued.
+    void lock_async(const std::string &key);
+    bool lock_wait();
+    void unlock_async(const std::string &key);
+    void unlock_wait();
+
+private:
+    /* Unique ID for this client. */
+    uint64_t client_id;
+
+    /* Transport layer and thread. */
+    Transport *transport;
+
+    /* Function to run the transport thread. */
+    void run_client();
+
+    /* Decide function for a lock server. */
+    string Decide(const std::map<string, std::size_t> &results);
+
+    /* IR client proxy. */
+    replication::ir::IRClient *client;
+
+    /* Promise to wait for pending operation. */
+    Promise *waiting = nullptr;
+
+    /* Callbacks for hearing back for an operation. */
+    void LockCallback(const std::string &, const std::string &);
+    void UnlockCallback(const std::string &, const std::string &);
+    void ErrorCallback(const std::string &, replication::ErrorCode);
+};
+
+} // namespace lockserver
+
+#endif /* _IR_LOCK_CLIENT_H_ */

lockserver/locks-proto.proto

+syntax = "proto2";
+
+package lockserver.proto;
+
+message Request {
+  required uint64 clientid = 1;
+  required string key = 2;
+  required bool type = 3;
+  // true = lock
+  // false = unlock
+}
+
+message Reply {
+  required string key = 1;
+  required int32 status = 2;
+  // 0 = Operation Success
+  // -1 = Held by someone else (for lock)
+  // -2 = Not held by you (for unlock)
+  // -3 = Invalid command
+}

lockserver/lockserver-repl.cc

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * lockserver-repl.cc: Step-by-step lock server evaluation.
+ *
+ * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ **********************************************************************/
+#include <thread>
+#include <memory>
+
+#include "lib/configuration.h"
+#include "lib/repltransport.h"
+#include "lockserver/client.h"
+#include "lockserver/server.h"
+#include "replication/ir/replica.h"
+
+int main() {
+    ReplTransport transport;
+    std::vector<transport::ReplicaAddress> replica_addrs = {
+        {"replica", "0"},
+        {"replica", "1"},
+        {"replica", "2"},
+        {"replica", "3"},
+        {"replica", "4"}};
+    transport::Configuration config(5 /* n */, 2 /* f */, replica_addrs);
+
+    // Clients.
+    lockserver::LockClient client_a(&transport, config);
+    lockserver::LockClient client_b(&transport, config);
+    lockserver::LockClient client_c(&transport, config);
+    client_a.lock_async("a");
+    client_b.lock_async("b");
+    client_c.lock_async("c");
+
+    // Servers.
+    std::vector<std::unique_ptr<lockserver::LockServer>> servers;
+    std::vector<std::unique_ptr<replication::ir::IRReplica>> replicas;
+    for (std::size_t i = 0; i < replica_addrs.size(); ++i) {
+        auto server = std::unique_ptr<lockserver::LockServer>(
+            new lockserver::LockServer());
+        servers.push_back(std::move(server));
+        auto replica = std::unique_ptr<replication::ir::IRReplica>(
+            new replication::ir::IRReplica(config, i, &transport,
+                                           servers[i].get()));
+        replicas.push_back(std::move(replica));
+    }
+
+    // Launch REPL.
+    transport.Run();
+
+    // Remove persisted files.
+    for (std::size_t i = 0; i < replica_addrs.size(); ++i) {
+        const transport::ReplicaAddress &addr = replica_addrs[i];
+        const std::string filename =
+            addr.host + ":" + addr.port + "_" + std::to_string(i) + ".bin";
+        int success = std::remove(filename.c_str());
+        ASSERT(success == 0);
+    }
+}

lockserver/server-main.cc

+#include "lockserver/server.h"
+
+int
+main(int argc, char **argv)
+{
+    int index = -1;
+    const char *configPath = NULL;
+
+    // Parse arguments
+    int opt;
+    char *strtolPtr;
+    while ((opt = getopt(argc, argv, "c:i:")) != -1) {
+        switch (opt) {
+        case 'c':
+            configPath = optarg;
+            break;
+
+        case 'i':
+            index = strtol(optarg, &strtolPtr, 10);
+            if ((*optarg == '\0') || (*strtolPtr != '\0') || (index < 0)) {
+                fprintf(stderr, "option -i requires a numeric arg\n");
+            }
+            break;
+
+        default:
+            fprintf(stderr, "Unknown argument %s\n", argv[optind]);
+        }
+    }
+
+    if (!configPath) {
+        fprintf(stderr, "option -c is required\n");
+        return EXIT_FAILURE;
+    }
+
+    if (index == -1) {
+        fprintf(stderr, "option -i is required\n");
+        return EXIT_FAILURE;
+    }
+
+    // Load configuration
+    std::ifstream configStream(configPath);
+    if (configStream.fail()) {
+        fprintf(stderr, "unable to read configuration file: %s\n", configPath);
+        return EXIT_FAILURE;
+    }
+    transport::Configuration config(configStream);
+
+    if (index >= config.n) {
+        fprintf(stderr, "replica index %d is out of bounds; "
+                "only %d replicas defined\n", index, config.n);
+        return EXIT_FAILURE;
+    }
+
+    UDPTransport transport(0.0, 0.0, 0);
+
+    lockserver::LockServer server;
+    replication::ir::IRReplica replica(config, index, &transport, &server);
+
+    transport.Run();
+
+    return EXIT_SUCCESS;
+}

lockserver/server.cc

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * lockserver/server.cc:
+ *   A lockserver replica.
+ *
+ * Copyright 2015 Naveen Kr. Sharma <naveenks@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 "lockserver/server.h"
+
+#include <algorithm>
+#include <iterator>
+#include <unordered_set>
+
+namespace lockserver {
+
+using namespace proto;
+
+LockServer::LockServer() { }
+LockServer::~LockServer() { }
+
+void
+LockServer::ExecInconsistentUpcall(const string &str1)
+{
+    Debug("ExecInconsistent: %s\n", str1.c_str());
+
+    Request request;
+
+    request.ParseFromString(str1);
+    string key = request.key();
+    uint64_t client_id = request.clientid();
+
+    if (request.type()) { // Lock operation.
+        Warning("Lock operation being sent as Inconsistent. Ignored.");
+    } else {
+        if (locks.find(key) != locks.end()) {
+            if (client_id == locks[key]) {
+                Debug("Releasing lock %lu: %s", client_id, key.c_str());
+                locks.erase(key);
+            } else {
+                Debug("Lock held by someone else %lu: %s, %lu",
+                        client_id, key.c_str(), locks[key]);
+            }
+        } else {
+            Debug("Lock held by no one.");
+        }
+    }
+}
+
+void
+LockServer::ExecConsensusUpcall(const string &str1, string &str2)
+{
+    Debug("ExecConsensus: %s\n", str1.c_str());
+
+    Request request;
+    Reply reply;
+
+    request.ParseFromString(str1);
+    string key = request.key();
+    uint64_t client_id = request.clientid();
+    reply.set_key(key);
+    int status = 0;
+
+    if (request.type()) { // Lock operation.
+        if (locks.find(key) == locks.end()) {
+            Debug("Assigning lock %lu: %s", client_id, key.c_str());
+            locks[key] = client_id;
+        } else if (locks[key] != client_id) {
+            Debug("Lock already held %lu: %s", client_id, key.c_str());
+            status = -1;
+        }
+    } else {
+        Warning("Unlock operation being sent as Consensus");
+        if (locks.find(key) == locks.end()) {
+            Debug("Lock held by no-one %lu: %s", client_id, key.c_str());
+            status = -2;
+        } else if (locks[key] != client_id) {
+            Debug("Lock held by someone else %lu: %s, %lu",
+                    client_id, key.c_str(), locks[key]);
+            status = -2;
+        } else {
+            Debug("Releasing lock %lu: %s", client_id, key.c_str());
+            locks.erase(key);
+        }
+    }
+
+    reply.set_status(status);
+    reply.SerializeToString(&str2);
+}
+
+void
+LockServer::UnloggedUpcall(const string &str1, string &str2)
+{
+    Debug("Unlogged: %s\n", str1.c_str());
+}
+
+void
+LockServer::Sync(const std::map<opid_t, RecordEntry>& record) {
+    locks.clear();
+
+    struct KeyLockInfo {
+        std::unordered_set<uint64_t> locked;
+        std::unordered_set<uint64_t> unlocked;
+    };
+    std::unordered_map<std::string, KeyLockInfo> key_lock_info;
+
+    for (const std::pair<const opid_t, RecordEntry> &p : record) {
+        const opid_t &opid = p.first;
+        const RecordEntry &entry = p.second;
+
+        Request request;
+        request.ParseFromString(entry.request.op());
+        Reply reply;
+        reply.ParseFromString(entry.result);
+        KeyLockInfo &info = key_lock_info[request.key()];
+
+        Debug("Sync opid=(%lu, %lu), clientid=%lu, key=%s, type=%d, status=%d.",
+              opid.first, opid.second, request.clientid(),
+              request.key().c_str(), request.type(), reply.status());
+
+        if (request.type() && reply.status() == 0) {
+            // Lock.
+            info.locked.insert(request.clientid());
+        } else if (!request.type() && reply.status() == 0) {
+            // Unlock.
+            info.unlocked.insert(request.clientid());
+        }
+    }
+
+    for (const std::pair<const std::string, KeyLockInfo> &p : key_lock_info) {
+        const std::string &key = p.first;
+        const KeyLockInfo &info = p.second;
+        std::unordered_set<uint64_t> diff;
+        std::set_difference(std::begin(info.locked), std::end(info.locked),
+                            std::begin(info.unlocked), std::end(info.unlocked),
+                            std::inserter(diff, diff.begin()));
+
+        ASSERT(diff.size() == 0 || diff.size() == 1);
+        if (diff.size() == 1) {
+            uint64_t client_id = *std::begin(diff);
+            Debug("Assigning lock %lu: %s", client_id, key.c_str());
+            locks[key] = client_id;
+        }
+    }
+}
+
+std::map<opid_t, std::string>
+LockServer::Merge(const std::map<opid_t, std::vector<RecordEntry>> &d,
+                  const std::map<opid_t, std::vector<RecordEntry>> &u,
+                  const std::map<opid_t, std::string> &majority_results_in_d) {
+    // First note that d and u only contain consensus operations, and lock
+    // requests are the only consensus operations (unlock is an inconsistent
+    // operation), so d and u only contain lock requests. To merge, we grant
+    // any majority successful lock request in d if it does not conflict with a
+    // currently held lock. We do not grant any other lock request.
+
+    std::map<opid_t, std::string> results;
+
+    using EntryVec = std::vector<RecordEntry>;
+    for (const std::pair<const opid_t, EntryVec>& p: d) {
+        const opid_t &opid = p.first;
+        const EntryVec &entries = p.second;
+
+        // Get the request and reply.
+        const RecordEntry &entry = *std::begin(entries);
+
+        Request request;
+        request.ParseFromString(entry.request.op());
+
+        Reply reply;
+        auto iter = majority_results_in_d.find(opid);
+        ASSERT(iter != std::end(majority_results_in_d));
+        reply.ParseFromString(iter->second);
+
+        // Form the final result.
+        const bool operation_successful = reply.status() == 0;
+        if (operation_successful) {
+            // If the lock was successful, then we acquire the lock so long as
+            // it is not already held.
+            const std::string &key = reply.key();
+            if (locks.count(key) == 0) {
+                Debug("Assigning lock %lu: %s", request.clientid(),
+                      key.c_str());
+                locks[key] = request.clientid();
+                results[opid] = iter->second;
+            } else {
+                Debug("Rejecting lock %lu: %s", request.clientid(),
+                      key.c_str());
+                reply.set_status(-1);
+                std::string s;
+                reply.SerializeToString(&s);
+                results[opid] = s;
+            }
+        } else {
+            // If the lock was not successful, then we maintain this as the
+            // majority result.
+            results[opid] = iter->second;
+        }
+    }
+
+    // We reject all lock requests in u. TODO: We could acquire a lock if
+    // it is free, but it's simplest to just reject them unilaterally.
+    for (const std::pair<const opid_t, EntryVec>& p: u) {
+        const opid_t &opid = p.first;
+        const EntryVec &entries = p.second;
+
+        const RecordEntry &entry = *std::begin(entries);
+        Request request;
+        request.ParseFromString(entry.request.op());
+
+        Debug("Rejecting lock %lu: %s", request.clientid(),
+              request.key().c_str());
+        Reply reply;
+        reply.set_key(request.key());
+        reply.set_status(-1);
+        std::string s;
+        reply.SerializeToString(&s);
+        results[opid] = s;
+    }
+
+    return results;
+}
+
+} // namespace lockserver

lockserver/server.h

+// -*- mode: c++; c-file-style: "k&r"; c-basic-offset: 4 -*-
+/***********************************************************************
+ *
+ * lockserver/server.h:
+ *   A lockserver replica interface.
+ *
+ * Copyright 2015 Naveen Kr. Sharma <naveenks@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 _IR_LOCK_SERVER_H_
+#define _IR_LOCK_SERVER_H_
+
+#include <string>
+#include <unordered_map>
+
+#include "lib/transport.h"
+#include "replication/ir/replica.h"
+#include "lockserver/locks-proto.pb.h"
+
+namespace lockserver {
+
+using opid_t = replication::ir::opid_t;
+using RecordEntry = replication::ir::RecordEntry;
+
+class LockServer : public replication::ir::IRAppReplica
+{
+public:
+    LockServer();
+    ~LockServer();
+
+    // Invoke inconsistent operation, no return value
+    void ExecInconsistentUpcall(const string &str1) override;
+
+    // Invoke consensus operation
+    void ExecConsensusUpcall(const string &str1, string &str2) override;
+
+    // Invoke unreplicated operation
+    void UnloggedUpcall(const string &str1, string &str2) override;
+
+    // Sync
+    void Sync(const std::map<opid_t, RecordEntry>& record) override;
+
+    // Merge
+    std::map<opid_t, std::string> Merge(
+        const std::map<opid_t, std::vector<RecordEntry>> &d,
+        const std::map<opid_t, std::vector<RecordEntry>> &u,
+        const std::map<opid_t, std::string> &majority_results_in_d) override;
+
+private:
+    std::unordered_map<std::string, uint64_t> locks;
+};
+
+} // namespace lockserver
+
+#endif /* _IR_LOCK_SERVER_H_ */