[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]

[Libguestfs] [PATCH libnbd] examples: Include an example of integrating with the glib main loop.



---
 .gitignore                |   1 +
 README                    |   2 +
 configure.ac              |   9 +
 examples/Makefile.am      |  22 ++
 examples/glib-main-loop.c | 501 ++++++++++++++++++++++++++++++++++++++
 5 files changed, 535 insertions(+)

diff --git a/.gitignore b/.gitignore
index edbf941..79b95b5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -41,6 +41,7 @@ Makefile.in
 /docs/libnbd-api.3
 /docs/libnbd-api.pod
 /examples/batched-read-write
+/examples/glib-main-loop
 /examples/threaded-reads-and-writes
 /examples/simple-fetch-first-sector
 /examples/simple-reads-and-writes
diff --git a/README b/README
index 3045312..443bb99 100644
--- a/README
+++ b/README
@@ -72,6 +72,8 @@ Optional:
 
  * qemu, qemu-io, qemu-img for interoperability testing.
 
+ * glib2 for examples that interoperate with the glib main loop.
+
  * OCaml and ocamlfind are both needed to generate the OCaml bindings.
 
  * Perl Pod::Man and Pod::Simple to generate the documentation.
diff --git a/configure.ac b/configure.ac
index 9c706c9..ccb19b5 100644
--- a/configure.ac
+++ b/configure.ac
@@ -153,6 +153,15 @@ AM_CONDITIONAL([HAVE_NBD_SERVER], [test "x$NBD_SERVER" != "x"])
 AC_CHECK_PROG([QEMU_NBD], [qemu-nbd], [qemu-nbd])
 AM_CONDITIONAL([HAVE_QEMU_NBD], [test "x$QEMU_NBD" != "x"])
 
