Allow 2 and 4 byte per block chain_info (allow more than 255 snapshots with encryption)

src/client/cluster_client.cpp

@@ -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
+            // read chain_info - max 4 bytes per block
-            + (op->enc ? op->len/pool_cfg.bitmap_granularity : 0));
+            + (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/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);
+        uint32_t key_index = key_index_bytes == 1
-        key = key_chain[key_indexes[offset/block_size]];
+            ? ((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)

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_receive.cpp

@@ -688,7 +688,7 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
         // Read data. In this case we assume that the buffer is preallocated by the caller (!)
         unsigned bmp_len = (op->reply.hdr.opcode == OSD_OP_SEC_READ ? op->reply.sec_rw.attr_len : op->reply.rw.bitmap_len);
         unsigned expected_size = (op->reply.hdr.opcode == OSD_OP_SEC_READ ? op->req.sec_rw.len : op->req.rw.len);
-        if (op->reply.hdr.retval >= 0 && (op->reply.hdr.retval != expected_size || bmp_len > op->bitmap_len))
+        if (op->reply.hdr.retval >= 0 && (op->reply.hdr.retval != expected_size || bmp_len != op->bitmap_len))
         {
             // Check reply length to not overflow the buffer
             fprintf(stderr, "Client %ju read reply of different length: expected %u+%u, got %jd+%u\n",

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/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);
         }
     }
@@ -138,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)

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

@@ -595,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;
@@ -606,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;
                 }
             }