Fixed test_mt and made TCPClient threadsafe

Makefile.am

@@ -1,7 +1,8 @@
 AUTOMAKE_OPTIONS = foreign serial-tests
 SUBDIRS = src tests benchmarks
 TESTS = ./tests/basic_test.py ./tests/test_mem_exhaustion.py ./tests/test_store_simple.py \
-	./tests/test_cache_manager.py ./tests/test_iterator.py ./tests/test_store.py
+	./tests/test_cache_manager.py ./tests/test_iterator.py ./tests/test_store.py \
+	./tests/test_mt.py
 
 source_dir = src/
 DEFAULT_INCLUDES = -I$(source_dir)

src/client/TCPClient.cpp

@@ -14,6 +14,7 @@
 #include "utils/utils.h"
 #include "common/Future.h"
 #include "common/Exception.h"
+#include "common/Synchronization.h"
 
 namespace cirrus {
 
@@ -103,15 +104,18 @@ cirrus::Future TCPClient::write_async(ObjectID oid, const void* data,
     auto msg_contents = message::TCPBladeMessage::CreateWrite(*builder,
                                                               oid,
                                                               data_fb_vector);
+    curr_txn_id_lock.wait();
+    const int txn_id = curr_txn_id++;
+    curr_txn_id_lock.signal();
     auto msg = message::TCPBladeMessage::CreateTCPBladeMessage(
                                         *builder,
-                                        curr_txn_id,
+                                        txn_id,
                                         0,
                                         message::TCPBladeMessage::Message_Write,
                                         msg_contents.Union());
     builder->Finish(msg);
 
-    return enqueue_message(builder);
+    return enqueue_message(builder, txn_id);
 }
 
 /**
@@ -134,15 +138,19 @@ cirrus::Future TCPClient::read_async(ObjectID oid, void* data,
     // Create and send read request
     auto msg_contents = message::TCPBladeMessage::CreateRead(*builder, oid);
 
+    curr_txn_id_lock.wait();
+    const int txn_id = curr_txn_id++;
+    curr_txn_id_lock.signal();
+
     auto msg = message::TCPBladeMessage::CreateTCPBladeMessage(
                                         *builder,
-                                        curr_txn_id,
+                                        txn_id,
                                         0,
                                         message::TCPBladeMessage::Message_Read,
                                         msg_contents.Union());
     builder->Finish(msg);
 
-    return enqueue_message(builder, data);
+    return enqueue_message(builder, txn_id, data);
 }
 
 /**
@@ -191,15 +199,19 @@ bool TCPClient::remove(ObjectID oid) {
     // Create and send removal request
     auto msg_contents = message::TCPBladeMessage::CreateRemove(*builder, oid);
 
+    curr_txn_id_lock.wait();
+    const int txn_id = curr_txn_id++;
+    curr_txn_id_lock.signal();
+
     auto msg = message::TCPBladeMessage::CreateTCPBladeMessage(
                                     *builder,
-                                    curr_txn_id,
+                                    txn_id,
                                     0,
                                     message::TCPBladeMessage::Message_Remove,
                                     msg_contents.Union());
     builder->Finish(msg);
 
-    cirrus::Future future = enqueue_message(builder);
+    cirrus::Future future = enqueue_message(builder, txn_id);
     return future.get();
 }
 
@@ -424,7 +436,7 @@ void TCPClient::process_send() {
   */
 cirrus::Future TCPClient::enqueue_message(
             std::shared_ptr<flatbuffers::FlatBufferBuilder> builder,
-            void *ptr) {
+            const int txn_id, void *ptr) {
     std::shared_ptr<struct txn_info> txn = std::make_shared<struct txn_info>();
 
     txn->mem_for_read = ptr;
@@ -433,7 +445,7 @@ cirrus::Future TCPClient::enqueue_message(
     map_lock.wait();
 
     // Add to map
-    txn_map[curr_txn_id++] = txn;
+    txn_map[txn_id] = txn;
 
     // Release lock on map
     map_lock.signal();

src/client/TCPClient.h

@@ -38,6 +38,7 @@ class TCPClient : public BladeClient {
     ssize_t send_all(int, const void*, size_t, int);
     cirrus::Future enqueue_message(
                         std::shared_ptr<flatbuffers::FlatBufferBuilder> builder,
+                        const int txn_id,
                         void *ptr = nullptr);
     void process_received();
     void process_send();
@@ -63,9 +64,12 @@ class TCPClient : public BladeClient {
             error_code = std::make_shared<cirrus::ErrorCodes>();
         }
     };
-
-    int sock = 0;  /**< fd of the socket used to communicate w/ remote store */
-    TxnID curr_txn_id = 0;  /**< next txn_id to assign */
+    /** fd of the socket used to communicate w/ remote store */
+    int sock = 0;
+    /**< next txn_id to assign */
+    TxnID curr_txn_id = 0;
+    /** Lock on the current transaction id. */
+    cirrus::SpinLock curr_txn_id_lock;
 
     /**
       * Map that allows receiver thread to map transactions to their

tests/object_store/test_mt.cpp

@@ -28,8 +28,8 @@ cirrus::ostore::FullBladeObjectStoreTempl<cirrus::Dummy<SIZE>> store(IP, PORT,
 
 /**
   * Tests that behavior is as expected when multiple threads make get and put
-  * requests to the remote store. These clients all use the same instance of
-  * the store to connect. Currently not working.
+  * requests to the remote store. These threads all use the same instance of
+  * the store to connect.
   */
 void test_mt() {
     cirrus::TimerFunction tf("connect time", true);
@@ -39,20 +39,23 @@ void test_mt() {
     std::random_device rd;
     std::mt19937 gen(rd());
     std::uniform_int_distribution<> dis(1, 10);
-
+    int start = 0;
+    int stop = 10;
     for (int i = 0; i < N_THREADS; ++i) {
-        threads[i] = new std::thread([dis, gen]() {
-            for (int i = 0; i < 100; ++i) {
+        threads[i] = new std::thread([dis, gen, start, stop]() {
+            for (int i = start; i < stop; i++) {
                 int rnd = std::rand();
                 struct cirrus::Dummy<SIZE> d(rnd);
 
-                store.put(1, d);
-                cirrus::Dummy<SIZE> d2 = store.get(1);
+                store.put(i, d);
+                cirrus::Dummy<SIZE> d2 = store.get(i);
 
                 if (d2.id != rnd)
-                    throw std::runtime_error("mismatch");
+                    throw std::runtime_error("Incorrect value returned.");
             }
         });
+        start += 10;
+        stop += 10;
     }
 
     for (int i = 0; i < N_THREADS; ++i)

tests/test_mt.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+
+import sys
+import subprocess
+import time
+import test_runner
+
+# Set name of test to run
+testPath = "./tests/object_store/test_mt"
+# Call script to run the test
+test_runner.runTest(testPath)