WIP Batch encrypted records to reduce EncryptInit/EncryptFinal costs

Seems slightly overcomplicated...

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 2.8...3.30)
 
 project(vitastor)
 
-set(VITASTOR_VERSION "3.0.8")
+set(VITASTOR_VERSION "3.0.9")
 
 include(CTest)
 

csi/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.0.8
+VITASTOR_VERSION ?= v3.0.9
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

@@ -49,7 +49,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v3.0.8
+          image: vitalif/vitastor-csi:v3.0.9
           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:v3.0.8
+          image: vitalif/vitastor-csi:v3.0.9
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

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

debian/changelog

@@ -1,4 +1,4 @@
-vitastor (3.0.8-1) unstable; urgency=medium
+vitastor (3.0.9-1) unstable; urgency=medium
 
   * Bugfixes
 

docker/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.0.8
+VITASTOR_VERSION ?= v3.0.9
 
 all: build push
 

docker/etc/vitastor/docker.conf

@@ -4,7 +4,7 @@
 #
 
 # Desired Vitastor version
-VITASTOR_VERSION=v3.0.8
+VITASTOR_VERSION=v3.0.9
 
 # Additional arguments for all containers
 # For example, you may want to specify a custom logging driver here

docs/config/osd.en.md

@@ -70,6 +70,7 @@ with an OSD restart or, for some of them, even without restarting by updating co
 - [use_atomic_flag](#use_atomic_flag)
 - [pg_reshard_chunk_size](#pg_reshard_chunk_size)
 - [pg_reshard_chunk_pause_ms](#pg_reshard_chunk_pause_ms)
+- [gc_on_start](#gc_on_start)
 
 ## bind_address
 
@@ -753,3 +754,9 @@ This option sets the maximum number of object is a chunk. Moving 100k objects us
 - Default: 100
 
 This option sets the interval between handling two PG count change chunks.
+
+## gc_on_start
+
+- Type: boolean
+
+Forcibly clean all garbage entries in the new store on every OSD restart.

docs/config/osd.ru.md

@@ -71,6 +71,7 @@
 - [use_atomic_flag](#use_atomic_flag)
 - [pg_reshard_chunk_size](#pg_reshard_chunk_size)
 - [pg_reshard_chunk_pause_ms](#pg_reshard_chunk_pause_ms)
+- [gc_on_start](#gc_on_start)
 
 ## bind_address
 
@@ -793,3 +794,9 @@ pg_minsize OSD во время переключений, что может по
 - Значение по умолчанию: 100
 
 Данная опция задаёт интервал между обработкой двух порций изменения числа PG пулов.
+
+## gc_on_start
+
+- Тип: булево (да/нет)
+
+Принудительно очищать все мусорные записи в новом хранилище при каждом запуске OSD.

docs/config/src/osd.yml

@@ -938,3 +938,7 @@
     This option sets the interval between handling two PG count change chunks.
   info_ru: |
     Данная опция задаёт интервал между обработкой двух порций изменения числа PG пулов.
+- name: gc_on_start
+  type: bool
+  info: Forcibly clean all garbage entries in the new store on every OSD restart.
+  info_ru: Принудительно очищать все мусорные записи в новом хранилище при каждом запуске OSD.

docs/installation/docker.en.md

@@ -26,9 +26,9 @@ at Vitastor Kubernetes operator: https://github.com/Antilles7227/vitastor-operat
 The instruction is very simple.
 
 1. Download a Docker image of the desired version: \
-   `docker pull vitalif/vitastor:v3.0.8`
+   `docker pull vitalif/vitastor:v3.0.9`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.8 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.9 install.sh`
 3. Reload udev rules: \
    `udevadm control --reload-rules`
 4. Enable the vitastor-host service: \

docs/installation/docker.ru.md

@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
 Инструкция по установке максимально простая.
 
 1. Скачайте Docker-образ желаемой версии: \
-   `docker pull vitalif/vitastor:v3.0.8`
+   `docker pull vitalif/vitastor:v3.0.9`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.8 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.0.9 install.sh`
 3. Перезагрузите правила udev: \
    `udevadm control --reload-rules`
 4. Включите сервис vitastor-host: \

mon/package.json

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

mon/stats.js

@@ -37,6 +37,7 @@ function derive_osd_stats(st, prev, prev_diff)
         const n = c.count - BigInt(pr && pr.count||0);
         diff.recovery_stats[op] = { ...c, bps: n > 0 ? b*1000n/timediff : 0n, iops: n > 0 ? n*1000n/timediff : 0n };
     }
+    diff.inode_stats = {};
     for (const pool_id in st.inode_stats||{})
     {
         diff.inode_stats[pool_id] = {};

node-binding/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor",
-  "version": "3.0.8",
+  "version": "3.0.9",
   "description": "Low-level native bindings to Vitastor client library",
   "main": "index.js",
   "keywords": [

patches/cinder-vitastor.py

@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils
 
-VITASTOR_VERSION = '3.0.8'
+VITASTOR_VERSION = '3.0.9'
 
 LOG = logging.getLogger(__name__)
 

rpm/vitastor-el10.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.0.8
+Version:        3.0.9
 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-3.0.8.el10.tar.gz
+Source0:        vitastor-3.0.9.el10.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.0.8
+Version:        3.0.9
 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-3.0.8.el7.tar.gz
+Source0:        vitastor-3.0.9.el7.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.0.8
+Version:        3.0.9
 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-3.0.8.el8.tar.gz
+Source0:        vitastor-3.0.9.el8.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.0.8
+Version:        3.0.9
 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-3.0.8.el9.tar.gz
+Source0:        vitastor-3.0.9.el9.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

src/CMakeLists.txt

@@ -20,7 +20,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 endif()
 set(ENABLE_COVERAGE false CACHE BOOL "Enable code coverage")
 
-add_definitions(-DVITASTOR_VERSION="3.0.8")
+add_definitions(-DVITASTOR_VERSION="3.0.9")
 add_definitions(-D_GNU_SOURCE -D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -fno-omit-frame-pointer -fvisibility=hidden -I ${CMAKE_SOURCE_DIR}/src)
 add_link_options(-fno-omit-frame-pointer)
 if (${WITH_ASAN})

src/blockstore/blockstore_heap.cpp

@@ -344,19 +344,12 @@ corrupted_block:
                     block_num, block_offset, wr->size, sizeof(heap_entry_t));
                 goto corrupted_block;
             }
-            if (wr->is_garbage())
-            {
-                // Garbage collection is only performed when writing new entries into the block
-                // because it needs a fake LSN and modified blocks require consecutive modified LSNs
-                // That's why garbage entries may persist on disk
-                if (log_level > 5)
-                {
-                    fprintf(stderr, "Notice: skipping garbage entry %jx:%jx v%ju l%ju in metadata block %u at %u\n",
-                        wr->inode, wr->stripe, wr->version, wr->lsn, block_num, block_offset);
-                }
-                block_offset += wr->size;
-                continue;
-            }
+            // Garbage collection is only performed when writing new entries into the block
+            // because it needs a fake LSN and modified blocks require consecutive modified LSNs
+            // At the same time, further modifications _after_ putting new entries into the block,
+            // but _before_ writing it, may mark some entries in it as garbage. That's why garbage
+            // entries may still be present on disk.
+            wr->entry_type &= ~BS_HEAP_GARBAGE;
             if ((wr->entry_type & BS_HEAP_TYPE) < BS_HEAP_BIG_WRITE ||
                 (wr->entry_type & BS_HEAP_TYPE) > BS_HEAP_ROLLBACK ||
                 (wr->entry_type & ~(BS_HEAP_TYPE|BS_HEAP_STABLE)) ||

src/blockstore/blockstore_init.cpp

@@ -311,7 +311,7 @@ resume_8:
         uint32_t block_num = recheck_mod[i];
         uint64_t block_offset = bs->dsk.meta_offset + (uint64_t)(block_num+1) * bs->dsk.meta_block_size;
         data = ((ring_data_t*)sqe->user_data);
-        uint8_t *buf = (uint8_t*)malloc_or_die(bs->dsk.meta_block_size);
+        uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, bs->dsk.meta_block_size);
         bs->heap->get_meta_block(block_num, buf);
         data->iov = { buf, bs->dsk.meta_block_size };
         data->callback = [this, buf, block_offset](ring_data_t *data)

src/blockstore/v1/impl.cpp

@@ -877,7 +877,7 @@ uint64_t blockstore_impl_t::get_garbage_entries()
 
 uint64_t blockstore_impl_t::get_garbage_memory()
 {
-    return sizeof(blockstore_dirty_db_t::node_type) * dirty_db.size();
+    return (sizeof(obj_ver_id) + sizeof(dirty_entry) + 32) * dirty_db.size();
 }
 
 } // namespace v1

src/client/cluster_client.cpp

@@ -590,7 +590,7 @@ void cluster_client_t::on_change_pool_config_hook()
         {
             if (log_level > 2 && pg_counts[pool_item.first])
             {
-                printf("Pool %u (%s) PG count changed from %lu to %lu\n", pool_item.first, pool_item.second.name.c_str(),
+                fprintf(stderr, "Pool %u (%s) PG count changed from %lu to %lu\n", pool_item.first, pool_item.second.name.c_str(),
                     pg_counts[pool_item.first], pool_item.second.real_pg_count);
             }
             // At this point, all pool operations should have been suspended
@@ -1294,8 +1294,8 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
         unsigned bitmap_mem = object_bitmap_size +
             op->parts.size() * pg_data_size *
             (pool_cfg.data_block_size / pool_cfg.bitmap_granularity / 8
-            // read chain info - 1 byte per block
-            + (op->enc ? op->len/pool_cfg.bitmap_granularity : 0));
+            // read chain_info - max 4 bytes per block
+            + (op->enc ? osd_op_rw_t::chain_info_bytes(op->enc->chain_size)*op->len/pool_cfg.bitmap_granularity : 0));
         if (!op->bitmap_buf || op->bitmap_buf_size < bitmap_mem)
         {
             op->bitmap_buf = realloc_or_die(op->bitmap_buf, bitmap_mem);
@@ -1452,7 +1452,10 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
             op->inflight_count++;
             uint32_t pg_data_size = (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks);
             uint64_t pg_bitmap_size = pg_data_size * (pool_cfg.data_block_size / pool_cfg.bitmap_granularity / 8
-                + (op->opcode == OSD_OP_READ && op->enc ? pool_cfg.data_block_size/pool_cfg.bitmap_granularity : 0));
+                // read chain_info - max 4 bytes per block
+                + (op->opcode == OSD_OP_READ && op->enc
+                    ? osd_op_rw_t::chain_info_bytes(op->enc->chain_size)*pool_cfg.data_block_size/pool_cfg.bitmap_granularity
+                    : 0));
             uint64_t meta_rev = 0;
             if (op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_DELETE && !op->deoptimise_snapshot)
             {

src/client/cluster_client.h

@@ -84,7 +84,7 @@ class writeback_cache_t;
 
 struct inode_cache_t
 {
-    std::vector<inode_t> chain;
+    std::vector<inode_t> chain; // only parents from the same pool
     uint8_t *key_data = NULL;
     osd_op_enc_t *op_enc = NULL;
     bool readonly = false;

src/client/http_client.cpp

@@ -164,30 +164,43 @@ void http_ares_cb(void *data, ares_socket_t socket_fd, int readable, int writabl
 }
 
 #ifdef WITH_OPENSSL
+bool openssl_ctx_add_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem)
+{
+    std::string pem;
+    BIO *bio = NULL;
+    if (file_or_pem.substr(0, 5) != "-----")
+    {
+        pem = read_file(file_or_pem);
+        bio = BIO_new_mem_buf(pem.data(), pem.size());
+    }
+    else
+        bio = BIO_new_mem_buf(file_or_pem.data(), file_or_pem.size());
+    if (!bio)
+        return false;
+    X509 *x509 = PEM_read_bio_X509(bio, NULL, 0, NULL);
+    bool ok = !!x509;
+    if (x509)
+    {
+        X509_STORE *store = SSL_CTX_get_cert_store(ssl_ctx);
+        X509_STORE_add_cert(store, x509);
+        X509_free(x509);
+    }
+    BIO_free(bio);
+    return ok;
+}
+
 bool openssl_ctx_use_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem)
 {
     if (file_or_pem.substr(0, 5) == "-----")
     {
-        BIO *bio = BIO_new_mem_buf(file_or_pem.data(), file_or_pem.size());
-        if (!bio)
-            return false;
-        X509 *x509 = PEM_read_bio_X509(bio, NULL, 0, NULL);
-        bool ok = !!x509;
-        if (x509)
-        {
-            X509_STORE *store = SSL_CTX_get_cert_store(ssl_ctx);
-            X509_STORE_add_cert(store, x509);
-            X509_free(x509);
-        }
-        BIO_free(bio);
-        return ok;
+        return openssl_ctx_add_ca(ssl_ctx, file_or_pem);
     }
     return file_or_pem.empty()
         ? !!SSL_CTX_set_default_verify_paths(ssl_ctx)
         : !!SSL_CTX_load_verify_locations(ssl_ctx, file_or_pem.c_str(), NULL);
 }
 
-static std::string openssl_get_cn(X509 *x509)
+std::string openssl_get_cn(X509 *x509)
 {
     X509_NAME* subj = X509_get_subject_name(x509);
     int pos = X509_NAME_get_index_by_NID(subj, NID_commonName, -1);

src/client/http_client.h

@@ -8,6 +8,10 @@
 #include <functional>
 #include "json11/json11.hpp"
 
+#ifdef WITH_OPENSSL
+#include <openssl/types.h>
+#endif
+
 #define WS_CONTINUATION 0
 #define WS_TEXT 1
 #define WS_BINARY 2
@@ -69,3 +73,11 @@ void http_close(http_co_t *co);
 void http_destroy(http_co_t *co);
 
 #pragma GCC visibility pop
+
+#ifdef WITH_OPENSSL
+bool openssl_ctx_add_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem);
+bool openssl_ctx_use_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem);
+std::string openssl_get_cn(X509 *x509);
+bool openssl_ctx_use_cert(SSL_CTX *ssl_ctx, const std::string & file_or_pem, std::string & common_name);
+bool openssl_ctx_use_key(SSL_CTX *ssl_ctx, const std::string & file_or_pem);
+#endif

src/client/messenger.cpp

@@ -10,10 +10,18 @@
 #include <stdexcept>
 
 #include "addr_util.h"
+#include "str_util.h"
 #include "messenger.h"
 #ifdef WITH_RDMA
 #include "msgr_rdma.h"
 #endif
+#include "http_client.h"
+#ifdef WITH_OPENSSL
+#include <openssl/bio.h>
+#include <openssl/err.h>
+#include <openssl/pem.h>
+#include <openssl/ssl.h>
+#endif
 
 #include <sys/poll.h>
 
@@ -117,6 +125,50 @@ void msgr_iothread_t::run()
 
 void osd_messenger_t::init()
 {
+    if (!tls_cert.empty() || !tls_key.empty() || !osd_tls_ca.empty() || !client_tls_ca.empty())
+    {
+        // Initialize TLS context
+        // FIXME: require OpenSSL
+#ifndef WITH_OPENSSL
+        fprintf(stderr, "Vitastor is built without OpenSSL support\n");
+        exit(1);
+#else
+        if (tls_cert.empty() || tls_key.empty() || osd_tls_ca.empty() || osd_num && client_tls_ca.empty())
+        {
+            if (osd_num)
+                fprintf(stderr, "Vitastor OSD TLS requires osd_tls_cert, osd_tls_key, osd_tls_ca, client_tls_ca\n");
+            else
+                fprintf(stderr, "Vitastor client TLS requires tls_cert, tls_key and osd_tls_ca\n");
+            exit(1);
+        }
+        else
+        {
+            ssl_ctx = SSL_CTX_new(TLS_method());
+            if (!ssl_ctx)
+            {
+init_err:
+                fprintf(stderr, "OpenSSL initialization failed: %s\n", ERR_error_string(ERR_get_error(), NULL));
+                exit(1);
+            }
+            SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER, NULL);
+            bool ok = SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_3_VERSION);
+            ok = ok && openssl_ctx_add_ca(ssl_ctx, osd_tls_ca);
+            if (osd_num)
+            {
+                // OSD uses 2 separate root certificates to distinguish between clients and peer OSDs
+                ok = ok && openssl_ctx_add_ca(ssl_ctx, client_tls_ca);
+            }
+            ok = ok && openssl_ctx_use_cert(ssl_ctx, tls_cert, tls_cn);
+            ok = ok && openssl_ctx_use_key(ssl_ctx, tls_key);
+            if (!ok)
+            {
+                SSL_CTX_free(ssl_ctx);
+                ssl_ctx = NULL;
+                goto init_err;
+            }
+        }
+#endif
+    }
 #ifdef WITH_RDMACM
     if (use_rdmacm)
     {
@@ -303,6 +355,13 @@ osd_messenger_t::~osd_messenger_t()
     {
         destroy_aes_xts_decrypt(decrypt_ctx);
     }
+#ifdef WITH_OPENSSL
+    if (ssl_ctx)
+    {
+        SSL_CTX_free(ssl_ctx);
+        ssl_ctx = NULL;
+    }
+#endif
 }
 
 void osd_messenger_t::parse_config(const json11::Json & config)
@@ -340,7 +399,30 @@ void osd_messenger_t::parse_config(const json11::Json & config)
     this->max_aes_xts_pool_size = config["max_aes_xts_pool_size"].uint64_value();
     if (!this->max_aes_xts_pool_size)
         this->max_aes_xts_pool_size = 256;
-    this->use_proto_checksums = config["use_proto_checksums"].is_null() || config["use_proto_checksums"].bool_value();
+    if (config["proto_checksums"].is_null())
+        this->use_proto_checksums = MSGR_CSUM_PAYLOAD;
+    else if (config["proto_checksums"].is_bool())
+        this->use_proto_checksums = config["proto_checksums"].bool_value() ? MSGR_CSUM_FULL : 0;
+    else if (config["proto_checksums"].string_value() != "")
+        this->use_proto_checksums = config["proto_checksums"].string_value() == "full" ? MSGR_CSUM_FULL : MSGR_CSUM_PAYLOAD;
+    else
+        this->use_proto_checksums = 0;
+    if (!osd_num)
+    {
+        tls_cert = config["tls_cert"].string_value();
+        tls_key = config["tls_key"].string_value();
+        osd_tls_ca = config["osd_tls_ca"].string_value();
+    }
+    else
+    {
+        tls_cert = config["osd_tls_cert"].string_value();
+        tls_key = config["osd_tls_key"].string_value();
+        osd_tls_ca = config["osd_tls_ca"].string_value();
+        client_tls_ca = config["client_tls_ca"].string_value();
+    }
+    test_osd_aes_key.resize(32);
+    if (fromhexstr(config["test_osd_aes_key"].string_value(), 32, (uint8_t*)test_osd_aes_key.data()) != 32)
+        test_osd_aes_key.clear();
     if (!osd_num)
         this->iothread_count = (uint32_t)config["client_iothread_count"].uint64_value();
     else
@@ -349,7 +431,7 @@ void osd_messenger_t::parse_config(const json11::Json & config)
     if (!this->receive_buffer_size || this->receive_buffer_size > 1024*1024*1024)
         this->receive_buffer_size = 65536;
     this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
-        config["use_sync_send_recv"].uint64_value();
+        config["use_sync_send_recv"].uint64_value() || !ringloop;
     this->min_zerocopy_send_size = config["min_zerocopy_send_size"].is_null()
         ? DEFAULT_MIN_ZEROCOPY_SEND_SIZE
         : (int)config["min_zerocopy_send_size"].int64_value();
@@ -571,6 +653,7 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
         handle_peer_epoll(peer_fd, epoll_events);
     });
     // Check OSD number
+    ssl_init(cl, false);
     check_peer_config(cl);
 }
 
@@ -663,7 +746,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
     auto features = json11::Json::object{ { "check_sequencing", true } };
     if (use_proto_checksums)
     {
-        features["proto_checksums"] = true;
+        features["proto_checksums"] = use_proto_checksums;
     }
     payload["features"] = features;
 #ifdef WITH_RDMA
@@ -740,9 +823,13 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
             delete op;
             return;
         }
-        if (use_proto_checksums && config["features"]["proto_checksums"].bool_value())
+        if (use_proto_checksums)
         {
-            cl->proto_csum_status = MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT;
+            auto peer_csums = config["features"]["proto_checksums"].uint64_value();
+            if (peer_csums == MSGR_CSUM_FULL && use_proto_checksums == MSGR_CSUM_FULL)
+                cl->proto_csum_status = MSGR_CSUM_FULL;
+            else if (peer_csums && use_proto_checksums)
+                cl->proto_csum_status = MSGR_CSUM_PAYLOAD;
         }
 #ifdef WITH_RDMA
         if (!use_rdmacm && cl->rdma_conn && config["rdma_address"].is_string())
@@ -808,6 +895,7 @@ void osd_messenger_t::accept_connections(int listen_fd)
         cl->peer_fd = peer_fd;
         cl->peer_state = PEER_CONNECTED;
         cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
+        ssl_init(cl, true);
         // Add FD to epoll
         tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
         {
@@ -822,6 +910,52 @@ void osd_messenger_t::accept_connections(int listen_fd)
     }
 }
 
