Experiment: zero-copy TCP send

CMakeLists.txt

@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
-set(VERSION "1.2.0")
+set(VERSION "1.1.0")
 
 add_subdirectory(src)

csi/Makefile

@@ -1,4 +1,4 @@
-VERSION ?= v1.2.0
+VERSION ?= v1.1.0
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

@@ -49,7 +49,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v1.2.0
+          image: vitalif/vitastor-csi:v1.1.0
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/deploy/007-csi-provisioner.yaml

@@ -121,7 +121,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v1.2.0
+          image: vitalif/vitastor-csi:v1.1.0
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

@@ -5,7 +5,7 @@ package vitastor
 
 const (
     vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "1.2.0"
+    vitastorCSIDriverVersion = "1.1.0"
 )
 
 // Config struct fills the parameters of request or user input

debian/changelog

@@ -1,10 +1,10 @@
-vitastor (1.2.0-1) unstable; urgency=medium
+vitastor (1.1.0-1) unstable; urgency=medium
 
   * Bugfixes
 
  -- Vitaliy Filippov <vitalif@yourcmc.ru>  Fri, 03 Jun 2022 02:09:44 +0300
 
-vitastor (1.2.0-1) unstable; urgency=medium
+vitastor (1.1.0-1) unstable; urgency=medium
 
   * Implement NFS proxy
   * Add documentation

debian/vitastor.Dockerfile

@@ -35,8 +35,8 @@ RUN set -e -x; \
     mkdir -p /root/packages/vitastor-$REL; \
     rm -rf /root/packages/vitastor-$REL/*; \
     cd /root/packages/vitastor-$REL; \
-    cp -r /root/vitastor vitastor-1.2.0; \
-    cd vitastor-1.2.0; \
+    cp -r /root/vitastor vitastor-1.1.0; \
+    cd vitastor-1.1.0; \
     ln -s /root/fio-build/fio-*/ ./fio; \
     FIO=$(head -n1 fio/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     ls /usr/include/linux/raw.h || cp ./debian/raw.h /usr/include/linux/raw.h; \
@@ -49,8 +49,8 @@ RUN set -e -x; \
     rm -rf a b; \
     echo "dep:fio=$FIO" > debian/fio_version; \
     cd /root/packages/vitastor-$REL; \
-    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_1.2.0.orig.tar.xz vitastor-1.2.0; \
-    cd vitastor-1.2.0; \
+    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_1.1.0.orig.tar.xz vitastor-1.1.0; \
+    cd vitastor-1.1.0; \
     V=$(head -n1 debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     DEBFULLNAME="Vitaliy Filippov <vitalif@yourcmc.ru>" dch -D $REL -v "$V""$REL" "Rebuild for $REL"; \
     DEB_BUILD_OPTIONS=nocheck dpkg-buildpackage --jobs=auto -sa; \

docs/config/src/network.yml

@@ -30,6 +30,18 @@
     будут использоваться обычные синхронные системные вызовы send/recv. Для OSD
     это бессмысленно, так как OSD в любом случае нуждается в io_uring, но, в
     принципе, это может применяться для клиентов со старыми версиями ядра.
+- name: use_zerocopy_send
+  type: bool
+  default: false
+  info: |
+    If true, OSDs and clients will attempt to use TCP zero-copy send
+    (MSG_ZEROCOPY) for big buffers. It's recommended to raise net.ipv4.tcp_wmem
+    and net.core.wmem_max sysctls when using this mode.
+  info_ru: |
+    Если установлено в true, то OSD и клиенты будут стараться использовать
+    TCP-отправку без копирования (MSG_ZEROCOPY) для больших буферов данных.
+    Рекомендуется поднять значения sysctl net.ipv4.tcp_wmem и net.core.wmem_max
+    при использовании этого режима.
 - name: use_rdma
   type: bool
   default: true

docs/installation/kubernetes.en.md

@@ -17,15 +17,4 @@ and apply all `NNN-*.yaml` manifests to your Kubernetes installation:
 for i in ./???-*.yaml; do kubectl apply -f $i; done
 ```
 
-After that you'll be able to create PersistentVolumes.
-
-## Features
-
-Vitastor CSI supports:
-- Kubernetes starting with 1.20 (or 1.17 for older vitastor-csi <= 1.1.0)
-- Filesystem RWO (ReadWriteOnce) volumes. Example: [PVC](../../csi/deploy/example-pvc.yaml), [pod](../../csi/deploy/example-test-pod.yaml)
-- Raw block RWX (ReadWriteMany) volumes. Example: [PVC](../../csi/deploy/example-pvc-block.yaml), [pod](../../csi/deploy/example-test-pod-block.yaml)
-- Volume expansion
-- Volume snapshots. Example: [snapshot class](../../csi/deploy/example-snapshot-class.yaml), [snapshot](../../csi/deploy/example-snapshot.yaml), [clone](../../csi/deploy/example-snapshot-clone.yaml)
-
-Remember that to use snapshots with CSI you also have to install [Snapshot Controller and CRDs](https://kubernetes-csi.github.io/docs/snapshot-controller.html#deployment).
+After that you'll be able to create PersistentVolumes. See example in [csi/deploy/example-pvc.yaml](../../csi/deploy/example-pvc.yaml).

docs/installation/kubernetes.ru.md

@@ -17,15 +17,4 @@
 for i in ./???-*.yaml; do kubectl apply -f $i; done
 ```
 
-После этого вы сможете создавать PersistentVolume.
-
-## Возможности
-
-CSI-плагин Vitastor поддерживает:
-- Версии Kubernetes, начиная с 1.20 (или с 1.17 для более старых vitastor-csi <= 1.1.0)
-- Файловые RWO (ReadWriteOnce) тома. Пример: [PVC](../../csi/deploy/example-pvc.yaml), [под](../../csi/deploy/example-test-pod.yaml)
-- Сырые блочные RWX (ReadWriteMany) тома. Пример: [PVC](../../csi/deploy/example-pvc-block.yaml), [под](../../csi/deploy/example-test-pod-block.yaml)
-- Расширение размера томов
-- Снимки томов. Пример: [класс снимков](../../csi/deploy/example-snapshot-class.yaml), [снимок](../../csi/deploy/example-snapshot.yaml), [клон снимка](../../csi/deploy/example-snapshot-clone.yaml)
-
-Не забывайте, что для использования снимков нужно сначала установить [контроллер снимков и CRD](https://kubernetes-csi.github.io/docs/snapshot-controller.html#deployment).
+После этого вы сможете создавать PersistentVolume. Пример смотрите в файле [csi/deploy/example-pvc.yaml](../../csi/deploy/example-pvc.yaml).

mon/mon.js

@@ -65,6 +65,7 @@ const etcd_tree = {
             // client and osd
             tcp_header_buffer_size: 65536,
             use_sync_send_recv: false,
+            use_zerocopy_send: false,
             use_rdma: true,
             rdma_device: null, // for example, "rocep5s0f0"
             rdma_port_num: 1,
@@ -403,7 +404,6 @@ class Mon
         this.ws_alive = false;
         this.ws_keepalive_timer = null;
         this.on_stop_cb = () => this.on_stop(0).catch(console.error);
-        this.recheck_pgs_active = false;
     }
 
     parse_etcd_addresses(addrs)
@@ -693,27 +693,8 @@ class Mon
         });
     }
 
