implement list_keys "starting key" feature

Honors "starting key" in BerkelyDB and LevelDB backends.  Note that
we are following RADOS semantics for starting key and excluding it from
returned list of keys.

closes #9

src/datastore/berkeleydb_datastore.cc

@@ -99,7 +99,7 @@ void BerkeleyDBDataStore::createDatabase(std::string db_name) {
       status = _dbm->open(NULL, // txn pointer
 			  NULL, // NULL for in-memory DB
 			  NULL, // logical DB name
-			  DB_HASH, // DB type (e.g. BTREE, HASH)
+			  DB_BTREE, // DB type (e.g. BTREE, HASH)
 			  flags,
 			  0);
       if (status == 0) {
@@ -111,7 +111,7 @@ void BerkeleyDBDataStore::createDatabase(std::string db_name) {
       status = _dbm->open(NULL, // txn pointer
 			  dbname.c_str(), // file name
 			  NULL, // logical DB name
-			  DB_HASH, // DB type (e.g. BTREE, HASH)
+			  DB_BTREE, // DB type (e.g. BTREE, HASH)
 			  flags,
 			  0);
     }
@@ -219,15 +219,32 @@ std::vector<ds_bulk_t> BerkeleyDBDataStore::list_keys(const ds_bulk_t &start, si
     std::vector<ds_bulk_t> keys;
     Dbc * cursorp;
     Dbt key, data;
+    int ret;
     _dbm->cursor(NULL, &cursorp, 0);
-    for (size_t i=0; i< count; i++) {
-        int ret = cursorp->get(&key, &data, DB_NEXT);
-        if (ret !=0 ) break;
 
-        ds_bulk_t k(key.get_size() );
-        memcpy(k.data(), key.get_data(), key.get_size() );
-        /* I hope this is a deep copy! */
-        keys.push_back(std::move(k));
+    /* 'start' is like RADOS: not inclusive  */
+    if (start.size()) {
+	key.set_size(start.size());
+	key.set_data((void *)start.data());
+	ret = cursorp->get(&key, &data, DB_SET_RANGE);
+	if (ret != 0) {
+	    cursorp->close();
+	    return keys;
+	}
+	ds_bulk_t k(key.get_size() );
+	memcpy(k.data(), key.get_data(), key.get_size() );
+	/* SET_RANGE will return the smallest key greater than or equal to the
+	 * requested key, but we want strictly greater than */
+	if (k != start)
+	    keys.push_back(std::move(k));
+    }
+    while (keys.size() < count) {
+	ret = cursorp->get(&key, &data, DB_NEXT);
+	if (ret !=0 ) break;
+
+	ds_bulk_t k(key.get_size() );
+	memcpy(k.data(), key.get_data(), key.get_size() );
+	keys.push_back(std::move(k));
     }
     cursorp->close();
     return keys;

src/datastore/leveldb_datastore.cc

@@ -132,12 +132,24 @@ std::vector<ds_bulk_t> LevelDBDataStore::list_keys(const ds_bulk_t &start, size_
     std::vector<ds_bulk_t> keys;
 
     leveldb::Iterator *it = _dbm->NewIterator(leveldb::ReadOptions());
-    size_t i=0;
-    for (it->SeekToFirst(); it->Valid(); it->Next() ) {
+    leveldb::Slice start_slice(start.data(), start.size());
+
+    if (start.size() > 0) {
+	it->Seek(start_slice);
+	/* we treat 'start' the way RADOS treats it: excluding it from returned
+	 * keys. LevelDB treats start inclusively, so skip over it if we found
+	 * an exact match */
+	if ( start.size() == it->key().size() &&
+		(memcmp(it->key().data(), start.data(), start.size()) == 0))
+	    it->Next();
+    } else {
+	it->SeekToFirst();
+    }
+    /* note: iterator initialized above, not in for loop */
+    for (; it->Valid() && keys.size() < count; it->Next() ) {
         ds_bulk_t k(it->key().size());
         memcpy(k.data(), it->key().data(), it->key().size() );
-        keys.push_back(k);
-        if (i++ > count) break;
+        keys.push_back(std::move(k));
     }
     delete it;
     return keys;

src/kv-server.cc

@@ -357,8 +357,9 @@ static hg_return_t list_handler(hg_handle_t handle)
     list_in_t list_in;
     list_out_t list_out;
 
-    std::vector<char> start{};
     margo_get_input(handle, &list_in);
+    ds_bulk_t start(list_in.list_in.start_key,
+	    list_in.list_in.start_key + list_in.list_in.start_ksize);
 
     auto keys = datastore->list_keys(start, list_in.list_in.max_keys);