[Qemu-devel] [PATCH 1/2] QMP: add get_events(wait=True) option

[Qemu-devel] [PATCH 1/2] QMP: add get_events(wait=True) option

[Stefan Hajnoczi]

The get_events() function polls for new QMP events and then returns.  It
can be useful to wait for the next QMP event so add the boolean 'wait'
keyword argument.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 QMP/qmp.py |   11 ++++++++---
 1 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/QMP/qmp.py b/QMP/qmp.py
index 14ce8b0..2565508 100644
--- a/QMP/qmp.py
+++ b/QMP/qmp.py
@@ -43,7 +43,7 @@ class QEMUMonitorProtocol:
             family = socket.AF_UNIX
         return socket.socket(family, socket.SOCK_STREAM)
 
-    def __json_read(self):
+    def __json_read(self, only_event=False):
         while True:
             data = self.__sockfile.readline()
             if not data:
@@ -51,7 +51,8 @@ class QEMUMonitorProtocol:
             resp = json.loads(data)
             if 'event' in resp:
                 self.__events.append(resp)
-                continue
+                if not only_event:
+                    continue
             return resp
 
     error = socket.error
@@ -106,9 +107,11 @@ class QEMUMonitorProtocol:
             qmp_cmd['id'] = id
         return self.cmd_obj(qmp_cmd)
 
-    def get_events(self):
+    def get_events(self, wait=False):
         """
         Get a list of available QMP events.
+
+        @param wait: block until an event is available (bool)
         """
         self.__sock.setblocking(0)
         try:
@@ -118,6 +121,8 @@ class QEMUMonitorProtocol:
                 # No data available
                 pass
         self.__sock.setblocking(1)
+        if not self.__events and wait:
+            self.__json_read(only_event=True)
         return self.__events
 
     def clear_events(self):
-- 
1.7.4.4

[Qemu-devel] [PATCH 2/2] QMP: add server mode to QEMUMonitorProtocol

[Stefan Hajnoczi]

QEMU supports socket chardevs that establish connections like a server
or a client.  The QEMUMonitorProtocol class only supports connecting as
a client.  It is not possible to connect race-free when launching QEMU
since trying to connect before QEMU has bound and is listening on the
socket results in failure.

Add the QEMUMonitorProtocol(server=True) argument to bind and listen on
the socket.  The QEMU process can then be launched and connects to the
already existing QMP socket without a race condition:

  qmp = qmp.QEMUMonitorProtocol(monitor_path, server=True)
  popen = subprocess.Popen(args)
  qmp.accept()

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
 QMP/qmp.py |   43 ++++++++++++++++++++++++++++++++-----------
 1 files changed, 32 insertions(+), 11 deletions(-)

diff --git a/QMP/qmp.py b/QMP/qmp.py
index 2565508..c7dbea0 100644
--- a/QMP/qmp.py
+++ b/QMP/qmp.py
@@ -22,19 +22,24 @@ class QMPCapabilitiesError(QMPError):
     pass
 
 class QEMUMonitorProtocol:
-    def __init__(self, address):
+    def __init__(self, address, server=False):
         """
         Create a QEMUMonitorProtocol class.
 
         @param address: QEMU address, can be either a unix socket path (string)
                         or a tuple in the form ( address, port ) for a TCP
                         connection
-        @note No connection is established, this is done by the connect() method
+        @param server: server mode listens on the socket (bool)
+        @raise socket.error on socket connection errors
+        @note No connection is established, this is done by the connect() or
+              accept() methods
         """
         self.__events = []
         self.__address = address
         self.__sock = self.__get_sock()
-        self.__sockfile = self.__sock.makefile()
+        if server:
+            self.__sock.bind(self.__address)
+            self.__sock.listen(1)
 
     def __get_sock(self):
         if isinstance(self.__address, tuple):
@@ -43,6 +48,17 @@ class QEMUMonitorProtocol:
             family = socket.AF_UNIX
         return socket.socket(family, socket.SOCK_STREAM)
 
