crimson/osd: add support for attr retrieval of MOSDECSubOpRead

Signed-off-by: Radosław Zarzyński <rzarzyns@redhat.com>

diff --git a/src/crimson/osd/ec_backend.cc b/src/crimson/osd/ec_backend.cc
index 77b693a6aafbf1dba7f929ab93ee3250b38aed0a..4600296795382b29856128e872633482db1dce57 100644 (file)
--- a/src/crimson/osd/ec_backend.cc
+++ b/src/crimson/osd/ec_backend.cc
@@ -128,10 +128,10 @@ ECBackend::handle_rep_read_op(Ref<MOSDECSubOpRead> m)
     const ECSubRead &op = m->op;
     reply.from = whoami;
     reply.tid = op.tid;
+    using read_ertr = crimson::os::FuturizedStore::Shard::read_errorator;
     return interruptor::do_for_each(op.to_read, [&op, &reply, this] (auto read_item) {
       const auto& [obj, op_list] = read_item;
       return interruptor::do_for_each(op_list, [&op, &reply, obj, this] (auto op_spec) {
-        using read_ertr = crimson::os::FuturizedStore::Shard::read_errorator;
         const auto& [off, size, flags] = op_spec;
         return maybe_chunked_read(
           obj, op, off, size, flags
@@ -152,8 +152,27 @@ ECBackend::handle_rep_read_op(Ref<MOSDECSubOpRead> m)
           return read_ertr::now();
         }));
       });
-    }).si_then([this] {
-      return ll_read_ierrorator::now();
+    }).si_then([&op, &reply, FNAME, this] {
+      DEBUG("attrs_to_read {}", op.attrs_to_read.size());
+      return interruptor::do_for_each(op.attrs_to_read,
+                                     [&reply, FNAME, this] (auto obj_attr) {
+       DEBUG("fulfilling attr request on obj {}", obj_attr);
+       if (reply.errors.count(obj_attr)) {
+          return read_ertr::now();
+       }
+        return store->get_attrs(
+          coll, ghobject_t{obj_attr, ghobject_t::NO_GEN, get_shard()}
+       ).safe_then([&reply, obj_attr] (auto&& attrs) {
+         reply.attrs_read[obj_attr] = std::move(attrs);
+          return read_ertr::now();
+        }, read_ertr::all_same_way([&reply, obj_attr] (const auto& e) {
+          assert(e.value() > 0);
+          reply.attrs_read.erase(obj_attr);
+          reply.buffers_read.erase(obj_attr);
+          reply.errors[obj_attr] = -e.value();
+          return read_ertr::now();
+       }));
+      });
     });
   });
 }