search for: aio_in_flight

...rocess 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']))...

...to remain valid). It may matter more for functions that don't take a buf (such as nbd_aio_trim). > +++ b/lib/rw.c Yep, definitely a mechanical conversion. > +++ b/ocaml/examples/asynch_copy.ml > @@ -48,7 +48,7 @@ let asynch_copy src dst = > if !soff < size && NBD.aio_in_flight src < max_reads_in_flight then ( > let bs = min bs (size -^ !soff) in > let buf = NBD.Buffer.alloc (Int64.to_int bs) in > - ignore (NBD.aio_pread_callback src buf !soff > + ignore (NBD.aio_pread src buf !soff > ~completion:(read_completed b...

...+ global bytes_written + bytes_written += buf.size () + # 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 lambda...

.... > >+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['leng...

This is based on top of: https://github.com/nertpinx/v2v-conversion-host/commit/0bb2efdcacd975a2cae7380080991ac7fc238d2b The first 4 patches are fairly uncontroversial miscellaneous cleanups. Patch 5 is the interesting one. (Note it doesn't quite work yet, so it's for discussion only.) Patch 5 converts the inner loop to use asynchronous libnbd calls. performance improves quite a bit for

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

...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...

Basically the same as this patch series, but for Python: https://www.redhat.com/archives/libguestfs/2019-August/msg00235.html plus adding the 590 asynch test at the end. Rich.

This applies on top of the OClosure v2 series posted a few minutes ago. Rich.

..."aio_is_connecting", (1, 0); + "aio_is_ready", (1, 0); + "aio_is_processing", (1, 0); + "aio_is_dead", (1, 0); + "aio_is_closed", (1, 0); + "aio_command_completed", (1, 0); + "aio_peek_command_completed", (1, 0); + "aio_in_flight", (1, 0); + "connection_state", (1, 0); + "get_package_name", (1, 0); + "get_version", (1, 0); + "kill_subprocess", (1, 0); + "supports_tls", (1, 0); + "supports_uri", (1, 0); + + (* Added in 1.1 development series. *) + &quot...

This doesn't include Eric's new APIs, but if you push those then I can rebase this one on top. Rich.

...We should be able to issue multiple requests in parallel, +# and the total time taken should not be much more than 10 seconds +# because all sleeps in the plugin should happen in parallel. +start_t = time.time() +for i in range (10): + buf = nbd.Buffer (512) + h.aio_pread (buf, 0) + +while h.aio_in_flight() > 0: + h.poll(-1) +end_t = time.time() + +t = end_t - start_t +print (t) + +# Since we launched 10 requests, if we serialized on them we +# would have waited at least 100 seconds. We would expect to +# wait around 10 seconds, but for flexibility on slow servers +# any test < 100 should...

I still need to wire in the use of *_notify functions into nbdkit to prove whether it makes the code any faster or easier to maintain, but at least the added example shows one good use case for the new API. Eric Blake (6): api: Add nbd_aio_in_flight generator: Allow DEAD state actions to run generator: Allow Int64 in callbacks states: Prepare for aio notify callback api: Add new nbd_aio_FOO_notify functions examples: New example for strict read validations .gitignore | 1 + docs/libnbd.pod...

This doesn't yet work. However it does make me more convinced than ever that we really need to sort out persistent buffer lifetimes in the library (similar to what we did for closures). Rich.

...dles[1] = nbd_aio_pwrite (nbd, out, packetsize, packetsize, 0); if (handles[1] == -1) { fprintf (stderr, "%s\n", nbd_get_error ()); goto error; } - in_flight++; /* Now wait for commands to retire, or for deadlock to occur */ - while (in_flight > 0) { + while (nbd_aio_in_flight (nbd) > 0) { if (nbd_aio_is_dead (nbd) || nbd_aio_is_closed (nbd)) { fprintf (stderr, "connection is dead or closed\n"); goto error; @@ -96,23 +92,20 @@ try_deadlock (void *arg) /* If a command is ready to retire, retire it. */ while ((done = nbd_aio_peek_...

....fn_user_data = extent_user_data, diff --git a/ocaml/examples/asynch_copy.ml b/ocaml/examples/asynch_copy.ml index 5aa6e60..8057118 100644 --- a/ocaml/examples/asynch_copy.ml +++ b/ocaml/examples/asynch_copy.ml @@ -48,7 +48,7 @@ let asynch_copy src dst = if !soff < size && NBD.aio_in_flight src < max_reads_in_flight then ( let bs = min bs (size -^ !soff) in let buf = NBD.Buffer.alloc (Int64.to_int bs) in - ignore (NBD.aio_pread_callback src buf !soff + ignore (NBD.aio_pread src buf !soff ~completion:(read_completed buf !soff)); soff :...

This is working for me now, although possibly only on Python 3.9. Dan suggested PyEval_InitThreads but that was deprecated in Python 3.7. Rich.

This is a combination of these two earlier series: https://www.redhat.com/archives/libguestfs/2019-August/msg00235.html https://www.redhat.com/archives/libguestfs/2019-August/msg00240.html plus changes to allow .callback = NULL / .free != NULL, and to reduce the complexity of freeing callbacks. Although it's rather long there's nothing complex here. We might consider squashing some

Patch 2 certainly allows you to set the thread model. However patch 3 shows that if you set it to nbdkit.THREAD_MODEL_PARALLEL it will crash. If you look closely at the stack trace (attached below) you can see that ignoring threads which are in parts of nbdkit unrelated to Python: Thread 4: In pread, waiting in time.sleep(). This thread has released the GIL. Thread 2: Started to

..., 116 insertions(+), 61 deletions(-) diff --git a/examples/glib-main-loop.c b/examples/glib-main-loop.c index c633c1d..2230077 100644 --- a/examples/glib-main-loop.c +++ b/examples/glib-main-loop.c @@ -188,6 +188,8 @@ finalize (GSource *sp) DEBUG (source, "finalize"); + assert (nbd_aio_in_flight (source->nbd) == 0); + assert (nbd_aio_peek_command_completed (source->nbd) == -1); nbd_close (source->nbd); } @@ -418,7 +420,7 @@ finished_read (void *vp, int64_t rcookie, int *error) /* Create a writer idle handler. */ g_idle_add (write_data, NULL); - return 0; + return 1;...