added tests

Makefile.am

@@ -7,7 +7,8 @@ AM_CPPFLAGS = -I${srcdir}/src
 
 bin_PROGRAMS = bin/sdskv-server-daemon \
 	       bin/sdskv-shutdown \
-	       test/sdskv-client-test
+	       test/sdskv-open-test \
+	       test/sdskv-put-test
 
 bin_sdskv_server_daemon_SOURCES = src/sdskv-server-daemon.c
 bin_sdskv_server_daemon_DEPENDENCIES = lib/libsdskv-server.la
@@ -76,16 +77,22 @@ noinst_HEADERS = src/datastore/datastore.h \
 
 lib_libkvgroupserver_la_SOURCES = src/kvgroup-server.cc
 
-check_PROGRAMS = test/sdskv-client-test
+check_PROGRAMS = test/sdskv-open-test
 
-TESTS = test/basic.sh
+TESTS = test/basic.sh \
+	test/open-test.sh \
+       	test/put-test.sh
 
 TESTS_ENVIRONMENT = TIMEOUT="$(TIMEOUT)" \
 		    MKTEMP="$(MKTEMP)"
 
-test_sdskv_client_test_SOURCES = test/sdskv-client-test.cc
-test_sdskv_client_test_DEPENDENCIES = lib/libsdskv-client.la
-test_sdskv_client_test_LDFLAGS = -Llib -lsdskv-client
+test_sdskv_open_test_SOURCES = test/sdskv-open-test.cc
+test_sdskv_open_test_DEPENDENCIES = lib/libsdskv-client.la
+test_sdskv_open_test_LDFLAGS = -Llib -lsdskv-client
+
+test_sdskv_put_test_SOURCES = test/sdskv-put-test.cc
+test_sdskv_put_test_DEPENDENCIES = lib/libsdskv-client.la
+test_sdskv_put_test_LDFLAGS = -Llib -lsdskv-client
 
 #############################################################
 ## tests bellow correspond to old tests (see old-test folder)

src/datastore/bwtree_datastore.cc

