Change in vdsm[ovirt-3.3]: utils: Create AsyncProcessOperation
smizrahi at redhat.com
smizrahi at redhat.com
Mon Nov 11 13:53:56 UTC 2013
Saggi Mizrahi has uploaded a new change for review.
Change subject: utils: Create AsyncProcessOperation
......................................................................
utils: Create AsyncProcessOperation
This is an object to wrap a process so it can be treated as an
AsyncOperation.
Features:
* Takes care of pid collection
* Has pluggable output parsing
* Process appropriate implementation of wait() and stop()
Related-Bug-Url: http://bugzilla.redhat.com/964595
Change-Id: I79d0eefc9d917a4a93916d52867fb4f1e793c60e
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M lib/vdsm/utils.py
M tests/utilsTests.py
2 files changed, 113 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/35/21135/1
diff --git a/lib/vdsm/utils.py b/lib/vdsm/utils.py
index d3e0c2b..14f9dfa 100644
--- a/lib/vdsm/utils.py
+++ b/lib/vdsm/utils.py
@@ -48,6 +48,7 @@
import subprocess
import threading
import time
+import zombiereaper
from cpopen import CPopen as BetterPopen
from config import config
@@ -914,6 +915,55 @@
time.sleep(sleep)
+class AsyncProcessOperation(object):
+ def __init__(self, proc, resultParser=None):
+ """Wraps a running process operation.
+
+ resultParser should be of type callback(rc, out, err) and can return
+ anything or throw exceptions."""
+ self._lock = threading.Lock()
+
+ self._result = None
+ self._resultParser = resultParser
+
+ self._proc = proc
+
+ def wait(self, timeout=None, cond=None):
+ """Waits until the process has exited, the timeout has been reached or
+ the condition has been met"""
+ return self._proc.wait(timeout, cond)
+
+ def stop(self):
+ """Stops the running operation, effectively sending a kill signal to
+ the process"""
+ self._proc.kill()
+
+ def result(self):
+ """Returns the result in the as a tuple of (result, error).
+ If the operation is still running it will block until it returns.
+
+ If no resultParser has been set the default result
+ is (rc, out, err) """
+ with self._lock:
+ if self._result is None:
+ out, err = self._proc.communicate()
+ rc = self._proc.returncode
+ if self._resultParser is not None:
+ try:
+ self._result = (self._resultParser(rc, out, err),
+ None)
+ except Exception as e:
+ self._result = (None, e)
+ else:
+ self._result = ((rc, out, err), None)
+
+ return self._result
+
+ def __del__(self):
+ if self._proc.returncode is None:
+ zombiereaper.autoReapPID(self._proc.pid)
+
+
def panic(msg):
logging.error("Panic: %s", msg, exc_info=True)
os.killpg(0, 9)
diff --git a/tests/utilsTests.py b/tests/utilsTests.py
index a6a222d..656c121 100644
--- a/tests/utilsTests.py
+++ b/tests/utilsTests.py
@@ -22,6 +22,7 @@
from testrunner import VdsmTestCase as TestCaseBase
from vdsm import utils
from storage import misc
+import time
class RetryTests(TestCaseBase):
@@ -80,3 +81,65 @@
class GeneralUtilsTests(TestCaseBase):
def testPanic(self):
self.assertRaises(AssertionError, utils.panic, "panic test")
+
+
+class AsyncProcessOperationTests(TestCaseBase):
+ def _echo(self, text):
+ proc = misc.execCmd(["echo", "-n", "test"], sync=False)
+
+ def parse(rc, out, err):
+ return out
+
+ return utils.AsyncProcessOperation(proc, parse)
+
+ def _sleep(self, t):
+ proc = misc.execCmd(["sleep", str(t)], sync=False)
+ return utils.AsyncProcessOperation(proc)
+
+ def _fail(self, t):
+ proc = misc.execCmd(["sleep", str(t)], sync=False)
+
+ def parse(rc, out, err):
+ raise Exception("TEST!!!")
+
+ return utils.AsyncProcessOperation(proc, parse)
+
+ def test(self):
+ aop = self._sleep(1)
+ self.assertEquals(aop.result(), ((0, "", ""), None))
+
+ def testAlreadyExitedSuccess(self):
+ aop = self._sleep(0)
+ time.sleep(1)
+ self.assertEquals(aop.result(), ((0, "", ""), None))
+
+ def testAlreadyExitedFail(self):
+ aop = self._sleep("hello")
+ time.sleep(1)
+ ((rc, out, err), err) = aop.result()
+ self.assertEquals(err, None)
+ self.assertEquals(rc, 1)
+
+ def testWait(self):
+ aop = self._sleep(1)
+ aop.wait(timeout=2)
+
+ def testParser(self):
+ aop = self._echo("test")
+ self.assertEquals(aop.result(), ("test", None))
+
+ def testStop(self):
+ aop = self._sleep(10)
+ aop.stop()
+
+ start = time.time()
+ aop.result()
+ end = time.time()
+ duration = end - start
+ self.assertTrue(duration < 2)
+
+ def testException(self):
+ aop = self._fail(1)
+ res, err = aop.result()
+ self.assertEquals(res, None)
+ self.assertNotEquals(err, None)
--
To view, visit http://gerrit.ovirt.org/21135
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I79d0eefc9d917a4a93916d52867fb4f1e793c60e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: ovirt-3.3
Gerrit-Owner: Saggi Mizrahi <smizrahi at redhat.com>
More information about the vdsm-patches
mailing list