Libguestfs - Aug 2019 - [PATCH libnbd v2 0/3] python: Add test for doing asynch copy.

v1 was here:
https://www.redhat.com/archives/libguestfs/2019-August/msg00103.html

In v2 I've made several changes:

 - Fix Python callbacks so if they don't return something
   which is int-like, we assume they mean to return 0.

 - Add nbd.Buffer free() method.  Read commit message in
   patch 2 to see what this is about.

 - Fixed the asynch copy test to deal with the unbelievably
   braindead behaviour of Python closures.  It took me many
   hours this morning to work out what was going on.  It
   turns out that Python closures aren't closed (yeah, the clue
   is in the name, Python developers).

Rich.

See equivalent change for OCaml in
commit d881d160e1cd9c9964782300a7652ffb4e506c27.

If the Python callback doesn't return something which looks like an
integer, assume 0 instead of returning an error.
---
 generator/generator | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/generator/generator b/generator/generator
index 55c4dfc..e5d9aaa 100755
--- a/generator/generator
+++ b/generator/generator
@@ -4234,7 +4234,12 @@ let print_python_binding name { args; optargs; ret;
may_set_error }         pr "    Py_DECREF (py_args);\n";
        pr "\n";
        pr "    if (py_ret != NULL) {\n";
-       pr "      Py_DECREF (py_ret); /* return value is discarded
*/\n";
+       pr "      if (PyLong_Check (py_ret))\n";
+       pr "        ret = PyLong_AsLong (py_ret);\n";
+       pr "      else\n";
+       pr "        /* If it's not a long, just assume it's 0.
*/\n";
+       pr "        ret = 0;\n";
+       pr "      Py_DECREF (py_ret);\n";
        pr "    }\n";
        pr "    else {\n";
        pr "      ret = -1;\n";
-- 
2.22.0

As in OCaml we need to manually free the buffer, so this commit adds a
free() method to the buffer.

Note (as in OCaml) you must keep a reference to the persistent buffer.
The following code will trivially segfault:

  buf = nbd.Buffer (512)
  nbd.aio_pread (buf, 0)
  # allow buf to go out of scope here

Writing correct code is therefore difficult, but doing it properly
will require cooperation from libnbd so we can increment the refcount
when aio_pread() is called and decrement the refcount when libnbd no
longer holds a reference to the buffer internally.
---
 generator/generator | 14 +++++++++++++-
 python/handle.c     | 23 +++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/generator/generator b/generator/generator
index e5d9aaa..f6c1454 100755
--- a/generator/generator
+++ b/generator/generator
@@ -4081,6 +4081,7 @@ raise_exception ()
          name;
   ) ([ "create"; "close";
        "alloc_aio_buffer";
+       "free_aio_buffer";
        "aio_buffer_from_bytearray";
        "aio_buffer_to_bytearray";
        "aio_buffer_size" ] @ List.map fst handle_calls);
@@ -4111,6 +4112,7 @@ let generate_python_libnbdmod_c ()           name name;
   ) ([ "create"; "close";
        "alloc_aio_buffer";
+       "free_aio_buffer";
        "aio_buffer_from_bytearray";
        "aio_buffer_to_bytearray";
        "aio_buffer_size" ] @ List.map fst handle_calls);
@@ -4394,7 +4396,12 @@ let print_python_binding name { args; optargs; ret;
may_set_error }      | BytesOut (n, count) ->
        pr "  %s = malloc (%s);\n" n count
     | BytesPersistIn (n, _) | BytesPersistOut (n, _) ->
-       pr "  %s_buf = nbd_internal_py_get_aio_buffer (%s);\n" n n
+       pr "  %s_buf = nbd_internal_py_get_aio_buffer (%s);\n" n n;
+       pr "  if (%s_buf->data == NULL) {\n" n;
+       pr "    PyErr_SetString (PyExc_RuntimeError,\n";
+       pr "                     \"nbd.Buffer has been
freed\");\n";
+       pr "    return NULL;\n";
+       pr "  }\n";
     | Closure { cbname } ->
        pr "  /* Increment refcount since pointer may be saved by libnbd.
