lib/simtransport.h

@@ -4,7 +4,9 @@
  * simtransport.h:
  *   simulated message-passing interface for testing use
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * 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
@@ -31,8 +33,8 @@
 #ifndef _LIB_SIMTRANSPORT_H_
 #define _LIB_SIMTRANSPORT_H_
 
-#include "paxos-lib/lib/transport.h"
-#include "paxos-lib/lib/transportcommon.h"
+#include "lib/transport.h"
+#include "lib/transportcommon.h"
 
 #include <deque>
 #include <map>
@@ -65,7 +67,7 @@ public:
     SimulatedTransport();
     ~SimulatedTransport();
     void Register(TransportReceiver *receiver,
-                  const specpaxos::Configuration &config,
+                  const transport::Configuration &config,
                   int replicaIdx);
     void Run();
     void AddFilter(int id, filter_t filter);
@@ -81,9 +83,9 @@ protected:
                              bool multicast);
     
     SimulatedTransportAddress
-    LookupAddress(const specpaxos::Configuration &cfg, int idx);
+    LookupAddress(const transport::Configuration &cfg, int idx);
     const SimulatedTransportAddress *
-    LookupMulticastAddress(const specpaxos::Configuration *cfg);
+    LookupMulticastAddress(const transport::Configuration *cfg);
     
 private:
     struct QueuedMessage {

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/configuration-test.cc

@@ -28,11 +28,11 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/configuration.h"
+#include "lib/configuration.h"
 
 #include <gtest/gtest.h>
 
-using namespace specpaxos;
+using namespace transport;
 using std::vector;
 
 TEST(Configuration, Basic)

lib/tests/simtransport-test.cc

@@ -28,14 +28,14 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/configuration.h"
-#include "paxos-lib/lib/message.h"
-#include "paxos-lib/lib/simtransport.h"
-#include "paxos-lib/lib/tests/simtransport-testmessage.pb.h"
+#include "lib/configuration.h"
+#include "lib/message.h"
+#include "lib/simtransport.h"
+#include "lib/tests/simtransport-testmessage.pb.h"
 
 #include <gtest/gtest.h>
 
-using namespace specpaxos::test;
+using namespace transport::test;
 using ::google::protobuf::Message;
 
 
@@ -67,11 +67,9 @@ TestReceiver::ReceiveMessage(const TransportAddress &src,
 class SimTransportTest : public testing::Test
 {
 protected:
-    std::vector<specpaxos::ReplicaAddress> replicaAddrs =
-    { { "localhost", "12345" },
-      { "localhost", "12346" },
-      { "localhost", "12347" }};
-    specpaxos::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

-package specpaxos.test;
+syntax = "proto2";
+
+package transport.test;
 
 message TestMessage {
     required string test = 1;

lib/timeval.h

@@ -4,7 +4,9 @@
  * timeval.h:
  *   utility functions for manipulating timevals
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * Copyright 2013-2015 Irene Zhang <iyzhang@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
  *
  * Permission is hereby granted, free of charge, to any person

lib/transport.cc

@@ -4,7 +4,9 @@
  * transport.cc:
  *   message-passing network interface; common definitions
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * 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
@@ -28,8 +30,8 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/assert.h"
-#include "paxos-lib/lib/transport.h"
+#include "lib/assert.h"
+#include "lib/transport.h"
 
 TransportReceiver::~TransportReceiver()
 {

lib/transport.h

@@ -31,7 +31,7 @@
 #ifndef _LIB_TRANSPORT_H_
 #define _LIB_TRANSPORT_H_
 
-#include "paxos-lib/lib/configuration.h"
+#include "lib/configuration.h"
 
 #include <google/protobuf/message.h>
 #include <functional>
@@ -74,7 +74,7 @@ protected:
 public:
     virtual ~Transport() {}
     virtual void Register(TransportReceiver *receiver,
-                          const specpaxos::Configuration &config,
+                          const transport::Configuration &config,
                           int replicaIdx) = 0;
     virtual bool SendMessage(TransportReceiver *src, const TransportAddress &dst,
                              const Message &m) = 0;

lib/transportcommon.h

@@ -4,7 +4,9 @@
  * transport-common.h:
  *   template support for implementing transports
  *
- * Copyright 2013 Dan R. K. Ports  <drkp@cs.washington.edu>
+ * 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
@@ -31,9 +33,9 @@
 #ifndef _LIB_TRANSPORTCOMMON_H_
 #define _LIB_TRANSPORTCOMMON_H_
 
-#include "paxos-lib/lib/assert.h"
-#include "paxos-lib/lib/configuration.h"
-#include "paxos-lib/lib/transport.h"
+#include "lib/assert.h"
+#include "lib/configuration.h"
+#include "lib/transport.h"
 
 #include <map>
 #include <unordered_map>
@@ -68,7 +70,7 @@ public:
     SendMessageToReplica(TransportReceiver *src, int replicaIdx,
                          const Message &m)
     {
-        const specpaxos::Configuration *cfg = configurations[src];
+        const transport::Configuration *cfg = configurations[src];
         ASSERT(cfg != NULL);
 
         if (!replicaAddressesInitialized) {
@@ -84,7 +86,7 @@ public:
     virtual bool
     SendMessageToAll(TransportReceiver *src, const Message &m)
     {
-        const specpaxos::Configuration *cfg = configurations[src];
+        const transport::Configuration *cfg = configurations[src];
         ASSERT(cfg != NULL);
 
         if (!replicaAddressesInitialized) {
@@ -115,25 +117,25 @@ protected:
                                      const ADDR &dst,
                                      const Message &m,
                                      bool multicast = false) = 0;
-    virtual ADDR LookupAddress(const specpaxos::Configuration &cfg,
+    virtual ADDR LookupAddress(const transport::Configuration &cfg,
                                int replicaIdx) = 0;
     virtual const ADDR *
-    LookupMulticastAddress(const specpaxos::Configuration *cfg) = 0;
+    LookupMulticastAddress(const transport::Configuration *cfg) = 0;
 
-    std::unordered_map<specpaxos::Configuration,
-                       specpaxos::Configuration *> canonicalConfigs;
+    std::unordered_map<transport::Configuration,
+                       transport::Configuration *> canonicalConfigs;
     std::map<TransportReceiver *,
-             specpaxos::Configuration *> configurations;
-    std::map<const specpaxos::Configuration *,
+             transport::Configuration *> configurations;
+    std::map<const transport::Configuration *,
              std::map<int, ADDR> > replicaAddresses;
-    std::map<const specpaxos::Configuration *,
+    std::map<const transport::Configuration *,
              std::map<int, TransportReceiver *> > replicaReceivers;
-    std::map<const specpaxos::Configuration *, ADDR> multicastAddresses;
+    std::map<const transport::Configuration *, ADDR> multicastAddresses;
     bool replicaAddressesInitialized;
 
-    virtual specpaxos::Configuration *
+    virtual transport::Configuration *
     RegisterConfiguration(TransportReceiver *receiver,
-                          const specpaxos::Configuration &config,
+                          const transport::Configuration &config,
                           int replicaIdx)
     {
         ASSERT(receiver != NULL);
@@ -142,10 +144,10 @@ protected:
         // pointer to the canonical copy; if not, create one. This
         // allows us to use that pointer as a key in various
         // structures. 
-        specpaxos::Configuration *canonical
+        transport::Configuration *canonical
             = canonicalConfigs[config];
         if (canonical == NULL) {
-            canonical = new specpaxos::Configuration(config);
+            canonical = new transport::Configuration(config);
             canonicalConfigs[config] = canonical;
         }
 
@@ -175,7 +177,7 @@ protected:
         // For every configuration, look up all addresses and cache
         // them.
         for (auto &kv : canonicalConfigs) {
-            specpaxos::Configuration *cfg = kv.second;
+            transport::Configuration *cfg = kv.second;
 
             for (int i = 0; i < cfg->n; i++) {
                 const ADDR addr = LookupAddress(*cfg, i);

lib/udptransport.cc

@@ -29,15 +29,16 @@
  *
  **********************************************************************/
 
-#include "paxos-lib/lib/assert.h"
-#include "paxos-lib/lib/configuration.h"
-#include "paxos-lib/lib/message.h"
-#include "paxos-lib/lib/udptransport.h"
+#include "lib/assert.h"
+#include "lib/configuration.h"
+#include "lib/message.h"
+#include "lib/udptransport.h"
 
 #include <google/protobuf/message.h>
 #include <event2/event.h>
 #include <event2/thread.h>
 
+#include <memory>
 #include <random>
 
 #include <arpa/inet.h>
@@ -83,7 +84,7 @@ bool operator<(const UDPTransportAddress &a, const UDPTransportAddress &b)
 }
 
 UDPTransportAddress
-UDPTransport::LookupAddress(const specpaxos::ReplicaAddress &addr)
+UDPTransport::LookupAddress(const transport::ReplicaAddress &addr)
 {
     int res;
     struct addrinfo hints;
@@ -106,15 +107,15 @@ UDPTransport::LookupAddress(const specpaxos::ReplicaAddress &addr)
 }
 
 UDPTransportAddress
-UDPTransport::LookupAddress(const specpaxos::Configuration &config,
+UDPTransport::LookupAddress(const transport::Configuration &config,
                             int idx)
 {
-    const specpaxos::ReplicaAddress &addr = config.replica(idx);
+    const transport::ReplicaAddress &addr = config.replica(idx);
     return LookupAddress(addr);
 }
 
 const UDPTransportAddress *
-UDPTransport::LookupMulticastAddress(const specpaxos::Configuration
+UDPTransport::LookupMulticastAddress(const transport::Configuration
                                      *config)
 {
     if (!config->multicast()) {
@@ -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);
+        }
     }
 }
 
@@ -232,7 +228,7 @@ UDPTransport::~UDPTransport()
 }
 
 void
-UDPTransport::ListenOnMulticastPort(const specpaxos::Configuration
+UDPTransport::ListenOnMulticastPort(const transport::Configuration
                                     *canonicalConfig)
 {
     if (!canonicalConfig->multicast()) {
@@ -298,13 +294,13 @@ UDPTransport::ListenOnMulticastPort(const specpaxos::Configuration
 
 void
 UDPTransport::Register(TransportReceiver *receiver,
-                       const specpaxos::Configuration &config,
+                       const transport::Configuration &config,
                        int replicaIdx)
 {
     ASSERT(replicaIdx < config.n);
     struct sockaddr_in sin;
 
-    const specpaxos::Configuration *canonicalConfig =
+    const transport::Configuration *canonicalConfig =
         RegisterConfiguration(receiver, config, replicaIdx);
 
     // Create socket
@@ -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
@@ -624,7 +618,7 @@ UDPTransport::OnReadable(int fd)
             // If so, deliver the message to all replicas for that
             // config, *except* if that replica was the sender of the
             // message.
-            const specpaxos::Configuration *cfg = it->second;
+            const transport::Configuration *cfg = it->second;
             for (auto &kv : replicaReceivers[cfg]) {
                 TransportReceiver *receiver = kv.second;
                 const UDPTransportAddress &raddr = 

lib/udptransport.h

@@ -32,9 +32,9 @@
 #ifndef _LIB_UDPTRANSPORT_H_
 #define _LIB_UDPTRANSPORT_H_
 
-#include "paxos-lib/lib/configuration.h"
-#include "paxos-lib/lib/transport.h"
-#include "paxos-lib/lib/transportcommon.h"
+#include "lib/configuration.h"
+#include "lib/transport.h"
+#include "lib/transportcommon.h"
 
 #include <event2/event.h>
 
@@ -66,10 +66,10 @@ 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 specpaxos::Configuration &config,
+                  const transport::Configuration &config,
                   int replicaIdx);
     void Run();
     void Stop();
@@ -106,8 +106,8 @@ private:
     std::vector<event *> signalEvents;
     std::map<int, TransportReceiver*> receivers; // fd -> receiver
     std::map<TransportReceiver*, int> fds; // receiver -> fd
-    std::map<const specpaxos::Configuration *, int> multicastFds;
-    std::map<int, const specpaxos::Configuration *> multicastConfigs;
+    std::map<const transport::Configuration *, int> multicastFds;
+    std::map<int, const transport::Configuration *> multicastConfigs;
     int lastTimerId;
     std::map<int, UDPTransportTimerInfo *> timers;
     uint64_t lastFragMsgId;
@@ -122,13 +122,13 @@ private:
                              const UDPTransportAddress &dst,
                              const Message &m, bool multicast = false);
     UDPTransportAddress
-    LookupAddress(const specpaxos::ReplicaAddress &addr);
+    LookupAddress(const transport::ReplicaAddress &addr);
     UDPTransportAddress
-    LookupAddress(const specpaxos::Configuration &cfg,
+    LookupAddress(const transport::Configuration &cfg,
                   int replicaIdx);
     const UDPTransportAddress *
-    LookupMulticastAddress(const specpaxos::Configuration *cfg);
-    void ListenOnMulticastPort(const specpaxos::Configuration
+    LookupMulticastAddress(const transport::Configuration *cfg);
+    void ListenOnMulticastPort(const transport::Configuration
                                *canonicalConfig);
     void OnReadable(int fd);
     void OnTimer(UDPTransportTimerInfo *info);

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;
+}