+void osd_messenger_t::ssl_init(osd_client_t *cl, bool server_mode)
+{
+    if (!tls_cert.empty())
+    {
+        cl->write_to_ssl = BIO_new(BIO_s_mem());
+        cl->read_from_ssl = BIO_new(BIO_s_mem());
+        cl->ssl_cli = SSL_new(ssl_ctx);
+        if (!cl->ssl_cli)
+        {
+            fprintf(stderr, "OpenSSL initialization failed: %s\n", ERR_error_string(ERR_get_error(), NULL));
+            exit(1);
+        }
+        if (server_mode)
+        {
+            SSL_set_accept_state(cl->ssl_cli);
+        }
+        else
+        {
+            SSL_set_connect_state(cl->ssl_cli);
+        }
+        SSL_set_bio(cl->ssl_cli, cl->write_to_ssl, cl->read_from_ssl);
+        bool ok = ssl_do_handshake(cl);
+        assert(ok);
+    }
+    else if (!test_osd_aes_key.empty())
+    {
+        int r;
+        cl->enc_ctx = EVP_CIPHER_CTX_new();
+        assert(cl->enc_ctx);
+        r = EVP_EncryptInit_ex(cl->enc_ctx, EVP_aes_256_gcm(), NULL, NULL, NULL);
+        assert(r == 1);
+        r = EVP_CIPHER_CTX_set_padding(cl->enc_ctx, 0);
+        assert(r == 1);
+        r = EVP_CIPHER_CTX_ctrl(cl->enc_ctx, EVP_CTRL_GCM_SET_IVLEN, 12, NULL);
+        assert(r == 1);
+        cl->dec_ctx = EVP_CIPHER_CTX_new();
+        assert(cl->dec_ctx);
+        r = EVP_DecryptInit_ex(cl->dec_ctx, EVP_aes_256_gcm(), NULL, NULL, NULL);
+        assert(r == 1);
+        r = EVP_CIPHER_CTX_set_padding(cl->dec_ctx, 0);
+        assert(r == 1);
+        r = EVP_CIPHER_CTX_ctrl(cl->dec_ctx, EVP_CTRL_GCM_SET_IVLEN, 12, NULL);
+        assert(r == 1);
+    }
+}
+
 #ifdef WITH_RDMA
 msgr_rdma_context_t* osd_messenger_t::choose_rdma_context(osd_client_t *cl)
 {

src/client/messenger.h

@@ -12,6 +12,10 @@
 #include <deque>
 #include <vector>
 
+#ifdef WITH_OPENSSL
+#include <openssl/types.h>
+#endif
+
 #include "../util/xxh_x86dispatch.h"
 #include "../util/robin_hood.h"
 #include "malloc_or_die.h"
@@ -32,8 +36,9 @@
 #define PEER_RDMA 4
 #define PEER_STOPPED 5
 
-#define MSGR_PEER_CSUM_IN 1
-#define MSGR_PEER_CSUM_OUT 2
+#define MSGR_CSUM_PAYLOAD 1
+#define MSGR_CSUM_FULL 2
+#define MSGR_CSUM_NEG 4
 
 #define VITASTOR_CONFIG_PATH "/etc/vitastor/vitastor.conf"
 
@@ -55,6 +60,14 @@ struct op_aes_xts_decrypt_t;
 void destroy_aes_xts_encrypt(op_aes_xts_encrypt_t *encrypt_ctx);
 void destroy_aes_xts_decrypt(op_aes_xts_decrypt_t *decrypt_ctx);
 
+// Standard TLS record header. We are only interested in the record size
+struct __attribute__((__packed__)) msgr_tls_record_hdr_t
+{
+    uint8_t content_type;
+    uint16_t version;
+    uint16_t size;
+};
+
 struct osd_client_t
 {
     uint64_t client_id = 0;
@@ -77,7 +90,32 @@ struct osd_client_t
     msgr_rdma_connection_t *rdma_conn = NULL;
 #endif
 
+#ifdef WITH_OPENSSL
+    SSL *ssl_cli = NULL;
+    BIO *write_to_ssl = NULL;
+    // FIXME: use custom bio to avoid 1 more memory copy?
+    BIO *read_from_ssl = NULL;
+    uint8_t *ssl_out_buf = NULL;
+    size_t ssl_out_buf_size = 0, ssl_out_buf_cap = 0;
+    bool ssl_handshake_done = false;
+    msgr_tls_record_hdr_t ssl_read_record;
+    size_t ssl_read_header_size = 0;
+    bool ssl_more_to_buffer = false;
+
+    EVP_CIPHER_CTX *enc_ctx = NULL;
+    uint8_t enc_tag[16];
+    size_t enc_tag_size = 0;
+    bool enc_batch = false;
+    EVP_CIPHER_CTX *dec_ctx = NULL;
+    uint8_t dec_tag[16];
+    size_t dec_tag_size = 0;
+    uint32_t dec_batch_size = 0;
+    size_t dec_batch_size_size = 0;
+    std::vector<osd_op_t*> unverified_ops;
+#endif
+
     // Read state
+    bool io_error = false;
     int read_ready = 0;
     osd_op_t *read_op = NULL;
     size_t read_op_size = 0;
@@ -85,7 +123,7 @@ struct osd_client_t
     iovec read_iov = { 0 };
     msghdr read_msg = { 0 };
     std::vector<iovec> recv_list;
-    size_t recv_list_size = 0;
+    std::vector<int> recv_flags;
     uint64_t read_op_id = 1;
     bool check_sequencing = false;
     bool enable_pg_locks = false;
@@ -194,9 +232,21 @@ struct osd_messenger_t;
 struct rdmacm_connecting_t;
 #endif
 
+class msgr_op_reader_t;
+class msgr_op_writer_t;
+
 struct __attribute__((visibility("default"))) osd_messenger_t
 {
 protected:
+    friend class copy_op_reader_t;
+    friend class ssl_op_reader_t;
+    friend class gcm_op_reader_t;
+    friend class get_op_reader_t;
+    friend class copy_op_writer_t;
+    friend class ssl_op_writer_t;
+    friend class gcm_op_writer_t;
+    friend class get_op_writer_t;
+
     int keepalive_timer_id = -1;
 
     uint32_t receive_buffer_size = 0;
@@ -210,6 +260,12 @@ protected:
     int iothread_count = 0;
     int max_aes_xts_pool_size = 256;
 
+    std::string tls_cert;
+    std::string tls_key;
+    std::string osd_tls_ca;
+    std::string client_tls_ca;
+    std::string test_osd_aes_key; // FIXME Insecure, only for PoC tests
+
 #ifdef WITH_RDMA
     bool use_rdma = true;
     bool use_rdmacm = false;
@@ -226,6 +282,14 @@ protected:
     robin_hood::unordered_flat_map<rdma_cm_id*, rdmacm_connecting_t*> rdmacm_connecting;
 #endif
 
+#ifdef WITH_OPENSSL
+    SSL_CTX *ssl_ctx = NULL;
+    std::string tls_cn;
+
+    void ssl_init(osd_client_t *cl, bool server_mode);
+    bool ssl_do_handshake(osd_client_t *cl);
+#endif
+
     std::vector<msgr_iothread_t*> iothreads;
     std::vector<uint64_t> read_ready_clients;
     std::vector<uint64_t> write_ready_clients;
@@ -252,7 +316,7 @@ public:
     std::vector<addr_mask_t> osd_cluster_network_masks;
     std::vector<std::string> all_osd_networks;
     std::vector<addr_mask_t> all_osd_network_masks;
-    bool use_proto_checksums = true;
+    int use_proto_checksums = 0;
     // op statistics
     osd_op_stats_t stats, recovery_stats;
 
@@ -302,29 +366,33 @@ protected:
 
     bool try_send(osd_client_t *cl);
     void handle_send(int result, bool prev, bool more, osd_client_t *cl);
-    size_t op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len);
-    void op_get_write_buffers(osd_client_t *cl, std::vector<iovec> & lst);
+    bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr);
+    void next_write_op(osd_client_t *cl);
+    bool op_write_buf(osd_client_t *cl, uint8_t *src, size_t src_len, uint8_t *dst, size_t dst_len, bool skip_csum, size_t & from, size_t & done);
+    bool op_copy_data_to(osd_client_t *cl, uint8_t *dst, size_t dst_len, size_t & from, size_t & done);
+    size_t copy_ops_to(osd_client_t *cl, uint8_t *dst, size_t dst_len);
+    template<typename T> size_t copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t dst_len);
 
     void handle_read(int result, osd_client_t *cl);
     bool handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize);