+    def __negotiate_capabilities(self):
+        self.__sockfile = self.__sock.makefile()
+        greeting = self.__json_read()
+        if greeting is None or not greeting.has_key('QMP'):
+            raise QMPConnectError
+        # Greeting seems ok, negotiate capabilities
+        resp = self.cmd('qmp_capabilities')
+        if "return" in resp:
+            return greeting
+        raise QMPCapabilitiesError
+
     def __json_read(self, only_event=False):
         while True:
             data = self.__sockfile.readline()
@@ -67,14 +83,19 @@ class QEMUMonitorProtocol:
         @raise QMPCapabilitiesError if fails to negotiate capabilities
         """
         self.__sock.connect(self.__address)
-        greeting = self.__json_read()
-        if greeting is None or not greeting.has_key('QMP'):
-            raise QMPConnectError
-        # Greeting seems ok, negotiate capabilities
-        resp = self.cmd('qmp_capabilities')
-        if "return" in resp:
-            return greeting
-        raise QMPCapabilitiesError
+        return self.__negotiate_capabilities()
+
+    def accept(self):
+        """
+        Await connection from QMP Monitor and perform capabilities negotiation.
+
+        @return QMP greeting dict
+        @raise socket.error on socket connection errors
+        @raise QMPConnectError if the greeting is not received
+        @raise QMPCapabilitiesError if fails to negotiate capabilities
+        """
+        self.__sock, _ = self.__sock.accept()
+        return self.__negotiate_capabilities()
 
     def cmd_obj(self, qmp_cmd):
         """
-- 
1.7.4.4

Re: [Qemu-devel] [PATCH 1/2] QMP: add get_events(wait=True) option

[Luiz Capitulino]

On Wed, 25 May 2011 19:48:00 +0100
Stefan Hajnoczi <stefanha@linux.vnet.ibm.com> wrote:

> The get_events() function polls for new QMP events and then returns.  It
> can be useful to wait for the next QMP event so add the boolean 'wait'
> keyword argument.
> 
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> ---
>  QMP/qmp.py |   11 ++++++++---
>  1 files changed, 8 insertions(+), 3 deletions(-)
> 
> diff --git a/QMP/qmp.py b/QMP/qmp.py
> index 14ce8b0..2565508 100644
> --- a/QMP/qmp.py
> +++ b/QMP/qmp.py
> @@ -43,7 +43,7 @@ class QEMUMonitorProtocol:
>              family = socket.AF_UNIX
>          return socket.socket(family, socket.SOCK_STREAM)
>  
> -    def __json_read(self):
> +    def __json_read(self, only_event=False):
>          while True:
>              data = self.__sockfile.readline()
>              if not data:
> @@ -51,7 +51,8 @@ class QEMUMonitorProtocol:
>              resp = json.loads(data)
>              if 'event' in resp:
>                  self.__events.append(resp)
> -                continue
> +                if not only_event:
> +                    continue
>              return resp
>  
>      error = socket.error
> @@ -106,9 +107,11 @@ class QEMUMonitorProtocol:
>              qmp_cmd['id'] = id
>          return self.cmd_obj(qmp_cmd)
>  
> -    def get_events(self):
> +    def get_events(self, wait=False):
>          """
>          Get a list of available QMP events.
> +
> +        @param wait: block until an event is available (bool)
>          """
>          self.__sock.setblocking(0)
>          try:
> @@ -118,6 +121,8 @@ class QEMUMonitorProtocol:
>                  # No data available
>                  pass
>          self.__sock.setblocking(1)
> +        if not self.__events and wait:
> +            self.__json_read(only_event=True)
>          return self.__events

Maybe this is better (untested):

    def get_events(self, wait=False):
        """
        Get a list of available QMP events.

        @param wait: block until an event is available (bool)
        """
        if not wait:
            self.__sock.setblocking(0)
        try:
            self.__json_read(only_event=wait)
        except socket.error, err:
            if err[0] == errno.EAGAIN:
                # No data available
                pass
        if not wait:
            self.__sock.setblocking(1)
        return self.__events

>  
>      def clear_events(self):

Re: [Qemu-devel] [PATCH 2/2] QMP: add server mode to QEMUMonitorProtocol