-    // Schedule save_last_clean() to to run after a small timeout (1s) (to not spam etcd)
-    schedule_save_last_clean()
-    {
-        if (!this.save_last_clean_timer)
-        {
-            this.save_last_clean_timer = setTimeout(() =>
-            {
-                this.save_last_clean_timer = null;
-                this.save_last_clean().catch(this.die);
-            }, this.config.mon_change_timeout || 1000);
-        }
-    }
-
     async save_last_clean()
     {
-        if (this.save_last_clean_running)
-        {
-            this.schedule_save_last_clean();
-            return;
-        }
-        this.save_last_clean_running = true;
         // last_clean_pgs is used to avoid extra data move when observing a series of changes in the cluster
         const new_clean_pgs = { items: {} };
     next_pool:
@@ -750,7 +731,6 @@ class Mon
                 value: b64(JSON.stringify(this.state.history.last_clean_pgs))
             } } ],
         }, this.etcd_start_timeout, 0);
-        this.save_last_clean_running = false;
     }
 
     get_mon_state()
@@ -1224,12 +1204,6 @@ class Mon
 
     async recheck_pgs()
     {
-        if (this.recheck_pgs_active)
-        {
-            this.schedule_recheck();
-            return;
-        }
-        this.recheck_pgs_active = true;
         // Take configuration and state, check it against the stored configuration hash
         // Recalculate PGs and save them to etcd if the configuration is changed
         // FIXME: Do not change anything if the distribution is good and random enough and no PGs are degraded
@@ -1251,7 +1225,6 @@ class Mon
                     // Pool deleted. Delete all PGs, but first stop them.
                     if (!await this.stop_all_pgs(pool_id))
                     {
-                        this.recheck_pgs_active = false;
                         this.schedule_recheck();
                         return;
                     }
@@ -1320,16 +1293,9 @@ class Mon
                         // PG count changed. Need to bring all PGs down.
                         if (!await this.stop_all_pgs(pool_id))
                         {
-                            this.recheck_pgs_active = false;
                             this.schedule_recheck();
                             return;
                         }
-                    }
-                    if (prev_pgs.length != pool_cfg.pg_count)
-                    {
-                        // Scale PG count
-                        // Do it even if old_pg_count is already equal to pool_cfg.pg_count,
-                        // because last_clean_pgs may still contain the old number of PGs
                         const new_pg_history = [];
                         PGUtil.scale_pg_count(prev_pgs, real_prev_pgs, pg_history, new_pg_history, pool_cfg.pg_count);
                         pg_history = new_pg_history;
@@ -1431,7 +1397,6 @@ class Mon
                 await this.save_pg_config(new_config_pgs);
             }
         }