+    template<typename T> bool handle_buffer_with(osd_client_t *cl, uint8_t *curbuf, size_t bufsize);
     bool handle_hdr(osd_client_t *cl);
     bool allocate_op_buffers(osd_client_t *cl);
     bool allocate_reply_buffers(osd_client_t *cl, osd_op_t *op);
-    bool op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done);
-    void op_get_read_buffers(osd_client_t *cl, std::vector<iovec> & lst);
-    void op_alloc_temp_buffers(osd_op_t *op, int i);
+    bool op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr);
     bool handle_finished_op(osd_client_t *cl);
+    void execute_verified_op(osd_client_t *cl, osd_op_t *op);
     void handle_immediate_ops();
 
-    bool op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *buf, size_t len, size_t from, size_t & done);
-    bool op_decrypted_copy_data_from(osd_client_t* cl, uint8_t *buf, size_t len, size_t from, size_t & done);
+    void op_encrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc);
+    void op_encrypt_free(osd_client_t* cl);
+    void op_decrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc);
     void op_decrypt_start(osd_client_t* cl);
     void op_decrypt_inline(osd_client_t* cl);
     void op_decrypt_free(osd_client_t* cl);
 
 #ifdef WITH_RDMA
     void try_send_rdma(osd_client_t *cl);
-    int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len);
     bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);