[Luiz Capitulino]

On Wed, 25 May 2011 19:48:01 +0100
Stefan Hajnoczi <stefanha@linux.vnet.ibm.com> wrote:

> QEMU supports socket chardevs that establish connections like a server
> or a client.  The QEMUMonitorProtocol class only supports connecting as
> a client.  It is not possible to connect race-free when launching QEMU
> since trying to connect before QEMU has bound and is listening on the
> socket results in failure.
> 
> Add the QEMUMonitorProtocol(server=True) argument to bind and listen on
> the socket.  The QEMU process can then be launched and connects to the
> already existing QMP socket without a race condition:
> 
>   qmp = qmp.QEMUMonitorProtocol(monitor_path, server=True)
>   popen = subprocess.Popen(args)
>   qmp.accept()

Would be cool to get the demo shell doing this (not a required for the
initial merge, of course).

> 
> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> ---
>  QMP/qmp.py |   43 ++++++++++++++++++++++++++++++++-----------
>  1 files changed, 32 insertions(+), 11 deletions(-)
> 
> diff --git a/QMP/qmp.py b/QMP/qmp.py
> index 2565508..c7dbea0 100644
> --- a/QMP/qmp.py
> +++ b/QMP/qmp.py
> @@ -22,19 +22,24 @@ class QMPCapabilitiesError(QMPError):
>      pass
>  
>  class QEMUMonitorProtocol:
> -    def __init__(self, address):
> +    def __init__(self, address, server=False):
>          """
>          Create a QEMUMonitorProtocol class.
>  
>          @param address: QEMU address, can be either a unix socket path (string)
>                          or a tuple in the form ( address, port ) for a TCP
>                          connection
> -        @note No connection is established, this is done by the connect() method
> +        @param server: server mode listens on the socket (bool)
> +        @raise socket.error on socket connection errors
> +        @note No connection is established, this is done by the connect() or
> +              accept() methods
>          """
>          self.__events = []
>          self.__address = address
>          self.__sock = self.__get_sock()
> -        self.__sockfile = self.__sock.makefile()
> +        if server:
> +            self.__sock.bind(self.__address)
> +            self.__sock.listen(1)
>  
>      def __get_sock(self):
>          if isinstance(self.__address, tuple):
> @@ -43,6 +48,17 @@ class QEMUMonitorProtocol:
>              family = socket.AF_UNIX
>          return socket.socket(family, socket.SOCK_STREAM)
>  
> +    def __negotiate_capabilities(self):
> +        self.__sockfile = self.__sock.makefile()
> +        greeting = self.__json_read()
> +        if greeting is None or not greeting.has_key('QMP'):
> +            raise QMPConnectError
> +        # Greeting seems ok, negotiate capabilities
> +        resp = self.cmd('qmp_capabilities')
> +        if "return" in resp:
> +            return greeting
> +        raise QMPCapabilitiesError
> +
>      def __json_read(self, only_event=False):
>          while True:
>              data = self.__sockfile.readline()
> @@ -67,14 +83,19 @@ class QEMUMonitorProtocol:
>          @raise QMPCapabilitiesError if fails to negotiate capabilities
>          """
>          self.__sock.connect(self.__address)
> -        greeting = self.__json_read()
> -        if greeting is None or not greeting.has_key('QMP'):
> -            raise QMPConnectError
> -        # Greeting seems ok, negotiate capabilities
> -        resp = self.cmd('qmp_capabilities')
> -        if "return" in resp:
> -            return greeting
> -        raise QMPCapabilitiesError
> +        return self.__negotiate_capabilities()
> +
> +    def accept(self):
> +        """
> +        Await connection from QMP Monitor and perform capabilities negotiation.
> +
> +        @return QMP greeting dict
> +        @raise socket.error on socket connection errors
> +        @raise QMPConnectError if the greeting is not received
> +        @raise QMPCapabilitiesError if fails to negotiate capabilities
> +        """
> +        self.__sock, _ = self.__sock.accept()
> +        return self.__negotiate_capabilities()
>  
>      def cmd_obj(self, qmp_cmd):
>          """

