This was the first version with the stable API and all the fixes required to make the Python bindings not crash when used asynchronously. --- wrapper/disk_sync.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/wrapper/disk_sync.py b/wrapper/disk_sync.py index e830193..e655ead 100644 --- a/wrapper/disk_sync.py +++ b/wrapper/disk_sync.py @@ -4,6 +4,8 @@ from pyVmomi import vim from pyVim.connect import SmartConnect, Disconnect from pyVim.task import WaitForTask +from packaging import version +

import nbd from six.moves.urllib.parse import urlparse, unquote @@ -15,6 +17,8 @@ import ssl import sys import json +NBD_MIN_VERSION = version.parse("0.9.8") + LOG_FORMAT_TIME = '[%(asctime)s] ' LOG_FORMAT_MSG = ': %(message)s' @@ -582,6 +586,13 @@ def main(): '''TODO: Add some description here ''' args = parse_args() + + nbd_version = version.parse(nbd.NBD().get_version()) + if nbd_version < NBD_MIN_VERSION: + logging.error("version on libnbd is too old. Version found = %s. Min version required = %s" % + (nbd_version, NBD_MIN_VERSION)) + sys.exit(1) +

state = State(args).instance validate_state(args.sync_type) parse_input(args) -- 2.22.0

Previously the inner loop would issue nbd.pread() requests synchronously, meaning that we would issue a request for each data block from the nbdkit server (which would in turn synchronously request the data from VMware) and wait until nbdkit replies before continuing. This converts the inner loop so it issues as many pread requests asychronously to nbdkit as the server can handle (any extra are queued in nbd_handle). The server will answer these in parallel and possibly out of order. This results in somewhat better throughput (for me: 13 minutes down to 5 minutes for an "initial" run). Although unfortunately we are still limited by VDDK's braindead threading model. --- wrapper/disk_sync.py | 55 +++++++++++++++++++++++++++++++------------- 1 file changed, 39 insertions(+), 16 deletions(-) diff --git a/wrapper/disk_sync.py b/wrapper/disk_sync.py index e655ead..e854009 100644 --- a/wrapper/disk_sync.py +++ b/wrapper/disk_sync.py @@ -409,6 +409,26 @@ def get_block_status(nbd_handle, extent): return blocks +# This is called back when nbd_aio_pread completes. +def read_completed(fd, buf, offset, err): + logging.debug('Writing %d B to offset %d B' % (buf.size(), offset)) + os.pwrite(fd, buf.to_bytearray(), offset) + # By returning 1 here we auto-retire the aio_pread command. + return 1 + + +# Process any AIO requests without blocking. +def process_aio_requests(nbd_handle): + while nbd_handle.poll(0) == 1: + pass + + +# Block until all AIO commands on the handle have finished. +def wait_for_aio_commands_to_finish(nbd_handle): + while nbd_handle.aio_in_flight() > 0: + nbd_handle.poll(-1) + + def sync_data(): state = State().instance for key, disk in state['disks'].items(): @@ -491,25 +511,28 @@ def sync_data(): (block['length'], block['offset'])) # Optimize for memory usage, maybe? os.pwrite(fd, [0] * block['length'], block['offset']) - copied += block['length'] - disk['progress']['copied'] = copied - state.write() else: - wrote = 0 - while wrote < block['length']: - length = min(block['length'] - wrote, MAX_PREAD_LEN) - offset = block['offset'] + wrote + count = 0 + while count < block['length']: + length = min(block['length'] - count, MAX_PREAD_LEN) + offset = block['offset'] + count + logging.debug('Reading %d B from offset %d B' % (length, offset)) - # Ideally use mmap() without any temporary buffer - data = nbd_handle.pread(length, offset) - logging.debug('Writing %d B to offset %d B' % - (length, offset)) - os.pwrite(fd, data, offset) - copied += length - wrote += length - disk['progress']['copied'] = copied - state.write() + buf = nbd.Buffer(length) + nbd_handle.aio_pread( + buf, offset, + lambda err, fd=fd, buf=buf, offset=offset: + read_completed(fd, buf, offset, err))

+ count += length + + process_aio_requests(nbd_handle)

