Change in vdsm[master]: storage.misc: Move execCmd and its dependencies to vdsm.util...

zhshzhou at linux.vnet.ibm.com zhshzhou at linux.vnet.ibm.com
Thu Apr 25 04:29:18 UTC 2013


Zhou Zheng Sheng has uploaded a new change for review.

Change subject: storage.misc: Move execCmd and its dependencies to vdsm.utils
......................................................................

storage.misc: Move execCmd and its dependencies to vdsm.utils

The original vdsm.utils.execCmd actually imports and invokes
storage.misc.execCmd, which means if a process does not have
/usr/share/vdsm in its PYTHONPATH and want to use vdsm.utils.execCmd, it
will fail. Furthermore, lib/vdsm/utils.py should not use anything that
is not present in the python standard library or in a dependency of the
vdsm-python package.

execCmd is very useful for modules in vdsm-tool, currently these modules
use ad-hoc execCmd implementation. This patch moves storage.misc.execCmd
and its dependencies to vdsm.utils, so that other modules can import the
'official' execCmd implementation without adding /usr/share/vdsm to
PYTHONPATH.

Change-Id: Id8fb42035b24bc915aaa700ab128580bdb609d9e
Signed-off-by: Zhou Zheng Sheng <zhshzhou at linux.vnet.ibm.com>
---
M lib/vdsm/utils.py
M vdsm/storage/misc.py
2 files changed, 314 insertions(+), 314 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/21/14221/1

