Merge PR ceph#33526 into nautilus

* refs/pull/33526/head:
	test: verify purge queue w/ large number of subvolumes
	test: pass timeout argument to mount::wait_for_dir_empty()
	mgr/volumes: access volume in lockless mode when fetching async job

Reviewed-by: Patrick Donnelly <pdonnell@redhat.com>

qa/tasks/cephfs/test_volumes.py

@@ -155,7 +155,7 @@ def _wait_for_trash_empty(self, timeout=30):
         # XXX: construct the trash dir path (note that there is no mgr
         # [sub]volume interface for this).
         trashdir = os.path.join("./", "volumes", "_deleting")
-        self.mount_a.wait_for_dir_empty(trashdir)
+        self.mount_a.wait_for_dir_empty(trashdir, timeout=timeout)
 
     def setUp(self):
         super(TestVolumes, self).setUp()
@@ -1221,23 +1221,23 @@ def test_subvolume_group_snapshot_ls(self):
                 raise RuntimeError("Error creating or listing subvolume group snapshots")
 
     def test_async_subvolume_rm(self):
-        subvolume = self._generate_random_subvolume_name()
-
-        # create subvolume
-        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+        subvolumes = self._generate_random_subvolume_name(100)
 
-        # fill subvolume w/ some data
-        self._do_subvolume_io(subvolume)
+        # create subvolumes
+        for subvolume in subvolumes:
+            self._fs_cmd("subvolume", "create", self.volname, subvolume)
+            self._do_subvolume_io(subvolume, number_of_files=10)
 
         self.mount_a.umount_wait()
 
-        # remove subvolume
-        self._fs_cmd("subvolume", "rm", self.volname, subvolume)
+        # remove subvolumes
+        for subvolume in subvolumes:
+            self._fs_cmd("subvolume", "rm", self.volname, subvolume)
 
         self.mount_a.mount()
 
         # verify trash dir is clean
-        self._wait_for_trash_empty()
+        self._wait_for_trash_empty(timeout=300)
 
     def test_subvolume_upgrade(self):
         """

src/pybind/mgr/volumes/fs/async_cloner.py

@@ -24,7 +24,7 @@ def get_next_clone_entry(volume_client, volname, running_jobs):
     log.debug("fetching clone entry for volume '{0}'".format(volname))
 
     try:
-        with open_volume(volume_client, volname) as fs_handle:
+        with open_volume_lockless(volume_client, volname) as fs_handle:
             try:
                 with open_clone_index(fs_handle, volume_client.volspec) as clone_index:
                     job = clone_index.get_oldest_clone_entry(running_jobs)

src/pybind/mgr/volumes/fs/purge_queue.py

@@ -13,7 +13,7 @@ def get_trash_entry_for_volume(volume_client, volname, running_jobs):
     log.debug("fetching trash entry for volume '{0}'".format(volname))
 
     try:
-        with open_volume(volume_client, volname) as fs_handle:
+        with open_volume_lockless(volume_client, volname) as fs_handle:
             try:
                 with open_trashcan(fs_handle, volume_client.volspec) as trashcan:
                     path = trashcan.get_trash_entry(running_jobs)