Re: [Qemu-devel] [PATCH 1/2] QMP: add get_events(wait=True) option

[Stefan Hajnoczi]

On Wed, May 25, 2011 at 10:15 PM, Luiz Capitulino
<lcapitulino@redhat.com> wrote:
> On Wed, 25 May 2011 19:48:00 +0100
> Stefan Hajnoczi <stefanha@linux.vnet.ibm.com> wrote:
>
>> The get_events() function polls for new QMP events and then returns.  It
>> can be useful to wait for the next QMP event so add the boolean 'wait'
>> keyword argument.
>>
>> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
>> ---
>>  QMP/qmp.py |   11 ++++++++---
>>  1 files changed, 8 insertions(+), 3 deletions(-)
>>
>> diff --git a/QMP/qmp.py b/QMP/qmp.py
>> index 14ce8b0..2565508 100644
>> --- a/QMP/qmp.py
>> +++ b/QMP/qmp.py
>> @@ -43,7 +43,7 @@ class QEMUMonitorProtocol:
>>              family = socket.AF_UNIX
>>          return socket.socket(family, socket.SOCK_STREAM)
>>
>> -    def __json_read(self):
>> +    def __json_read(self, only_event=False):
>>          while True:
>>              data = self.__sockfile.readline()
>>              if not data:
>> @@ -51,7 +51,8 @@ class QEMUMonitorProtocol:
>>              resp = json.loads(data)
>>              if 'event' in resp:
>>                  self.__events.append(resp)
>> -                continue
>> +                if not only_event:
>> +                    continue
>>              return resp
>>
>>      error = socket.error
>> @@ -106,9 +107,11 @@ class QEMUMonitorProtocol:
>>              qmp_cmd['id'] = id
>>          return self.cmd_obj(qmp_cmd)
>>
>> -    def get_events(self):
>> +    def get_events(self, wait=False):
>>          """
>>          Get a list of available QMP events.
>> +
>> +        @param wait: block until an event is available (bool)
>>          """
>>          self.__sock.setblocking(0)
>>          try:
>> @@ -118,6 +121,8 @@ class QEMUMonitorProtocol:
>>                  # No data available
>>                  pass
>>          self.__sock.setblocking(1)
>> +        if not self.__events and wait:
>> +            self.__json_read(only_event=True)
>>          return self.__events
>
> Maybe this is better (untested):

I've tested it because that's how I implemented it first too :).
However, I don't want to block until the QMP monitor sends the next
event when there is already a received event pending.

The patch I posted polls for new events first, then only blocks if
there are no events available.

Stefan

Re: [Qemu-devel] [PATCH 2/2] QMP: add server mode to QEMUMonitorProtocol

[Stefan Hajnoczi]

The test-stream.py script performs several automated tests of the image
streaming QMP interface, including exercising both the incremental and
background streaming modes.

This should probably be ported to KVM-Autotest rather than reinventing
the wheel.

Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
---
This is an example of how I am using qmp.py, especially the new
QEMUMonitorProtocol(server=True) and QEMUMonitorProtocol.get_events(wait=True)
arguments.

I am not submitting this for merge.  Like the commit message says, this should
probably be ported to KVM-Autotest because it duplicates basic QEMU testing
infrastructure.

 test-stream.py |  145 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 145 insertions(+), 0 deletions(-)
 create mode 100644 test-stream.py

