[ntuple] DAOS support: applied minor changes from the code review

jalopezg-git · jblomer · commit 89deeb0dab9f · 2021-05-03T23:07:30.000+02:00
diff --git a/tree/ntuple/v7/inc/ROOT/RDaos.hxx b/tree/ntuple/v7/inc/ROOT/RDaos.hxx
@@ -6,7 +6,7 @@
 /// is welcome!
 
 /*************************************************************************
- * Copyright (C) 1995-2020, Rene Brun and Fons Rademakers.               *
+ * Copyright (C) 1995-2021, Rene Brun and Fons Rademakers.               *
  * All rights reserved.                                                  *
  *                                                                       *
  * For the licensing terms see $ROOTSYS/LICENSE.                         *
@@ -43,13 +43,14 @@ class RDaosPool {
    friend class RDaosContainer;
 private:
    daos_handle_t fPoolHandle{};
-   daos_pool_info_t fPoolInfo{};
    uuid_t fPoolUuid{};
 
 public:
-   RDaosPool() = delete;
+   RDaosPool(const RDaosPool&) = delete;
    RDaosPool(std::string_view poolUuid, std::string_view serviceReplicas);
    ~RDaosPool();
+
+   RDaosPool& operator=(const RDaosPool&) = delete;
 };
 
 /**
@@ -85,9 +86,15 @@ public:
       FetchUpdateArgs(const FetchUpdateArgs&) = delete;
       FetchUpdateArgs(FetchUpdateArgs&& fua);
       FetchUpdateArgs(DistributionKey_t &d, AttributeKey_t &a, std::vector<d_iov_t> &v, daos_event_t *p = nullptr);
+      FetchUpdateArgs& operator=(const FetchUpdateArgs&) = delete;
 
+      /// \brief A `daos_key_t` is a type alias of `d_iov_t`. This type stores a pointer and a length.
+      /// In order for `fDistributionKey` and `fIods` to point to memory that we own, `fDkey` and
+      /// `fAkey` store a copy of the distribution and attribute key, respectively.
       DistributionKey_t fDkey{};
       AttributeKey_t fAkey{};
+
+      /// \brief The distribution key, as used by the `daos_obj_{fetch,update}` functions.
       daos_key_t fDistributionKey{};
       daos_iod_t fIods[1] = {};
       d_sg_list_t fSgls[1] = {};
@@ -130,19 +137,24 @@ private:
       daos_handle_t fQueue;
       DaosEventQueue(std::size_t size);
       ~DaosEventQueue();
+      /**
+        \brief Wait for all events in this event queue to complete.
+        \return Number of events still in the queue. This should be 0 on success.
+       */
       int Poll();
    };
 
    daos_handle_t fContainerHandle{};
-   daos_cont_info_t fContainerInfo{};
    uuid_t fContainerUuid{};
    std::shared_ptr<RDaosPool> fPool;
    /// OID that will be used by the next call to `WriteObject(const void *, std::size_t, DKeyT, AKeyT)`.
    daos_obj_id_t fSequentialWrOid{};
 
