Change in vdsm[master]: monitoring: Add process helpers

nsoffer at redhat.com nsoffer at redhat.com
Fri Mar 25 17:21:30 UTC 2016


Nir Soffer has uploaded a new change for review.

Change subject: monitoring: Add process helpers
......................................................................

monitoring: Add process helpers

Add helpers for running processes using the event loop:

BufferedReader   read data from file descriptor or a pipe and notify
                 when the read was completed. Useful for collecting
                 process output.

Reaper           wait until process exits and return its exit code.
                 Useful when process has closed th standard streams
                 before terminating.

Change-Id: Ic7a8e452ce2acd3aebdc76212a534e6d4a5638d9
Bug-Url: https://bugzilla.redhat.com/1081962
Signed-off-by: Nir Soffer <nsoffer at redhat.com>
---
M lib/vdsm/storage/asyncevent.py
M tests/storage_asyncevent_test.py
2 files changed, 161 insertions(+), 0 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/54/55254/1

diff --git a/lib/vdsm/storage/asyncevent.py b/lib/vdsm/storage/asyncevent.py
index 6d4c29f..cc8cab2 100644
--- a/lib/vdsm/storage/asyncevent.py
+++ b/lib/vdsm/storage/asyncevent.py
@@ -455,3 +455,70 @@
         self._wfd = -1
         os.close(wfd)
         asyncore.file_dispatcher.close(self)
+
+
+class BufferedReader(asyncore.file_dispatcher):
+    """
+    Read from file until file is close and notify when read was completed.
+    """
+
+    def __init__(self, fd, complete, bufsize=4096, map=None):
+        asyncore.file_dispatcher.__init__(self, fd, map=map)
+        filecontrol.set_close_on_exec(self._fileno)
+        self._complete = complete
+        self._bufsize = bufsize
+        self._data = bytearray()
+
+    def handle_read(self):
+        chunk = self.socket.read(self._bufsize)
+        if not chunk:
+            self.handle_close()
+            return
+        self._data += chunk
+
+    def handle_close(self):
+        self._complete(self._data)
+        self.close()
+
+    def handle_error(self):
+        log.exception("Unhandled error in %s", self)
+        self.handle_close()
+
+    def close(self):
+        # asyncore.dispatcher define closing attribute, but doe not use it.
+        if self.closing:
+            return
+        self.closing = True
+        self._complete = None
+        asyncore.file_dispatcher.close(self)
+
+    def writable(self):
+        return False
+
+
+class Reaper(object):
+    """
+    Wait for process and notify when it has terminated.
+    """
+
+    def __init__(self, loop, proc, complete, min_interval=2**-5,
+                 max_interval=1.0):
+        self._loop = loop
+        self._proc = proc
+        self._complete = complete
+        self._interval = min_interval
+        self._max_interval = max_interval
+        self._count = 0
+        self._loop.call_later(self._interval, self.reap)
+
+    def reap(self):
+        self._count += 1
+        rc = self._proc.poll()
+        if rc is None:
+            if self._interval < self._max_interval:
+                self._interval *= 2
+            self._loop.call_later(self._interval, self.reap)
+            return
+        log.debug("Process %s terminated (count=%d)", self._proc, self._count)
+        self._complete(rc)
+        self._complete = None
diff --git a/tests/storage_asyncevent_test.py b/tests/storage_asyncevent_test.py
index eaa50be..ec18412 100644
--- a/tests/storage_asyncevent_test.py
+++ b/tests/storage_asyncevent_test.py
@@ -21,9 +21,12 @@
 from __future__ import print_function
 
 import asyncore
+import os
 import random
 import socket
+import subprocess
 import time
+from contextlib import closing
 
 from testValidation import slowtest
 from testlib import VdsmTestCase
@@ -487,3 +490,94 @@
 
     def writable(self):
         return False
+
+
+ at expandPermutations
+class TestBufferedReader(VdsmTestCase):
+
+    def setUp(self):
+        self.loop = asyncevent.EventLoop()
+        self.received = None
+
+    def tearDown(self):
+        self.loop.close()
+
+    def complete(self, data):
+        self.received = data
+        self.loop.stop()
+
+    @permutations([
+        # size, bufsize
+        (0, 1),
+        (1, 32),
+        (1024, 256),
+        (4096, 1024),
+        (16384, 4096),
+        (65536, 16384),
+    ])
+    def test_read(self, size, bufsize):
+        data = b"x" * size
+        r, w = os.pipe()
+        reader = self.loop.create_dispatcher(
+            asyncevent.BufferedReader, r, self.complete, bufsize=bufsize)
+        with closing(reader):
+            os.close(r)  # Dupped by BufferedReader
+            Sender(self.loop, w, data, bufsize)
+            self.loop.run_forever()
+            self.assertEqual(self.received, data)
+
+
+class Sender(object):
+
+    def __init__(self, loop, fd, data, bufsize):
+        self.loop = loop
+        self.fd = fd
+        self.data = data
+        self.pos = 0
+        self.bufsize = bufsize
+        self.loop.call_soon(self.send)
+
+    def send(self):
+        if self.pos == len(self.data):
+            os.close(self.fd)
+            return
+        buf = memoryview(self.data)[self.pos:self.pos + self.bufsize]
+        self.pos += os.write(self.fd, buf)
+        self.loop.call_soon(self.send)
+
+
+ at expandPermutations
+class TestReaper(VdsmTestCase):
+
+    def setUp(self):
+        self.loop = asyncevent.EventLoop()
+        self.rc = None
+
+    def tearDown(self):
+        self.loop.close()
+
+    def complete(self, rc):
+        self.rc = rc
+        self.loop.stop()
+
+    def test_success(self):
+        self.reap(["true"])
+        self.assertEqual(0, self.rc)
+
+    def test_failure(self):
+        self.reap(["false"])
+        self.assertEqual(1, self.rc)
+
+    @slowtest
+    @permutations([[0.1], [0.2], [0.4], [0.8], [1.6]])
+    def test_slow(self, delay):
+        start = time.time()
+        self.reap(["sleep", "%.1f" % delay])
+        reap_time = time.time() - start - delay
+        print("reap time: %.3f" % reap_time)
+        self.assertLess(reap_time, 1.0)
+
+    def reap(self, cmd):
+        proc = subprocess.Popen(cmd, stdin=None, stdout=None, stderr=None)
+        asyncevent.Reaper(self.loop, proc, self.complete)
+        self.loop.run_forever()


-- 
To view, visit https://gerrit.ovirt.org/55254
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic7a8e452ce2acd3aebdc76212a534e6d4a5638d9
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer at redhat.com>


More information about the vdsm-patches mailing list