+ + copied += block['length'] + disk['progress']['copied'] = copied + state.write() + + wait_for_aio_commands_to_finish(nbd_handle) if copied == 0: logging.debug('Nothing to copy for disk: %s(key=%s)' % -- 2.22.0

On Mon, Aug 26, 2019 at 01:36:41PM +0200, Martin Kletzander wrote: > On Thu, Aug 22, 2019 at 03:39:35PM +0100, Richard W.M. Jones wrote: > >Previously the inner loop would issue nbd.pread() requests > >synchronously, meaning that we would issue a request for each data > >block from the nbdkit server (which would in turn synchronously > >request the data from VMware) and wait until nbdkit replies before > >continuing. > > > >This converts the inner loop so it issues as many pread requests > >asychronously to nbdkit as the server can handle (any extra are queued > >in nbd_handle). The server will answer these in parallel and possibly > >out of order. > > > >This results in somewhat better throughput (for me: 13 minutes down to > >5 minutes for an "initial" run). Although unfortunately we are still > >limited by VDDK's braindead threading model. > >--- > >wrapper/disk_sync.py | 55 +++++++++++++++++++++++++++++++------------- > >1 file changed, 39 insertions(+), 16 deletions(-) > > > >diff --git a/wrapper/disk_sync.py b/wrapper/disk_sync.py > >index e655ead..e854009 100644 > >--- a/wrapper/disk_sync.py > >+++ b/wrapper/disk_sync.py > >@@ -409,6 +409,26 @@ def get_block_status(nbd_handle, extent): > > return blocks > > > > > >+# This is called back when nbd_aio_pread completes. > >+def read_completed(fd, buf, offset, err): > >+ logging.debug('Writing %d B to offset %d B' % (buf.size(), offset)) > >+ os.pwrite(fd, buf.to_bytearray(), offset) > >+ # By returning 1 here we auto-retire the aio_pread command. > >+ return 1 > >+ > >+ > >+# Process any AIO requests without blocking. > >+def process_aio_requests(nbd_handle): > >+ while nbd_handle.poll(0) == 1: > >+ pass > >+ > >+ > >+# Block until all AIO commands on the handle have finished. > >+def wait_for_aio_commands_to_finish(nbd_handle): > >+ while nbd_handle.aio_in_flight() > 0: > >+ nbd_handle.poll(-1) > >+ > >+ > >def sync_data(): > > state = State().instance > > for key, disk in state['disks'].items(): > >@@ -491,25 +511,28 @@ def sync_data(): > > (block['length'], block['offset'])) > > # Optimize for memory usage, maybe? > > os.pwrite(fd, [0] * block['length'], block['offset']) > >- copied += block['length'] > >- disk['progress']['copied'] = copied > >- state.write() > > else: > >- wrote = 0 > >- while wrote < block['length']: > >- length = min(block['length'] - wrote, MAX_PREAD_LEN) > >- offset = block['offset'] + wrote > >+ count = 0 > >+ while count < block['length']: > >+ length = min(block['length'] - count, MAX_PREAD_LEN) > >+ offset = block['offset'] + count > >+ > > logging.debug('Reading %d B from offset %d B' % > > (length, offset)) > >- # Ideally use mmap() without any temporary buffer > >- data = nbd_handle.pread(length, offset) > >- logging.debug('Writing %d B to offset %d B' % > >- (length, offset)) > >- os.pwrite(fd, data, offset) > >- copied += length > >- wrote += length > >- disk['progress']['copied'] = copied > >- state.write() > >+ buf = nbd.Buffer(length) > >+ nbd_handle.aio_pread( > >+ buf, offset, > >+ lambda err, fd=fd, buf=buf, offset=offset: > >+ read_completed(fd, buf, offset, err)) > > If the order of parameters is changed, there is no need for the anonymous > function here, but that's just a small thing I noticed. The initialized parameters seem to be needed to work around a bug (more properly "serious implementation flaw") in Python: https://stackoverflow.com/questions/2295290

> >+ count += length > >+ > >+ process_aio_requests(nbd_handle) > > In order to allow less requests in flight, would it be enough to just do > something like this here (similarly to wait_for_aio_commands_to_finish)? > > while nbd_handle.aio_in_flight() > NUM_IN_FLIGHT: > nbd_handle.poll(-1) Yes we could do this. The eventual solution may be to replace the whole loop with a polling loop, but this would be a more invasive change. For comparison of this approach see: https://github.com/libguestfs/libnbd/blob/7ef893735937cd7ae62d0b41171ec14... > Also, I presume all of the locking is left to libnbd to be done (and > as you can see I don't concern myself with any locking in the whole > file), but if that was to be improved, is there some python part > that would require it? For example when cleaning up the code? There's a lock automatically held whenever any method on nbd_handle is called. I don't believe any other locking is needed. In any case it's all single-threaded, libnbd does not create any threads itself. Rich. -- Richard Jones, Virtualization Group, Red Hat http://people.redhat.com/~rjones Read my programming and virtualization blog: http://rwmj.wordpress.com virt-df lists disk usage of guests without needing to install any software inside the virtual machine. Supports Linux and Windows. http://people.redhat.com/~rjones/virt-df/