-        this.recheck_pgs_active = false;
     }
 
     async save_pg_config(new_config_pgs, etcd_request = { compare: [], success: [] })
@@ -1481,6 +1446,7 @@ class Mon
     }
 
     // Schedule a recheck to run after a small timeout (1s)
+    // If already scheduled, cancel previous timer and schedule it again
     // This is required for multiple change events to trigger at most 1 recheck in 1s
     schedule_recheck()
     {

mon/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor-mon",
-  "version": "1.2.0",
+  "version": "1.1.0",
   "description": "Vitastor SDS monitor service",
   "main": "mon-main.js",
   "scripts": {

patches/cinder-vitastor.py

@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils
 
-VERSION = '1.2.0'
+VERSION = '1.1.0'
 
 LOG = logging.getLogger(__name__)
 

rpm/build-tarball.sh

@@ -24,4 +24,4 @@ rm fio
 mv fio-copy fio
 FIO=`rpm -qi fio | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Version[\s:]+(\S+)/ && print $1; /^Release[\s:]+(\S+)/ && print "-$1"; }'`
 perl -i -pe 's/(Requires:\s*fio)([^\n]+)?/$1 = '$FIO'/' $VITASTOR/rpm/vitastor-el$EL.spec
-tar --transform 's#^#vitastor-1.2.0/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-1.2.0$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-1.1.0/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-1.1.0$(rpm --eval '%dist').tar.gz *

rpm/vitastor-el7.Dockerfile

@@ -35,7 +35,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-1.2.0.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-1.1.0.el7.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el7.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.2.0
+Version:        1.1.0
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.2.0.el7.tar.gz
+Source0:        vitastor-1.1.0.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el8.Dockerfile

@@ -35,7 +35,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-1.2.0.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-1.1.0.el8.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el8.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.2.0
+Version:        1.1.0
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.2.0.el8.tar.gz
+Source0:        vitastor-1.1.0.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el9.Dockerfile

@@ -18,7 +18,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-1.2.0.el9.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-1.1.0.el9.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el9.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        1.2.0
+Version:        1.1.0
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-1.2.0.el9.tar.gz
+Source0:        vitastor-1.1.0.el9.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

src/CMakeLists.txt

@@ -16,7 +16,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()
 
-add_definitions(-DVERSION="1.2.0")
+add_definitions(-DVERSION="1.1.0")
 add_definitions(-Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -I ${CMAKE_SOURCE_DIR}/src)
 if (${WITH_ASAN})
 	add_definitions(-fsanitize=address -fno-omit-frame-pointer)

src/blockstore_flush.cpp

@@ -1372,8 +1372,7 @@ bool journal_flusher_co::trim_journal(int wait_base)
                     ? (uint32_t)JE_START_V1_SIZE : (uint32_t)JE_START_V2_SIZE),
                 .reserved = 0,
                 .journal_start = new_trim_pos,
-                .version = (uint64_t)(!bs->dsk.data_csum_type && ((journal_entry_start*)flusher->journal_superblock)->version == JOURNAL_VERSION_V1
-                    ? JOURNAL_VERSION_V1 : JOURNAL_VERSION_V2),
+                .version = JOURNAL_VERSION_V2,
                 .data_csum_type = bs->dsk.data_csum_type,
                 .csum_block_size = bs->dsk.csum_block_size,
             };

src/blockstore_init.cpp

@@ -553,7 +553,7 @@ resume_1:
         }
         if (je_start->size == JE_START_V0_SIZE ||
             (je_start->version != JOURNAL_VERSION_V1 || je_start->size != JE_START_V1_SIZE) &&
-            (je_start->version != JOURNAL_VERSION_V2 || je_start->size != JE_START_V2_SIZE && je_start->size != JE_START_V1_SIZE))
+            (je_start->version != JOURNAL_VERSION_V2 || je_start->size != JE_START_V2_SIZE))
         {
             fprintf(
                 stderr, "The code only supports journal versions 2 and 1, but it is %lu on disk."
@@ -562,8 +562,7 @@ resume_1:
             );
             exit(1);
         }
