Allow to return chain_info in response to reads

src/client/cluster_client.cpp

@@ -958,14 +958,23 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
     {
         op->flags |= OP_IMMEDIATE_COMMIT;
     }
-    auto ino_it = st_cli.inode_config.find(op->inode);
+    // FIXME: Rework client API by adding open/close and cache inode information in the "FD"
-    if (ino_it != st_cli.inode_config.end() && ino_it->second.enc)
+    bool searched = false;
+    std::map<inode_t, inode_config_t>::iterator ino_it;
+    if (op->opcode == OSD_OP_READ || op->opcode == OSD_OP_WRITE)
     {
-        // FIXME: Rework client API by adding open/close and cache inode information in the "FD"
+        if (!searched)
-        op->enc = ino_it->second.enc;
-        if (!op->enc->bitmap_granularity)
         {
-            op->enc->bitmap_granularity = pool_it->second.bitmap_granularity;
+            ino_it = st_cli.inode_config.find(op->inode);
+            searched = true;
+        }
+        if (ino_it != st_cli.inode_config.end() && ino_it->second.enc)
+        {
+            op->enc = ino_it->second.enc;
+            if (!op->enc->bitmap_granularity)
+            {
+                op->enc->bitmap_granularity = pool_it->second.bitmap_granularity;
+            }
         }
     }
     else
@@ -974,6 +983,11 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
     }
     if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && !(op->flags & OSD_OP_IGNORE_READONLY))
     {
+        if (!searched)
+        {
+            ino_it = st_cli.inode_config.find(op->inode);
+            searched = true;
+        }
         if (ino_it != st_cli.inode_config.end() && ino_it->second.readonly)
         {
             op->retval = -EROFS;
@@ -985,6 +999,11 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
     op->deoptimise_snapshot = false;
     if (enable_writeback && (op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP))
     {
+        if (!searched)
+        {
+            ino_it = st_cli.inode_config.find(op->inode);
+            searched = true;
+        }
         if (ino_it != st_cli.inode_config.end())
         {
             int chain_size = 0;
@@ -1264,7 +1283,11 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
         // Allocate memory for the bitmap
         unsigned object_bitmap_size = ((op->len / pool_cfg.bitmap_granularity + 7) / 8);
         object_bitmap_size = (object_bitmap_size < 8 ? 8 : object_bitmap_size);
-        unsigned bitmap_mem = object_bitmap_size + (pool_cfg.data_block_size / pool_cfg.bitmap_granularity / 8 * pg_data_size) * op->parts.size();
+        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));
         if (!op->bitmap_buf || op->bitmap_buf_size < bitmap_mem)
         {
             op->bitmap_buf = realloc_or_die(op->bitmap_buf, bitmap_mem);
@@ -1419,9 +1442,9 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
             int peer_fd = peer_it->second;
             part->flags |= PART_SENT|PART_VALID;
             op->inflight_count++;
-            uint64_t pg_bitmap_size = (pool_cfg.data_block_size / pool_cfg.bitmap_granularity / 8) * (
+            uint32_t pg_data_size = (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks);
-                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));
             uint64_t meta_rev = 0;
             if (op->opcode != OSD_OP_READ_BITMAP && op->opcode != OSD_OP_DELETE && !op->deoptimise_snapshot)
             {
@@ -1440,6 +1463,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                     .inode = op->cur_inode,
                     .offset = part->offset,
                     .len = part->len,
+                    .flags = op->opcode == OSD_OP_READ && op->enc ? OSD_OP_RETURN_CHAIN : 0,
                     .meta_revision = meta_rev,
                     .version = op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE ? op->version : 0,
                 } },

src/client/osd_ops.h

@@ -37,6 +37,7 @@
 
 #define OSD_OP_RECOVERY_RELATED     (uint32_t)1
 #define OSD_OP_IGNORE_PG_LOCK       (uint32_t)2
+#define OSD_OP_RETURN_CHAIN         (uint32_t)4
 
 // Memory alignment for direct I/O (usually 512 bytes)
 #ifndef DIRECT_IO_ALIGNMENT
@@ -228,9 +229,10 @@ struct __attribute__((__packed__)) osd_op_rw_t
     uint64_t offset;
     // length. 0 means to read all bitmaps of the specified range, but no data.
     uint32_t len;
-    // flags (for future)
+    // flags
+    // OSD_OP_RETURN_CHAIN for chained reads: return parent number in chain for each block
     uint32_t flags;
-    // inode metadata revision
+    // 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)
@@ -245,6 +247,15 @@ struct __attribute__((__packed__)) osd_reply_rw_t
     uint32_t pad0;
     // for reads and writes: object version
     uint64_t version;
+    // for reads: chain info size for OSD_OP_RETURN_CHAIN
+    // (parent number is returned as a variable 2^N number of bits)
+    // i.e. 0 bits = everything is read from the inode itself
+    // 1 bit = inode(0) or its parent(1)
+    // 2 bits = inode(0) or its 1-3 parents
+    // 4 bits = inode(0) or its 1-7 parents
+    // and etc
+    // chain size in bits is (op->req.rw.len / bitmap_granularity * chain_bits_per_block + 7) / 8
+    uint32_t chain_bits_per_block;
 };
 
 struct __attribute__((__packed__)) osd_reply_del_t

src/osd/osd_primary.cpp

@@ -74,6 +74,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
     int stripe_count = (cur_op->req.hdr.opcode == OSD_OP_SCRUB ? 0 :
         (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pg_it->second.pg_size));
     int chain_size = 0;
+    int chain_info_len = 0;
     if (cur_op->req.hdr.opcode == OSD_OP_READ && cur_op->req.rw.meta_revision > 0)
     {
         // Chained read
@@ -94,9 +95,10 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             // Check for loops - FIXME check it in etcd_state_client
             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 > st_cli.inode_config.size() ||
+                chain_size > 255)
             {
-                printf("Inode %ju from pool %u has a parent_id loop, returning EINVAL in response to read\n",
+                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));
                 finish_op(cur_op, -EINVAL);
                 return false;
@@ -108,6 +110,9 @@ 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)
+                : 0);
         }
     }
     osd_primary_op_data_t *op_data = (osd_primary_op_data_t*)calloc_or_die(
@@ -125,7 +130,9 @@ 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;
@@ -157,6 +164,11 @@ 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;

src/osd/osd_primary.h

@@ -56,6 +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;
         };
     };
 };

src/osd/osd_primary_chain.cpp

@@ -584,6 +584,11 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
     // Send bitmap
     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
     uint64_t sent = 0;
     int prev_pos = 0, pos = 0;
@@ -600,7 +605,11 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
             {
                 has_bit = (((uint8_t*)op_data->snapshot_bitmaps)[pos*stripe_count*clean_entry_bitmap_size + cur/8] >> (cur%8)) & 1;
                 if (has_bit)
+                {
+                    if (op_data->chain_info)
+                        op_data->chain_info[cur] = pos;
                     break;
+                }
             }
         }
         if (has_bit != prev_set || pos != prev_pos || cur == end)