DAOS-17681 test: add EC coverage to the datamover tests

src/client/pydaos/raw/daos_api.py

@@ -874,6 +874,172 @@ def punch_array(self, dkey, akey, rec_idx, rec_count,
             raise DaosApiError("Array punch returned non-zero. RC: {0}"
                                .format(ret))
 
+    def insert_recx(self, dkey, akey, rec_size, rx_idx, rx_nr, data,
+                    txn=daos_cref.DAOS_TX_NONE):
+        """Update one record extent from a single contiguous buffer.
+
+        insert_array describes an extent with one scatter gather entry per record, which
+        caps the extent at the scatter gather list limit. This uses a single entry for
+        the whole extent, so a byte granular extent can span an erasure coded stripe.
+
+        dkey      --1st level key for the array value
+        akey      --2nd level key for the array value
+        rec_size  --size in bytes of a single record
+        rx_idx    --index of the first record
+        rx_nr     --how many records
+        data      --buffer holding rx_nr * rec_size bytes
+        txn       --which transaction to write in.
+                    Default is independent transaction (DAOS_TX_NONE)
+        """
+        buf_len = rx_nr * rec_size
+
+        extent = daos_cref.Extent()
+        extent.rx_idx = rx_idx
+        extent.rx_nr = rx_nr
+
+        self.iod.iod_name.iov_buf = ctypes.cast(akey, ctypes.c_void_p)
+        self.iod.iod_name.iov_buf_len = ctypes.sizeof(akey)
+        self.iod.iod_name.iov_len = ctypes.sizeof(akey)
+        self.iod.iod_type = 2
+        self.iod.iod_size = rec_size
+        self.iod.iod_flags = 0
+        self.iod.iod_nr = 1
+        self.iod.iod_recxs = ctypes.pointer(extent)
+
+        sgl_iov = daos_cref.IOV()
+        sgl_iov.iov_len = buf_len
+        sgl_iov.iov_buf_len = buf_len
+        sgl_iov.iov_buf = ctypes.cast(data, ctypes.c_void_p)
+        self.sgl.sg_iovs = ctypes.pointer(sgl_iov)
+        self.sgl.sg_nr = 1
+        self.sgl.sg_nr_out = 0
+
+        dkey_iov = daos_cref.IOV()
+        dkey_iov.iov_buf = ctypes.cast(dkey, ctypes.c_void_p)
+        dkey_iov.iov_buf_len = ctypes.sizeof(dkey)
+        dkey_iov.iov_len = ctypes.sizeof(dkey)
+
+        func = self.context.get_function('update-obj')
+
+        ret = func(self.obj.obj_handle, txn, 0, ctypes.byref(dkey_iov),
+                   1, ctypes.byref(self.iod), ctypes.byref(self.sgl), None)
+        if ret != 0:
+            raise DaosApiError("Recx update returned non-zero. RC: {0}"
+                               .format(ret))
+
+    def fetch_recx(self, dkey, akey, rec_size, rx_idx, rx_nr,
+                   txn=daos_cref.DAOS_TX_NONE):
+        """Fetch one record extent into a single contiguous buffer.
+
+        dkey      --1st level key for the array value
+        akey      --2nd level key for the array value
+        rec_size  --size in bytes of a single record
+        rx_idx    --index of the first record
+        rx_nr     --how many records
+        txn       --which transaction to read from.
+                    Default is independent transaction (DAOS_TX_NONE)
+
+        Returns:
+            bytes: rx_nr * rec_size bytes. The fetch leaves holes untouched, and the
+                buffer starts zeroed, so a hole reads back as zeros.
+
+        """
+        data, _ = self._fetch_recx(dkey, akey, rec_size, rx_idx, rx_nr, 0, txn)
+        return data
+
+    def fetch_recx_map(self, dkey, akey, rec_size, rx_idx, rx_nr, max_extents=64,
+                       txn=daos_cref.DAOS_TX_NONE):
+        """Fetch one record extent and the io map describing what really holds data.
+
+        A hole and a range someone wrote zeros over read back the same, so the io map
+        is the only way to tell them apart.
+
+        dkey        --1st level key for the array value
+        akey        --2nd level key for the array value
+        rec_size    --size in bytes of a single record
+        rx_idx      --index of the first record
+        rx_nr       --how many records
+        max_extents --how many extents the io map can hold. Raises if the server has
+                      more to report, since a truncated map would silently under
+                      report what is there
+        txn         --which transaction to read from.
+                      Default is independent transaction (DAOS_TX_NONE)
+
+        Returns:
+            tuple: (bytes, extents), where extents is a list of (rx_idx, rx_nr)
+                covering only the records that really hold data.
+
+        """
+        return self._fetch_recx(dkey, akey, rec_size, rx_idx, rx_nr, max_extents, txn)
+
+    def _fetch_recx(self, dkey, akey, rec_size, rx_idx, rx_nr, max_extents, txn):
+        """Fetch one record extent, optionally asking for the io map.
+
+        See fetch_recx and fetch_recx_map.
+
+        Returns:
+            tuple: (bytes, extents), extents is None when max_extents is 0.
+
+        """
+        buf_len = rx_nr * rec_size
+        buf = ctypes.create_string_buffer(buf_len)
+
+        extent = daos_cref.Extent()
+        extent.rx_idx = rx_idx
+        extent.rx_nr = rx_nr
+
+        self.iod.iod_name.iov_buf = ctypes.cast(akey, ctypes.c_void_p)
+        self.iod.iod_name.iov_buf_len = ctypes.sizeof(akey)
+        self.iod.iod_name.iov_len = ctypes.sizeof(akey)
+        self.iod.iod_type = 2
+        self.iod.iod_size = rec_size
+        self.iod.iod_flags = 0
+        self.iod.iod_nr = 1
+        self.iod.iod_recxs = ctypes.pointer(extent)
+
+        sgl_iov = daos_cref.IOV()
+        sgl_iov.iov_len = buf_len
+        sgl_iov.iov_buf_len = buf_len
+        sgl_iov.iov_buf = ctypes.cast(buf, ctypes.c_void_p)
+        self.sgl.sg_iovs = ctypes.pointer(sgl_iov)
+        self.sgl.sg_nr = 1
+        self.sgl.sg_nr_out = 0
+
+        dkey_iov = daos_cref.IOV()
+        dkey_iov.iov_buf = ctypes.cast(dkey, ctypes.c_void_p)
+        dkey_iov.iov_buf_len = ctypes.sizeof(dkey)
+        dkey_iov.iov_len = ctypes.sizeof(dkey)
+
+        iom_ptr = None
+        iom = None
+        if max_extents:
+            iom = daos_cref.DaosIOMap()
+            iom_recxs = (daos_cref.Extent * max_extents)()
+            iom.iom_flags = daos_cref.DAOS_IOMF_DETAIL
+            iom.iom_nr = max_extents
+            iom.iom_recxs = ctypes.cast(iom_recxs, ctypes.POINTER(daos_cref.Extent))
+            iom_ptr = ctypes.byref(iom)
+
+        func = self.context.get_function('fetch-obj')
+
+        ret = func(self.obj.obj_handle, txn, 0, ctypes.byref(dkey_iov), 1,
+                   ctypes.byref(self.iod), ctypes.byref(self.sgl), iom_ptr, None)
+        if ret != 0:
+            raise DaosApiError("Recx fetch returned non-zero. RC: {0}"
+                               .format(ret))
+
+        data = ctypes.string_at(buf, buf_len)
+        if iom is None:
+            return data, None
+
+        if iom.iom_nr_out > iom.iom_nr:
+            raise DaosApiError(
+                "Io map holds {0} extents, need room for {1}".format(
+                    iom.iom_nr, iom.iom_nr_out))
+
+        return data, [(iom.iom_recxs[i].rx_idx, iom.iom_recxs[i].rx_nr)
+                      for i in range(iom.iom_nr_out)]
+
     def fetch_array(self, dkey, akey, rec_count, rec_size,
                     txn=daos_cref.DAOS_TX_NONE):
         """Retrieve an array data from a dkey/akey pair.

src/client/pydaos/raw/daos_cref.py

@@ -209,6 +209,22 @@ class DaosIODescriptor(ctypes.Structure):
                 ("iod_recxs", ctypes.POINTER(Extent))]
 
 
+class DaosIOMap(ctypes.Structure):
+    """Represents struct: daos_iom_t"""
+    _fields_ = [("iom_type", ctypes.c_int),  # enum
+                ("iom_nr", ctypes.c_uint32),
+                ("iom_nr_out", ctypes.c_uint32),
+                ("iom_flags", ctypes.c_uint32),
+                ("iom_size", ctypes.c_uint64),
+                ("iom_recx_lo", Extent),
+                ("iom_recx_hi", Extent),
+                ("iom_recxs", ctypes.POINTER(Extent))]
+
+
+# daos_iom_t iom_flags, ask for the full list of extents rather than just lo/hi
+DAOS_IOMF_DETAIL = 0x1
+
+
 class Anchor(ctypes.Structure):
     """ Class to represent a C daos_anchor_t struct. """
     _fields_ = [('da_type', ctypes.c_uint16),

src/tests/ftest/datamover/obj_ec.py

@@ -11,15 +11,15 @@
 
 class DmvrObjEcTest(DataMoverTestBase):
     # pylint: disable=too-many-ancestors
-    """Object Data Mover validation for cloning erasure coded containers.
+    """Object Data Mover validation for erasure coded containers.
 
     Test Class Description:
-        Cloning a container whose objects are erasure coded and hold punched
-        extents. Enumeration of an erasure coded object is served by a parity
-        shard, where one parity block stands for a whole stripe, so the listed
-        extents are an upper bound on the data and cover the punched records.
-        A clone that copies the listed extents instead of what the fetch
-        actually returned writes uninitialized bytes over those holes.
+        Copying a container whose objects are erasure coded and hold punched
+        stripes. Enumeration of an erasure coded object is served by a parity
+        shard, where one parity block stands for a whole stripe, so a stripe
+        that has been punched is still reported as holding data. A copy that
+        trusts the listed extents instead of the io map the fetch returned
+        writes over those holes.
     :avocado: recursive
     """
 
@@ -29,24 +29,57 @@ def setUp(self):
 
         self.num_objs = self.params.get("num_objs", "/run/dataset/*")
         self.num_dkeys = self.params.get("num_dkeys", "/run/dataset/*")
-        self.num_akeys_array = self.params.get("num_akeys_array", "/run/dataset/*")
-        self.num_akeys_single = self.params.get("num_akeys_single", "/run/dataset/*")
-        self.akey_sizes = self.params.get("akey_sizes", "/run/dataset/*")
-        self.akey_extents = self.params.get("akey_extents", "/run/dataset/*")
-        self.punch_extents = self.params.get("punch_extents", "/run/dataset/*")
+        self.num_akeys = self.params.get("num_akeys", "/run/dataset/*")
+        self.stripe_size = self.params.get("stripe_size", "/run/dataset/*")
+        self.num_stripes = self.params.get("num_stripes", "/run/dataset/*")
+        self.punch_stripes = self.params.get("punch_stripes", "/run/dataset/*")
+        self.full_punch_akeys = self.params.get("full_punch_akeys", "/run/dataset/*", 0)
         # an EC class is required, a replicated one would make the test vacuous
         self.obj_class = self.params.get("obj_class", "/run/dataset/*", "OC_EC_2P1G1")
 
+    def _gen_dataset(self, cont):
+        """Create the punched EC dataset and confirm the source reads back as expected.
+
+        Args:
+            cont (TestContainer): the container to create the dataset in.
+
+        Returns:
+            list: a list of DaosObj created.
+
+        """
+        obj_list = self.dataset_gen_ec(
+            cont, self.num_objs, self.num_dkeys, self.num_akeys, self.obj_class,
+            self.stripe_size, self.num_stripes, self.punch_stripes,
+            full_punch_akeys=self.full_punch_akeys)
+
+        # the source itself must report the punched stripes as holes, otherwise
+        # the run below would pass without ever exercising the hole handling
+        self._verify_dataset(obj_list, cont)
+
+        return obj_list
+
+    def _verify_dataset(self, obj_list, cont):
+        """Verify a dataset created by _gen_dataset.
+
+        Args:
+            obj_list (list): obj_list returned from _gen_dataset.
+            cont (TestContainer): the container to verify.
+        """
+        self.dataset_verify_ec(
+            obj_list, cont, self.num_objs, self.num_dkeys, self.num_akeys,
+            self.stripe_size, self.num_stripes, self.punch_stripes,
+            full_punch_akeys=self.full_punch_akeys)
+
     def run_dm_obj_ec(self, tool):
         """
         Test Description:
-            Tests cloning a container of erasure coded objects with holes.
+            Tests copying a container of erasure coded objects with holes.
         Use Cases:
             Create pool1 and cont1.
             Create a dataset of erasure coded objects in cont1, where each
-            array akey spans whole stripes and then has its leading records
+            array akey spans whole stripes and then has alternate stripes
             punched back out.
-            Clone cont1 to a new cont2 and verify that the holes are still
+            Copy cont1 to a new cont2 and verify that the holes are still
             holes in cont2.
 
         Args:
@@ -57,19 +90,7 @@ def run_dm_obj_ec(self, tool):
         pool1 = self.get_pool()
         cont1 = self.get_container(pool1)
 
-        obj_list = self.dataset_gen(
-            cont1,
-            self.num_objs, self.num_dkeys, self.num_akeys_single,
-            self.num_akeys_array, self.akey_sizes, self.akey_extents,
-            oclass=self.obj_class, punch_extents=self.punch_extents)
-
-        # the source itself must report the punched records as holes, otherwise
-        # the run below would pass without ever exercising the hole handling
-        self.dataset_verify(
-            obj_list, cont1,
-            self.num_objs, self.num_dkeys, self.num_akeys_single,
-            self.num_akeys_array, self.akey_sizes, self.akey_extents,
-            punch_extents=self.punch_extents)
+        obj_list = self._gen_dataset(cont1)
 
         result = self.run_datamover(
             self.test_id + " (cont1->cont2) (same pool)",
@@ -78,11 +99,35 @@ def run_dm_obj_ec(self, tool):
         cont2_label = self.parse_create_cont_label(result.stdout_text)
 
         cont2 = get_existing_container(self, pool1, cont2_label)
-        self.dataset_verify(
-            obj_list, cont2,
-            self.num_objs, self.num_dkeys, self.num_akeys_single,
-            self.num_akeys_array, self.akey_sizes, self.akey_extents,
-            punch_extents=self.punch_extents)
+        self._verify_dataset(obj_list, cont2)
+
+    def run_dm_obj_ec_dsync(self):
+        """
+        Test Description:
+            Tests syncing a container of erasure coded objects with holes.
+        Use Cases:
+            Create pool1, cont1 and an empty cont2.
+            Create a dataset of erasure coded objects with punched stripes in cont1.
+            Sync cont1 to cont2 and verify the holes are still holes in cont2.
+            Sync a second time, so the comparison of the destination runs against
+            a populated container rather than an empty one.
+        """
+        self.set_tool("DSYNC")
+
+        pool1 = self.get_pool()
+        cont1 = self.get_container(pool1)
+
+        # dsync does not create the destination
+        cont2 = self.get_container(pool1)
+
+        obj_list = self._gen_dataset(cont1)
+
+        for run in ("first", "second"):
+            self.run_datamover(
+                self.test_id + " (cont1->cont2) ({} sync)".format(run),
+                "DAOS_UUID", None, pool1, cont1,
+                "DAOS_UUID", None, pool1, cont2)
+            self._verify_dataset(obj_list, cont2)
 
     @avocado.fail_on(DaosApiError)
     def test_dm_obj_ec_cont_clone(self):
@@ -95,3 +140,39 @@ def test_dm_obj_ec_cont_clone(self):
         :avocado: tags=DmvrObjEcTest,test_dm_obj_ec_cont_clone
         """
         self.run_dm_obj_ec("CONT_CLONE")
+
+    @avocado.fail_on(DaosApiError)
+    def test_dm_obj_ec_dcp(self):
+        """
+        Test Description:
+            Verify copying an erasure coded container with punched extents.
+        :avocado: tags=all,daily_regression
+        :avocado: tags=hw,medium
+        :avocado: tags=datamover,mfu,mfu_dcp
+        :avocado: tags=DmvrObjEcTest,test_dm_obj_ec_dcp
+        """
+        self.run_dm_obj_ec("DCP")
+
+    @avocado.fail_on(DaosApiError)
+    def test_dm_obj_ec_dserialize(self):
+        """
+        Test Description:
+            Verify serializing an erasure coded container with punched extents.
+        :avocado: tags=all,daily_regression
+        :avocado: tags=hw,medium
+        :avocado: tags=datamover,mfu,mfu_serialize,mfu_deserialize,hdf5
+        :avocado: tags=DmvrObjEcTest,test_dm_obj_ec_dserialize
+        """
+        self.run_dm_obj_ec("DSERIAL")
+
+    @avocado.fail_on(DaosApiError)
+    def test_dm_obj_ec_dsync(self):
+        """
+        Test Description:
+            Verify syncing an erasure coded container with punched extents.
+        :avocado: tags=all,daily_regression
+        :avocado: tags=hw,medium
+        :avocado: tags=datamover,mfu,mfu_dsync
+        :avocado: tags=DmvrObjEcTest,test_dm_obj_ec_dsync
+        """
+        self.run_dm_obj_ec_dsync()