-        if (je_start->version == JOURNAL_VERSION_V1 ||
-            je_start->version == JOURNAL_VERSION_V2 && je_start->size == JE_START_V1_SIZE)
+        if (je_start->version == JOURNAL_VERSION_V1)
         {
             je_start->data_csum_type = 0;
             je_start->csum_block_size = 0;

src/blockstore_write.cpp

@@ -21,7 +21,7 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
         dyn = calloc_or_die(1, dyn_size+sizeof(int));
         *((int*)dyn) = 1;
     }
-    uint8_t *dyn_ptr = (alloc_dyn_data ? (uint8_t*)dyn+sizeof(int) : (uint8_t*)&dyn);
+    uint8_t *dyn_ptr = (uint8_t*)(alloc_dyn_data ? dyn+sizeof(int) : &dyn);
     uint64_t version = 1;
     if (dirty_db.size() > 0)
     {

src/cli.cpp

@@ -17,7 +17,7 @@
 static const char *exe_name = NULL;
 
 static const char* help_text =
-    "Vitastor command-line tool " VERSION "\n"
+    "Vitastor command-line tool\n"
     "(c) Vitaliy Filippov, 2019+ (VNPL-1.1)\n"
     "\n"
     "COMMANDS:\n"

src/cli_df.cpp

@@ -109,7 +109,7 @@ resume_1:
             }
             for (auto pg_per_pair: pg_per_osd)
             {
-                uint64_t pg_free = osd_free[pg_per_pair.first] * pool_cfg.real_pg_count / pg_per_pair.second;
+                uint64_t pg_free = osd_free[pg_per_pair.first] * pool_cfg.pg_count / pg_per_pair.second;
                 if (pool_avail > pg_free)
                 {
                     pool_avail = pg_free;
@@ -127,7 +127,6 @@ resume_1:
                 { "id", (uint64_t)pool_cfg.id },
                 { "name", pool_cfg.name },
                 { "pg_count", pool_cfg.pg_count },
-                { "real_pg_count", pool_cfg.real_pg_count },
                 { "scheme", pool_cfg.scheme == POOL_SCHEME_REPLICATED ? "replicated" : "ec" },
                 { "scheme_name", pool_cfg.scheme == POOL_SCHEME_REPLICATED
                     ? std::to_string(pool_cfg.pg_size)+"/"+std::to_string(pool_cfg.pg_minsize)
@@ -178,7 +177,7 @@ resume_1:
             { "title", "SCHEME" },
         });
         cols.push_back(json11::Json::object{
-            { "key", "pg_count_fmt" },
+            { "key", "pg_count" },
             { "title", "PGS" },
         });
         cols.push_back(json11::Json::object{
@@ -207,9 +206,6 @@ resume_1:
             double raw_to = kv.second["raw_to_usable"].number_value();
             if (raw_to < 0.000001 && raw_to > -0.000001)
                 raw_to = 1;
-            kv.second["pg_count_fmt"] = kv.second["real_pg_count"] == kv.second["pg_count"]
-                ? kv.second["real_pg_count"].as_string()
-                : kv.second["real_pg_count"].as_string()+"->"+kv.second["pg_count"].as_string();
             kv.second["total_fmt"] = format_size(kv.second["total_raw"].uint64_value() / raw_to);
             kv.second["used_fmt"] = format_size(kv.second["used_raw"].uint64_value() / raw_to);
             kv.second["max_avail_fmt"] = format_size(kv.second["max_available"].uint64_value());

src/cli_status.cpp

@@ -158,7 +158,12 @@ resume_2:
         for (auto & pool_pair: parent->cli->st_cli.pool_config)
         {
             auto & pool_cfg = pool_pair.second;
-            bool active = pool_cfg.real_pg_count > 0;
+            bool active = true;
+            if (pool_cfg.pg_config.size() != pool_cfg.pg_count)
+            {
+                active = false;
+                pgs_by_state["offline"] += pool_cfg.pg_count-pool_cfg.pg_config.size();
+            }
             pool_count++;
             for (auto pg_it = pool_cfg.pg_config.begin(); pg_it != pool_cfg.pg_config.end(); pg_it++)
             {

src/cluster_client.cpp

@@ -64,7 +64,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
 
 cluster_client_t::~cluster_client_t()
 {
-    msgr.repeer_pgs = [](osd_num_t){};
+    msgr.repeer_pgs = [this](osd_num_t){};
     if (ringloop)
     {
         ringloop->unregister_consumer(&consumer);
@@ -454,7 +454,7 @@ bool cluster_client_t::flush()
             wb->start_writebacks(this, 0);
             cluster_op_t *sync = new cluster_op_t;
             sync->opcode = OSD_OP_SYNC;
-            sync->callback = [](cluster_op_t *sync)
+            sync->callback = [this](cluster_op_t *sync)
             {
                 delete sync;
             };
@@ -465,7 +465,7 @@ bool cluster_client_t::flush()
     bool sync_done = false;
     cluster_op_t *sync = new cluster_op_t;
     sync->opcode = OSD_OP_SYNC;
-    sync->callback = [&sync_done](cluster_op_t *sync)
+    sync->callback = [this, &sync_done](cluster_op_t *sync)
     {
         delete sync;
         sync_done = true;

src/cluster_client_wb.cpp

@@ -263,7 +263,7 @@ void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from
     }
     assert(calc_len == op->len);
     writebacks_active++;
-    op->callback = [this, flush_id](cluster_op_t* op)
+    op->callback = [this, cli, flush_id](cluster_op_t* op)
     {
         // Buffer flushes should be always retried, regardless of the error,
         // so they should never result in an error here
@@ -383,7 +383,7 @@ static void copy_to_op(cluster_op_t *op, uint64_t offset, uint8_t *buf, uint64_t
             auto begin = (cur_offset < offset ? offset : cur_offset);
             auto end = (cur_offset+v.iov_len > offset+len ? offset+len : cur_offset+v.iov_len);
             memcpy(
-                (uint8_t*)v.iov_base + begin - cur_offset,
+                v.iov_base + begin - cur_offset,
                 buf + (cur_offset <= offset ? 0 : cur_offset-offset),
                 end - begin
             );

src/disk_tool.cpp

@@ -5,7 +5,7 @@
 #include "str_util.h"
 
 static const char *help_text =
-    "Vitastor disk management tool " VERSION "\n"
+    "Vitastor disk management tool\n"
     "(c) Vitaliy Filippov, 2022+ (VNPL-1.1)\n"
     "\n"
     "COMMANDS:\n"

src/messenger.cpp

@@ -42,6 +42,12 @@ void osd_messenger_t::init()
             handle_rdma_events();
         }
     }
+#endif
+#ifndef SO_ZEROCOPY
+    if (log_level > 0)
+    {
+        fprintf(stderr, "Zero-copy TCP send is not supported in this build, ignoring\n");
+    }
 #endif
     keepalive_timer_id = tfd->set_timer(1000, true, [this](int)
     {
@@ -173,6 +179,8 @@ void osd_messenger_t::parse_config(const json11::Json & config)
         this->receive_buffer_size = 65536;
     this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
         config["use_sync_send_recv"].uint64_value();
+    this->use_zerocopy_send = config["use_zerocopy_send"].bool_value() ||
+        config["use_zerocopy_send"].uint64_value();
     this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
     if (!this->peer_connect_interval)
         this->peer_connect_interval = 5;
@@ -303,8 +311,7 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
         on_connect_peer(peer_osd, -result);
         return;
     }
-    int one = 1;
-    setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+    set_socket_options(cl);
     cl->peer_state = PEER_CONNECTED;
     tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
     {
@@ -314,6 +321,23 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
     check_peer_config(cl);
 }
 
+void osd_messenger_t::set_socket_options(osd_client_t *cl)
+{
+    int one = 1;
+    setsockopt(cl->peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+#ifdef SO_ZEROCOPY
+    if (!use_zerocopy_send)
+        cl->zerocopy_send = false;
+    else if (setsockopt(cl->peer_fd, SOL_SOCKET, SO_ZEROCOPY, &one, sizeof(one)) != 0)
+    {
+        if (log_level > 0)
+            fprintf(stderr, "[OSD %lu] Failed to enable zero-copy send for client %d: %s\n", this->osd_num, cl->peer_fd, strerror(errno));
+    }
+    else
+        cl->zerocopy_send = true;
+#endif
+}
+
 void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
 {
     // Mark client as ready (i.e. some data is available)
@@ -490,14 +514,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
                     fprintf(stderr, "Connected to OSD %lu using RDMA\n", cl->osd_num);
                 }
                 cl->peer_state = PEER_RDMA;
-                tfd->set_fd_handler(cl->peer_fd, false, [this](int peer_fd, int epoll_events)
-                {
-                    // Do not miss the disconnection!
-                    if (epoll_events & EPOLLRDHUP)
-                    {
-                        handle_peer_epoll(peer_fd, epoll_events);
-                    }
-                });
+                tfd->set_fd_handler(cl->peer_fd, false, NULL);
                 // Add the initial receive request
                 try_recv_rdma(cl);
             }
@@ -522,14 +539,13 @@ void osd_messenger_t::accept_connections(int listen_fd)
         fprintf(stderr, "[OSD %lu] new client %d: connection from %s\n", this->osd_num, peer_fd,
             addr_to_string(addr).c_str());
         fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
-        int one = 1;
-        setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
-        clients[peer_fd] = new osd_client_t();
-        clients[peer_fd]->peer_addr = addr;
-        clients[peer_fd]->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port);
-        clients[peer_fd]->peer_fd = peer_fd;
-        clients[peer_fd]->peer_state = PEER_CONNECTED;
-        clients[peer_fd]->in_buf = malloc_or_die(receive_buffer_size);
+        auto cl = clients[peer_fd] = new osd_client_t();
+        cl->peer_addr = addr;
+        cl->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port);
+        cl->peer_fd = peer_fd;
+        cl->peer_state = PEER_CONNECTED;
+        cl->in_buf = malloc_or_die(receive_buffer_size);
+        set_socket_options(cl);
         // Add FD to epoll
         tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
         {

src/messenger.h

@@ -45,6 +45,12 @@ struct msgr_rdma_connection_t;
 struct msgr_rdma_context_t;
 #endif
 
+struct msgr_zc_not_t
+{
+    osd_op_t *op;
+    uint32_t nsend;
+};
+
 struct osd_client_t
 {
     int refs = 0;
@@ -57,6 +63,7 @@ struct osd_client_t
     int ping_time_remaining = 0;
     int idle_time_remaining = 0;
     osd_num_t osd_num = 0;
+    bool zerocopy_send = false;
 
     void *in_buf = NULL;
 
@@ -87,6 +94,12 @@ struct osd_client_t
     int write_state = 0;
     std::vector<iovec> send_list, next_send_list;
     std::vector<msgr_sendp_t> outbox, next_outbox;
+    std::vector<msgr_zc_not_t> zerocopy_sent;
+    uint64_t outbox_size = 0, next_outbox_size = 0;
+    uint32_t zerocopy_notification_idx = 0;
+    uint32_t zerocopy_notification_prev = 0;
+    uint8_t zerocopy_notification_buf[256];
+    struct msghdr zerocopy_notification_msg;
 
     ~osd_client_t();
 };
@@ -123,6 +136,7 @@ protected:
     int osd_ping_timeout = 0;
     int log_level = 0;
     bool use_sync_send_recv = false;
+    bool use_zerocopy_send = false;
 
 #ifdef WITH_RDMA
     bool use_rdma = true;
@@ -185,9 +199,11 @@ protected:
     void check_peer_config(osd_client_t *cl);
     void cancel_osd_ops(osd_client_t *cl);
     void cancel_op(osd_op_t *op);
+    void set_socket_options(osd_client_t *cl);
 
     bool try_send(osd_client_t *cl);
     void handle_send(int result, osd_client_t *cl);
+    void handle_zerocopy_notification(osd_client_t *cl, int res);
 
     bool handle_read(int result, osd_client_t *cl);
     bool handle_read_buffer(osd_client_t *cl, void *curbuf, int remain);

src/msgr_send.cpp

@@ -3,10 +3,15 @@
 
 #define _XOPEN_SOURCE
 #include <limits.h>
-#include <sys/epoll.h>
 
 #include "messenger.h"
 
+#include <linux/errqueue.h>
+
+#ifndef MSG_ZEROCOPY
+#define MSG_ZEROCOPY 0
+#endif
+
 void osd_messenger_t::outbox_push(osd_op_t *cur_op)
 {
     assert(cur_op->peer_fd);
@@ -37,6 +42,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     }
     auto & to_send_list = cl->write_msg.msg_iovlen ? cl->next_send_list : cl->send_list;
     auto & to_outbox = cl->write_msg.msg_iovlen ? cl->next_outbox : cl->outbox;
+    auto & to_size = cl->write_msg.msg_iovlen ? cl->next_outbox_size : cl->outbox_size;
     if (cur_op->op_type == OSD_OP_IN)
     {
         measure_exec(cur_op);
@@ -47,6 +53,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         to_send_list.push_back((iovec){ .iov_base = cur_op->req.buf, .iov_len = OSD_PACKET_SIZE });
         cl->sent_ops[cur_op->req.hdr.id] = cur_op;
     }
+    to_size += OSD_PACKET_SIZE;
     to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = MSGR_SENDP_HDR });
     // Bitmap
     if (cur_op->op_type == OSD_OP_IN &&
@@ -58,6 +65,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
             .iov_len = cur_op->reply.sec_rw.attr_len,
         });
         to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
+        to_size += cur_op->reply.sec_rw.attr_len;
     }
     else if (cur_op->op_type == OSD_OP_OUT &&
         (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) &&
@@ -68,6 +76,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
             .iov_len = cur_op->req.sec_rw.attr_len,
         });
         to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
+        to_size += cur_op->req.sec_rw.attr_len;
     }
     // Operation data
     if ((cur_op->op_type == OSD_OP_IN
@@ -90,15 +99,22 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
                 assert(cur_op->iov.buf[i].iov_base);
                 to_send_list.push_back(cur_op->iov.buf[i]);
                 to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
+                to_size += cur_op->iov.buf[i].iov_len;
             }
         }
     }
     if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
     {
         if (cur_op->op_type == OSD_OP_IN && cur_op->reply.hdr.retval > 0)
+        {
             to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->reply.hdr.retval });