src/client/msgr_encrypt.cpp

@@ -186,13 +186,14 @@ op_aes_xts_decrypt_t::~op_aes_xts_decrypt_t()
         free(tmp);
 }
 
-void op_aes_xts_decrypt_t::start(uint8_t **key_chain, size_t chain_size, uint8_t *key_indexes, uint64_t start_offset, size_t block_size)
+void op_aes_xts_decrypt_t::start(uint8_t **key_chain, size_t chain_size, void *key_indexes, uint64_t start_offset, size_t block_size)
 {
     assert(!decrypted);
     this->start_offset = start_offset;
     this->key_chain = chain_size > 1 ? key_chain : 0;
     this->chain_size = chain_size > 1 ? chain_size : 0;
     this->key_indexes = chain_size > 1 ? key_indexes : NULL;
+    this->key_index_bytes = osd_op_rw_t::chain_info_bytes(chain_size);
     assert(chain_size <= 1 || key_indexes != NULL);
     this->block_size = block_size;
     this->offset = 0;
@@ -217,8 +218,15 @@ void op_aes_xts_decrypt_t::decrypt_block(uint8_t *in, uint8_t *out)
     uint8_t *key = NULL;
     if (chain_size)
     {
-        assert(key_indexes[offset/block_size] < chain_size);
-        key = key_chain[key_indexes[offset/block_size]];
+        uint32_t key_index = key_index_bytes == 1
+            ? ((uint8_t*)key_indexes)[offset/block_size]
+            : (key_index_bytes == 2
+                ? ((uint16_t*)key_indexes)[offset/block_size]
+                : (key_index_bytes == 4
+                    ? ((uint32_t*)key_indexes)[offset/block_size]
+                    : UINT32_MAX));
+        assert(key_index < chain_size);
+        key = key_chain[key_index];
         if (!key)
         {
             if (in != out)
@@ -325,86 +333,49 @@ void destroy_aes_xts_decrypt(op_aes_xts_decrypt_t *decrypt_ctx)
     delete decrypt_ctx;
 }
 
-bool osd_messenger_t::op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *enc_buf, size_t enc_len, size_t from, size_t & done)
+void osd_messenger_t::op_encrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc)
 {
-    auto op = cl->write_op;
-    auto & op_pos = cl->write_op_pos;
-    assert(op->req.hdr.opcode == OSD_OP_WRITE);
-    if (!from)
+    if (!cl->encrypt_ctx)
     {
-        if (!cl->encrypt_ctx)
+        if (encrypt_ctx_pool.size())
         {
-            if (encrypt_ctx_pool.size())
-            {
-                cl->encrypt_ctx = encrypt_ctx_pool.back();
-                encrypt_ctx_pool.pop_back();
-            }
-            else
-                cl->encrypt_ctx = new op_aes_xts_encrypt_t();
+            cl->encrypt_ctx = encrypt_ctx_pool.back();
+            encrypt_ctx_pool.pop_back();
         }
-        assert(op->enc->key_chain[0]);
-        cl->encrypt_ctx->start(op->enc->key_chain[0], op->req.rw.offset, op->enc->bitmap_granularity);
-    }
-    for (int i = 0; i < op->iov.count; i++)
-    {
-        uint8_t *plain = (uint8_t*)op->iov.buf[i].iov_base;
-        size_t plain_len = op->iov.buf[i].iov_len;
-        while (from < plain_len || cl->encrypt_ctx->has_buffered())
-        {
-            if (done >= enc_len)
-                return false;
-            size_t done_in = 0;
-            size_t done_out = 0;
-            cl->encrypt_ctx->update(plain+from, plain_len-from, enc_buf+done, enc_len-done, done_in, done_out);
-            if (cl->write_csum_state && done_out > 0)
-                XXH3_64bits_update(cl->write_csum_state, enc_buf+done, done_out);
-            done += done_out;
-            op_pos += done_in;
-            from += done_in;
-        }
-        from -= plain_len;
-    }
-    if (cl->encrypt_ctx)
-    {
-        if (encrypt_ctx_pool.size() > max_aes_xts_pool_size)
-            delete cl->encrypt_ctx;
         else
-            encrypt_ctx_pool.push_back(cl->encrypt_ctx);
-        cl->encrypt_ctx = NULL;
+            cl->encrypt_ctx = new op_aes_xts_encrypt_t();
+        assert(cl->write_op->enc->key_chain[0]);
+        cl->encrypt_ctx->start(cl->write_op->enc->key_chain[0], cl->write_op->req.rw.offset, cl->write_op->enc->bitmap_granularity);
+    }
+    while (done_enc < enc_len && (done_plain < plain_len || cl->encrypt_ctx->has_buffered()))
+    {
+        size_t done_in = 0;
+        size_t done_out = 0;
+        cl->encrypt_ctx->update(plain+done_plain, plain_len-done_plain, enc_buf+done_enc, enc_len-done_enc, done_in, done_out);
+        if (cl->write_csum_state && done_out > 0)
+            XXH3_64bits_update(cl->write_csum_state, enc_buf+done_enc, done_out);
+        done_enc += done_out;
+        cl->write_op_pos += done_in;
+        done_plain += done_in;
     }
-    return true;
 }
 