diff --git a/test-stream.py b/test-stream.py
new file mode 100644
index 0000000..eb9d9d6
--- /dev/null
+++ b/test-stream.py
@@ -0,0 +1,145 @@
+#!/usr/bin/env python
+import unittest
+import subprocess
+import re
+import os
+import sys; sys.path.append('QMP/')
+import qmp
+
+def qemu_img(*args):
+    devnull = open('/dev/null', 'r+')
+    return subprocess.call(['./qemu-img'] + list(args), stdin=devnull, stdout=devnull)
+
+class VM(object):
+    def __init__(self):
+        self._monitor_path = '/tmp/qemu-mon.%d' % os.getpid()
+        self._qemu_log_path = '/tmp/qemu-log.%d' % os.getpid()
+        self._args = ['x86_64-softmmu/qemu-system-x86_64',
+                      '-chardev', 'socket,id=mon,path=' + self._monitor_path,
+                      '-mon', 'chardev=mon,mode=control',
+                      '-nographic']
+        self._num_drives = 0
+
+    def add_drive(self, path):
+        self._args.append('-drive')
+        self._args.append('if=virtio,cache=none,file=%s,id=drive%d' % (path, self._num_drives))
+        self._num_drives += 1
+        return self
+
+    def launch(self):
+        devnull = open('/dev/null', 'rb')
+        qemulog = open(self._qemu_log_path, 'wb')
+        self._qmp = qmp.QEMUMonitorProtocol(self._monitor_path, server=True)
+        self._popen = subprocess.Popen(self._args, stdin=devnull, stdout=qemulog,
+                                       stderr=subprocess.STDOUT)
+        self._qmp.accept()
+
+    def shutdown(self):
+        self._qmp.cmd('quit')
+        self._popen.wait()
+        os.remove(self._monitor_path)
+        os.remove(self._qemu_log_path)
+
+    def qmp(self, cmd, **args):
+        return self._qmp.cmd(cmd, args=args)
+
+    def get_qmp_events(self, wait=False):
+        events = self._qmp.get_events(wait=wait)
+        self._qmp.clear_events()
+        return events
+
+index_re = re.compile(r'([^\[]+)\[([^\]]+)\]')
+
+class QMPTestCase(unittest.TestCase):
+    def dictpath(self, d, path):
+        """Traverse a path in a nested dict"""
+        for component in path.split('/'):
+            m = index_re.match(component)
+            if m:
+                component, idx = m.groups()
+                idx = int(idx)
+
+            if not isinstance(d, dict) or component not in d:
+                self.fail('failed path traversal for "%s" in "%s"' % (path, str(d)))
+            d = d[component]
+
+            if m:
+                if not isinstance(d, list):
+                    self.fail('path component "%s" in "%s" is not a list in "%s"' % (component, path, str(d)))
+                try:
+                    d = d[idx]
+                except IndexError:
+                    self.fail('invalid index "%s" in path "%s" in "%s"' % (idx, path, str(d)))
+        return d
+
+    def assert_qmp(self, d, path, value):
+        result = self.dictpath(d, path)
+        self.assertEqual(result, value, 'values not equal "%s" and "%s"' % (str(result), str(value)))
+
+    def assert_no_active_streams(self):
+        result = self.vm.qmp('query-block-stream')
+        self.assert_qmp(result, 'return', [])
+
+class TestSingleDrive(QMPTestCase):
+    image_len = 1 * 1024 * 1024
+
+    def setUp(self):
+        qemu_img('create', 'backing.img', str(TestSingleDrive.image_len))
+        qemu_img('create', '-f', 'qed', '-o', 'backing_file=backing.img,copy_on_read=on', 'test.qed')
+        self.vm = VM().add_drive('test.qed')
+        self.vm.launch()
+
+    def tearDown(self):
+        self.vm.shutdown()
+        os.remove('test.qed')
+        os.remove('backing.img')
+
+    def test_stream_incremental(self):
+        self.assert_no_active_streams()
+
+        completed = False
+        while not completed:
+            result = self.vm.qmp('block_stream', device='drive0')
+            self.assert_qmp(result, 'return/device', 'drive0')
+            self.assert_qmp(result, 'return/len', self.image_len)
+
+            for event in self.vm.get_qmp_events():
+                if event['event'] == 'BLOCK_STREAM_COMPLETED':
+                    self.assert_qmp(event, 'data/device', 'drive0')
+                    self.assert_qmp(event, 'data/offset', self.image_len)
+                    self.assert_qmp(event, 'data/len', self.image_len)
+                    completed = True
+
+            # Compare result against query-block-stream output
+            if not completed:
+                query = self.vm.qmp('query-block-stream')
+                self.assert_qmp(query, 'return[0]/device', 'drive0')
+                self.assert_qmp(query, 'return[0]/offset',
+                                self.dictpath(result, 'return/offset'))
+                self.assert_qmp(query, 'return[0]/len', self.image_len)
+
+        self.assert_no_active_streams()
+
+    def test_stream_all(self):
+        self.assert_no_active_streams()
+
+        result = self.vm.qmp('block_stream', device='drive0', all=True)
+        self.assert_qmp(result, 'return', {})
+
+        completed = False
+        while not completed:
+            for event in self.vm.get_qmp_events(wait=True):
+                if event['event'] == 'BLOCK_STREAM_COMPLETED':
+                    self.assert_qmp(event, 'data/device', 'drive0')
+                    self.assert_qmp(event, 'data/offset', self.image_len)
+                    self.assert_qmp(event, 'data/len', self.image_len)
+                    completed = True
+
+        self.assert_no_active_streams()
+
+    def test_device_not_found(self):
+        result = self.vm.qmp('block_stream', device='nonexistent')
+        self.assert_qmp(result, 'error/class', 'DeviceNotFound')
+
+if __name__ == '__main__':
+    unittest.main()
-- 
1.7.4.4

