From: Venky Shankar <vshankar@redhat.com>
Date: Tue, 14 Jan 2020 09:19:42 +0000 (-0500)
Subject: mgr/volumes: allow canceling a single asynchronous job for a volume
X-Git-Tag: v14.2.10~189^2~3
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=90fc0d0aabe8bf8f008783347acb308dcabb25c7;p=ceph.git

mgr/volumes: allow canceling a single asynchronous job for a volume

Signed-off-by: Venky Shankar <vshankar@redhat.com>
(cherry picked from commit 89a16c7867415085dd20c06d87e97bfa7de34fb2)
---

diff --git a/src/pybind/mgr/volumes/fs/async_job.py b/src/pybind/mgr/volumes/fs/async_job.py
index a83fc6dc91698..b64590e587871 100644
--- a/src/pybind/mgr/volumes/fs/async_job.py
+++ b/src/pybind/mgr/volumes/fs/async_job.py
@@ -156,7 +156,7 @@ class AsyncJobs(object):
         thread_id.reset_cancel()
 
         # wake up cancellation waiters if needed
-        if not self.jobs[volname] and cancelled:
+        if cancelled:
             logging.info("waking up cancellation waiters")
             self.cancel_cv.notifyAll()
 
@@ -194,6 +194,32 @@ class AsyncJobs(object):
         except (KeyError, ValueError):
             pass
 
+    def _cancel_job(self, volname, job):
+        """
+        cancel a executing job for a given volume. return True if canceled, False
+        otherwise (volume/job not found).
+        """
+        canceled = False
+        log.info("canceling job {0} for volume {1}".format(job, volname))
+        try:
+            if not volname in self.q and not volname in self.jobs and not job in self.jobs[volname]:
+                return canceled
+            for j in self.jobs[volname]:
+                if j[0] == job:
+                    j[1].cancel_job()
+                    # be safe against _cancel_jobs() running concurrently
+                    while j in self.jobs.get(volname, []):
+                        self.cancel_cv.wait()
+                    canceled = True
+                    break
+        except (KeyError, ValueError):
+            pass
+        return canceled
+
+    def cancel_job(self, volname, job):
+        with self.lock:
+            return self._cancel_job(volname, job)
+
     def cancel_jobs(self, volname):
         """
         cancel all executing jobs for a given volume.