-bool osd_messenger_t::op_decrypted_copy_data_from(osd_client_t* cl, uint8_t *enc_buf, size_t enc_len, size_t from, size_t & done)
+void osd_messenger_t::op_decrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc)
 {
     op_decrypt_start(cl);
-    auto op = cl->read_op;
-    assert(op->req.hdr.opcode == OSD_OP_READ);
-    for (int i = 0; i < op->iov.count; i++)
+    while (done_plain < plain_len && done_enc < enc_len)
     {
-        uint8_t *plain = (uint8_t*)op->iov.buf[i].iov_base;
-        size_t plain_len = op->iov.buf[i].iov_len;
-        while (from < plain_len)
-        {
-            if (done >= enc_len)
-                return false;
-            size_t done_in = 0;
-            size_t done_out = 0;
-            // plain == NULL means skip output
-            cl->decrypt_ctx->update(enc_buf+done, enc_len-done, plain ? plain+from : NULL, plain_len-from, done_in, done_out);
-            if (cl->read_csum_state && done_in > 0)
-                XXH3_64bits_update(cl->read_csum_state, enc_buf+done, done_in);
-            done += done_in;
-            cl->read_op_pos += done_out;
-            cl->read_op_inline_decrypt_in += done_in;
-            from += done_out;
-            if (!done_out)
-                return false;
-        }
-        from -= plain_len;
+        size_t done_in = 0;
+        size_t done_out = 0;
+        // plain == NULL means skip output
+        cl->decrypt_ctx->update(enc_buf+done_enc, enc_len-done_enc, plain ? plain+done_plain : NULL, plain_len-done_plain, done_in, done_out);
+        if (cl->read_csum_state && done_in > 0)
+            XXH3_64bits_update(cl->read_csum_state, enc_buf+done_enc, done_in);
+        done_enc += done_in;
+        cl->read_op_pos += done_out;
+        cl->read_op_inline_decrypt_in += done_in;
+        done_plain += done_out;
     }
-    op_decrypt_free(cl);
-    return true;
 }
 
 void osd_messenger_t::op_decrypt_start(osd_client_t* cl)
@@ -419,6 +390,7 @@ void osd_messenger_t::op_decrypt_start(osd_client_t* cl)
         else
             cl->decrypt_ctx = new op_aes_xts_decrypt_t();
         auto & enc = cl->read_op->enc;
+        assert(cl->read_op->req.hdr.opcode == OSD_OP_READ);
         cl->decrypt_ctx->start(enc->key_chain, enc->chain_size,
             (cl->read_op->req.rw.flags & OSD_OP_RETURN_CHAIN) ? (uint8_t*)cl->read_op->bitmap + enc->read_chain_bitmap_pos : 0,
             cl->read_op->req.rw.offset, enc->bitmap_granularity);
@@ -468,7 +440,6 @@ void osd_messenger_t::op_decrypt_inline(osd_client_t* cl)
             from_out += done_out;
     }
     assert(j >= op->iov.count);
-    op_decrypt_free(cl);
 }
 
 void osd_messenger_t::op_decrypt_free(osd_client_t* cl)
@@ -482,3 +453,15 @@ void osd_messenger_t::op_decrypt_free(osd_client_t* cl)
         cl->decrypt_ctx = NULL;
     }
 }
+
+void osd_messenger_t::op_encrypt_free(osd_client_t* cl)
+{
+    if (cl->encrypt_ctx)
+    {
+        if (encrypt_ctx_pool.size() > max_aes_xts_pool_size)
+            delete cl->encrypt_ctx;
+        else
+            encrypt_ctx_pool.push_back(cl->encrypt_ctx);
+        cl->encrypt_ctx = NULL;
+    }
+}

src/client/msgr_encrypt.h

@@ -46,7 +46,8 @@ class op_aes_xts_decrypt_t
     uint64_t start_offset = 0;
     uint8_t **key_chain = NULL;
     size_t chain_size = 0;