+            to_size += cur_op->reply.hdr.retval;
+        }
         else if (cur_op->op_type == OSD_OP_OUT && cur_op->req.sec_read_bmp.len > 0)
+        {
             to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->req.sec_read_bmp.len });
+            to_size += cur_op->req.sec_read_bmp.len;
+        }
         to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
     }
     if (cur_op->op_type == OSD_OP_IN)
@@ -184,17 +200,19 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
         }
         cl->write_msg.msg_iov = cl->send_list.data();
         cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+        cl->write_msg.msg_flags = (cl->zerocopy_send && (cl->outbox_size/cl->send_list.size()) >= 4096 ? MSG_ZEROCOPY : 0);
         cl->refs++;
         ring_data_t* data = ((ring_data_t*)sqe->user_data);
         data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, cl); };
-        my_uring_prep_sendmsg(sqe, peer_fd, &cl->write_msg, 0);
+        my_uring_prep_sendmsg(sqe, peer_fd, &cl->write_msg, cl->write_msg.msg_flags);
     }
     else
     {
         cl->write_msg.msg_iov = cl->send_list.data();
         cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+        cl->write_msg.msg_flags = (cl->zerocopy_send && (cl->outbox_size/cl->send_list.size()) >= 4096 ? MSG_ZEROCOPY : 0);
         cl->refs++;
-        int result = sendmsg(peer_fd, &cl->write_msg, MSG_NOSIGNAL);
+        int result = sendmsg(peer_fd, &cl->write_msg, MSG_NOSIGNAL | cl->write_msg.msg_flags);
         if (result < 0)
         {
             result = -errno;
@@ -204,6 +222,62 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
     return true;
 }
 
+void osd_messenger_t::handle_zerocopy_notification(osd_client_t *cl, int res)
+{
+    cl->refs--;
+    if (cl->peer_state == PEER_STOPPED)
+    {
+        if (cl->refs <= 0)
+        {
+            delete cl;
+        }
+        return;
+    }
+    if (res != 0)
+    {
+        return;
+    }
+    if (cl->zerocopy_notification_msg.msg_flags & MSG_CTRUNC)
+    {
+        fprintf(stderr, "zero-copy send notification truncated on client socket %d\n", cl->peer_fd);
+        return;
+    }
+    for (struct cmsghdr *cm = CMSG_FIRSTHDR(&cl->zerocopy_notification_msg); cm; cm = CMSG_NXTHDR(&cl->zerocopy_notification_msg, cm))
+    {
+        if (cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR)
+        {
+            struct sock_extended_err *serr = (struct sock_extended_err*)CMSG_DATA(cm);
+            if (serr->ee_errno == 0 && serr->ee_origin == SO_EE_ORIGIN_ZEROCOPY)
+            {
+                // completed sends numbered serr->ee_info .. serr->ee_data
+                int start = 0;
+                while (start < cl->zerocopy_sent.size() && cl->zerocopy_sent[start].nsend < serr->ee_info)
+                    start++;
+                int end = start;
+                if (serr->ee_data < serr->ee_info)
+                {
+                    // counter has wrapped around
+                    while (end < cl->zerocopy_sent.size() && cl->zerocopy_sent[end].nsend >= cl->zerocopy_sent[start].nsend)
+                        end++;
+                }
+                while (end < cl->zerocopy_sent.size() && cl->zerocopy_sent[end].nsend <= serr->ee_data)
+                    end++;
+                if (end > start)
+                {
+                    for (int i = start; i < end; i++)
+                    {
+                        delete cl->zerocopy_sent[i].op;
+                    }
+                    cl->zerocopy_sent.erase(
+                        cl->zerocopy_sent.begin() + start,
+                        cl->zerocopy_sent.begin() + end
+                    );
+                }
+            }
+        }
+    }
+}
+
 void osd_messenger_t::send_replies()
 {
     for (int i = 0; i < write_ready_clients.size(); i++)
@@ -231,16 +305,19 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
         }
         return;
     }