Re: [Qemu-devel] [PATCH 1/2] QMP: add get_events(wait=True) option

[Luiz Capitulino]

On Thu, 26 May 2011 09:03:49 +0100
Stefan Hajnoczi <stefanha@gmail.com> wrote:

> On Wed, May 25, 2011 at 10:15 PM, Luiz Capitulino
> <lcapitulino@redhat.com> wrote:
> > On Wed, 25 May 2011 19:48:00 +0100
> > Stefan Hajnoczi <stefanha@linux.vnet.ibm.com> wrote:
> >
> >> The get_events() function polls for new QMP events and then returns.  It
> >> can be useful to wait for the next QMP event so add the boolean 'wait'
> >> keyword argument.
> >>
> >> Signed-off-by: Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
> >> ---
> >>  QMP/qmp.py |   11 ++++++++---
> >>  1 files changed, 8 insertions(+), 3 deletions(-)
> >>
> >> diff --git a/QMP/qmp.py b/QMP/qmp.py
> >> index 14ce8b0..2565508 100644
> >> --- a/QMP/qmp.py
> >> +++ b/QMP/qmp.py
> >> @@ -43,7 +43,7 @@ class QEMUMonitorProtocol:
> >>              family = socket.AF_UNIX
> >>          return socket.socket(family, socket.SOCK_STREAM)
> >>
> >> -    def __json_read(self):
> >> +    def __json_read(self, only_event=False):
> >>          while True:
> >>              data = self.__sockfile.readline()
> >>              if not data:
> >> @@ -51,7 +51,8 @@ class QEMUMonitorProtocol:
> >>              resp = json.loads(data)
> >>              if 'event' in resp:
> >>                  self.__events.append(resp)
> >> -                continue
> >> +                if not only_event:
> >> +                    continue
> >>              return resp
> >>
> >>      error = socket.error
> >> @@ -106,9 +107,11 @@ class QEMUMonitorProtocol:
> >>              qmp_cmd['id'] = id
> >>          return self.cmd_obj(qmp_cmd)
> >>
> >> -    def get_events(self):
> >> +    def get_events(self, wait=False):
> >>          """
> >>          Get a list of available QMP events.
> >> +
> >> +        @param wait: block until an event is available (bool)
> >>          """
> >>          self.__sock.setblocking(0)
> >>          try:
> >> @@ -118,6 +121,8 @@ class QEMUMonitorProtocol:
> >>                  # No data available
> >>                  pass
> >>          self.__sock.setblocking(1)
> >> +        if not self.__events and wait:
> >> +            self.__json_read(only_event=True)
> >>          return self.__events
> >
> > Maybe this is better (untested):
> 
> I've tested it because that's how I implemented it first too :).
> However, I don't want to block until the QMP monitor sends the next
> event when there is already a received event pending.
> 
> The patch I posted polls for new events first, then only blocks if
> there are no events available.

Ok, pushed both patches to the QMP queue.

Thanks.