-    uint8_t *key_indexes = NULL;
+    void *key_indexes = NULL;
+    int key_index_bytes = 0;
     size_t offset = 0;
     size_t block_size = 0;
     uint8_t *tmp = NULL;
@@ -61,7 +62,7 @@ public:
     ~op_aes_xts_decrypt_t();
 
     inline bool has_buffered() { return decrypted; };
-    void start(uint8_t **key_chain, size_t chain_size, uint8_t *key_indexes, uint64_t start_offset, size_t block_size);
+    void start(uint8_t **key_chain, size_t chain_size, void *key_indexes, uint64_t start_offset, size_t block_size);
     void update(uint8_t *in, size_t max_in, uint8_t *out, size_t max_out, size_t & done_in, size_t & done_out);
 };
 

src/client/msgr_op.h

@@ -182,7 +182,7 @@ struct __attribute__((visibility("default"))) osd_op_t
     void *bitmap = NULL;
     unsigned bitmap_len = 0;
     unsigned bmp_data = 0;
-    void *bitmap_buf = NULL;
+    uint8_t *bitmap_buf = NULL;
     void *rmw_buf = NULL;
     std::shared_ptr<osd_op_enc_t> enc;
     uint8_t *enc_buf = NULL;

src/client/msgr_rdma.cpp

@@ -187,6 +187,8 @@ std::vector<msgr_rdma_context_t*> msgr_rdma_context_t::create_all(const std::vec
     ibv_device **raw_dev_list = NULL;
     ibv_device **dev_list = NULL;
     ibv_device *single_list[2] = {};
+    int up_ports = 0;
+    int single_port_num = 0;
 
     raw_dev_list = dev_list = ibv_get_device_list(NULL);
     if (!dev_list || !*dev_list)
@@ -221,6 +223,7 @@ std::vector<msgr_rdma_context_t*> msgr_rdma_context_t::create_all(const std::vec
         dev_list = single_list;
     }
 
+retry:
     for (int i = 0; dev_list[i]; ++i)
     {
         auto dev = dev_list[i];
@@ -258,6 +261,9 @@ std::vector<msgr_rdma_context_t*> msgr_rdma_context_t::create_all(const std::vec
                 fprintf(stderr, "RDMA device %s port %d GID %d does not exist\n", ibv_get_device_name(dev), port_num, sel_gid_index);
                 continue;
             }
+            up_ports++;
+            single_port_num = port_num;
+            single_list[0] = dev;
             uint32_t port_mtu = sel_mtu ? sel_mtu : ibv_mtu_to_bytes(portinfo.active_mtu);
 #ifdef IBV_ADVISE_MR_ADVICE_PREFETCH_NO_FAULT
             if (sel_gid_index < 0)
@@ -298,6 +304,14 @@ cleanup_dev:
         ibv_close_device(context);
     }
 
+    if (!ret.size() && up_ports == 1 && dev_list != single_list)
+    {
+        // Auto-select the only available device/port if there is only one
+        dev_list = single_list;
+        sel_port_num = single_port_num;
+        goto retry;
+    }
+
 cleanup:
     if (raw_dev_list)
         ibv_free_device_list(raw_dev_list);
@@ -554,33 +568,6 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
     cl->rdma_conn->cur_send++;
 }
 
-int osd_messenger_t::try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len)
-{
-    int total_dst_len = dst_len;
-    while (dst_len > 0 && (cl->write_op || cl->write_ops.size()))
-    {
-        if (!cl->write_op)
-        {
-            cl->write_op = cl->write_ops.front();
-            cl->write_ops.pop_front();
-        }
-        osd_op_t *op = cl->write_op;
-        size_t copied = op_copy_to(cl, dst, dst_len);
-        if (!copied)
-        {
-            break;
-        }
-        dst += copied;
-        dst_len -= copied;
-        if (!cl->write_op && op->op_type == OSD_OP_IN)
-        {
-            // this is a reply, free the op after sending it
-            cl->send_free_ops.push_back(op);
-        }
-    }
-    return total_dst_len-dst_len;
-}
-
 void osd_messenger_t::try_send_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
@@ -603,10 +590,44 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
     while (!rc->send_out_full && copied > 0 && rc->cur_send < rc->max_send)
     {
         dst = (uint8_t*)rc->send_out.buf + rc->send_out_pos;
-        dst_len = (rc->send_out_pos < rc->send_out_size ? rc->send_out_size-rc->send_out_pos : rc->send_done_pos-rc->send_out_pos);
+        if (rc->send_out_pos >= rc->send_done_pos)
+        {
+            dst_len = rc->send_out_size-rc->send_out_pos;
+            if (dst_len < 4096)
+            {
+                // free end of the buffer is too small, skip
+                rc->send_out_pos = 0;
+                if (rc->send_out_pos >= rc->send_done_pos)
+                    rc->send_out_full = true;
+                if (!rc->send_sizes.size())
+                {
+                    rc->send_done_pos += dst_len;
+                    rc->send_out_full = false;
+                    if (rc->send_done_pos == rc->send_out_size)
+                        rc->send_done_pos = 0;
+                }
+                else
+                    rc->send_sizes.back() += dst_len;
+                continue;
+            }
+        }
+        else
+        {
+            dst_len = rc->send_done_pos-rc->send_out_pos;
+            if (dst_len < 4096)
+            {
+                // too small buffer, stop
+                break;
+            }
+        }
         if (dst_len > rc->max_msg)
             dst_len = rc->max_msg;
-        copied = try_send_rdma_copy(cl, dst, dst_len);
+        copied = copy_ops_to(cl, dst, dst_len);
+        if (cl->io_error)
+        {
+            stop_client(cl->client_id);
+            return;
+        }
         if (copied > 0)
         {
             rc->send_out_pos += copied;
@@ -744,10 +765,10 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
                     cl->send_free_ops.pop_front();
                 }
                 cl->send_free_ops.pop_front();
-                if (cl->proto_csum_status == MSGR_PEER_CSUM_IN && !cl->write_op && !cl->write_ops.size())
+                if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
                 {
                     // Checksums negotiated, enable
-                    cl->proto_csum_status = MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT;
+                    cl->proto_csum_status = cl->proto_csum_status & (~MSGR_CSUM_NEG);
                 }
                 try_send_rdma(cl);
             }

src/client/msgr_rdma.h

@@ -79,7 +79,7 @@ struct msgr_rdma_connection_t
     msgr_rdma_buf_t recv_buf;
     std::deque<uint64_t> send_sizes;
     msgr_rdma_buf_t send_out;
-    int send_out_pos = 0, send_done_pos = 0, send_out_size = 0;
+    size_t send_out_pos = 0, send_done_pos = 0, send_out_size = 0;
     bool send_out_full = false;
 
     ~msgr_rdma_connection_t();

src/client/msgr_stop.cpp

@@ -9,6 +9,12 @@
 #ifdef WITH_RDMA
 #include "msgr_rdma.h"
 #endif
+#ifdef WITH_OPENSSL
+#include <openssl/bio.h>
+#include <openssl/err.h>
+#include <openssl/pem.h>
+#include <openssl/ssl.h>
+#endif
 
 void osd_client_t::cancel_ops()
 {
@@ -204,6 +210,12 @@ osd_client_t::~osd_client_t()
         read_op->cancel();
         read_op = NULL;
     }
+    while (unverified_ops.size())
+    {
+        auto op = unverified_ops.back();
+        unverified_ops.pop_back();
+        op->cancel();
+    }
     // Cancel outbound ops
     cancel_ops();
     for (osd_op_t *op: send_free_ops)
@@ -217,7 +229,10 @@ osd_client_t::~osd_client_t()
     {
         if (op)
         {
-            delete op;
+            if (!((size_t)op & 7))
+                delete op;
+            else
+                free((void*)((size_t)op & ~(size_t)7));
         }
     }
 #ifndef __MOCK__
@@ -239,4 +254,28 @@ osd_client_t::~osd_client_t()
         XXH3_freeState(write_csum_state);
         write_csum_state = NULL;
     }