-   /** \brief Perform a vector read/write operation on different objects.
+   /**
+     \brief Perform a vector read/write operation on different objects.
      \param vec A `std::vector<RWOperation>` that describes read/write operations to perform.
      \param fn Either `std::mem_fn<&RDaosObject::Fetch>` (read) or `std::mem_fn<&RDaosObject::Update>` (write).
+     \return Number of requests that did not complete; this should be 0 after a successful call.
      */
    template <typename Fn, typename DKeyT, typename AKeyT>
    int VectorReadWrite(std::vector<RWOperation<DKeyT, AKeyT>> &vec, Fn fn) {
@@ -168,7 +180,15 @@ public:
    RDaosContainer(std::shared_ptr<RDaosPool> pool, std::string_view containerUuid, bool create = false);
    ~RDaosContainer();
 
-   /** \brief Read data from an object in this container to the given buffer. */
+   /**
+     \brief Read data from an object in this container to the given buffer.
+     \param oid A 128-bit DAOS object identifier.
+     \param buffer The address of a buffer that has capacity for at least `length` bytes.
+     \param length Length of the buffer.
+     \param dkey The distribution key used for this operation.
+     \param akey The attribute key used for this operation.
+     \return 0 if the operation succeeded; a negative DAOS error number otherwise.
+     */
    template <typename DKeyT, typename AKeyT>
    int ReadObject(daos_obj_id_t oid, void *buffer, std::size_t length, DKeyT dkey, AKeyT akey)
    {
@@ -178,7 +198,15 @@ public:
       return RDaosObject<DKeyT, AKeyT>(*this, oid).Fetch(args);
    }
 
-   /** \brief Write the given buffer to an object in this container. */
+   /**
+     \brief Write the given buffer to an object in this container.
+     \param oid A 128-bit DAOS object identifier.
+     \param buffer The address of the source buffer.
+     \param length Length of the buffer.
+     \param dkey The distribution key used for this operation.
+     \param akey The attribute key used for this operation.
+     \return 0 if the operation succeeded; a negative DAOS error number otherwise.
+     */
    template <typename DKeyT, typename AKeyT>
    int WriteObject(daos_obj_id_t oid, const void *buffer, std::size_t length, DKeyT dkey, AKeyT akey)
    {
@@ -188,7 +216,14 @@ public:
       return RDaosObject<DKeyT, AKeyT>(*this, oid).Update(args);
    }
 
-   /** \brief Write the given buffer to an object in this container and return a generated OID. */
+   /**
+     \brief Write the given buffer to an object in this container and return a generated OID.
+     \param buffer The address of the source buffer.
+     \param length Length of the buffer.
+     \param dkey The distribution key used for this operation.
+     \param akey The attribute key used for this operation.
+     \return A `std::tuple<>` that contains the generated OID and a DAOS error number (0 if the operation succeeded).
+     */
    template <typename DKeyT, typename AKeyT>
    std::tuple<daos_obj_id_t, int> WriteObject(const void *buffer, std::size_t length, DKeyT dkey, AKeyT akey)
    {
@@ -198,14 +233,20 @@ public:
       return ret;
    }
 
-   /** \brief Perform a vector read operation on (possibly) multiple objects.
-     \param vec A `std::vector<RWOperation>` that describes read operations to perform. */
+   /**
+     \brief Perform a vector read operation on (possibly) multiple objects.
+     \param vec A `std::vector<RWOperation>` that describes read operations to perform.
+     \return Number of operations that could not complete.
+     */
    template <typename DKeyT, typename AKeyT>
    int ReadV(std::vector<RWOperation<DKeyT, AKeyT>> &vec)
    { return VectorReadWrite(vec, std::mem_fn(&RDaosObject<DKeyT, AKeyT>::Fetch)); }
 
-   /** \brief Perform a vector write operation on (possibly) multiple objects.
-     \param vec A `std::vector<RWOperation>` that describes write operations to perform. */
+   /**
+     \brief Perform a vector write operation on (possibly) multiple objects.
+     \param vec A `std::vector<RWOperation>` that describes write operations to perform.
+     \return Number of operations that could not complete.
+     */
    template <typename DKeyT, typename AKeyT>
    int WriteV(std::vector<RWOperation<DKeyT, AKeyT>> &vec)
    { return VectorReadWrite(vec, std::mem_fn(&RDaosObject<DKeyT, AKeyT>::Update)); }
diff --git a/tree/ntuple/v7/src/RDaos.cxx b/tree/ntuple/v7/src/RDaos.cxx
@@ -14,6 +14,7 @@
  *************************************************************************/
 
 #include <ROOT/RDaos.hxx>
+#include <ROOT/RError.hxx>
 
 #include <numeric>
 #include <stdexcept>
@@ -24,10 +25,11 @@ ROOT::Experimental::Detail::RDaosPool::RDaosPool(std::string_view poolUuid, std:
       SvcRAII(std::string_view ranks) { rankList = daos_rank_list_parse(ranks.data(), "_"); }
       ~SvcRAII() { d_rank_list_free(rankList); }
    } Svc(serviceReplicas);
+   daos_pool_info_t poolInfo{};
 
    uuid_parse(poolUuid.data(), fPoolUuid);
-   if (int err = daos_pool_connect(fPoolUuid, nullptr, Svc.rankList, DAOS_PC_RW, &fPoolHandle, &fPoolInfo, nullptr))
-      throw std::runtime_error("daos_pool_connect: error: " + std::string(d_errstr(err)));
+   if (int err = daos_pool_connect(fPoolUuid, nullptr, Svc.rankList, DAOS_PC_RW, &fPoolHandle, &poolInfo, nullptr))
+      throw RException(R__FAIL("daos_pool_connect: error: " + std::string(d_errstr(err))));
 }
 
 ROOT::Experimental::Detail::RDaosPool::~RDaosPool() {
@@ -77,7 +79,7 @@ ROOT::Experimental::Detail::RDaosObject<DKeyT, AKeyT>::RDaosObject(RDaosContaine
       ofeats |= DAOS_OF_AKEY_UINT64;
    daos_obj_generate_id(&oid, ofeats /*| DAOS_OF_ARRAY_BYTE*/, cid, 0);
    if (int err = daos_obj_open(container.fContainerHandle, oid, DAOS_OO_RW, &fObjectHandle, nullptr))
-      throw std::runtime_error("daos_obj_open: error: " + std::string(d_errstr(err)));
+      throw RException(R__FAIL("daos_obj_open: error: " + std::string(d_errstr(err))));
 }
 
 template <typename DKeyT, typename AKeyT>
@@ -139,14 +141,16 @@ ROOT::Experimental::Detail::RDaosContainer::RDaosContainer(std::shared_ptr<RDaos
                                                            std::string_view containerUuid, bool create)
   : fPool(pool)
 {
+   daos_cont_info_t containerInfo{};
+
    uuid_parse(containerUuid.data(), fContainerUuid);
    if (create) {
       if (int err = daos_cont_create(fPool->fPoolHandle, fContainerUuid, nullptr, nullptr))
-         throw std::runtime_error("daos_cont_create: error: " + std::string(d_errstr(err)));
+         throw RException(R__FAIL("daos_cont_create: error: " + std::string(d_errstr(err))));
    }
    if (int err = daos_cont_open(fPool->fPoolHandle, fContainerUuid, DAOS_COO_RW,
-         &fContainerHandle, &fContainerInfo, nullptr))
-      throw std::runtime_error("daos_cont_open: error: " + std::string(d_errstr(err)));
+         &fContainerHandle, &containerInfo, nullptr))
+      throw RException(R__FAIL("daos_cont_open: error: " + std::string(d_errstr(err))));
 }
 
 ROOT::Experimental::Detail::RDaosContainer::~RDaosContainer() {
diff --git a/tree/ntuple/v7/src/RPageStorage.cxx b/tree/ntuple/v7/src/RPageStorage.cxx
@@ -68,7 +68,7 @@ std::unique_ptr<ROOT::Experimental::Detail::RPageSource> ROOT::Experimental::Det
 #ifdef R__ENABLE_DAOS
       return std::make_unique<RPageSourceDaos>(ntupleName, location, options);
 #else
-      throw std::runtime_error("This RNTuple build does not support DAOS.");
+      throw RException(R__FAIL("This RNTuple build does not support DAOS."));
 #endif
 
    return std::make_unique<RPageSourceFile>(ntupleName, location, options);
@@ -160,7 +160,7 @@ std::unique_ptr<ROOT::Experimental::Detail::RPageSink> ROOT::Experimental::Detai
 #ifdef R__ENABLE_DAOS
       realSink = std::make_unique<RPageSinkDaos>(ntupleName, location, options);
 #else
-      throw std::runtime_error("This RNTuple build does not support DAOS.");
+      throw RException(R__FAIL("This RNTuple build does not support DAOS."));
 #endif
    } else {
       realSink = std::make_unique<RPageSinkFile>(ntupleName, location, options);
diff --git a/tree/ntuple/v7/src/libdaos_mock/libdaos_mock.cxx b/tree/ntuple/v7/src/libdaos_mock/libdaos_mock.cxx
@@ -6,7 +6,7 @@
 /// is welcome!
 
 /*************************************************************************
- * Copyright (C) 1995-2020, Rene Brun and Fons Rademakers.               *
+ * Copyright (C) 1995-2021, Rene Brun and Fons Rademakers.               *
  * All rights reserved.                                                  *
  *                                                                       *
  * For the licensing terms see $ROOTSYS/LICENSE.                         *
@@ -84,6 +84,7 @@ int RDaosFakeObject::Fetch(daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
       || sgls[0].sg_nr != 1)
       return -DER_INVAL;
 
+   std::lock_guard<std::mutex> lock(fMutexStorage);
    auto it = fStorage.find(GetKey(dkey, &iods[0].iod_name));
    if (it == fStorage.end())
       return -DER_INVAL;
@@ -99,6 +100,7 @@ int RDaosFakeObject::Update(daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
       || sgls[0].sg_nr != 1)
       return -DER_INVAL;
 
+   std::lock_guard<std::mutex> lock(fMutexStorage);
    auto &data = fStorage[GetKey(dkey, &iods[0].iod_name)];
    d_iov_t &iov = sgls[0].sg_iovs[0];
    data.assign(reinterpret_cast<char *>(iov.iov_buf), iov.iov_buf_len);
@@ -186,18 +188,12 @@ indirection layer is added in order to detect the use of invalidated handles.
 // clang-format on
 class RDaosHandle {
 private:
-   static constexpr uint32_t kCookieMagic = 0xfee1dead;
-   static constexpr uint32_t kCookiePoison = 0x00001000;
-
-   /// \brief Wrapper over a `void *` that helps to detect the use of invalid handles.
-   /// A pointer is considered valid only if the magic number matches; this magic
-   /// number is clobbered on destruction.
+   /// \brief Wrapper over a `void *` that may help to detect the use of invalid handles.
    struct Cookie {
-      Cookie(void *p) : fMagic(kCookieMagic), fPointer(p) {}
-      ~Cookie() { fMagic = kCookiePoison; }
-      void *GetPointer() { return (fMagic == kCookieMagic) ? fPointer : nullptr; }
+      Cookie(void *p) : fPointer(p) {}
+      ~Cookie() { fPointer = nullptr; }
+      void *GetPointer() { return fPointer; }
 
-      uint32_t fMagic;
       void *fPointer;
    };
 
@@ -251,7 +247,7 @@ const char *d_errstr(int rc)
 
 
 int daos_cont_create(daos_handle_t poh, const uuid_t uuid, daos_prop_t *cont_prop,
-		 daos_event_t *ev)
+                     daos_event_t *ev)
 {
    (void)cont_prop;
    (void)ev;
@@ -266,7 +262,7 @@ int daos_cont_create(daos_handle_t poh, const uuid_t uuid, daos_prop_t *cont_pro
 }
 
 int daos_cont_open(daos_handle_t poh, const uuid_t uuid, unsigned int flags,
-	       daos_handle_t *coh, daos_cont_info_t *info, daos_event_t *ev)
+                   daos_handle_t *coh, daos_cont_info_t *info, daos_event_t *ev)
 {
    (void)flags;
    (void)info;
@@ -310,7 +306,7 @@ int daos_eq_destroy(daos_handle_t eqh, int flags)
 }
 
 int daos_eq_poll(daos_handle_t eqh, int wait_running,
-	     int64_t timeout, unsigned int nevents, daos_event_t **events)
+                 int64_t timeout, unsigned int nevents, daos_event_t **events)
 {
    (void)eqh;
    (void)wait_running;
@@ -338,7 +334,7 @@ int daos_event_fini(daos_event_t *ev)
 
 
 int daos_obj_open(daos_handle_t coh, daos_obj_id_t oid, unsigned int mode,
-	      daos_handle_t *oh, daos_event_t *ev)
+                  daos_handle_t *oh, daos_event_t *ev)
 {
    (void)ev;
 
@@ -358,8 +354,8 @@ int daos_obj_close(daos_handle_t oh, daos_event_t *ev)
 }
 
 int daos_obj_fetch(daos_handle_t oh, daos_handle_t th, uint64_t flags,
-	       daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
-	       d_sg_list_t *sgls, daos_iom_t *ioms, daos_event_t *ev)
+                   daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
+                   d_sg_list_t *sgls, daos_iom_t *ioms, daos_event_t *ev)
 {
    (void)th;
    (void)flags;
@@ -373,8 +369,8 @@ int daos_obj_fetch(daos_handle_t oh, daos_handle_t th, uint64_t flags,
 }
 
 int daos_obj_update(daos_handle_t oh, daos_handle_t th, uint64_t flags,
-		daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
-		d_sg_list_t *sgls, daos_event_t *ev)
+                    daos_key_t *dkey, unsigned int nr, daos_iod_t *iods,
+                    d_sg_list_t *sgls, daos_event_t *ev)
 {
    (void)th;
    (void)flags;
@@ -391,8 +387,8 @@ int daos_obj_update(daos_handle_t oh, daos_handle_t th, uint64_t flags,
 
 
 int daos_pool_connect(const uuid_t uuid, const char *grp,
-		  const d_rank_list_t *svc, unsigned int flags,
-		  daos_handle_t *poh, daos_pool_info_t *info, daos_event_t *ev)
+                      const d_rank_list_t *svc, unsigned int flags,
+                      daos_handle_t *poh, daos_pool_info_t *info, daos_event_t *ev)
 {
    (void)grp;
    (void)svc;
diff --git a/tree/ntuple/v7/test/CMakeLists.txt b/tree/ntuple/v7/test/CMakeLists.txt
@@ -52,7 +52,7 @@ if(daos OR daos_mock)
     set(daos_test_pool 1b245c52-765e-449d-80b1-8dce93bafc4b)
   endif()
   set_property(SOURCE ntuple_storage_daos.cxx
-              APPEND PROPERTY COMPILE_OPTIONS -DR__DAOS_TEST_POOL="${daos_test_pool}")
+               APPEND PROPERTY COMPILE_OPTIONS -DR__DAOS_TEST_POOL="${daos_test_pool}")
 
   ROOT_ADD_GTEST(ntuple_storage_daos ntuple_storage_daos.cxx LIBRARIES ROOTDataFrame ROOTNTuple MathCore CustomStruct)
 endif()
diff --git a/tree/ntuple/v7/test/ntuple_storage_daos.cxx b/tree/ntuple/v7/test/ntuple_storage_daos.cxx
@@ -9,8 +9,7 @@ TEST(RNTuple, Basics)
 
    {
       RNTupleWriteOptions options;
-      options.SetContainerFormat(ENTupleContainerFormat::kBare);
-      auto ntuple = RNTupleWriter::Recreate(std::move(model), "f", daosUri, options);
+      auto ntuple = RNTupleWriter::Recreate(std::move(model), "ntuple", daosUri, options);
       ntuple->Fill();
       ntuple->CommitCluster();
       *wrPt = 24.0;
@@ -42,8 +41,7 @@ TEST(RNTuple, Extended)
    double chksumWrite = 0.0;
    {
       RNTupleWriteOptions options;
-      options.SetContainerFormat(ENTupleContainerFormat::kBare);
-      auto ntuple = RNTupleWriter::Recreate(std::move(model), "f", daosUri, options);
+      auto ntuple = RNTupleWriter::Recreate(std::move(model), "ntuple", daosUri, options);
       constexpr unsigned int nEvents = 32000;
       for (unsigned int i = 0; i < nEvents; ++i) {
          auto nVec = 1 + floor(rnd.Rndm() * 1000.);
