.gitea/workflows/test.yml

@@ -1067,6 +1067,24 @@ jobs:
           echo ""
         done
 
+  test_kv_list_bounds:
+    runs-on: ubuntu-latest
+    needs: build
+    container: ${{env.TEST_IMAGE}}:${{github.sha}}
+    steps:
+    - name: Run test
+      id: test
+      timeout-minutes: 3
+      run: /root/vitastor/tests/test_kv_list_bounds.sh
+    - name: Print logs
+      if: always() && steps.test.outcome == 'failure'
+      run: |
+        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
+          echo "-------- $i --------"
+          cat $i
+          echo ""
+        done
+
   test_splitbrain:
     runs-on: ubuntu-latest
     needs: build
@@ -2453,6 +2471,24 @@ jobs:
           echo ""
         done
 
+  test_nfs_list_bounds:
+    runs-on: ubuntu-latest
+    needs: build
+    container: ${{env.TEST_IMAGE}}:${{github.sha}}
+    steps:
+    - name: Run test
+      id: test
+      timeout-minutes: 3
+      run: /root/vitastor/tests/test_nfs_list_bounds.sh
+    - name: Print logs
+      if: always() && steps.test.outcome == 'failure'
+      run: |
+        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
+          echo "-------- $i --------"
+          cat $i
+          echo ""
+        done
+
   test_write_encrypted:
     runs-on: ubuntu-latest
     needs: build

src/kv/kv_cli.cpp

@@ -664,7 +664,7 @@ void kv_cli_t::handle_cmd(const std::vector<std::string> & cmd, std::function<vo
         kv_cli_list_t *lst = new kv_cli_list_t;
         std::string start = cmd.size() >= 2 ? cmd[1] : "";
         std::string end = cmd.size() >= 3 ? cmd[2] : "";
-        lst->handle = db->list_start(start);
+        lst->handle = db->list_start(start, end);
         lst->db = db;
         lst->format = opname == "dump" ? 1 : (opname == "dumpjson" ? 2 : 0);
         lst->cb = std::move(cb);

src/kv/kv_db.cpp

@@ -190,6 +190,8 @@ struct kv_op_t
     kv_db_t *db;
     int opcode;
     std::string key, value;
+    // for list: stop before the first key >= list_end, unless list_end is empty
+    std::string list_end;
     int res;
     bool in_next = false;
     bool next_retry_pending = false;
@@ -2021,6 +2023,12 @@ void kv_op_t::next_get()
     }
     if (kv_it != blk->data.end())
     {
+        if (list_end != "" && kv_it->first >= list_end)
+        {
+            // End of the listing range
+            finish(-ENOENT);
+            return;
+        }
         // Send this item
         assert(blk->type == KV_LEAF || blk->type == KV_LEAF_SPLIT);
         this->res = 0;
@@ -2032,6 +2040,12 @@ void kv_op_t::next_get()
     // Find next block
     else if (blk->type == KV_LEAF_SPLIT)
     {
+        if (list_end != "" && blk->right_half >= list_end)
+        {
+            // The right half is beyond the listing range
+            finish(-ENOENT);
+            return;
+        }
         // Left half finished, go to the right
         recheck_policy = KV_RECHECK_LEAF;
         key = blk->right_half;
@@ -2063,6 +2077,12 @@ void kv_op_t::next_go_up()
             // Block can be updated and key_lt may be lower. Don't rewind it
             key = blk->key_lt;
         }
+        if (list_end != "" && key >= list_end)
+        {
+            // Next keys are beyond the listing range, don't read the next block
+            finish(-ENOENT);
+            return;
+        }
         skip_equal = false;
         path.pop_back();
         cur_level--;
@@ -2194,6 +2214,11 @@ void vitastorkv_dbw_t::del(const std::string & key, std::function<void(int res)>
 }
 
 void* vitastorkv_dbw_t::list_start(const std::string & start)
+{
+    return list_start(start, "");
+}
+
+void* vitastorkv_dbw_t::list_start(const std::string & start, const std::string & end)
 {
     if (!db->inode_id || db->closing)
         return NULL;
@@ -2201,6 +2226,7 @@ void* vitastorkv_dbw_t::list_start(const std::string & start)
     op->db = db;
     op->opcode = KV_LIST;
     op->key = start;
+    op->list_end = end;
     op->callback = [](kv_op_t *){};
     op->exec();
     return op;

src/kv/vitastor_kv.h

@@ -40,6 +40,8 @@ struct __attribute__((visibility("default"))) vitastorkv_dbw_t
         std::function<bool(int res, const std::string & value)> cas_compare = NULL);
 
     void* list_start(const std::string & start);
+    // list keys >= start and < end (all keys >= start if end is empty)
+    void* list_start(const std::string & start, const std::string & end);
     void list_next(void *handle, std::function<void(int res, const std::string & key, const std::string & value)> cb);
     void list_close(void *handle);
 

src/nfs/nfs_kv.cpp

@@ -110,6 +110,16 @@ std::string kv_direntry_key(uint64_t dir_ino, const std::string & filename)
     return (char*)key + filename;
 }
 