+#ifdef WITH_OPENSSL
+    if (enc_ctx)
+    {
+        EVP_CIPHER_CTX_free(enc_ctx);
+        enc_ctx = NULL;
+    }
+    if (dec_ctx)
+    {
+        EVP_CIPHER_CTX_free(dec_ctx);
+        dec_ctx = NULL;
+    }
+    if (ssl_cli)
+    {
+        SSL_free(ssl_cli);
+        ssl_cli = NULL;
+        write_to_ssl = NULL;
+        read_from_ssl = NULL;
+    }
+    if (ssl_out_buf)
+    {
+        free(ssl_out_buf);
+        ssl_out_buf = NULL;
+    }
+#endif
 }

src/client/osd_ops.h

@@ -231,12 +231,25 @@ struct __attribute__((__packed__)) osd_op_rw_t
     uint32_t len;
     // flags
     // OSD_OP_RETURN_CHAIN for chained reads: return parent number in chain for each block
+    // read_chain size comes after bitmap, takes 0 bytes / 1 byte / 2 byte / 4 byte per each block,
+    // depending on the number of parent inodes (0 parents = 0 bytes, up to 255 parents = 1 byte, etc)
     uint32_t flags;
     // inode metadata revision for chained reads
     uint64_t meta_revision;
     // object version for atomic "CAS" (compare-and-set) writes
     // writes and deletes fail with -EINTR if object version differs from (version-1)
     uint64_t version;
+
+    static inline size_t chain_info_bytes(size_t chain_size)
+    {
+        if (chain_size <= 1)
+            return 0;
+        if (chain_size <= 256)
+            return 1;
+        if (chain_size <= 65536)
+            return 2;
+        return 4;
+    }
 };
 
 struct __attribute__((__packed__)) osd_reply_rw_t

src/client/vitastor.pc.in

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

src/cmd/cli_rm_data.cpp

@@ -286,6 +286,8 @@ struct rm_inode_t
                     .data = data,
                 };
             }
+            // Wakeup callers (otherwise they wake up only on next loop())
+            parent->ringloop->wakeup();
         }
         in_continue = false;
     }

src/osd/osd.cpp

@@ -8,7 +8,6 @@
 #include <netinet/tcp.h>
 #include <arpa/inet.h>
 
-#include "xxhash.h"
 #include "addr_util.h"
 #include "osd_primary.h"
 #include "osd.h"
@@ -16,46 +15,8 @@
 #include "str_util.h"
 #include "json_util.h"
 
-void xxhash_run_test(XXH3_state_t *st, void *buf, size_t size)
-{
-    timespec tv, tv1;
-    clock_gettime(CLOCK_REALTIME, &tv);
-    uint64_t res = 1;
-    uint64_t done = 0;
-    while (true)
-    {
-        XXH3_64bits_reset(st);
-        XXH3_64bits_update(st, buf, size);
-        res *= XXH3_64bits_digest(st);
-        done++;
-        if (!(done % 100000))
-        {
-            clock_gettime(CLOCK_REALTIME, &tv1);
-            if ((tv1.tv_sec-tv.tv_sec)*1000 + (tv1.tv_nsec-tv.tv_nsec)/1000000 > 1000)
-                break;
-        }
-    }
-    double sec = (tv1.tv_sec-tv.tv_sec) + (tv1.tv_nsec-tv.tv_nsec)/1000000000.0;
-    fprintf(stderr, "xxhash3: %lu %zu byte blocks in %.2f sec = %.1f MB/s\n",
-        done, size, sec, done*size/sec/1024/1024);
-}
-
-void xxhash_bench()
-{
-    std::vector<uint8_t> buf;
-    buf.resize(128*1024);
-    for (size_t i = 0; i < buf.size(); i++)
-        buf[i] = ((i+37)*3) % 255;
-    XXH3_state_t *st = XXH3_createState();
-    xxhash_run_test(st, buf.data(), 4096);
-    xxhash_run_test(st, buf.data(), buf.size());
-    XXH3_freeState(st);
-}
-
 osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
 {
-    xxhash_bench();
-
     zero_buffer_size = 1<<20;
     zero_buffer = malloc_or_die(zero_buffer_size);
     memset(zero_buffer, 0, zero_buffer_size);

src/osd/osd_primary.cpp

@@ -96,7 +96,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             if (inode_it->second.parent_id == cur_op->req.rw.inode ||
                 inode_it->second.parent_id == inode_it->second.num ||
                 chain_size > st_cli.inode_config.size() ||
-                chain_size > 255)
+                chain_size > UINT32_MAX)
             {
                 printf("Inode %ju from pool %u has too many parents, returning EINVAL in response to read\n",
                     INODE_NO_POOL(cur_op->req.rw.inode), INODE_POOL(cur_op->req.rw.inode));
@@ -111,7 +111,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             // Add the original inode
             chain_size++;
             chain_info_len = (cur_op->req.rw.flags & OSD_OP_RETURN_CHAIN
-                ? (cur_op->req.rw.len / bs_bitmap_granularity)
+                ? osd_op_rw_t::chain_info_bytes(chain_size) * (cur_op->req.rw.len / bs_bitmap_granularity)
                 : 0);
         }
     }
@@ -130,9 +130,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             stripe_count * clean_entry_bitmap_size +
             // - 'missing' flags for chained reads
             (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 0 : pg_it->second.pg_size)
-        ) +
-        // read chain info
-        chain_info_len
+        )
     );
     void *data_buf = (uint8_t*)op_data + sizeof(osd_primary_op_data_t);
     op_data->pg_num = pg_num;
@@ -140,6 +138,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
     op_data->oid = oid;
     op_data->stripes = (osd_rmw_stripe_t*)data_buf;
     op_data->stripe_count = stripe_count;
+    op_data->chain_info = NULL;
     data_buf = (uint8_t*)data_buf + sizeof(osd_rmw_stripe_t) * stripe_count;
     cur_op->op_data = op_data;
     if (cur_op->req.hdr.opcode != OSD_OP_SCRUB)
@@ -147,10 +146,17 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
         split_stripes(pg_data_size, bs_block_size, (uint32_t)(cur_op->req.rw.offset - oid.stripe), cur_op->req.rw.len, op_data->stripes);
         // Resulting bitmaps have to survive op_data and be freed with the op itself
         assert(!cur_op->bitmap_buf);
-        cur_op->bitmap_buf = calloc_or_die(1, clean_entry_bitmap_size * stripe_count);
+        cur_op->bitmap_buf = (uint8_t*)calloc_or_die(1, clean_entry_bitmap_size*stripe_count + chain_info_len);
+        uint8_t *buf = cur_op->bitmap_buf;
         for (int i = 0; i < stripe_count; i++)
         {
-            op_data->stripes[i].bmp_buf = (uint8_t*)cur_op->bitmap_buf + clean_entry_bitmap_size * i;
+            op_data->stripes[i].bmp_buf = buf;
+            buf += clean_entry_bitmap_size;
+            if (i == pg_data_size-1 && chain_info_len)
+            {
+                op_data->chain_info = buf;
+                buf += chain_info_len;
+            }
         }
     }
     op_data->chain_size = chain_size;
@@ -164,11 +170,6 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
         data_buf = (uint8_t*)data_buf + chain_size * stripe_count * clean_entry_bitmap_size;
         op_data->missing_flags = (uint8_t*)data_buf;
         data_buf = (uint8_t*)data_buf + chain_size * (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 0 : pg_it->second.pg_size);
-        if (chain_info_len)
-        {
-            op_data->chain_info = (uint8_t*)data_buf;
-            data_buf = (uint8_t*)data_buf + chain_info_len;
-        }
         // Copy chain
         int chain_num = 0;
         op_data->read_chain[chain_num] = cur_op->req.rw.inode;
@@ -319,6 +320,7 @@ resume_2:
     }
     cur_op->reply.rw.version = op_data->fact_ver;
     cur_op->reply.rw.bitmap_len = (pg ? pg->pg_data_size : 1) * clean_entry_bitmap_size;