+dnl glib2 main loop for examples that interoperate with the glib main loop.
+PKG_CHECK_MODULES([GLIB], [glib-2.0], [
+    AC_SUBST([GLIB_CFLAGS])
+    AC_SUBST([GLIB_LIBS])
+],[
+    AC_MSG_WARN([glib2 not found, some examples will not be compiled])
+])
+AM_CONDITIONAL([HAVE_GLIB], [test "x$GLIB_LIBS" != "x"])
+
 dnl Check we have enough to run podwrapper.
 AC_CHECK_PROG([PERL],[perl],[perl],[no])
 AS_IF([test "x$PERL" != "xno"],[
diff --git a/examples/Makefile.am b/examples/Makefile.am
index 7560855..de3f090 100644
--- a/examples/Makefile.am
+++ b/examples/Makefile.am
@@ -27,6 +27,11 @@ noinst_PROGRAMS = \
 	strict-structured-reads \
 	$(NULL)
 
+if HAVE_GLIB
+noinst_PROGRAMS += \
+	glib-main-loop
+endif
+
 simple_fetch_first_sector_SOURCES = \
 	simple-fetch-first-sector.c \
 	$(NULL)
@@ -93,3 +98,20 @@ batched_read_write_CFLAGS = \
 batched_read_write_LDADD = \
 	$(top_builddir)/lib/libnbd.la \
 	$(NULL)
+
+if HAVE_GLIB
+glib_main_loop_SOURCES = \
+	glib-main-loop.c \
+	$(NULL)
+glib_main_loop_CPPFLAGS = \
+	-I$(top_srcdir)/include \
+	$(NULL)
+glib_main_loop_CFLAGS = \
+	$(WARNINGS_CFLAGS) \
+	$(GLIB_CFLAGS) \
+	$(NULL)
+glib_main_loop_LDADD = \
+	$(top_builddir)/lib/libnbd.la \
+	$(GLIB_LIBS) \
+	$(NULL)
+endif
diff --git a/examples/glib-main-loop.c b/examples/glib-main-loop.c
new file mode 100644
index 0000000..dea4666
--- /dev/null
+++ b/examples/glib-main-loop.c
@@ -0,0 +1,501 @@
+/* This example shows you how to make libnbd interoperate with the
+ * glib main loop.  For more information about glib main loop see:
+ *
+ * https://developer.gnome.org/glib/stable/glib-The-Main-Event-Loop.html
+ *
+ * To run it, simply do:
+ *
+ *   ./examples/glib-main-loop
+ *
+ * For debugging, do:
+ *
+ *   LIBNBD_DEBUG=1 ./examples/glib-main-loop
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <assert.h>
+
+#include <libnbd.h>
+
+#include <glib.h>
+
+struct NBDSource;
+typedef void (*connecting_callback_t) (struct NBDSource *);
+typedef void (*connected_callback_t) (struct NBDSource *);
+
+/* This is the derived GSource type. */
+struct NBDSource {
+  /* The base type.  This MUST be the first element in this struct. */
+  GSource source;
+
+  /* The underlying libnbd handle. */
+  struct nbd_handle *nbd;
+  bool debug;                   /* true if handle has debug set */
+
+  /* The poll file descriptor, only valid when poll_registered == true. */
+  bool poll_registered;
+  GPollFD pollfd;
+
+  /* You can optionally register callbacks to be called when the
+   * handle changes state:
+   *
+   * connecting_callback is called once when the handle moves from
+   * created to connecting state.
+   *
+   * connected_callback is called once when the handle moves from
+   * connecting to connected (ready) state.
+   */
+  connecting_callback_t connecting_callback;
+  connected_callback_t connected_callback;
+  bool called_connected_callback;
+
+  /* Arbitrary pointer for use by caller. */
+  gpointer user_data;
+};
+
+/* Print debug statements when debugging is set for the handle. */
+#define DEBUG(source, fs, ...)                                          \
+  do {                                                                  \
+    if ((source)->debug)                                                \
+      fprintf (stderr, "glib: debug: " fs "\n", ## __VA_ARGS__);        \
+  } while (0)
+
+/* These are the GSource functions for libnbd handles. */
+static inline int
+events_from_nbd (struct nbd_handle *nbd)
+{
+  int dir = nbd_aio_get_direction (nbd);
+  int r = 0;
+
+  if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0)
+    r |= G_IO_IN;
+  if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0)
+    r |= G_IO_OUT;
+  return r;
+}
+
+static gboolean
+prepare (GSource *sp, gint *timeout_)
+{
+  struct NBDSource *source = (struct NBDSource *) sp;
+
+  /* When the NBD handle moves out of the created state (which means
+   * that it first has a socket associated with it) we must initialize
+   * and register the pollfd.
+   */
+  if (!source->poll_registered && !nbd_aio_is_created (source->nbd)) {
+    int fd;
+
+    if (source->connecting_callback) {
+      DEBUG (source, "calling connecting_callback");
+      source->connecting_callback (source);
+    }
+
+    fd = nbd_aio_get_fd (source->nbd);
+    assert (fd >= 0);
+
+    source->pollfd.fd = fd;
+    source->pollfd.events = events_from_nbd (source->nbd);
+    g_source_add_poll ((GSource *)source, &source->pollfd);
+
+    DEBUG (source, "registered pollfd");
+
+    source->poll_registered = true;
+  }
+
+  if (!source->poll_registered)
+    return FALSE;
+
+  if (source->connected_callback &&
+      !source->called_connected_callback &&
+      nbd_aio_is_ready (source->nbd)) {
+    DEBUG (source, "calling connected_callback");
+    source->connected_callback (source);
+    source->called_connected_callback = true;
+  }
+
+  source->pollfd.events = events_from_nbd (source->nbd);
+  *timeout_ = -1;
+
+  DEBUG (source, "prepare: events = 0x%x%s%s",
+         source->pollfd.events,
+         source->pollfd.events & G_IO_IN ? " G_IO_IN" : "",
+         source->pollfd.events & G_IO_OUT ? " G_IO_OUT" : "");
+
+  return FALSE;
+}
+
+static gboolean
+check (GSource *sp)
+{
+  struct NBDSource *source = (struct NBDSource *) sp;
+  int dir;
+
+  if (!source->poll_registered)
+    return FALSE;
+
+  dir = nbd_aio_get_direction (source->nbd);
+
+  DEBUG (source, "check: direction = 0x%x%s%s, revents = 0x%x%s%s",
+         dir,
+         dir & LIBNBD_AIO_DIRECTION_READ ? " READ" : "",
+         dir & LIBNBD_AIO_DIRECTION_WRITE ? " WRITE" : "",
+         source->pollfd.revents,
+         source->pollfd.revents & G_IO_IN ? " G_IO_IN" : "",
+         source->pollfd.revents & G_IO_OUT ? " G_IO_OUT" : "");
+
+  if ((source->pollfd.revents & G_IO_IN) != 0 &&
+      (dir & LIBNBD_AIO_DIRECTION_READ) != 0)
+    return TRUE;
+  if ((source->pollfd.revents & G_IO_OUT) != 0 &&
+      (dir & LIBNBD_AIO_DIRECTION_WRITE) != 0)
+    return TRUE;
+
+  return FALSE;
+}
+
+static gboolean
+dispatch (GSource *sp,
+          GSourceFunc callback,
+          gpointer user_data)
+{
+  struct NBDSource *source = (struct NBDSource *) sp;
+
+  DEBUG (source, "dispatch: revents = 0x%x%s%s",
+         source->pollfd.revents,
+         source->pollfd.revents & G_IO_IN ? " G_IO_IN" : "",
+         source->pollfd.revents & G_IO_OUT ? " G_IO_OUT" : "");
+
+  if ((source->pollfd.revents & G_IO_IN) != 0)
+    nbd_aio_notify_read (source->nbd);
+  else if ((source->pollfd.revents & G_IO_OUT) != 0)
+    nbd_aio_notify_write (source->nbd);
+
+  return TRUE;
+}
+
+static void
+finalize (GSource *sp)
+{
+  struct NBDSource *source = (struct NBDSource *) sp;
+
+  DEBUG (source, "finalize");
+
+  nbd_close (source->nbd);
+}
+
+GSourceFuncs nbd_source_funcs = {
+  .prepare = prepare,
+  .check = check,
+  .dispatch = dispatch,
+  .finalize = finalize,
+};
+
+/* Create a libnbd GSource from a libnbd handle.
+ *
+ * Note that the return value is also a ‘GSource *’, you just have to
+ * cast the return value if you need a GSource pointer.
+ */
+static struct NBDSource *
+create_libnbd_gsource (struct nbd_handle *nbd)
+{
+  struct NBDSource *source;
+
+  source =
+    (struct NBDSource *) g_source_new (&nbd_source_funcs, sizeof *source);
+  source->nbd = nbd;
+  source->debug = nbd_get_debug (nbd);
+  source->poll_registered = false;
+
+  return source;
+}
+
+/*----------------------------------------------------------------------*/
+
+/* The rest of this file is an example showing how to use the GSource
+ * defined above to control two nbdkit subprocesses, copying from one
+ * to the other in parallel.
+ */
+
+/* Source and destination nbdkit instances. */
+static struct NBDSource *gssrc, *gsdest;
+
+#define SIZE (1024*1024*1024)
+
+static const char *src_args[] = {
+  "nbdkit", "-s", "-r", "pattern", "size=1G", NULL
+};
+
+static const char *dest_args[] = {
+  "nbdkit", "-s", "memory", "size=1G", NULL
+};
+
+/* The list of buffers waiting to be written.  Note that the source
+ * server can answer requests out of order so these buffers may not be
+ * sorted by offset.
+ */
+#define MAX_BUFFERS 16
+#define BUFFER_SIZE 65536
+
+enum buffer_state {
+  BUFFER_READING,
+  BUFFER_READ_COMPLETED,
+  BUFFER_WRITING,
+};
+
+struct buffer {
+  uint64_t offset;
+  int64_t cookie;
+  enum buffer_state state;
+  char *data;
+};
+
+static struct buffer buffers[MAX_BUFFERS];
+static size_t nr_buffers;
+
+static bool finished, reader_paused;
+
+static GMainLoop *loop;
+
+static void connected (struct NBDSource *source);
+static gboolean read_data (gpointer user_data);
+static int finished_read (void *vp, int64_t cookie, int *error);
+static gboolean write_data (gpointer user_data);
+static int finished_write (void *vp, int64_t cookie, int *error);
+
+int
+main (int argc, char *argv[])
+{
+  struct nbd_handle *src, *dest;
+  GMainContext *loopctx = NULL;
+
+  /* Create the main loop. */
+  loop = g_main_loop_new (loopctx, FALSE);
+
+  /* Create the two NBD handles and nbdkit instances. */
+  src = nbd_create ();
+  if (!src) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+  dest = nbd_create ();
+  if (!dest) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+
+  /* Create the GSource main loop sources from each handle. */
+  gssrc = create_libnbd_gsource (src);
+  gsdest = create_libnbd_gsource (dest);
+  loopctx = g_main_loop_get_context (loop);
+  g_source_attach ((GSource *) gssrc, loopctx);
+  g_source_attach ((GSource *) gsdest, loopctx);
+
+  /* Make sure we get called back when each handle connects. */
+  gssrc->connected_callback = connected;
+  gsdest->connected_callback = connected;
+
+  /* Asynchronously start each handle connecting. */
+  if (nbd_aio_connect_command (src, (char **) src_args) == -1) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+  if (nbd_aio_connect_command (dest, (char **) dest_args) == -1) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+
+  /* Run the main loop until quit. */
+  g_main_loop_run (loop);
+  exit (EXIT_SUCCESS);
+}
+
+/* This is called back when either handle becomes connected.  By
+ * counting the number of times this happens (there are two handles)
+ * we can tell when both handles have finished connecting.
+ */
+static void
+connected (struct NBDSource *source)
+{
+  static int count = 0;
+
+  count++;
+  if (count == 2) {
+    DEBUG (source, "both handles are connected");
+
+    /* Now that both handles are connected, we can begin copying.
+     * Register an idle handler that will repeatedly read from the
+     * source.
+     */
+    g_idle_add (read_data, NULL);
+  }
+}
+
+/* This idle callback reads data from the source nbdkit until the ring
+ * is full.
+ */
+static gboolean
+read_data (gpointer user_data)
+{
+  static uint64_t posn = 0;
+  const size_t i = nr_buffers;
+
+  if (gssrc == NULL)
+    return FALSE;
+
+  /* Finished reading from the source nbdkit? */
+  if (posn >= SIZE) {
+    DEBUG (gssrc, "read_data: finished reading from source");
+    finished = true;
+    return FALSE;
+  }
+
+  /* If too many read requests are in flight, return FALSE so this
+   * idle callback is unregistered.  It will be registered by the
+   * write callback when nr_buffers decreases.
+   */
+  if (nr_buffers >= MAX_BUFFERS) {
+    DEBUG (gssrc, "read_data: buffer full, pausing reads from source");
+    reader_paused = true;
+    return FALSE;
+  }
+
+  /* Begin reading into the new buffer. */
+  assert (buffers[i].data == NULL);
+  buffers[i].data = g_new (char, BUFFER_SIZE);
+  buffers[i].state = BUFFER_READING;
+  buffers[i].offset = posn;
+  nr_buffers++;
+  posn += BUFFER_SIZE;
+
+  buffers[i].cookie =
+    nbd_aio_pread_callback (gssrc->nbd, buffers[i].data,
+                            BUFFER_SIZE, buffers[i].offset,
+                            NULL, finished_read, 0);
+  if (buffers[i].cookie == -1) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+
+  return TRUE;
+}
+
+/* This callback is called from libnbd when any read command finishes. */
+static int
+finished_read (void *vp, int64_t cookie, int *error)
+{
+  size_t i;
+
+  if (gssrc == NULL)
+    return 0;
+
+  DEBUG (gssrc, "finished_read: read completed");
+
+  /* Find the corresponding buffer and mark it as completed. */
+  for (i = 0; i < nr_buffers; ++i) {
+    if (buffers[i].cookie == cookie)
+      goto found;
+  }
+  /* This should never happen. */
+  abort ();
+
+ found:
+  buffers[i].state = BUFFER_READ_COMPLETED;
+
+  /* Create a writer idle handler. */
+  g_idle_add (write_data, NULL);
+
+  return 0;
+}
+
+/* This idle callback schedules a write. */
+static gboolean
+write_data (gpointer user_data)
+{
+  size_t i;
+
+  if (gsdest == NULL)
+    return FALSE;
+
+  /* Find the first read-completed buffer and schedule it to be
+   * written.
+   */
+  for (i = 0; i < nr_buffers; ++i) {
+    if (buffers[i].state == BUFFER_READ_COMPLETED)
+      goto found;
+  }
+  goto out;
+
+ found:
+  buffers[i].cookie =
+    nbd_aio_pwrite_callback (gsdest->nbd, buffers[i].data,
+                             BUFFER_SIZE, buffers[i].offset,
+                             NULL, finished_write, 0);
+  if (buffers[i].cookie == -1) {
+    fprintf (stderr, "%s\n", nbd_get_error ());
+    exit (EXIT_FAILURE);
+  }
+  buffers[i].state = BUFFER_WRITING;
+
+ out:
+  /* We always unregister this idle handler because the read side
+   * creates a new idle handler for every buffer that has to be
+   * written.
+   */
+  return FALSE;
+}
+
+/* This callback is called from libnbd when any write command finishes. */
+static int
+finished_write (void *vp, int64_t cookie, int *error)
+{
+  size_t i;
+
+  if (gsdest == NULL)
+    return 0;
+
+  DEBUG (gsdest, "finished_write: write completed");
+
+  /* Find the corresponding buffer and free it. */
+  for (i = 0; i < nr_buffers; ++i) {
+    if (buffers[i].cookie == cookie)
+      goto found;
+  }
+  /* This should never happen. */
+  abort ();
+
+ found:
+  g_free (buffers[i].data);
+  memmove (&buffers[i], &buffers[i+1],
+           sizeof (struct buffer) * (nr_buffers-(i+1)));
+  nr_buffers--;
+  buffers[nr_buffers].data = NULL;
+
+  /* If the number of buffers was MAX_BUFFERS and has now gone down to
+   * MAX_BUFFERS-1 then we need to restart the read handler.
+   */
+  if (nr_buffers == MAX_BUFFERS-1 && reader_paused) {
+    DEBUG (gsdest, "finished_write: restarting reader");
+    g_idle_add (read_data, NULL);
+    reader_paused = false;
+  }
+
+  /* If the reader has finished and there are no more buffers then we
+   * have done.
+   */
+  if (finished && nr_buffers == 0) {
+    DEBUG (gsdest, "finished_write: all finished");
+    g_source_remove (g_source_get_id ((GSource *) gssrc));
+    g_source_unref ((GSource *) gssrc);
+    gssrc = NULL;
+    g_source_remove (g_source_get_id ((GSource *) gsdest));
+    g_source_unref ((GSource *) gsdest);
+    gsdest = NULL;
+    g_main_loop_quit (loop);
+  }
+
+  return 0;
+}
-- 
2.22.0


[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]