-    if (result < 0 && result != -EAGAIN && result != -EINTR)
+    if (result < 0 && result != -EAGAIN && result != -EINTR && result != -ENOBUFS)
     {
         // this is a client socket, so don't panic. just disconnect it
         fprintf(stderr, "Client %d socket write error: %d (%s). Disconnecting client\n", cl->peer_fd, -result, strerror(-result));
         stop_client(cl->peer_fd);
         return;
     }
+    bool used_zerocopy = false;
     if (result >= 0)
     {
+        used_zerocopy = (cl->write_msg.msg_flags & MSG_ZEROCOPY) ? true : false;
         int done = 0;
+        int bytes_written = result;
         while (result > 0 && done < cl->send_list.size())
         {
             iovec & iov = cl->send_list[done];
@@ -249,7 +326,19 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
                 if (cl->outbox[done].flags & MSGR_SENDP_FREE)
                 {
                     // Reply fully sent
-                    delete cl->outbox[done].op;
+                    if (!used_zerocopy)
+                    {
+                        delete cl->outbox[done].op;
+                    }
+                    else
+                    {
+                        // With zero-copy send the difference is that we must keep the buffer (i.e. the operation)
+                        // allocated until we get send notification from MSG_ERRQUEUE
+                        cl->zerocopy_sent.push_back((msgr_zc_not_t){
+                            .op = cl->outbox[done].op,
+                            .nsend = cl->zerocopy_notification_idx,
+                        });
+                    }
                 }
                 result -= iov.iov_len;
                 done++;
@@ -261,6 +350,11 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
                 break;
             }
         }