@@ -20,7 +20,9 @@ BwTreeDataStore::BwTreeDataStore(Duplicates duplicates, bool eraseOnGet, bool de
 
 BwTreeDataStore::~BwTreeDataStore() {
   // deleting BwTree can cause core dump
-  delete _tree;
+#if 0 // letting leak, for now
+    delete _tree;
+#endif
 };
 
 void BwTreeDataStore::createDatabase(std::string db_name) {

src/datastore/datastore_factory.h

 #ifndef datastore_factory_h
 #define datastore_factory_h
+#include <string>
 
 #ifdef SDSKV
 #include "sdskv-common.h"
@@ -22,25 +23,31 @@
 
 class datastore_factory {
 
-    static AbstractDataStore* create_bwtree_datastore() {
+    static AbstractDataStore* create_bwtree_datastore(const std::string& name) {
 #ifdef USE_BWTREE
-        return new BwTreeDataStore();
+        auto db = new BwTreeDataStore();
+        db->createDatabase(name);
+        return db;
 #else
         return nullptr;
 #endif
     }
 
-    static AbstractDataStore* create_berkeleydb_datastore() {
+    static AbstractDataStore* create_berkeleydb_datastore(const std::string& name) {
 #ifdef USE_BDB
-        return new BerkeleyDBDataStore();
+        auto db = new BerkeleyDBDataStore();
+        db->createDatabase(name);
+        return db;
 #else
         return nullptr;
 #endif
     }
 
-    static AbstractDataStore* create_leveldb_datastore() {
+    static AbstractDataStore* create_leveldb_datastore(const std::string& name) {
 #ifdef USE_LEVELDB
-        return new LevelDBDataStore();
+        auto db = new LevelDBDataStore();
+        db->createDatabase(name);
+        return db;
 #else
         return nullptr;
 #endif
@@ -49,18 +56,18 @@ class datastore_factory {
     public:
 
 #ifdef SDSKV
-    static AbstractDataStore* create_datastore(sdskv_db_type_t type)
+    static AbstractDataStore* create_datastore(sdskv_db_type_t type, const std::string& name)
 #else
-    static AbstractDataStore* create_datastore(kv_db_type_t type)
+    static AbstractDataStore* create_datastore(kv_db_type_t type, const std::string& name="db")
 #endif
     {
         switch(type) {
             case KVDB_BWTREE:
-                return create_bwtree_datastore();
+                return create_bwtree_datastore(name);
             case KVDB_LEVELDB:
-                return create_berkeleydb_datastore();
+                return create_berkeleydb_datastore(name);
             case KVDB_BERKELEYDB:
-                return create_leveldb_datastore();
+                return create_leveldb_datastore(name);
         }
         return nullptr;
     };

src/sdskv-client.c

@@ -164,6 +164,8 @@ int sdskv_open(
         return -1;
     }
 
+    in.name = (char*)db_name;
+
     hret = margo_forward(handle, &in);
     if(hret != HG_SUCCESS) {
         margo_destroy(handle);
@@ -213,23 +215,29 @@ int sdskv_put(sdskv_provider_handle_t provider,
                 provider->addr,
                 provider->client->sdskv_put_id,
                 &handle);
-        if(hret != HG_SUCCESS) return -1;
+        if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_create() failed in sdskv_put()\n");
+            return -1;
+        }
 
         hret = margo_set_target_id(handle, provider->mplex_id);
 
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_set_target_id() failed in sdskv_put()\n");
             margo_destroy(handle);
             return -1;
         }
 
         hret = margo_forward(handle, &in);
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_forward() failed in sdskv_put()\n");
             margo_destroy(handle);
             return -1;
         }
 
         hret = margo_get_output(handle, &out);
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_get_output() failed in sdskv_put()\n");
             margo_destroy(handle);
             return -1;
         }
@@ -250,7 +258,10 @@ int sdskv_put(sdskv_provider_handle_t provider,
 
         hret = margo_bulk_create(provider->client->mid, 1, (void**)(&value), &in.bulk.vsize,
                                 HG_BULK_READ_ONLY, &in.bulk.handle);
-        if(ret != HG_SUCCESS) return -1;
+        if(ret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_bulk_create() failed in sdskv_put()\n");
+            return -1;
+        }
 
         /* create handle */
         hret = margo_create(
@@ -259,6 +270,7 @@ int sdskv_put(sdskv_provider_handle_t provider,
                 provider->client->sdskv_bulk_put_id,
                 &handle);
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_create() failed in sdskv_put()\n");
             margo_bulk_free(in.bulk.handle);
             return -1;
         }
@@ -266,6 +278,7 @@ int sdskv_put(sdskv_provider_handle_t provider,
         hret = margo_set_target_id(handle, provider->mplex_id);
 
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_set_target_id() failed in sdskv_put()\n");
             margo_bulk_free(in.bulk.handle);
             margo_destroy(handle);
             return -1;
@@ -273,6 +286,7 @@ int sdskv_put(sdskv_provider_handle_t provider,
 
         hret = margo_forward(handle, &in);
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_forward() failed in sdskv_put()\n");
             margo_bulk_free(in.bulk.handle);
             margo_destroy(handle);
             return -1;
@@ -280,6 +294,7 @@ int sdskv_put(sdskv_provider_handle_t provider,
 
         hret = margo_get_output(handle, &out);
         if(hret != HG_SUCCESS) {
+            fprintf(stderr,"[SDSKV] margo_get_output() failed in sdskv_put()\n");
             margo_bulk_free(in.bulk.handle);
             margo_destroy(handle);
             return -1;

src/sdskv-server-daemon.c

@@ -36,10 +36,11 @@ static void usage(int argc, char **argv)
     return;
 }
 
-static sdskv_db_type_t parse_db_type(const char* db_fullname) {
-    const char* column = strstr(db_fullname, ":");
+static sdskv_db_type_t parse_db_type(char* db_fullname) {
+    char* column = strstr(db_fullname, ":");
     if(column == NULL) return KVDB_BWTREE;
-    const char* db_type = column + 1;
+    *column = '\0';
+    char* db_type = column + 1;
     if(strcmp(db_type, "bwt") == 0) {
         return KVDB_BWTREE;
     } else if(strcmp(db_type, "bdb") == 0) {
@@ -185,7 +186,7 @@ int main(int argc, char **argv)
                 return(-1);
             }
 
-            printf("Provider %d managing new database at multiplex id %d\n", i, i+1);
+            printf("Provider %d managing database \"%s\" at multiplex id %d\n", i, opts.db_names[i], i+1);
         }
 
     } else {
@@ -215,7 +216,7 @@ int main(int argc, char **argv)
                 return(-1);
             }
 
-            printf("Provider 0 managing new database at multiplex id %d\n", 1);
+            printf("Provider 0 managing database \"%s\" at multiplex id %d\n", opts.db_names[i] , 1);
         }
     }
 

src/sdskv-server.cc

@@ -5,6 +5,7 @@
  */
 
 #include <map>
+#include <iostream>
 #include <unordered_map>
 #define SDSKV
 #include "datastore/datastore_factory.h"
@@ -98,7 +99,7 @@ extern "C" int sdskv_provider_add_database(
         sdskv_db_type_t db_type,
         sdskv_database_id_t* db_id)
 {
-    auto db = datastore_factory::create_datastore(db_type);
+    auto db = datastore_factory::create_datastore(db_type, std::string(db_name));
     if(db == nullptr) return -1;
     sdskv_database_id_t id = (sdskv_database_id_t)(db);
 
@@ -201,7 +202,7 @@ static void sdskv_put_ult(hg_handle_t handle)
     if(it->second->put(kdata, vdata)) {
         out.ret = 0;
     } else {
-        out.ret = 1;
+        out.ret = -1;
     }
 
     margo_respond(handle, &out);

test/open-test.sh

+#!/bin/bash -x
+
+if [ -z $srcdir ]; then
+    echo srcdir variable not set.
+    exit 1
+fi
+source $srcdir/test/test-util.sh
+
+# start a server with 2 second wait,
+# 20s timeout, and my_test_db as database
+test_start_server 2 20 my_test_db
+
+sleep 1
+
+#####################
+
+# tear down
+run_to 10 test/sdskv-open-test $svr_addr 1 my_test_db
+if [ $? -ne 0 ]; then
+    wait
+    exit 1
+fi
+
+wait
+
+echo cleaning up $TMPBASE
+rm -rf $TMPBASE
+
+exit 0

test/put-test.sh

+#!/bin/bash -x
+
+if [ -z $srcdir ]; then
+    echo srcdir variable not set.
+    exit 1
+fi
+source $srcdir/test/test-util.sh
+
+# start a server with 2 second wait,
+# 20s timeout, and my_test_db as database
+test_start_server 2 20 my_test_db
+
+sleep 1
+
+#####################
+
+run_to 20 test/sdskv-put-test $svr_addr 1 my_test_db 100
+if [ $? -ne 0 ]; then
+    wait
+    exit 1
+fi
+
+wait
+
+echo cleaning up $TMPBASE
+rm -rf $TMPBASE
+
+exit 0

test/sdskv-open-test.cc

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include <margo.h>
+#include <string>
+
+#include "sdskv-client.h"
+
+int main(int argc, char *argv[])
+{
+    char cli_addr_prefix[64] = {0};
+    char *sdskv_svr_addr_str;
+    char *db_name;
+    margo_instance_id mid;
+    hg_addr_t svr_addr;
+    uint8_t mplex_id;
+    sdskv_client_t kvcl;
+    sdskv_provider_handle_t kvph;
+    hg_return_t hret;
+    int ret;
+
+    if(argc != 4)
+    {
+        fprintf(stderr, "Usage: %s <sdskv_server_addr> <mplex_id> <db_name>\n", argv[0]);
+        fprintf(stderr, "  Example: %s tcp://localhost:1234 1 foo\n", argv[0]);
+        return(-1);
+    }
+    sdskv_svr_addr_str = argv[1];
+    mplex_id          = atoi(argv[2]);
+    db_name           = argv[3];
+
+    /* initialize Margo using the transport portion of the server
+     * address (i.e., the part before the first : character if present)
+     */
+    for(unsigned i=0; (i<63 && sdskv_svr_addr_str[i] != '\0' && sdskv_svr_addr_str[i] != ':'); i++)
+        cli_addr_prefix[i] = sdskv_svr_addr_str[i];
+
+    /* start margo */
+    mid = margo_init(cli_addr_prefix, MARGO_SERVER_MODE, 0, 0);
+    if(mid == MARGO_INSTANCE_NULL)
+    {
+        fprintf(stderr, "Error: margo_init()\n");
+        return(-1);
+    }
+
+    ret = sdskv_client_init(mid, &kvcl);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: sdskv_client_init()\n");
+        margo_finalize(mid);
+        return -1;
+    }
+
+    /* look up the SDSKV server address */
+    hret = margo_addr_lookup(mid, sdskv_svr_addr_str, &svr_addr);
+    if(hret != HG_SUCCESS)
+    {
+        fprintf(stderr, "Error: margo_addr_lookup()\n");
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* create a SDSKV provider handle */
+    ret = sdskv_provider_handle_create(kvcl, svr_addr, mplex_id, &kvph);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: sdskv_provider_handle_create()\n");
+        sdskv_shutdown_service(kvcl, svr_addr);
+        margo_addr_free(mid, svr_addr);
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* open the database */
+    sdskv_database_id_t id;
+    ret = sdskv_open(kvph, db_name, &id);
+    if(ret == 0) {
+        printf("Successfuly open database %s, id is %ld\n", db_name, id);
+    } else {
+        fprintf(stderr, "Error: could not open database %s\n", db_name);
+        sdskv_provider_handle_release(kvph);
+        sdskv_shutdown_service(kvcl, svr_addr);
+        margo_addr_free(mid, svr_addr);
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* shutdown the server */
+    ret = sdskv_shutdown_service(kvcl, svr_addr);
+
+    /**** cleanup ****/
+    sdskv_provider_handle_release(kvph);
+    margo_addr_free(mid, svr_addr);
+    sdskv_client_finalize(kvcl);
+    margo_finalize(mid);
+    return(ret);
+}
+

test/sdskv-put-test.cc

+/*
+ * (C) 2015 The University of Chicago
+ * 
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include <margo.h>
+#include <string>
+#include <vector>
+
+#include "sdskv-client.h"
+
+static std::string gen_random_string(size_t len);
+
+int main(int argc, char *argv[])
+{
+    char cli_addr_prefix[64] = {0};
+    char *sdskv_svr_addr_str;
+    char *db_name;
+    margo_instance_id mid;
+    hg_addr_t svr_addr;
+    uint8_t mplex_id;
+    uint32_t num_keys;
+    sdskv_client_t kvcl;
+    sdskv_provider_handle_t kvph;
+    hg_return_t hret;
+    int ret;
+
+    if(argc != 5)
+    {
+        fprintf(stderr, "Usage: %s <sdskv_server_addr> <mplex_id> <db_name> <num_keys>\n", argv[0]);
+        fprintf(stderr, "  Example: %s tcp://localhost:1234 1 foo 1000\n", argv[0]);
+        return(-1);
+    }
+    sdskv_svr_addr_str = argv[1];
+    mplex_id          = atoi(argv[2]);
+    db_name           = argv[3];
+    num_keys          = atoi(argv[4]);
+
+    /* initialize Margo using the transport portion of the server
+     * address (i.e., the part before the first : character if present)
+     */
+    for(unsigned i=0; (i<63 && sdskv_svr_addr_str[i] != '\0' && sdskv_svr_addr_str[i] != ':'); i++)
+        cli_addr_prefix[i] = sdskv_svr_addr_str[i];
+
+    /* start margo */
+    mid = margo_init(cli_addr_prefix, MARGO_SERVER_MODE, 0, 0);
+    if(mid == MARGO_INSTANCE_NULL)
+    {
+        fprintf(stderr, "Error: margo_init()\n");
+        return(-1);
+    }
+
+    ret = sdskv_client_init(mid, &kvcl);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: sdskv_client_init()\n");
+        margo_finalize(mid);
+        return -1;
+    }
+
+    /* look up the SDSKV server address */
+    hret = margo_addr_lookup(mid, sdskv_svr_addr_str, &svr_addr);
+    if(hret != HG_SUCCESS)
+    {
+        fprintf(stderr, "Error: margo_addr_lookup()\n");
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* create a SDSKV provider handle */
+    ret = sdskv_provider_handle_create(kvcl, svr_addr, mplex_id, &kvph);
+    if(ret != 0)
+    {
+        fprintf(stderr, "Error: sdskv_provider_handle_create()\n");
+        margo_addr_free(mid, svr_addr);
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* open the database */
+    sdskv_database_id_t db_id;
+    ret = sdskv_open(kvph, db_name, &db_id);
+    if(ret == 0) {
+        printf("Successfuly open database %s, id is %ld\n", db_name, db_id);
+    } else {
+        fprintf(stderr, "Error: could not open database %s\n", db_name);
+        sdskv_provider_handle_release(kvph);
+        margo_addr_free(mid, svr_addr);
+        sdskv_client_finalize(kvcl);
+        margo_finalize(mid);
+        return(-1);
+    }
+
+    /* **** put keys ***** */
+    std::vector<std::string> keys;
+    std::vector<std::string> data;
+    for(unsigned i=0; i < num_keys; i++) {
+        auto k = gen_random_string(16);
+        // half of the entries will be put using bulk
+        auto v = gen_random_string(i*8000/num_keys);
+        ret = sdskv_put(kvph, db_id,
+                (const void *)k.data(), k.size()+1,
+                (const void *)v.data(), v.size()+1);
+        if(ret != 0) {
+            fprintf(stderr, "Error: sdskv_put() failed (iteration %d)\n", i);
+            sdskv_shutdown_service(kvcl, svr_addr);
+            sdskv_provider_handle_release(kvph);
+            margo_addr_free(mid, svr_addr);
+            sdskv_client_finalize(kvcl);
+            margo_finalize(mid);
+            return -1;
+        }
+    }
+
+    /* shutdown the server */
+    ret = sdskv_shutdown_service(kvcl, svr_addr);
+
+    /**** cleanup ****/
+    sdskv_provider_handle_release(kvph);
+    margo_addr_free(mid, svr_addr);
+    sdskv_client_finalize(kvcl);
+    margo_finalize(mid);
+    return(ret);
+}
+
+static std::string gen_random_string(size_t len) {
+    static const char alphanum[] =
+                "0123456789"
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+                "abcdefghijklmnopqrstuvwxyz";
+    std::string s(len, ' ');
+    for (unsigned i = 0; i < len; ++i) {
+        s[i] = alphanum[rand() % (sizeof(alphanum) - 1)];
+    }
+    return s;
+}

test/test-util.sh

@@ -35,7 +35,7 @@ function test_start_server ()
     startwait=${1:-15}
     maxtime=${2:-120}
 
-    run_to ${maxtime} src/sdskv-server-daemon -f $TMPBASE/sdskv.addr na+sm ${@:3} &
+    run_to ${maxtime} bin/sdskv-server-daemon -f $TMPBASE/sdskv.addr na+sm ${@:3} &
     # wait for server to start
     sleep ${startwait}