*/\n";
        pr "  Py_INCREF (%s_user_data);\n" cbname;
@@ -4651,6 +4658,11 @@ class Buffer (object):
         '''allocate an uninitialized AIO buffer used for
nbd.aio_pread'''
         self._o = libnbdmod.alloc_aio_buffer (len)
 
+    def free (self):
+        '''free the buffer:
+        you should call this in the completion callback'''
+        libnbdmod.free_aio_buffer (self._o)
+
     @classmethod
     def from_bytearray (cls, ba):
         '''create an AIO buffer from a bytearray'''
diff --git a/python/handle.c b/python/handle.c
index 4b510ad..2de8712 100644
--- a/python/handle.c
+++ b/python/handle.c
@@ -89,6 +89,29 @@ free_aio_buffer (PyObject *capsule)
   free (buf);
 }
 
+PyObject *
+nbd_internal_py_free_aio_buffer (PyObject *self, PyObject *args)
+{
+  PyObject *obj;
+  struct py_aio_buffer *buf;
+
+  if (!PyArg_ParseTuple (args,
+                         (char *)
"O:nbd_internal_py_free_aio_buffer",
+                         &obj))
+    return NULL;
+
+  buf = nbd_internal_py_get_aio_buffer (obj);
+  if (buf == NULL)
+    return NULL;
+
+  free (buf->data);
+  /* Set this to NULL to avoid double-free in the destructor. */
+  buf->data = NULL;
+
+  Py_INCREF (Py_None);
+  return Py_None;
+}
+
 /* Allocate a persistent buffer used for nbd_aio_pread. */
 PyObject *
 nbd_internal_py_alloc_aio_buffer (PyObject *self, PyObject *args)
-- 
2.22.0

---
 python/t/590-aio-copy.py | 123 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 123 insertions(+)

diff --git a/python/t/590-aio-copy.py b/python/t/590-aio-copy.py
new file mode 100644
index 0000000..999491b
--- /dev/null
+++ b/python/t/590-aio-copy.py
@@ -0,0 +1,123 @@
+# libnbd Python bindings
+# Copyright (C) 2010-2019 Red Hat Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+
+import select
+import nbd
+
+disk_size = 512 * 1024 * 1024
+bs = 65536
+max_reads_in_flight = 16
+bytes_read = 0
+bytes_written = 0
+
+def asynch_copy (src, dst):
+    size = src.get_size ()
+
+    # This is our reading position in the source.
+    soff = 0
+
+    # This callback is called when any pread from the source
+    # has completed.
+    writes = []
+    def read_completed (buf, offset, error):
+        global bytes_read
+        bytes_read += buf.size ()
+        wr = (buf, offset)
+        writes.append (wr)
+        # By returning 1 here we auto-retire the pread command.
+        return 1
+
+    # This callback is called when any pwrite to the destination
+    # has completed.
+    def write_completed (buf, error):
+        global bytes_written
+        bytes_written += buf.size ()
+        buf.free ()
+        # By returning 1 here we auto-retire the pwrite command.
+        return 1
+
+    # The main loop which runs until we have finished reading and
+    # there are no more commands in flight.
+    while soff < size or dst.aio_in_flight () > 0:
+        # If we're able to submit more reads from the source
+        # then do so now.
+        if soff < size and src.aio_in_flight () < max_reads_in_flight:
+            bufsize = min (bs, size - soff)
+            buf = nbd.Buffer (bufsize)
+            # NB: Python lambdas are BROKEN.
+            # https://stackoverflow.com/questions/2295290
+            src.aio_pread_callback (buf, soff,
+                                    lambda err, buf=buf, soff=soff:
+                                    read_completed (buf, soff, err))
+            soff += bufsize
+
+        # If there are any write commands waiting to be issued
+        # to the destination, send them now.
+        for buf, offset in writes:
+            # See above link about broken Python lambdas.
+            dst.aio_pwrite_callback (buf, offset,
+                                     lambda err, buf=buf:
+                                     write_completed (buf, err))
+        writes = []
+
+        poll = select.poll ()
+
+        sfd = src.aio_get_fd ()
+        dfd = dst.aio_get_fd ()
+
+        sevents = 0
+        devents = 0
+        if src.aio_get_direction () & nbd.AIO_DIRECTION_READ:
+            sevents += select.POLLIN
+        if src.aio_get_direction () & nbd.AIO_DIRECTION_WRITE:
+            sevents += select.POLLOUT
+        if dst.aio_get_direction () & nbd.AIO_DIRECTION_READ:
+            devents += select.POLLIN
+        if dst.aio_get_direction () & nbd.AIO_DIRECTION_WRITE:
+            devents += select.POLLOUT
+        poll.register (sfd, sevents)
+        poll.register (dfd, devents)
+        for (fd, revents) in poll.poll ():
+            # The direction of each handle can change since we
+            # slept in the select.
+            if fd == sfd and revents & select.POLLIN and \
+               src.aio_get_direction () & nbd.AIO_DIRECTION_READ:
+                src.aio_notify_read ()
+            elif fd == sfd and revents & select.POLLOUT and \
+                 src.aio_get_direction () & nbd.AIO_DIRECTION_WRITE:
+                src.aio_notify_write ()
+            elif fd == dfd and revents & select.POLLIN and \
+                 dst.aio_get_direction () & nbd.AIO_DIRECTION_READ:
+                dst.aio_notify_read ()
+            elif fd == dfd and revents & select.POLLOUT and \
+                 dst.aio_get_direction () & nbd.AIO_DIRECTION_WRITE:
+                dst.aio_notify_write ()
+
+src = nbd.NBD ()
+src.set_handle_name ("src")
+dst = nbd.NBD ()
+dst.set_handle_name ("dst")
+src.connect_command (["nbdkit", "-s",
"--exit-with-parent", "-r",
+                      "pattern", "size=%d" % disk_size])
+dst.connect_command (["nbdkit", "-s",
"--exit-with-parent",
+                      "memory", "size=%d" % disk_size])
+asynch_copy (src, dst)
+
+print ("bytes read: %d written: %d size: %d" %
+       (bytes_read, bytes_written, disk_size))
+assert bytes_read == disk_size
+assert bytes_written == disk_size
-- 
2.22.0