+        if (used_zerocopy)
+        {
+            cl->zerocopy_notification_idx++;
+        }
+        cl->outbox_size -= bytes_written;
         if (done > 0)
         {
             cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+done);
@@ -270,8 +364,10 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
         {
             cl->send_list.insert(cl->send_list.end(), cl->next_send_list.begin(), cl->next_send_list.end());
             cl->outbox.insert(cl->outbox.end(), cl->next_outbox.begin(), cl->next_outbox.end());
+            cl->outbox_size += cl->next_outbox_size;
             cl->next_send_list.clear();
             cl->next_outbox.clear();
+            cl->next_outbox_size = 0;
         }
         cl->write_state = cl->outbox.size() > 0 ? CL_WRITE_READY : 0;
 #ifdef WITH_RDMA
@@ -284,14 +380,7 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
                 fprintf(stderr, "Successfully connected with client %d using RDMA\n", cl->peer_fd);
             }
             cl->peer_state = PEER_RDMA;
-            tfd->set_fd_handler(cl->peer_fd, false, [this](int peer_fd, int epoll_events)
-            {
-                // Do not miss the disconnection!
-                if (epoll_events & EPOLLRDHUP)
-                {
-                    handle_peer_epoll(peer_fd, epoll_events);
-                }
-            });
+            tfd->set_fd_handler(cl->peer_fd, false, NULL);
             // Add the initial receive request
             try_recv_rdma(cl);
         }