diff --git a/lib/vdsm/utils.py b/lib/vdsm/utils.py
index e398a7e..cfec8cc 100644
--- a/lib/vdsm/utils.py
+++ b/lib/vdsm/utils.py
@@ -27,22 +27,35 @@
     Contains a reverse dictionary pointing from error string to its error code.
 """
 from SimpleXMLRPCServer import SimpleXMLRPCServer
+from StringIO import StringIO
+from weakref import proxy
 import SocketServer
-import threading
-import os
-import time
-import logging
 import errno
-import subprocess
-import pwd
 import fcntl
 import functools
-import stat
 import glob
+import io
+import logging
+import os
+import pwd
+import select
+import signal
+import stat
+import subprocess
+import threading
+import time
 
 
-import constants
+from betterPopen import BetterPopen
 from config import config
+import constants
+
+# Buffsize is 1K because I tested it on some use cases and 1k was fastets. If
+# you find this number to be a bottleneck in any way you are welcome to change
+# it
+BUFFSIZE = 1024
+
+SUDO_NON_INTERACTIVE_FLAG = "-n"
 
 _THP_STATE_PATH = '/sys/kernel/mm/transparent_hugepage/enabled'
 if not os.path.exists(_THP_STATE_PATH):
@@ -147,10 +160,296 @@
         return val
 
 
-def execCmd(*args, **kwargs):
-    # import only after config as been initialized
-    from storage.misc import execCmd
-    return execCmd(*args, **kwargs)
+# NOTE: it would be best to try and unify NoIntrCall and NoIntrPoll.
+# We could do so defining a new object that can be used as a placeholer
+# for the changing timeout value in the *args/**kwargs. This would
+# lead us to rebuilding the function arguments at each loop.
+def NoIntrPoll(pollfun, timeout=-1):
+    """
+    This wrapper is used to handle the interrupt exceptions that might
+    occur during a poll system call. The wrapped function must be defined
+    as poll([timeout]) where the special timeout value 0 is used to return
+    immediately and -1 is used to wait indefinitely.
+    """
+    # When the timeout < 0 we shouldn't compute a new timeout after an
+    # interruption.
+    endtime = None if timeout < 0 else time.time() + timeout
+
+    while True:
+        try:
+            return pollfun(timeout)
+        except (IOError, select.error) as e:
+            if e.args[0] != errno.EINTR:
+                raise
+
+        if endtime is not None:
+            timeout = max(0, endtime - time.time())
+
+
+class AsyncProc(object):
+    """
+    AsyncProc is a funky class. It warps a standard subprocess.Popen
+    Object and gives it super powers. Like the power to read from a stream
+    without the fear of deadlock. It does this by always sampling all
+    stream while waiting for data. By doing this the other process can freely
+    write data to all stream without the fear of it getting stuck writing
+    to a full pipe.
+    """
+    class _streamWrapper(io.RawIOBase):
+        def __init__(self, parent, streamToWrap, fd):
+            io.IOBase.__init__(self)
+            self._stream = streamToWrap
+            self._parent = proxy(parent)
+            self._fd = fd
+            self._closed = False
+            self._emptyCounter = 0
+
+        def close(self):
+            if not self._closed:
+                self._closed = True
+                while not self._streamClosed:
+                    self._parent._processStreams()
+
+        @property
+        def closed(self):
+            return self._closed
+
+        @property
+        def _streamClosed(self):
+            return (self.fileno() in self._parent._closedfds)
+
+        def fileno(self):
+            return self._fd
+
+        def seekable(self):
+            return False
+
+        def readable(self):
+            return True
+
+        def writable(self):
+            return True
+
+        def read(self, length):
+            hasNewData = (self._stream.len - self._stream.pos)
+            if hasNewData < length and not self._streamClosed:
+                self._parent._processStreams()
+
+            with self._parent._streamLock:
+                res = self._stream.read(length)
+                if self._stream.pos == self._stream.len:
+                    if self._streamClosed and res == "":
+                        self._emptyCounter += 1
+                        if self._emptyCounter > 2:
+                            self._closed = True
+
+                    self._stream.truncate(0)
+
+            return res
+
+        def readinto(self, b):
+            data = self.read(len(b))
+            bytesRead = len(data)
+            b[:bytesRead] = data
+
+            return bytesRead
+
+        def write(self, data):
+            if hasattr(data, "tobytes"):
+                data = data.tobytes()
+            with self._parent._streamLock:
+                oldPos = self._stream.pos
+                self._stream.pos = self._stream.len
+                self._stream.write(data)
+                self._stream.pos = oldPos
+
+            while self._stream.len > 0 and not self._streamClosed:
+                self._parent._processStreams()
+
+            if self._streamClosed:
+                self._closed = True
+
+            if self._stream.len != 0:
+                raise IOError(errno.EPIPE,
+                              "Could not write all data to stream")
+
+            return len(data)
+
+    def __init__(self, popenToWrap):
+        self._streamLock = threading.Lock()
+        self._proc = popenToWrap
+
+        self._stdout = StringIO()
+        self._stderr = StringIO()
+        self._stdin = StringIO()
+
+        fdout = self._proc.stdout.fileno()
+        fderr = self._proc.stderr.fileno()
+        self._fdin = self._proc.stdin.fileno()
+
+        self._closedfds = []
+
+        self._poller = select.epoll()
+        self._poller.register(fdout, select.EPOLLIN | select.EPOLLPRI)
+        self._poller.register(fderr, select.EPOLLIN | select.EPOLLPRI)
+        self._poller.register(self._fdin, 0)
+        self._fdMap = {fdout: self._stdout,
+                       fderr: self._stderr,
+                       self._fdin: self._stdin}
+
+        self.stdout = io.BufferedReader(self._streamWrapper(self,
+                                        self._stdout, fdout), BUFFSIZE)
+
+        self.stderr = io.BufferedReader(self._streamWrapper(self,
+                                        self._stderr, fderr), BUFFSIZE)
+
+        self.stdin = io.BufferedWriter(self._streamWrapper(self,
+                                       self._stdin, self._fdin), BUFFSIZE)
+
+        self._returncode = None
+
+    def _processStreams(self):
+        if len(self._closedfds) == 3:
+            return
+
+        if not self._streamLock.acquire(False):
+            self._streamLock.acquire()
+            self._streamLock.release()
+            return
+        try:
+            if self._stdin.len > 0 and self._stdin.pos == 0:
+                # Polling stdin is redundant if there is nothing to write
+                # trun on only if data is waiting to be pushed
+                self._poller.modify(self._fdin, select.EPOLLOUT)
+
+            pollres = NoIntrPoll(self._poller.poll, 1)
+
+            for fd, event in pollres:
+                stream = self._fdMap[fd]
+                if event & select.EPOLLOUT and self._stdin.len > 0:
+                    buff = self._stdin.read(BUFFSIZE)
+                    written = os.write(fd, buff)
+                    stream.pos -= len(buff) - written
+                    if stream.pos == stream.len:
+                        stream.truncate(0)
+                        self._poller.modify(fd, 0)
+
+                elif event & (select.EPOLLIN | select.EPOLLPRI):
+                    data = os.read(fd, BUFFSIZE)
+                    oldpos = stream.pos
+                    stream.pos = stream.len
+                    stream.write(data)
+                    stream.pos = oldpos
+
+                elif event & (select.EPOLLHUP | select.EPOLLERR):
+                    self._poller.unregister(fd)
+                    self._closedfds.append(fd)
+                    # I don't close the fd because the original Popen
+                    # will do it.
+
+            if self.stdin.closed and self._fdin not in self._closedfds:
+                self._poller.unregister(self._fdin)
+                self._closedfds.append(self._fdin)
+                self._proc.stdin.close()
+
+        finally:
+            self._streamLock.release()
+
+    @property
+    def pid(self):
+        return self._proc.pid
+
+    @property
+    def returncode(self):
+        if self._returncode is None:
+            self._returncode = self._proc.poll()
+        return self._returncode
+
+    def kill(self):
+        try:
+            self._proc.kill()
+        except OSError as ex:
+            if ex.errno != errno.EPERM:
+                raise
+            execCmd([constants.EXT_KILL, "-%d" % (signal.SIGTERM,),
+                    str(self.pid)], sudo=True)
+
+    def wait(self, timeout=None, cond=None):
+        startTime = time.time()
+        while self.returncode is None:
+            if timeout is not None and (time.time() - startTime) > timeout:
+                return False
+            if cond is not None and cond():
+                return False
+            self._processStreams()
+        return True
+
+    def communicate(self, data=None):
+        if data is not None:
+            self.stdin.write(data)
+            self.stdin.flush()
+        self.stdin.close()
+
+        self.wait()
+        return "".join(self.stdout), "".join(self.stderr)
+
+    def __del__(self):
+        self._poller.close()
+
+
+def execCmd(command, sudo=False, cwd=None, data=None, raw=False, logErr=True,
+            printable=None, env=None, sync=True, nice=None, ioclass=None,
+            ioclassdata=None, setsid=False, execCmdLogger=logging.root):
+    """
+    Executes an external command, optionally via sudo.
+    """
+    if ioclass is not None:
+        cmd = command
+        command = [constants.EXT_IONICE, '-c', str(ioclass)]
+        if ioclassdata is not None:
+            command.extend(("-n", str(ioclassdata)))
+
+        command = command + cmd
+
+    if nice is not None:
+        command = [constants.EXT_NICE, '-n', str(nice)] + command
+
+    if setsid:
+        command = [constants.EXT_SETSID] + command
+
+    if sudo:
+        command = [constants.EXT_SUDO, SUDO_NON_INTERACTIVE_FLAG] + command
+
+    if not printable:
+        printable = command
+
+    cmdline = repr(subprocess.list2cmdline(printable))
+    execCmdLogger.debug("%s (cwd %s)", cmdline, cwd)
+
+    p = BetterPopen(command, close_fds=True, cwd=cwd, env=env)
+    p = AsyncProc(p)
+    if not sync:
+        if data is not None:
+            p.stdin.write(data)
+            p.stdin.flush()
+
+        return p
+
+    (out, err) = p.communicate(data)
+
+    if out is None:
+        # Prevent splitlines() from barfing later on
+        out = ""
+
+    execCmdLogger.debug("%s: <err> = %s; <rc> = %d",
+                        {True: "SUCCESS", False: "FAILED"}[p.returncode == 0],
+                        repr(err), p.returncode)
+
+    if not raw:
+        out = out.splitlines(False)
+        err = err.splitlines(False)
+
+    return (p.returncode, out, err)
 
 
 def checkPathStat(pathToCheck):
diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py
index ae92b75..bee9daa 100644
--- a/vdsm/storage/misc.py
+++ b/vdsm/storage/misc.py
@@ -31,25 +31,19 @@
 from contextlib import contextmanager
 from functools import wraps, partial
 from itertools import chain, imap
-from StringIO import StringIO
-from weakref import proxy
 import contextlib
 import errno
 import glob
-import io
 import logging
 import os
 import Queue
 import random
 import re
 import select
-import signal
 import string
 import struct
-import subprocess
 import sys
 import threading
-import time
 import types
 import weakref
 import fcntl
@@ -58,7 +52,6 @@
 from vdsm import constants
 from vdsm import utils
 import storage_exception as se
-from betterPopen import BetterPopen
 import fileUtils
 import logUtils
 
@@ -70,7 +63,6 @@
 UUID_HYPHENS = [8, 13, 18, 23]
 OVIRT_NODE = False
 MEGA = 1 << 20
-SUDO_NON_INTERACTIVE_FLAG = "-n"
 UNLIMITED_THREADS = -1
 
 log = logging.getLogger('Storage.Misc')
@@ -86,11 +78,6 @@
                                          logger, logger.__class__)
 
     return logger
-
-# Buffsize is 1K because I tested it on some use cases and 1k was fastets. If
-# you find this number to be a bottleneck in any way you are welcome to change
-# it
-BUFFSIZE = 1024
 
 
 def stripNewLines(lines):
@@ -193,60 +180,8 @@
     LOW = 19
 
 
- at logskip("Storage.Misc.excCmd")
-def execCmd(command, sudo=False, cwd=None, data=None, raw=False, logErr=True,
-            printable=None, env=None, sync=True, nice=None, ioclass=None,
-            ioclassdata=None, setsid=False):
-    """
-    Executes an external command, optionally via sudo.
-    """
-    if ioclass is not None:
-        cmd = command
-        command = [constants.EXT_IONICE, '-c', str(ioclass)]
-        if ioclassdata is not None:
-            command.extend(("-n", str(ioclassdata)))
-
-        command = command + cmd
-
-    if nice is not None:
-        command = [constants.EXT_NICE, '-n', str(nice)] + command
-
-    if setsid:
-        command = [constants.EXT_SETSID] + command
-
-    if sudo:
-        command = [constants.EXT_SUDO, SUDO_NON_INTERACTIVE_FLAG] + command
-
-    if not printable:
-        printable = command
-
-    cmdline = repr(subprocess.list2cmdline(printable))
-    execCmdLogger.debug("%s (cwd %s)", cmdline, cwd)
-
-    p = BetterPopen(command, close_fds=True, cwd=cwd, env=env)
-    p = AsyncProc(p)
-    if not sync:
-        if data is not None:
-            p.stdin.write(data)
-            p.stdin.flush()
-
-        return p
-
-    (out, err) = p.communicate(data)
-
-    if out is None:
-        # Prevent splitlines() from barfing later on
-        out = ""
-
-    execCmdLogger.debug("%s: <err> = %s; <rc> = %d",
-                        {True: "SUCCESS", False: "FAILED"}[p.returncode == 0],
-                        repr(err), p.returncode)
-
-    if not raw:
-        out = out.splitlines(False)
-        err = err.splitlines(False)
-
-    return (p.returncode, out, err)
+execCmd = partial(logskip("Storage.Misc.excCmd")(utils.execCmd),
+                  execCmdLogger=execCmdLogger)
 
 
 def pidExists(pid):
@@ -786,217 +721,6 @@
         self._finally.insert(0, (func, args, kwargs))
 
 
-class AsyncProc(object):
-    """
-    AsyncProc is a funky class. It warps a standard subprocess.Popen
-    Object and gives it super powers. Like the power to read from a stream
-    without the fear of deadlock. It does this by always sampling all
-    stream while waiting for data. By doing this the other process can freely
-    write data to all stream without the fear of it getting stuck writing
-    to a full pipe.
-    """
-    class _streamWrapper(io.RawIOBase):
-        def __init__(self, parent, streamToWrap, fd):
-            io.IOBase.__init__(self)
-            self._stream = streamToWrap
-            self._parent = proxy(parent)
-            self._fd = fd
-            self._closed = False
-            self._emptyCounter = 0
-
-        def close(self):
-            if not self._closed:
-                self._closed = True
-                while not self._streamClosed:
-                    self._parent._processStreams()
-
-        @property
-        def closed(self):
-            return self._closed
-
-        @property
-        def _streamClosed(self):
-            return (self.fileno() in self._parent._closedfds)
-
-        def fileno(self):
-            return self._fd
-
-        def seekable(self):
-            return False
-
-        def readable(self):
-            return True
-
-        def writable(self):
-            return True
-
-        def read(self, length):
-            hasNewData = (self._stream.len - self._stream.pos)
-            if hasNewData < length and not self._streamClosed:
-                self._parent._processStreams()
-
-            with self._parent._streamLock:
-                res = self._stream.read(length)
-                if self._stream.pos == self._stream.len:
-                    if self._streamClosed and res == "":
-                        self._emptyCounter += 1
-                        if self._emptyCounter > 2:
-                            self._closed = True
-
-                    self._stream.truncate(0)
-
-            return res
-
-        def readinto(self, b):
-            data = self.read(len(b))
-            bytesRead = len(data)
-            b[:bytesRead] = data
-
-            return bytesRead
-
-        def write(self, data):
-            if hasattr(data, "tobytes"):
-                data = data.tobytes()
-            with self._parent._streamLock:
-                oldPos = self._stream.pos
-                self._stream.pos = self._stream.len
-                self._stream.write(data)
-                self._stream.pos = oldPos
-
-            while self._stream.len > 0 and not self._streamClosed:
-                self._parent._processStreams()
-
-            if self._streamClosed:
-                self._closed = True
-
-            if self._stream.len != 0:
-                raise IOError(errno.EPIPE,
-                              "Could not write all data to stream")
-
-            return len(data)
-
-    def __init__(self, popenToWrap):
-        self._streamLock = threading.Lock()
-        self._proc = popenToWrap
-
-        self._stdout = StringIO()
-        self._stderr = StringIO()
-        self._stdin = StringIO()
-
-        fdout = self._proc.stdout.fileno()
-        fderr = self._proc.stderr.fileno()
-        self._fdin = self._proc.stdin.fileno()
-
-        self._closedfds = []
-
-        self._poller = select.epoll()
-        self._poller.register(fdout, select.EPOLLIN | select.EPOLLPRI)
-        self._poller.register(fderr, select.EPOLLIN | select.EPOLLPRI)
-        self._poller.register(self._fdin, 0)
-        self._fdMap = {fdout: self._stdout,
-                       fderr: self._stderr,
-                       self._fdin: self._stdin}
-
-        self.stdout = io.BufferedReader(self._streamWrapper(self,
-                                        self._stdout, fdout), BUFFSIZE)
-
-        self.stderr = io.BufferedReader(self._streamWrapper(self,
-                                        self._stderr, fderr), BUFFSIZE)
-
-        self.stdin = io.BufferedWriter(self._streamWrapper(self,
-                                       self._stdin, self._fdin), BUFFSIZE)
-
-        self._returncode = None
-
-    def _processStreams(self):
-        if len(self._closedfds) == 3:
-            return
-
-        if not self._streamLock.acquire(False):
-            self._streamLock.acquire()
-            self._streamLock.release()
-            return
-        try:
-            if self._stdin.len > 0 and self._stdin.pos == 0:
-                # Polling stdin is redundant if there is nothing to write
-                # trun on only if data is waiting to be pushed
-                self._poller.modify(self._fdin, select.EPOLLOUT)
-
-            pollres = NoIntrPoll(self._poller.poll, 1)
-
-            for fd, event in pollres:
-                stream = self._fdMap[fd]
-                if event & select.EPOLLOUT and self._stdin.len > 0:
-                    buff = self._stdin.read(BUFFSIZE)
-                    written = os.write(fd, buff)
-                    stream.pos -= len(buff) - written
-                    if stream.pos == stream.len:
-                        stream.truncate(0)
-                        self._poller.modify(fd, 0)
-
-                elif event & (select.EPOLLIN | select.EPOLLPRI):
-                    data = os.read(fd, BUFFSIZE)
-                    oldpos = stream.pos
-                    stream.pos = stream.len
-                    stream.write(data)
-                    stream.pos = oldpos
-
-                elif event & (select.EPOLLHUP | select.EPOLLERR):
-                    self._poller.unregister(fd)
-                    self._closedfds.append(fd)
-                    # I don't close the fd because the original Popen
-                    # will do it.
-
-            if self.stdin.closed and self._fdin not in self._closedfds:
-                self._poller.unregister(self._fdin)
-                self._closedfds.append(self._fdin)
-                self._proc.stdin.close()
-
-        finally:
-            self._streamLock.release()
-
-    @property
-    def pid(self):
-        return self._proc.pid
-
-    @property
-    def returncode(self):
-        if self._returncode is None:
-            self._returncode = self._proc.poll()
-        return self._returncode
-
-    def kill(self):
-        try:
-            self._proc.kill()
-        except OSError as ex:
-            if ex.errno != errno.EPERM:
-                raise
-            execCmd([constants.EXT_KILL, "-%d" % (signal.SIGTERM,),
-                    str(self.pid)], sudo=True)
-
-    def wait(self, timeout=None, cond=None):
-        startTime = time.time()
-        while self.returncode is None:
-            if timeout is not None and (time.time() - startTime) > timeout:
-                return False
-            if cond is not None and cond():
-                return False
-            self._processStreams()
-        return True
-
-    def communicate(self, data=None):
-        if data is not None:
-            self.stdin.write(data)
-            self.stdin.flush()
-        self.stdin.close()
-
-        self.wait()
-        return "".join(self.stdout), "".join(self.stderr)
-
-    def __del__(self):
-        self._poller.close()
-
-
 class DynamicBarrier(object):
     def __init__(self):
         self._lock = threading.Lock()
@@ -1374,30 +1098,7 @@
         break
 
 
-# NOTE: it would be best to try and unify NoIntrCall and NoIntrPoll.
-# We could do so defining a new object that can be used as a placeholer
-# for the changing timeout value in the *args/**kwargs. This would
-# lead us to rebuilding the function arguments at each loop.
-def NoIntrPoll(pollfun, timeout=-1):
-    """
-    This wrapper is used to handle the interrupt exceptions that might
-    occur during a poll system call. The wrapped function must be defined
-    as poll([timeout]) where the special timeout value 0 is used to return
-    immediately and -1 is used to wait indefinitely.
-    """
-    # When the timeout < 0 we shouldn't compute a new timeout after an
-    # interruption.
-    endtime = None if timeout < 0 else time.time() + timeout
-
-    while True:
-        try:
-            return pollfun(timeout)
-        except (IOError, select.error) as e:
-            if e.args[0] != errno.EINTR:
-                raise
-
-        if endtime is not None:
-            timeout = max(0, endtime - time.time())
+NoIntrPoll = utils.NoIntrPoll
 
 
 def isAscii(s):


--
To view, visit http://gerrit.ovirt.org/14221
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Id8fb42035b24bc915aaa700ab128580bdb609d9e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Zhou Zheng Sheng <zhshzhou at linux.vnet.ibm.com>


More information about the vdsm-patches mailing list