On 8/11/19 5:26 AM, Richard W.M. Jones wrote:
> See equivalent change for OCaml in
> commit d881d160e1cd9c9964782300a7652ffb4e506c27.
> 
> If the Python callback doesn't return something which looks like an
> integer, assume 0 instead of returning an error.
> ---
>  generator/generator | 7 ++++++-
>  1 file changed, 6 insertions(+), 1 deletion(-)
> 
> diff --git a/generator/generator b/generator/generator
> index 55c4dfc..e5d9aaa 100755
> --- a/generator/generator
> +++ b/generator/generator
> @@ -4234,7 +4234,12 @@ let print_python_binding name { args; optargs; ret;
may_set_error } >         pr "    Py_DECREF (py_args);\n";
>         pr "\n";
>         pr "    if (py_ret != NULL) {\n";
> -       pr "      Py_DECREF (py_ret); /* return value is discarded
*/\n";
> +       pr "      if (PyLong_Check (py_ret))\n";
> +       pr "        ret = PyLong_AsLong (py_ret);\n";
If I'm reading the python documentation correctly, this only succeeds
for actual subtypes of PyLong (and not for other python types that have
an __int__ converter available).  But I don't find that too restricting;
'return 1' is easier to type than 'return myobject' where
myobject is
not a subtype of PyLong but could otherwise convert to 1.  And even if
the user insists on using some sort of myobject for tracking the logic
on whether auto-retire is needed, they can spell longhand:
  if myobject == 1:
    return 1
  return 0
to get the desired semantics if 'return myobject' doesn't do what
they want.
> +       pr "      else\n";
> +       pr "        /* If it's not a long, just assume it's 0.
*/\n";
> +       pr "        ret = 0;\n";
Makes sense. Thus, only an explicit return of -1 or throwing an
exception will turn into a C return of -1, only an explicit return of 1
will turn into auto-retire, and anything else makes no difference.

ACK.

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3226
Virtualization:  qemu.org | libvirt.org