+    cur_op->bitmap = op_data->stripes[0].bmp_buf;
     if (op_data->degraded)
     {
         // Reconstruct missing stripes
@@ -331,7 +333,6 @@ resume_2:
         {
             reconstruct_stripes_ec(stripes, pg->pg_size, pg->pg_data_size, clean_entry_bitmap_size);
         }
-        cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len);
         for (int role = 0; role < pg->pg_size; role++)
         {
             if (stripes[role].req_end != 0)
@@ -346,7 +347,6 @@ resume_2:
     }
     else
     {
-        cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len);
         cur_op->iov.push_back(cur_op->buf, cur_op->req.rw.len);
     }
     finish_op(cur_op, cur_op->req.rw.len);

src/osd/osd_primary.h

@@ -56,7 +56,7 @@ struct osd_primary_op_data_t
             int chain_size;
             osd_chain_read_t *chain_reads;
             int chain_read_count;
-            uint8_t *chain_info;
+            void *chain_info;
         };
     };
 };

src/osd/osd_primary_chain.cpp

@@ -582,11 +582,10 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
         }
     }
     // Send bitmap
+    cur_op->bitmap = op_data->stripes[0].bmp_buf;
     cur_op->reply.rw.bitmap_len = (pg ? pg->pg_data_size : 1) * clean_entry_bitmap_size;
-    cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len);
     if (cur_op->req.rw.flags & OSD_OP_RETURN_CHAIN)
     {
-        cur_op->iov.push_back(op_data->chain_info, (cur_op->req.rw.len / bs_bitmap_granularity));
         cur_op->reply.rw.bitmap_len += (cur_op->req.rw.len / bs_bitmap_granularity);
     }
     // And finally compose the result
@@ -596,6 +595,7 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
     int prev = (cur_op->req.rw.offset - op_data->oid.stripe) / bs_bitmap_granularity;
     int end = prev + cur_op->req.rw.len/bs_bitmap_granularity;
     int cur = prev;
+    size_t key_index_bytes = osd_op_rw_t::chain_info_bytes(op_data->chain_size);
     while (cur <= end)
     {
         bool has_bit = false;
@@ -607,7 +607,14 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
                 if (has_bit)
                 {
                     if (op_data->chain_info)
-                        op_data->chain_info[cur] = pos;
+                    {
+                        if (key_index_bytes == 1)
+                            ((uint8_t*)op_data->chain_info)[cur] = pos;
+                        else if (key_index_bytes == 2)
+                            ((uint16_t*)op_data->chain_info)[cur] = pos;
+                        else
+                            ((uint32_t*)op_data->chain_info)[cur] = pos;
+                    }
                     break;
                 }
             }

src/osd/osd_scrub.cpp

@@ -413,7 +413,7 @@ void osd_t::submit_scrub_subops(osd_op_t *cur_op)
         }
     }
     assert(!cur_op->bitmap_buf);
-    cur_op->bitmap_buf = calloc_or_die(1, clean_entry_bitmap_size * op_data->stripe_count);
+    cur_op->bitmap_buf = (uint8_t*)calloc_or_die(1, clean_entry_bitmap_size * op_data->stripe_count);
     for (int i = 0; i < op_data->stripe_count; i++)
     {
         op_data->stripes[i].bmp_buf = (uint8_t*)cur_op->bitmap_buf + clean_entry_bitmap_size * i;

src/osd/osd_secondary.cpp

@@ -348,10 +348,17 @@ void osd_t::exec_show_config(osd_op_t *cur_op)
         cl->read_op_id = cur_op->req.hdr.id + 1;
     }
     auto features = json11::Json::object{ { "pg_locks", true } };
-    if (req_json["features"]["proto_checksums"].bool_value() && msgr.use_proto_checksums)
+    if (msgr.use_proto_checksums)
     {
-        cl->proto_csum_status = MSGR_PEER_CSUM_IN;
-        features["proto_checksums"] = true;
+        auto peer_csums = req_json["features"]["proto_checksums"].uint64_value();
+        if (peer_csums == MSGR_CSUM_FULL || peer_csums == MSGR_CSUM_PAYLOAD)
+        {
+            if (msgr.use_proto_checksums == MSGR_CSUM_FULL && peer_csums == MSGR_CSUM_FULL)
+                cl->proto_csum_status = MSGR_CSUM_FULL|MSGR_CSUM_NEG;
+            else
+                cl->proto_csum_status = MSGR_CSUM_PAYLOAD|MSGR_CSUM_NEG;
+            features["proto_checksums"] = msgr.use_proto_checksums;
+        }
     }
     // Expose sensitive configuration values so peers can check them
     json11::Json::object wire_config = json11::Json::object {

src/test/CMakeLists.txt

@@ -41,7 +41,7 @@ add_executable(test_heap
 	../blockstore/multilist.cpp
 	../blockstore/blockstore_heap.cpp
 	../util/crc32c.c
-	../util/xxhash.c
+	../util/xxh_x86dispatch.c
 	../util/allocator.cpp
 	../blockstore/blockstore_disk.cpp
 	../util/str_util.cpp

src/test/test_heap.cpp

@@ -1893,7 +1893,8 @@ void test_big_intent_csums()
         heap.finish_recheck();
 
         auto mod = heap.get_recheck_modified_blocks();
-        assert(mod.size() == 0);
+        assert(mod.size() == 1);
+        assert(mod[0] == 0);
 
         // read object 1 - big_intent should be there
         object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };

tests/common.sh

@@ -27,6 +27,7 @@ ETCD_COUNT=${ETCD_COUNT:-1}
 ANTIETCD=${ANTIETCD}
 USE_RAMDISK=${USE_RAMDISK}
 ETCD_SCHEME=${ETCD_SCHEME:-http}
+OSD_TLS=${OSD_TLS}
 
 RAMDISK=/run/user/$(id -u)
 findmnt $RAMDISK >/dev/null || (sudo mkdir -p $RAMDISK && sudo mount -t tmpfs tmpfs $RAMDISK)
@@ -124,6 +125,24 @@ VITASTOR_CFG='"etcd_address":"'$ETCD_URL'"'"$VITASTOR_CFG"
 if [[ "$ETCD_SCHEME" = "https" ]]; then
     VITASTOR_CFG="$VITASTOR_CFG"',"etcd_ca":"'$(pwd)'/testdata/etcd.crt"'
 fi
+if [[ "$OSD_TLS" = "1" ]]; then
+    cd ./testdata
+    openssl req -days 3650 -x509 -new -newkey rsa:4096 -nodes -keyout client_ca.key -out client_ca.crt \
+        -subj '/C=RU/ST=Russia/L=Moscow/O=VitastorClientCA'
+    openssl req -days 3650 -x509 -new -newkey rsa:4096 -nodes -keyout osd.key -out osd.crt \
+        -subj '/C=RU/ST=Russia/L=Moscow/O=VitastorOSD' -addext "extendedKeyUsage = serverAuth, clientAuth"
+    openssl req -subj '/CN=test' -nodes -new -keyout cli.key -out cli.csr -addext "extendedKeyUsage = clientAuth"
+    openssl x509 -req -days 3650 -CA client_ca.crt -CAkey client_ca.key -CAcreateserial -in cli.csr -out cli.crt
+    rm cli.csr
+    cd $(dirname $0)/..
+    VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_cert":"'$(pwd)'/testdata/osd.crt"'
+    VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_key":"'$(pwd)'/testdata/osd.key"'
+    VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_ca":"'$(pwd)'/testdata/osd.crt"'
+    VITASTOR_CFG="$VITASTOR_CFG"',"client_tls_ca":"'$(pwd)'/testdata/client_ca.crt"'
+    VITASTOR_CFG="$VITASTOR_CFG"',"tls_cert":"'$(pwd)'/testdata/cli.crt"'
+    VITASTOR_CFG="$VITASTOR_CFG"',"tls_key":"'$(pwd)'/testdata/cli.key"'
+fi
+VITASTOR_CFG="$VITASTOR_CFG"',"test_osd_aes_key":"'$(openssl rand -hex 32)'"'
 echo "{$VITASTOR_CFG}" > ./testdata/vitastor.conf
 VITASTOR_CFG=./testdata/vitastor.conf
 VITASTOR_CLI="build/src/cmd/vitastor-cli --config_path $VITASTOR_CFG"