@@ -301,4 +390,34 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
     {
         write_ready_clients.push_back(cl->peer_fd);
     }
+    if (used_zerocopy && (cl->zerocopy_notification_idx-cl->zerocopy_notification_prev) >= 16 &&
+        cl->zerocopy_sent.size() > 0)
+    {
+        cl->zerocopy_notification_prev = cl->zerocopy_notification_idx;
+        cl->zerocopy_notification_msg = {
+            .msg_control = cl->zerocopy_notification_buf,
+            .msg_controllen = sizeof(cl->zerocopy_notification_buf),
+        };
+        cl->refs++;
+        io_uring_sqe* sqe = NULL;
+        if (ringloop && !use_sync_send_recv)
+        {
+            sqe = ringloop->get_sqe();
+        }
+        if (!sqe)
+        {
+            int res = recvmsg(cl->peer_fd, &cl->zerocopy_notification_msg, MSG_ERRQUEUE|MSG_DONTWAIT);
+            if (res < 0)
+            {
+                res = -errno;
+            }
+            handle_zerocopy_notification(cl, res);
+        }
+        else
+        {
+            ring_data_t* data = ((ring_data_t*)sqe->user_data);
+            data->callback = [this, cl](ring_data_t *data) { handle_zerocopy_notification(cl, data->res); };
+            my_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->zerocopy_notification_msg, MSG_ERRQUEUE);
+        }
+    }
 }

src/osd_main.cpp

@@ -19,14 +19,6 @@ static void handle_sigint(int sig)
     exit(0);
 }
 
-static const char* help_text =
-    "Vitastor OSD (block object storage daemon) " VERSION "\n"
-    "(c) Vitaliy Filippov, 2019+ (VNPL-1.1)\n"
-    "\n"
-    "OSDs are usually started by vitastor-disk.\n"
-    "Manual usage: vitastor-osd [--option value] ...\n"
-;
-
 int main(int narg, char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
@@ -45,16 +37,6 @@ int main(int narg, char *args[])
             char *opt = args[i]+2;
             config[std::string(opt)] = std::string(args[++i]);
         }
-        else if (!strcmp(args[i], "--help"))
-        {
-            printf("%s", help_text);
-            return 0;
-        }
-    }
-    if (!config.size())
-    {
-        printf("%s", help_text);
-        return 1;
     }
     signal(SIGINT, handle_sigint);
     signal(SIGTERM, handle_sigint);

src/vitastor.pc.in

@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
 
 Name: Vitastor
 Description: Vitastor client library
-Version: 1.2.0
+Version: 1.1.0
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}