finish the TapirDecide function

replication/ir/replica.cc

@@ -19,8 +19,7 @@ IRReplica::IRReplica(transport::Configuration config, int myIdx,
     view(0),
     myIdx(myIdx),
     transport(transport),
-    app(app),
-    status(STATUS_NORMAL)
+    app(app)
 {
     transport->Register(this, config, myIdx);
 }

replication/ir/replica.h

@@ -46,7 +46,6 @@ private:
     Transport *transport;
 
     IRAppReplica *app;
-    ReplicaStatus status;
 
     // record for this replica
     Record record;

store/tapir/shardclient.cc

@@ -30,6 +30,7 @@
  **********************************************************************/
 
 #include "store/tapir/shardclient.h"
+#include "lib/configuration.h"
 
 namespace tapir {
 
@@ -39,14 +40,13 @@ using namespace proto;
 TapirClient::TapirClient(const string &configPath,
                        Transport *transport, uint64_t client_id, int
                        shard, int closestReplica)
-    : transport(transport), client_id(client_id), shard(shard)
+    : transport(transport), config(configStream), client_id(client_id), shard(shard)
 { 
     ifstream configStream(configPath);
     if (configStream.fail()) {
         fprintf(stderr, "unable to read configuration file: %s\n",
                 configPath.c_str());
     }
-    transport::Configuration config(configStream);
 
     client = new replication::ir::IRClient(config, transport, client_id);
 
@@ -168,6 +168,44 @@ TapirClient::Prepare(uint64_t id, const Transaction &txn,
     });
 }
 
+std::string
+TapirClient::TapirDecide(const std::set<std::string> &results)
+{
+    // If a majority say prepare_ok, 
+    int ok_count = 0;
+    uint64_t ts = 0;
+    string final_reply_str;
+    Reply final_reply;
+
+    for (string s : results) {
+        Reply reply;
+        reply.ParseFromString(s);
+
+	switch(reply.status()) {
+	case REPLY_OK:
+	    ok_count++;
+	    continue;
+	case REPLY_FAIL:
+	    return s;
+	case REPLY_RETRY:
+	    if (reply.timestamp() > ts) {
+		ts = reply.timestamp();
+	    }
+	    continue;
+	default:
+	    continue;
+	}
+    }
+
+    if (ok_count >= config.QuorumSize()) {
+	final_reply.set_status(REPLY_OK);
+    } else {
+       final_reply.set_status(REPLY_RETRY);
+    }
+    final_reply.SerializeToString(&final_reply_str);
+    return final_reply_str;
+}
+
 void
 TapirClient::Commit(uint64_t id, const Transaction &txn,
                    uint64_t timestamp, Promise *promise)

store/tapir/shardclient.h

@@ -80,6 +80,7 @@ public:
 private:
     uint64_t client_id; // Unique ID for this client.
     Transport *transport; // Transport layer.
+    transport::Configuration config;
     int shard; // which shard this client accesses
     int replica; // which replica to use for reads