+std::string kv_prefix_end(const std::string & prefix)
+{
+    std::string end = prefix;
+    while (end.size() && (uint8_t)end.back() == 0xff)
+        end.pop_back();
+    if (end.size())
+        end.back()++;
+    return end;
+}
+
 std::string kv_direntry_filename(const std::string & key)
 {
     // decode as: d <length> <hex dir_ino> / <filename>

src/nfs/nfs_kv.h

@@ -130,6 +130,8 @@ int kv_map_type(const std::string & type);
 fattr3 get_kv_attributes(nfs_proxy_t *proxy, uint64_t ino, json11::Json attrs);
 std::string kv_direntry_key(uint64_t dir_ino, const std::string & filename);
 std::string kv_direntry_filename(const std::string & key);
+// The first key after all keys starting with prefix, to stop a listing at the end of the prefix
+std::string kv_prefix_end(const std::string & prefix);
 std::string kv_inode_prefix_key(uint64_t ino, const char *prefix);
 std::string kv_inode_key(uint64_t ino);
 uint64_t kv_key_inode(const std::string & key, int prefix_len = 1);

src/nfs/nfs_kv_defrag.cpp

@@ -317,7 +317,7 @@ void kv_fs_defrag_all_t::run(int st)
     else if (st == 6)
         goto resume_6;
     clock_gettime(CLOCK_REALTIME, &now);
-    list_shared = proxy->db->list_start("shared");
+    list_shared = proxy->db->list_start("shared", kv_prefix_end("shared"));
     proxy->db->list_next(list_shared, [this](int res, const std::string & key, const std::string & value)
     {
         if (res == -ENOENT || key.substr(0, 6) != "shared")
@@ -482,7 +482,7 @@ void kv_fs_state_t::upgrade_db(std::function<void(int)> cb)
             return;
         }
         // Create missing shared inode index keys
-        auto list_inodes = proxy->db->list_start("i");
+        auto list_inodes = proxy->db->list_start("i", kv_prefix_end("i"));
         proxy->db->list_next(list_inodes, [=](int res, const std::string & key, const std::string & value)
         {
             if (res == -ENOENT || key.substr(0, 1) != "i" || key == "id")

src/nfs/nfs_kv_readdir.cpp

@@ -241,7 +241,7 @@ resume_2:
         st->cookieverf = ((uint64_t)lrand48() | ((uint64_t)lrand48() << 31) | ((uint64_t)lrand48() << 62));
     }
     st->getattr_cur = st->entries.size();
-    st->list_handle = st->self->parent->db->list_start(st->start);
+    st->list_handle = st->self->parent->db->list_start(st->start, kv_prefix_end(st->prefix));
     st->self->parent->db->list_next(st->list_handle, [=](int res, const std::string & key, const std::string & value)
     {
         st->res = res;

src/nfs/nfs_kv_remove.cpp

@@ -181,7 +181,7 @@ resume_3:
     if (st->is_rmdir)
     {
         // (4) Check if the directory actually is not empty
-        st->list_handle = st->self->parent->db->list_start(kv_direntry_key(st->ino, ""));
+        st->list_handle = st->self->parent->db->list_start(kv_direntry_key(st->ino, ""), kv_prefix_end(kv_direntry_key(st->ino, "")));
         st->self->parent->db->list_next(st->list_handle, [st](int res, const std::string & key, const std::string & value)
         {
             st->res = res;

src/nfs/nfs_kv_rename.cpp

@@ -138,7 +138,7 @@ resume_2:
         {
             // Check that the destination directory is empty
             st->new_dir_prefix = kv_direntry_key(st->new_direntry["ino"].uint64_value(), "");
-            st->list_handle = st->self->parent->db->list_start(st->new_dir_prefix);
+            st->list_handle = st->self->parent->db->list_start(st->new_dir_prefix, kv_prefix_end(st->new_dir_prefix));
             st->self->parent->db->list_next(st->list_handle, [st](int res, const std::string & key, const std::string & value)
             {
                 st->res = res;

tests/run_tests.sh

@@ -85,6 +85,7 @@ OLD=1 SCHEME=ec ./test_snapshot_down.sh
 
 ./test_kv_stress.sh
 IMMEDIATE_COMMIT=1 ./test_kv_stress.sh
+./test_kv_list_bounds.sh
 
 ./test_splitbrain.sh
 
@@ -184,6 +185,7 @@ TEST_NAME=old_csum_4k      OLD=1 OSD_ARGS="--data_csum_type crc32c" OFFSET_ARGS=
 ./test_nfs.sh
 ./test_nfs_unaligned_append.sh
 ./test_nfs_long_readdir.sh
+./test_nfs_list_bounds.sh
 
 ./test_write_encrypted.sh
 SCHEME=ec ./test_write_encrypted.sh

tests/test_kv_list_bounds.sh

@@ -0,0 +1,45 @@
+#!/bin/bash -ex
+# Test K/V listings with an end key: they must return exactly the keys of their range,
+# also when the range spans several blocks or ends in the middle of a block.
+
+PG_COUNT=16
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_CLI create -s 1G kvlist
+KV="build/src/kv/vitastor-kv --config_path $VITASTOR_CFG kvlist"
+
+# 1000 keys with 100-byte values span several 4 KB blocks
+VALUE=$(printf 'v%.0s' $(seq 1 100))
+(
+    for i in $(seq -f '%03g' 0 999); do
+        echo "set t/a$i $VALUE"
+    done
+    echo "set t/b0 v"
+) | $KV > ./testdata/kv_list_set.log 2>&1
+if [ "$(grep -c '^OK' ./testdata/kv_list_set.log)" -ne 1001 ]; then
+    format_error "Failed to set the keys"
+fi
+
+list_keys()
+{
+    $KV list "$@" | sed 's/ = .*//'
+}
+
+if [ "$(list_keys t/a t/b)" != "$(seq -f 't/a%03g' 0 999)" ]; then
+    format_error "list t/a t/b did not return exactly t/a000..t/a999"
+fi
+if [ "$(list_keys t/a5 t/a6)" != "$(seq -f 't/a%03g' 500 599)" ]; then
+    format_error "list t/a5 t/a6 did not return exactly t/a500..t/a599"
+fi
+if [ -n "$(list_keys t/c t/d)" ]; then
+    format_error "list t/c t/d returned keys"
+fi
+if [ "$($KV dump t/a5 t/a6 | grep -c '^set ')" -ne 100 ]; then
+    format_error "dump t/a5 t/a6 did not return 100 keys"
+fi
+# Without an end, the listing goes on past the range
+if [ "$(list_keys t/a | head -n 1001)" != "$(seq -f 't/a%03g' 0 999; echo t/b0)" ]; then
+    format_error "list t/a did not return t/a000..t/a999, then t/b0"
+fi
+
+format_green OK

tests/test_nfs_list_bounds.sh

@@ -0,0 +1,103 @@
+#!/bin/bash -ex
+# Test that RMDIR does not slow down after many K/V keys were created and deleted,
+# and that RENAME of a directory over an empty directory succeeds.
+#
+# K/V blocks emptied by deletes are never freed. The check that a directory is empty lists
+# its entries, and the listing used to go on past the end of the directory through all
+# empty blocks up to the next existing key. The keys below are created and deleted right
+# after the entries of the newest directories (dD... with 13 hex digits of inode number in
+# pool 1), so the emptied blocks are exactly where these listings ran.
+
+PG_COUNT=16
+. `dirname $0`/run_3osds.sh
+
+DIR_COUNT=50
+KEY_COUNT=10000
+
+$VITASTOR_CLI create -s 10G fsmeta
+$VITASTOR_CLI modify-pool --used-for-app fs:fsmeta testpool
+
+start_nfs()
+{
+    build/src/nfs/vitastor-nfs --config_path $VITASTOR_CFG start --fs fsmeta \
+        --portmap 0 --port 2050 --foreground 1 >>./testdata/nfs.log 2>&1 &
+    NFS_PID=$!
+    for i in $(seq 1 30); do
+        sudo mount localhost:/ ./testdata/nfs -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp && return 0
+        sleep 1
+    done
+    format_error "Failed to mount vitastor-nfs"
+}
+
+stop_nfs()
+{
+    sudo umount ./testdata/nfs/
+    kill $NFS_PID
+    wait $NFS_PID || true
+}
+
+# Time RMDIR of $DIR_COUNT empty directories, in milliseconds
+rmdir_ms()
+{
+    local dir=$MNT/$1 start
+    mkdir $dir
+    for i in $(seq 1 $DIR_COUNT); do
+        mkdir $dir/d$i
+    done
+    start=$(date +%s%N)
+    for i in $(seq 1 $DIR_COUNT); do
+        rmdir $dir/d$i
+    done
+    echo $(( ($(date +%s%N) - start) / 1000000 ))
+    rmdir $dir
+}
+
+mkdir -p testdata/nfs
+MNT=$(pwd)/testdata/nfs
+trap "sudo umount -f $MNT"' || true; kill -9 $(jobs -p)' EXIT
+
+start_nfs
+BEFORE=$(rmdir_ms before)
+stop_nfs
+
+# Create and delete keys with 400-byte values after the entries of all directories
+VALUE=$(printf 'x%.0s' $(seq 1 400))
+(
+    for i in $(seq 1 $KEY_COUNT); do
+        echo "set dF100000000000000/k$i $VALUE"
+    done
+    for i in $(seq 1 $KEY_COUNT); do
+        echo "del dF100000000000000/k$i"
+    done
+) | build/src/kv/vitastor-kv --config_path $VITASTOR_CFG fsmeta > ./testdata/kv_churn.log 2>&1
+if [ "$(grep -c '^OK' ./testdata/kv_churn.log)" -ne $((KEY_COUNT*2)) ]; then
+    format_error "Failed to create and delete $KEY_COUNT keys"
+fi
+
+start_nfs
+AFTER=$(rmdir_ms after)
+if [ $AFTER -gt $((BEFORE*2)) ]; then
+    format_error "RMDIR of $DIR_COUNT directories took $AFTER ms after creating and deleting $KEY_COUNT keys, $BEFORE ms before"
+fi
+format_green "RMDIR of $DIR_COUNT directories: $BEFORE ms before, $AFTER ms after creating and deleting $KEY_COUNT keys"
+# READDIR still returns exactly the entries of a directory
+mkdir $MNT/list
+for i in $(seq 1 20); do
+    mkdir $MNT/list/e$i
+done
+if [ "$(ls -f $MNT/list | grep -c '^e')" -ne 20 ]; then
+    format_error "READDIR returned a wrong number of entries"
+fi
+# RENAME of a directory over an empty directory succeeds (the emptiness check used to take
+# the first key after the prefix, which belongs to another directory, for an entry)
+mkdir $MNT/list/e1/x
+if ! mv -T $MNT/list/e1 $MNT/list/e2; then
+    format_error "RENAME of a directory over an empty directory failed"
+fi
+if [ ! -d $MNT/list/e2/x ]; then
+    format_error "RENAME of a directory over an empty directory lost its contents"
+fi
+stop_nfs
+
+
+format_green OK