From: Daniel Elstner <redacted>
Date: Thu, 10 Sep 2015 07:46:45 +0000 (+0200)
Subject: session: Port to GLib main loop
X-Git-Tag: libsigrok-0.4.0~312
X-Git-Url: https://sigrok.org/gitaction?a=commitdiff_plain;h=c2bf5506ee7864975917a7b6e7e93b78226887ce;p=libsigrok.git

session: Port to GLib main loop

Replace the custom session main loop with the GLib main loop.
This is phase one of the port, which leaves the session and
driver APIs unchanged while replacing the internals.
---

diff --git a/src/libsigrok-internal.h b/src/libsigrok-internal.h
index 3bae560d..fc436014 100644
--- a/src/libsigrok-internal.h
+++ b/src/libsigrok-internal.h
@@ -205,7 +205,6 @@ struct sr_context {
 	struct sr_dev_driver **driver_list;
 #ifdef HAVE_LIBUSB_1_0
 	libusb_context *libusb_ctx;
-	gboolean usb_source_present;
 #endif
 };
 
@@ -710,42 +709,30 @@ struct sr_session {
 	GSList *transforms;
 	struct sr_trigger *trigger;
 
-	gboolean running;
+	/** Mutex protecting the main context pointer and ownership flag. */
+	GMutex main_mutex;
+	/** Context of the session main loop. */
+	GMainContext *main_context;
+	/** Whether we are using the thread's default context. */
+	gboolean main_context_is_default;
 
-	/*
-	 * Event sources and poll FDs are stored in the same order in the
-	 * the sources and pollfds arrays. However, each source may cover
-	 * any number of associated poll FDs, so the indices do not match.
-	 *
-	 * We cannot embed the GPollFD into the source struct since we want
-	 * to be able to pass the array of all poll descriptors to g_poll().
-	 */
-	GArray *sources;
-	GArray *pollfds;
+	/** Whether the session has been started. */
+	gboolean running;
 
-	/*
-	 * These are our synchronization primitives for stopping the session in
-	 * an async fashion. We need to make sure the session is stopped from
-	 * within the session thread itself.
-	 */
-	/** Mutex protecting access to abort_session. */
-	GMutex stop_mutex;
-	/** Abort current session. See sr_session_stop(). */
-	gboolean abort_session;
+	/** Registered event sources for this session. */
+	GHashTable *event_sources;
+	/** Session main loop. */
+	GMainLoop *main_loop;
 };
 
 SR_PRIV int sr_session_source_add_internal(struct sr_session *session,
-		int timeout, sr_receive_data_callback cb, void *cb_data,
-		gintptr poll_object);
-SR_PRIV int sr_session_source_poll_add(struct sr_session *session,
-		gintptr poll_object, gintptr fd, int events);
+		void *key, GSource *source);
 SR_PRIV int sr_session_source_remove_internal(struct sr_session *session,
-		gintptr poll_object);
-SR_PRIV int sr_session_source_poll_remove(struct sr_session *session,
-		gintptr poll_object, gintptr fd);
+		void *key);
+SR_PRIV int sr_session_source_destroyed(struct sr_session *session,
+		void *key, GSource *source);
 SR_PRIV int sr_session_send(const struct sr_dev_inst *sdi,
 		const struct sr_datafeed_packet *packet);
-SR_PRIV int sr_session_stop_sync(struct sr_session *session);
 SR_PRIV int sr_sessionfile_check(const char *filename);
 SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet,
 		struct sr_datafeed_packet **copy);
diff --git a/src/session.c b/src/session.c
index 58be4446..cc5f09dc 100644
--- a/src/session.c
+++ b/src/session.c
@@ -2,6 +2,7 @@
  * This file is part of the libsigrok project.
  *
  * Copyright (C) 2010-2012 Bert Vermeulen <bert@biot.com>
+ * Copyright (C) 2015 Daniel Elstner <daniel.kitta@gmail.com>
  *
  * 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
@@ -44,32 +45,163 @@
  * @{
  */
 
-struct source {
-	int64_t timeout;	/* microseconds */
-	int64_t due;		/* microseconds */
-
-	sr_receive_data_callback cb;
+struct datafeed_callback {
+	sr_datafeed_callback cb;
 	void *cb_data;
+};
 
-	/* This is used to keep track of the object (fd, pollfd or channel) which is
-	 * being polled and will be used to match the source when removing it again.
-	 */
-	gintptr poll_object;
+/** Custom GLib event source for generic descriptor I/O.
+ * @internal
+ */
+struct fd_source {
+	GSource base;
 
-	/* Number of fds to poll for this source. This will be 0 for timer
-	 * sources, 1 for normal I/O sources, and 1 or more for libusb I/O
-	 * sources on Unix platforms.
-	 */
-	int num_fds;
+	int64_t timeout_us;
+	int64_t due_us;
 
-	gboolean triggered;
-};
+	/* Meta-data needed to keep track of installed sources */
+	struct sr_session *session;
+	void *key;
 
-struct datafeed_callback {
-	sr_datafeed_callback cb;
-	void *cb_data;
+	GPollFD pollfd;
 };
 
+/** FD event source prepare() method.
+ */
+static gboolean fd_source_prepare(GSource *source, int *timeout)
+{
+	int64_t now_us;
+	struct fd_source *fsource;
+	int remaining_ms;
+
+	fsource = (struct fd_source *)source;
+
+	if (fsource->timeout_us >= 0) {
+		now_us = g_source_get_time(source);
+
+		if (fsource->due_us == 0) {
+			/* First-time initialization of the expiration time */
+			fsource->due_us = now_us + fsource->timeout_us;
+		}
+		remaining_ms = (MAX(0, fsource->due_us - now_us) + 999) / 1000;
+	} else {
+		remaining_ms = -1;
+	}
+	*timeout = remaining_ms;
+
+	return (remaining_ms == 0);
+}
+
+/** FD event source check() method.
+ */
+static gboolean fd_source_check(GSource *source)
+{
+	struct fd_source *fsource;
+	unsigned int revents;
+
+	fsource = (struct fd_source *)source;
+	revents = fsource->pollfd.revents;
+
+	return (revents != 0 || (fsource->timeout_us >= 0
+			&& fsource->due_us <= g_source_get_time(source)));
+}
+
+/** FD event source dispatch() method.
+ */
+static gboolean fd_source_dispatch(GSource *source,
+		GSourceFunc callback, void *user_data)
+{
+	struct fd_source *fsource;
+	const char *name;
+	unsigned int revents;
+	gboolean keep;
+
+	fsource = (struct fd_source *)source;
+	name = g_source_get_name(source);
+	revents = fsource->pollfd.revents;
+
+	if (revents != 0) {
+		sr_spew("%s: %s " G_POLLFD_FORMAT ", revents 0x%.2X",
+			__func__, name, fsource->pollfd.fd, revents);
+	} else {
+		sr_spew("%s: %s " G_POLLFD_FORMAT ", timed out",
+			__func__, name, fsource->pollfd.fd);
+	}
+	if (!callback) {
+		sr_err("Callback not set, cannot dispatch event.");
+		return G_SOURCE_REMOVE;
+	}
+	keep = (*(sr_receive_data_callback)callback)
+			(fsource->pollfd.fd, revents, user_data);
+
+	if (fsource->timeout_us >= 0 && G_LIKELY(keep)
+			&& G_LIKELY(!g_source_is_destroyed(source)))
+		fsource->due_us = g_source_get_time(source)
+				+ fsource->timeout_us;
+	return keep;
+}
+
+/** FD event source finalize() method.
+ */
+static void fd_source_finalize(GSource *source)
+{
+	struct fd_source *fsource;
+
+	fsource = (struct fd_source *)source;
+
+	sr_dbg("%s: key %p", __func__, fsource->key);
+
+	sr_session_source_destroyed(fsource->session, fsource->key, source);
+}
+
+/** Create an event source for I/O on a file descriptor.
+ *
+ * In order to maintain API compatibility, this event source also doubles
+ * as a timer event source.
+ *
+ * @param session The session the event source belongs to.
+ * @param key The key used to identify this source.
+ * @param fd The file descriptor or HANDLE.
+ * @param timeout_ms The timeout interval in ms, or -1 to wait indefinitely.
+ * @return A new event source object, or NULL on failure.
+ */
+static GSource *fd_source_new(struct sr_session *session, void *key,
+		gintptr fd, int events, int timeout_ms)
+{
+	static GSourceFuncs fd_source_funcs = {
+		.prepare  = &fd_source_prepare,
+		.check    = &fd_source_check,
+		.dispatch = &fd_source_dispatch,
+		.finalize = &fd_source_finalize
+	};
+	GSource *source;
+	struct fd_source *fsource;
+
+	source = g_source_new(&fd_source_funcs, sizeof(struct fd_source));
+	fsource = (struct fd_source *)source;
+
+	g_source_set_name(source, (fd < 0) ? "timer" : "fd");
+
+	if (timeout_ms >= 0) {
+		fsource->timeout_us = 1000 * (int64_t)timeout_ms;
+		fsource->due_us = 0;
+	} else {
+		fsource->timeout_us = -1;
+		fsource->due_us = INT64_MAX;
+	}
+	fsource->session = session;
+	fsource->key = key;
+
+	fsource->pollfd.fd = fd;
+	fsource->pollfd.events = events;
+	fsource->pollfd.revents = 0;
+
+	if (fd >= 0)
+		g_source_add_poll(source, &fsource->pollfd);
+
+	return source;
+}
+
 /**
  * Create a new session.
  *
@@ -95,10 +227,12 @@ SR_API int sr_session_new(struct sr_context *ctx,
 
 	session->ctx = ctx;
 
-	session->sources = g_array_new(FALSE, FALSE, sizeof(struct source));
-	session->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
+	g_mutex_init(&session->main_mutex);
 
-	g_mutex_init(&session->stop_mutex);
+	/* To maintain API compatibility, we need a lookup table
+	 * which maps poll_object IDs to GSource* pointers.
+	 */
+	session->event_sources = g_hash_table_new(NULL, NULL);
 
 	*new_session = session;
 
@@ -124,12 +258,11 @@ SR_API int sr_session_destroy(struct sr_session *session)
 	}
 
 	sr_session_dev_remove_all(session);
-	g_mutex_clear(&session->stop_mutex);
-
 	g_slist_free_full(session->owned_devs, (GDestroyNotify)sr_dev_inst_free);
 
-	g_array_unref(session->pollfds);
-	g_array_unref(session->sources);
+	g_hash_table_unref(session->event_sources);
+
+	g_mutex_clear(&session->main_mutex);
 
 	g_free(session);
 
@@ -371,192 +504,6 @@ SR_API int sr_session_trigger_set(struct sr_session *session, struct sr_trigger
 	return SR_OK;
 }
 
-static gboolean sr_session_check_aborted(struct sr_session *session)
-{
-	gboolean stop;
-
-	g_mutex_lock(&session->stop_mutex);
-	stop = session->abort_session;
-	if (stop) {
-		sr_session_stop_sync(session);
-		/* But once is enough. */
-		session->abort_session = FALSE;
-	}
-	g_mutex_unlock(&session->stop_mutex);
-
-	return stop;
-}
-
-/**
- * Poll the session's event sources.
- *
- * @param session The session to use. Must not be NULL.
- * @retval SR_OK Success.
- * @retval SR_ERR Error occurred.
- */
-static int sr_session_iteration(struct sr_session *session)
-{
-	int64_t start_time, stop_time, min_due, due;
-	int timeout_ms;
-	unsigned int i;
-	int k, fd_index;
-	int ret;
-	int fd;
-	int revents;
-	gboolean triggered, stopped;
-	struct source *source;
-	GPollFD *pollfd;
-	gintptr poll_object;
-#if HAVE_LIBUSB_1_0
-	int64_t usb_timeout;
-	int64_t usb_due;
-	struct timeval tv;
-#endif
-	if (session->sources->len == 0) {
-		sr_session_check_aborted(session);
-		return SR_OK;
-	}
-	start_time = g_get_monotonic_time();
-	min_due = INT64_MAX;
-
-	for (i = 0; i < session->sources->len; ++i) {
-		source = &g_array_index(session->sources, struct source, i);
-		if (source->due < min_due)
-			min_due = source->due;
-		source->triggered = FALSE;
-	}
-#if HAVE_LIBUSB_1_0
-	usb_due = INT64_MAX;
-	if (session->ctx->usb_source_present) {
-		ret = libusb_get_next_timeout(session->ctx->libusb_ctx, &tv);
-		if (ret < 0) {
-			sr_err("Error getting libusb timeout: %s",
-				libusb_error_name(ret));
-			return SR_ERR;
-		} else if (ret == 1) {
-			usb_timeout = (int64_t)tv.tv_sec * G_USEC_PER_SEC
-					+ tv.tv_usec;
-			usb_due = start_time + usb_timeout;
-			if (usb_due < min_due)
-				min_due = usb_due;
-
-			sr_spew("poll: next USB timeout %g ms",
-				1e-3 * usb_timeout);
-		}
-	}
-#endif
-	if (min_due == INT64_MAX)
-		timeout_ms = -1;
-	else if (min_due > start_time)
-		timeout_ms = MIN((min_due - start_time + 999) / 1000, INT_MAX);
-	else
-		timeout_ms = 0;
-
-	sr_spew("poll enter: %u sources, %u fds, %d ms timeout",
-		session->sources->len, session->pollfds->len, timeout_ms);
-
-	ret = g_poll((GPollFD *)session->pollfds->data,
-			session->pollfds->len, timeout_ms);
-#ifdef G_OS_UNIX
-	if (ret < 0 && errno != EINTR) {
-		sr_err("Error in poll: %s", g_strerror(errno));
-		return SR_ERR;
-	}
-#else
-	if (ret < 0) {
-		sr_err("Error in poll: %d", ret);
-		return SR_ERR;
-	}
-#endif
-	stop_time = g_get_monotonic_time();
-
-	sr_spew("poll leave: %g ms elapsed, %d events",
-		1e-3 * (stop_time - start_time), ret);
-
-	triggered = FALSE;
-	stopped = FALSE;
-	fd_index = 0;
-
-	for (i = 0; i < session->sources->len; ++i) {
-		source = &g_array_index(session->sources, struct source, i);
-
-		poll_object = source->poll_object;
-		fd = (int)poll_object;
-		revents = 0;
-
-		for (k = 0; k < source->num_fds; ++k) {
-			pollfd = &g_array_index(session->pollfds,
-					GPollFD, fd_index + k);
-			fd = pollfd->fd;
-			revents |= pollfd->revents;
-		}
-		fd_index += source->num_fds;
-
-		if (source->triggered)
-			continue; /* already handled */
-		if (ret > 0 && revents == 0)
-			continue; /* skip timeouts if any I/O event occurred */
-
-		/* Make invalid to avoid confusion in case of multiple FDs. */
-		if (source->num_fds > 1)
-			fd = -1;
-		if (ret <= 0)
-			revents = 0;
-
-		due = source->due;
-#if HAVE_LIBUSB_1_0
-		if (usb_due < due && poll_object
-				== (gintptr)session->ctx->libusb_ctx)
-			due = usb_due;
-#endif
-		if (revents == 0 && stop_time < due)
-			continue;
-		/*
-		 * The source may be gone after the callback returns,
-		 * so access any data now that needs accessing.
-		 */
-		if (source->timeout >= 0)
-			source->due = stop_time + source->timeout;
-		source->triggered = TRUE;
-		triggered = TRUE;
-		/*
-		 * Invoke the source's callback on an event or timeout.
-		 */
-		sr_spew("callback for event source %" G_GINTPTR_FORMAT " with"
-			" event mask 0x%.2X", poll_object, (unsigned)revents);
-		if (!source->cb(fd, revents, source->cb_data)) {
-#if HAVE_LIBUSB_1_0
-			/* Hackish, to be cleaned up when porting to
-			 * the GLib main loop.
-			 */
-			if (poll_object == (gintptr)session->ctx->libusb_ctx)
-				usb_source_remove(session, session->ctx);
-			else
-#endif
-				sr_session_source_remove_internal(session,
-						poll_object);
-		}
-		/*
-		 * We want to take as little time as possible to stop
-		 * the session if we have been told to do so. Therefore,
-		 * we check the flag after processing every source, not
-		 * just once per main event loop.
-		 */
-		if (!stopped)
-			stopped = sr_session_check_aborted(session);
-
-		/* Restart loop as the sources list may have changed. */
-		fd_index = 0;
-		i = 0;
-	}
-
-	/* Check for abort at least once per iteration. */
-	if (!triggered)
-		sr_session_check_aborted(session);
-
-	return SR_OK;
-}
-
 static int verify_trigger(struct sr_trigger *trigger)
 {
 	struct sr_trigger_stage *stage;
@@ -593,6 +540,85 @@ static int verify_trigger(struct sr_trigger *trigger)
 	return SR_OK;
 }
 
+/** Set up the main context the session will be executing in.
+ *
+ * Must be called just before the session starts, by the thread which
+ * will execute the session main loop. Once acquired, the main context
+ * pointer is immutable for the duration of the session run.
+ */
+static int set_main_context(struct sr_session *session)
+{
+	GMainContext *def_context;
+
+	/* May happen if sr_session_start() is called again after
+	 * sr_session_run(), but the session hasn't been stopped yet.
+	 */
+	if (session->main_loop) {
+		sr_err("Cannot set main context; main loop already created.");
+		return SR_ERR;
+	}
+
+	g_mutex_lock(&session->main_mutex);
+
+	def_context = g_main_context_get_thread_default();
+
+	if (!def_context)
+		def_context = g_main_context_default();
+	/*
+	 * Try to use an existing main context if possible, but only if we
+	 * can make it owned by the current thread. Otherwise, create our
+	 * own main context so that event source callbacks can execute in
+	 * the session thread.
+	 */
+	if (g_main_context_acquire(def_context)) {
+		g_main_context_release(def_context);
+
+		sr_dbg("Using thread-default main context.");
+
+		session->main_context = def_context;
+		session->main_context_is_default = TRUE;
+	} else {
+		sr_dbg("Creating our own main context.");
+
+		session->main_context = g_main_context_new();
+		session->main_context_is_default = FALSE;
+	}
+	g_mutex_unlock(&session->main_mutex);
+
+	return SR_OK;
+}
+
+/** Unset the main context used for the current session run.
+ *
+ * Must be called right after stopping the session. Note that if the
+ * session is stopped asynchronously, the main loop may still be running
+ * after the main context has been unset. This is OK as long as no new
+ * event sources are created -- the main loop holds its own reference
+ * to the main context.
+ */
+static int unset_main_context(struct sr_session *session)
+{
+	int ret;
+
+	g_mutex_lock(&session->main_mutex);
+
+	if (session->main_context) {
+		if (!session->main_context_is_default)
+			g_main_context_unref(session->main_context);
+
+		session->main_context = NULL;
+		ret = SR_OK;
+	} else {
+		/* May happen if the set/unset calls are not matched.
+		 */
+		sr_err("No main context to unset.");
+		ret = SR_ERR;
+	}
+	g_mutex_unlock(&session->main_mutex);
+
+	return ret;
+}
+
 /**
  * Start a session.
  *
@@ -624,9 +650,14 @@ SR_API int sr_session_start(struct sr_session *session)
 	if (session->trigger && verify_trigger(session->trigger) != SR_OK)
 		return SR_ERR;
 
+	ret = set_main_context(session);
+	if (ret != SR_OK)
+		return ret;
+
+	session->running = TRUE;
+
 	sr_info("Starting.");
 
-	ret = SR_OK;
 	for (l = session->devs; l; l = l->next) {
 		sdi = l->data;
 		enabled_channels = 0;
@@ -656,6 +687,10 @@ SR_API int sr_session_start(struct sr_session *session)
 		}
 	}
 
+	if (ret != SR_OK) {
+		unset_main_context(session);
+		session->running = FALSE;
+	}
 	/* TODO: What if there are multiple devices? Which return code? */
 
 	return ret;
@@ -674,70 +709,65 @@ SR_API int sr_session_start(struct sr_session *session)
  */
 SR_API int sr_session_run(struct sr_session *session)
 {
-	int ret;
-
 	if (!session) {
 		sr_err("%s: session was NULL", __func__);
 		return SR_ERR_ARG;
 	}
-
 	if (!session->devs) {
 		/* TODO: Actually the case? */
 		sr_err("%s: session->devs was NULL; a session "
 		       "cannot be run without devices.", __func__);
 		return SR_ERR_ARG;
 	}
-	session->running = TRUE;
+	if (session->main_loop) {
+		sr_err("Main loop already created.");
+		return SR_ERR;
+	}
+	if (g_hash_table_size(session->event_sources) == 0) {
+		sr_err("Refusing to run without any event sources.");
+		return SR_ERR;
+	}
 
+	g_mutex_lock(&session->main_mutex);
+	if (!session->main_context) {
+		sr_err("Cannot run without main context.");
+		g_mutex_unlock(&session->main_mutex);
+		return SR_ERR;
+	}
 	sr_info("Running.");
 
-	/* Poll event sources until none are left. */
-	while (session->sources->len > 0) {
-		ret = sr_session_iteration(session);
-		if (ret != SR_OK)
-			return ret;
-	}
+	session->main_loop = g_main_loop_new(session->main_context, FALSE);
+	g_mutex_unlock(&session->main_mutex);
+
+	g_main_loop_run(session->main_loop);
+
+	g_main_loop_unref(session->main_loop);
+	session->main_loop = NULL;
+
 	return SR_OK;
 }
 
-/**
- * Stop a session.
- *
- * The session is stopped immediately, with all acquisition sessions stopped
- * and hardware drivers cleaned up.
- *
- * This must be called from within the session thread, to prevent freeing
- * resources that the session thread will try to use.
- *
- * @param session The session to use. Must not be NULL.
- *
- * @retval SR_OK Success.
- * @retval SR_ERR_ARG Invalid session passed.
- *
- * @private
- */
-SR_PRIV int sr_session_stop_sync(struct sr_session *session)
+static gboolean session_stop_sync(void *user_data)
 {
+	struct sr_session *session;
 	struct sr_dev_inst *sdi;
-	GSList *l;
+	GSList *node;
 
-	if (!session) {
-		sr_err("%s: session was NULL", __func__);
-		return SR_ERR_ARG;
-	}
+	session = user_data;
+
+	if (!session->running)
+		return G_SOURCE_REMOVE;
 
 	sr_info("Stopping.");
 
-	for (l = session->devs; l; l = l->next) {
-		sdi = l->data;
-		if (sdi->driver) {
-			if (sdi->driver->dev_acquisition_stop)
-				sdi->driver->dev_acquisition_stop(sdi, sdi);
-		}
+	for (node = session->devs; node; node = node->next) {
+		sdi = node->data;
+		if (sdi->driver && sdi->driver->dev_acquisition_stop)
+			sdi->driver->dev_acquisition_stop(sdi, sdi);
 	}
 	session->running = FALSE;
 
-	return SR_OK;
+	return G_SOURCE_REMOVE;
 }
 
 /**
@@ -755,6 +785,7 @@ SR_PRIV int sr_session_stop_sync(struct sr_session *session)
  *
  * @retval SR_OK Success.
  * @retval SR_ERR_ARG Invalid session passed.
+ * @retval SR_ERR Other error.
  *
  * @since 0.4.0
  */
@@ -762,14 +793,19 @@ SR_API int sr_session_stop(struct sr_session *session)
 {
 	if (!session) {
 		sr_err("%s: session was NULL", __func__);
-		return SR_ERR_BUG;
+		return SR_ERR_ARG;
 	}
+	g_mutex_lock(&session->main_mutex);
 
-	g_mutex_lock(&session->stop_mutex);
-	session->abort_session = TRUE;
-	g_mutex_unlock(&session->stop_mutex);
+	if (session->main_context) {
+		g_main_context_invoke(session->main_context,
+				&session_stop_sync, session);
+	} else {
+		sr_err("No main context set; already stopped?");
+	}
+	g_mutex_unlock(&session->main_mutex);
 
-	return SR_OK;
+	return unset_main_context(session);
 }
 
 /**
@@ -910,107 +946,62 @@ SR_PRIV int sr_session_send(const struct sr_dev_inst *sdi,
  * Add an event source for a file descriptor.
  *
  * @param session The session to use. Must not be NULL.
- * @param[in] timeout Max time in ms to wait before the callback is called,
- *                    or -1 to wait indefinitely.
- * @param cb Callback function to add. Must not be NULL.
- * @param cb_data Data for the callback function. Can be NULL.
- * @param poll_object Handle by which the source is identified
- *
+ * @param key The key which identifies the event source.
+ * @param source An event source object. Must not be NULL.
  * @retval SR_OK Success.
  * @retval SR_ERR_ARG Invalid argument.
- * @retval SR_ERR An event source for @a poll_object is already installed.
+ * @retval SR_ERR_BUG Event source with @a key already installed.
+ * @retval SR_ERR Other error.
  */
 SR_PRIV int sr_session_source_add_internal(struct sr_session *session,
-		int timeout, sr_receive_data_callback cb, void *cb_data,
-		gintptr poll_object)
+		void *key, GSource *source)
 {
-	struct source src;
-	unsigned int i;
-
-	/* Note: cb_data can be NULL, that's not a bug. */
-	if (!cb) {
-		sr_err("%s: cb was NULL", __func__);
-		return SR_ERR_ARG;
-	}
-	/* Make sure that poll_object is unique.
+	int ret;
+	/*
+	 * This must not ever happen, since the source has already been
+	 * created and its finalize() method will remove the key for the
+	 * already installed source. (Well it would, if we did not have
+	 * another sanity check there.)
 	 */
-	for (i = 0; i < session->sources->len; ++i) {
-		if (g_array_index(session->sources, struct source, i)
-				.poll_object == poll_object) {
-			sr_err("Event source %" G_GINTPTR_FORMAT
-				" already installed.", poll_object);
-			return SR_ERR;
-		}
+	if (g_hash_table_contains(session->event_sources, key)) {
+		sr_err("Event source with key %p already exists.", key);
+		return SR_ERR_BUG;
 	}
-	sr_dbg("Installing event source %" G_GINTPTR_FORMAT
-		" with %d ms timeout.", poll_object, timeout);
-	src.cb = cb;
-	src.cb_data = cb_data;
-	src.poll_object = poll_object;
-	src.num_fds = 0;
-	src.triggered = FALSE;
-
-	if (timeout >= 0) {
-		src.timeout = INT64_C(1000) * timeout;
-		src.due = g_get_monotonic_time() + src.timeout;
+	g_hash_table_insert(session->event_sources, key, source);
+
+	g_mutex_lock(&session->main_mutex);
+
+	if (session->main_context) {
+		if (g_source_attach(source, session->main_context) > 0)
+			ret = SR_OK;
+		else
+			ret = SR_ERR;
 	} else {
-		src.timeout = -1;
-		src.due = INT64_MAX;
+		sr_err("Cannot add event source without main context.");
+		ret = SR_ERR;
 	}
-	g_array_append_val(session->sources, src);
+	g_mutex_unlock(&session->main_mutex);
 
-	return SR_OK;
+	return ret;
 }
 
-SR_PRIV int sr_session_source_poll_add(struct sr_session *session,
-		gintptr poll_object, gintptr fd, int events)
+static int attach_fd_source(struct sr_session *session,
+		void *key, int fd, int events, int timeout,
+		sr_receive_data_callback cb, void *cb_data)
 {
-	struct source *source;
-	GPollFD pollfd;
-	unsigned int i;
-	int fd_index, k;
-
-	source = NULL;
-	fd_index = 0;
+	GSource *source;
+	int ret;
 
-	/* Look up existing event source.
-	 */
-	for (i = 0; i < session->sources->len; ++i) {
-		source = &g_array_index(session->sources, struct source, i);
-		if (source->poll_object == poll_object)
-			break;
-		fd_index += source->num_fds;
-	}
-	if (!source) {
-		sr_err("Cannot add poll FD %" G_GINTPTR_FORMAT
-			" to non-existing event source %" G_GINTPTR_FORMAT
-			".",  fd, poll_object);
+	source = fd_source_new(session, key, fd, events, timeout);
+	if (!source)
 		return SR_ERR;
-	}
-	/* Make sure the FD is unique.
-	 */
-	for (k = 0; k < source->num_fds; ++k)
-		if (g_array_index(session->pollfds, GPollFD, fd_index + k)
-				.fd == fd) {
-			sr_err("Cannot add poll FD %" G_GINTPTR_FORMAT
-				" twice to event source %" G_GINTPTR_FORMAT
-				".", fd, poll_object);
-			return SR_ERR;
-		}
-
-	pollfd.fd = fd;
-	pollfd.events = events;
-	pollfd.revents = 0;
 
-	g_array_insert_val(session->pollfds,
-		fd_index + source->num_fds, pollfd);
-	++source->num_fds;
+	g_source_set_callback(source, (GSourceFunc)cb, cb_data, NULL);
 
-	sr_dbg("Added poll FD %" G_GINTPTR_FORMAT " with event mask 0x%.2X"
-		" to event source %" G_GINTPTR_FORMAT ".",
-		fd, (unsigned)events, poll_object);
+	ret = sr_session_source_add_internal(session, key, source);
+	g_source_unref(source);
 
-	return SR_OK;
+	return ret;
 }
 
 /**
@@ -1032,17 +1023,12 @@ SR_PRIV int sr_session_source_poll_add(struct sr_session *session,
 SR_API int sr_session_source_add(struct sr_session *session, int fd,
 		int events, int timeout, sr_receive_data_callback cb, void *cb_data)
 {
-	int ret;
-
 	if (fd < 0 && timeout < 0) {
-		sr_err("Timer source without timeout would block indefinitely");
+		sr_err("Cannot create timer source without timeout.");
 		return SR_ERR_ARG;
 	}
-	ret = sr_session_source_add_internal(session, timeout, cb, cb_data, fd);
-	if (ret != SR_OK || fd < 0)
-		return ret;
-
-	return sr_session_source_poll_add(session, fd, fd, events);
+	return attach_fd_source(session, GINT_TO_POINTER(fd),
+			fd, events, timeout, cb, cb_data);
 }
 
 /**
@@ -1064,19 +1050,12 @@ SR_API int sr_session_source_add_pollfd(struct sr_session *session,
 		GPollFD *pollfd, int timeout, sr_receive_data_callback cb,
 		void *cb_data)
 {
-	int ret;
-
 	if (!pollfd) {
 		sr_err("%s: pollfd was NULL", __func__);
 		return SR_ERR_ARG;
 	}
-	ret = sr_session_source_add_internal(session,
-			timeout, cb, cb_data, (gintptr)pollfd);
-	if (ret != SR_OK)
-		return ret;
-
-	return sr_session_source_poll_add(session,
-			(gintptr)pollfd, pollfd->fd, pollfd->events);
+	return attach_fd_source(session, pollfd, pollfd->fd,
+			pollfd->events, timeout, cb, cb_data);
 }
 
 /**
@@ -1099,108 +1078,51 @@ SR_API int sr_session_source_add_channel(struct sr_session *session,
 		GIOChannel *channel, int events, int timeout,
 		sr_receive_data_callback cb, void *cb_data)
 {
-	int ret;
+	GPollFD pollfd;
 
-	ret = sr_session_source_add_internal(session,
-			timeout, cb, cb_data, (gintptr)channel);
-	if (ret != SR_OK)
-		return ret;
+	if (!channel) {
+		sr_err("%s: channel was NULL", __func__);
+		return SR_ERR_ARG;
+	}
+	/* We should be using g_io_create_watch(), but can't without
+	 * changing the driver API, as the callback signature is different.
+	 */
 #ifdef G_OS_WIN32
-	GPollFD p;
-	g_io_channel_win32_make_pollfd(channel, events, &p);
-
-	return sr_session_source_poll_add(session,
-			(gintptr)channel, p.fd, p.events);
+	g_io_channel_win32_make_pollfd(channel, events, &pollfd);
 #else
-	return sr_session_source_poll_add(session, (gintptr)channel,
-			g_io_channel_unix_get_fd(channel), events);
+	pollfd.fd = g_io_channel_unix_get_fd(channel);
+	pollfd.events = events;
 #endif
+	return attach_fd_source(session, channel, pollfd.fd,
+			pollfd.events, timeout, cb, cb_data);
 }
 
 /**
  * Remove the source identified by the specified poll object.
  *
  * @param session The session to use. Must not be NULL.
- * @param poll_object The channel for which the source should be removed.
+ * @param key The key by which the source is identified.
  *
  * @retval SR_OK Success
  * @retval SR_ERR_BUG No event source for poll_object found.
  */
 SR_PRIV int sr_session_source_remove_internal(struct sr_session *session,
-		gintptr poll_object)
+		void *key)
 {
-	struct source *source;
-	unsigned int i;
-	int fd_index = 0;
+	GSource *source;
 
-	for (i = 0; i < session->sources->len; ++i) {
-		source = &g_array_index(session->sources, struct source, i);
-
-		if (source->poll_object == poll_object) {
-			if (source->num_fds > 0)
-				g_array_remove_range(session->pollfds,
-						fd_index, source->num_fds);
-			g_array_remove_index(session->sources, i);
-
-			sr_dbg("Removed event source %" G_GINTPTR_FORMAT ".",
-				poll_object);
-			return SR_OK;
-		}
-		fd_index += source->num_fds;
-	}
-	/* Trying to remove an already removed event source is problematic
+	source = g_hash_table_lookup(session->event_sources, key);
+	/*
+	 * Trying to remove an already removed event source is problematic
 	 * since the poll_object handle may have been reused in the meantime.
 	 */
-	sr_warn("Cannot remove non-existing event source %"
-		G_GINTPTR_FORMAT ".", poll_object);
-
-	return SR_ERR_BUG;
-}
-
-SR_PRIV int sr_session_source_poll_remove(struct sr_session *session,
-		gintptr poll_object, gintptr fd)
-{
-	struct source *source;
-	unsigned int i;
-	int fd_index, k;
-
-	source = NULL;
-	fd_index = 0;
-
-	/* Look up existing event source.
-	 */
-	for (i = 0; i < session->sources->len; ++i) {
-		source = &g_array_index(session->sources, struct source, i);
-		if (source->poll_object == poll_object)
-			break;
-		fd_index += source->num_fds;
-	}
 	if (!source) {
-		sr_err("Cannot remove poll FD %" G_GINTPTR_FORMAT
-			" from non-existing event source %" G_GINTPTR_FORMAT
-			".", fd, poll_object);
-		return SR_ERR;
+		sr_warn("Cannot remove non-existing event source %p.", key);
+		return SR_ERR_BUG;
 	}
-	/* Look up the FD in the poll set.
-	 */
-	for (k = 0; k < source->num_fds; ++k)
-		if (g_array_index(session->pollfds, GPollFD, fd_index + k)
-				.fd == fd) {
+	g_source_destroy(source);
 
-			g_array_remove_index(session->pollfds, fd_index + k);
-			--source->num_fds;
-
-			sr_dbg("Removed poll FD %" G_GINTPTR_FORMAT
-				" from event source %" G_GINTPTR_FORMAT ".",
-				fd, poll_object);
-			return SR_OK;
-		}
-
-	sr_err("Cannot remove non-existing poll FD %" G_GINTPTR_FORMAT
-		" from event source %" G_GINTPTR_FORMAT ".",
-		fd, poll_object);
-
-	return SR_ERR;
+	return SR_OK;
 }
 
 /**
@@ -1217,7 +1139,7 @@ SR_PRIV int sr_session_source_poll_remove(struct sr_session *session,
  */
 SR_API int sr_session_source_remove(struct sr_session *session, int fd)
 {
-	return sr_session_source_remove_internal(session, fd);
+	return sr_session_source_remove_internal(session, GINT_TO_POINTER(fd));
 }
 
 /**
@@ -1239,7 +1161,7 @@ SR_API int sr_session_source_remove_pollfd(struct sr_session *session,
 		sr_err("%s: pollfd was NULL", __func__);
 		return SR_ERR_ARG;
 	}
-	return sr_session_source_remove_internal(session, (gintptr)pollfd);
+	return sr_session_source_remove_internal(session, pollfd);
 }
 
 /**
@@ -1261,7 +1183,52 @@ SR_API int sr_session_source_remove_channel(struct sr_session *session,
 		sr_err("%s: channel was NULL", __func__);
 		return SR_ERR_ARG;
 	}
-	return sr_session_source_remove_internal(session, (gintptr)channel);
+	return sr_session_source_remove_internal(session, channel);
+}
+
+/** Unregister an event source that has been destroyed.
+ *
+ * This is intended to be called from a source's finalize() method.
+ *
+ * @param session The session to use. Must not be NULL.
+ * @param key The key used to identify @a source.
+ * @param source The source object that was destroyed.
+ *
+ * @retval SR_OK Success.
+ * @retval SR_ERR_BUG Event source for @a key does not match @a source.
+ */
+SR_PRIV int sr_session_source_destroyed(struct sr_session *session,
+		void *key, GSource *source)
+{
+	GSource *registered_source;
+
+	registered_source = g_hash_table_lookup(session->event_sources, key);
+	/*
+	 * Trying to remove an already removed event source is problematic
+	 * since the poll_object handle may have been reused in the meantime.
+	 */
+	if (!registered_source) {
+		sr_err("No event source for key %p found.", key);
+		return SR_ERR_BUG;
+	}
+	if (registered_source != source) {
+		sr_err("Event source for key %p does not match"
+			" destroyed source.", key);
+		return SR_ERR_BUG;
+	}
+	g_hash_table_remove(session->event_sources, key);
+	/*
+	 * Quit the main loop if we just removed the last event source.
+	 * TODO: This may need an idle callback depending on when event
+	 * sources are finalized. (The issue is remove followed by add
+	 * within the same main loop iteration.)
+	 */
+	if (session->main_loop
+			&& g_hash_table_size(session->event_sources) == 0) {
+		sr_dbg("Stopping main loop...");
+		g_main_loop_quit(session->main_loop);
+	}
+	return SR_OK;
 }
 
 static void copy_src(struct sr_config *src, struct sr_datafeed_meta *meta_copy)
diff --git a/src/usb.c b/src/usb.c
index 1b22de41..d66f1880 100644
--- a/src/usb.c
+++ b/src/usb.c
@@ -3,10 +3,11 @@
  *
  * Copyright (C) 2012 Uwe Hermann <uwe@hermann-uwe.de>
  * Copyright (C) 2012 Bert Vermeulen <bert@biot.com>
+ * Copyright (C) 2015 Daniel Elstner <daniel.kitta@gmail.com>
  *
- * This program is free software; you can redistribute it and/or modify
+ * 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
+ * the Free Software Foundation, either version 3 of the License, or
  * (at your option) any later version.
  *
  * This program is distributed in the hope that it will be useful,
@@ -15,8 +16,7 @@
  * 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 St, Fifth Floor, Boston, MA  02110-1301 USA
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
 #include <stdlib.h>
@@ -36,6 +36,264 @@
 typedef int libusb_os_handle;
 #endif
 
+/** Custom GLib event source for libusb I/O.
+ * @internal
+ */
+struct usb_source {
+	GSource base;
+
+	int64_t timeout_us;
+	int64_t due_us;
+
+	/* Needed to keep track of installed sources */
+	struct sr_session *session;
+
+	struct libusb_context *usb_ctx;
+	GPtrArray *pollfds;
+};
+
+/** USB event source prepare() method.
+ */
+static gboolean usb_source_prepare(GSource *source, int *timeout)
+{
+	int64_t now_us, usb_due_us;
+	struct usb_source *usource;
+	struct timeval usb_timeout;
+	int remaining_ms;
+	int ret;
+
+	usource = (struct usb_source *)source;
+
+	ret = libusb_get_next_timeout(usource->usb_ctx, &usb_timeout);
+	if (G_UNLIKELY(ret < 0)) {
+		sr_err("Failed to get libusb timeout: %s",
+			libusb_error_name(ret));
+	}
+	now_us = g_source_get_time(source);
+
+	if (usource->due_us == 0) {
+		/* First-time initialization of the expiration time */
+		usource->due_us = now_us + usource->timeout_us;
+	}
+	if (ret == 1) {
+		usb_due_us = (int64_t)usb_timeout.tv_sec * G_USEC_PER_SEC
+				+ usb_timeout.tv_usec + now_us;
+		if (usb_due_us < usource->due_us)
+			usource->due_us = usb_due_us;
+	}
+	if (usource->due_us != INT64_MAX)
+		remaining_ms = (MAX(0, usource->due_us - now_us) + 999) / 1000;
+	else
+		remaining_ms = -1;
+
+	*timeout = remaining_ms;
+
+	return (remaining_ms == 0);
+}
+
+/** USB event source check() method.
+ */
+static gboolean usb_source_check(GSource *source)
+{
+	struct usb_source *usource;
+	GPollFD *pollfd;
+	unsigned int revents;
+	unsigned int i;
+
+	usource = (struct usb_source *)source;
+	revents = 0;
+
+	for (i = 0; i < usource->pollfds->len; i++) {
+		pollfd = g_ptr_array_index(usource->pollfds, i);
+		revents |= pollfd->revents;
+	}
+	return (revents != 0 || (usource->due_us != INT64_MAX
+			&& usource->due_us <= g_source_get_time(source)));
+}
+
+/** USB event source dispatch() method.
+ */
+static gboolean usb_source_dispatch(GSource *source,
+		GSourceFunc callback, void *user_data)
+{
+	struct usb_source *usource;
+	GPollFD *pollfd;
+	unsigned int revents;
+	unsigned int i;
+	gboolean keep;
+
+	usource = (struct usb_source *)source;
+	revents = 0;
+	/*
+	 * This is somewhat arbitrary, but drivers use revents to distinguish
+	 * actual I/O from timeouts. When we remove the user timeout from the
+	 * driver API, this will no longer be needed.
+	 */
+	for (i = 0; i < usource->pollfds->len; i++) {
+		pollfd = g_ptr_array_index(usource->pollfds, i);
+		revents |= pollfd->revents;
+	}
+	if (revents != 0)
+		sr_spew("%s: revents 0x%.2X", __func__, revents);
+	else
+		sr_spew("%s: timed out", __func__);
+
+	if (!callback) {
+		sr_err("Callback not set, cannot dispatch event.");
+		return G_SOURCE_REMOVE;
+	}
+	keep = (*(sr_receive_data_callback)callback)(-1, revents, user_data);
+
+	if (G_LIKELY(keep) && G_LIKELY(!g_source_is_destroyed(source))) {
+		if (usource->timeout_us >= 0)
+			usource->due_us = g_source_get_time(source)
+					+ usource->timeout_us;
+		else
+			usource->due_us = INT64_MAX;
+	}
+	return keep;
+}
+
+/** USB event source finalize() method.
+ */
+static void usb_source_finalize(GSource *source)
+{
+	struct usb_source *usource;
+
+	usource = (struct usb_source *)source;
+
+	sr_spew("%s", __func__);
+
+	libusb_set_pollfd_notifiers(usource->usb_ctx, NULL, NULL, NULL);
+
+	g_ptr_array_unref(usource->pollfds);
+	usource->pollfds = NULL;
+
+	sr_session_source_destroyed(usource->session,
+			usource->usb_ctx, source);
+}
+
+/** Callback invoked when a new libusb FD should be added to the poll set.
+ */
+static LIBUSB_CALL void usb_pollfd_added(libusb_os_handle fd,
+		short events, void *user_data)
+{
+	struct usb_source *usource;
+	GPollFD *pollfd;
+
+	usource = user_data;
+
+	pollfd = g_slice_new(GPollFD);
+#ifdef G_OS_WIN32
+	events = G_IO_IN;
+#endif
+	pollfd->fd = (gintptr)fd;
+	pollfd->events = events;
+	pollfd->revents = 0;
+
+	g_ptr_array_add(usource->pollfds, pollfd);
+	g_source_add_poll((GSource *)usource, pollfd);
+}
+
+/** Callback invoked when a libusb FD should be removed from the poll set.
+ */
+static LIBUSB_CALL void usb_pollfd_removed(libusb_os_handle fd, void *user_data)
+{
+	struct usb_source *usource;
+	GPollFD *pollfd;
+	unsigned int i;
+
+	usource = user_data;
+
+	/* It's likely that the removed poll FD is at the end.
+	 */
+	for (i = usource->pollfds->len; G_LIKELY(i > 0); i--) {
+		pollfd = g_ptr_array_index(usource->pollfds, i - 1);
+
+		if ((libusb_os_handle)pollfd->fd == fd) {
+			g_source_remove_poll((GSource *)usource, pollfd);
+			g_ptr_array_remove_index_fast(usource->pollfds, i - 1);
+			return;
+		}
+	}
+	sr_err("FD to be removed (%" G_GINTPTR_FORMAT
+		") not found in event source poll set.", (gintptr)fd);
+}
+
+/** Destroy notify callback for FDs maintained by the USB event source.
+ */
+static void usb_source_free_pollfd(void *data)
+{
+	g_slice_free(GPollFD, data);
+}
+
+/** Create an event source for libusb I/O.
+ *
+ * TODO: The combination of the USB I/O source with a user timeout is
+ * conceptually broken. The user timeout supplied here is completely
+ * unrelated to I/O -- the actual I/O timeout is set when submitting
+ * a USB transfer.
+ * The sigrok drivers generally use the timeout to poll device state.
+ * Usually, this polling can be sensibly done only when there is no
+ * active USB transfer -- i.e. it's actually mutually exclusive with
+ * waiting for transfer completion.
+ * Thus, the user timeout should be removed from the USB event source
+ * API at some point. Instead, drivers should install separate timer
+ * event sources for their polling needs.
+ *
+ * @param session The session the event source belongs to.
+ * @param usb_ctx The libusb context for which to handle events.
+ * @param timeout_ms The timeout interval in ms, or -1 to wait indefinitely.
+ * @return A new event source object, or NULL on failure.
+ */
+static GSource *usb_source_new(struct sr_session *session,
+		struct libusb_context *usb_ctx, int timeout_ms)
+{
+	static GSourceFuncs usb_source_funcs = {
+		.prepare  = &usb_source_prepare,
+		.check    = &usb_source_check,
+		.dispatch = &usb_source_dispatch,
+		.finalize = &usb_source_finalize
+	};
+	GSource *source;
+	struct usb_source *usource;
+	const struct libusb_pollfd **upollfds, **upfd;
+
+	upollfds = libusb_get_pollfds(usb_ctx);
+	if (!upollfds) {
+		sr_err("Failed to get libusb file descriptors.");
+		return NULL;
+	}
+	source = g_source_new(&usb_source_funcs, sizeof(struct usb_source));
+	usource = (struct usb_source *)source;
+
+	g_source_set_name(source, "usb");
+
+	if (timeout_ms >= 0) {
+		usource->timeout_us = 1000 * (int64_t)timeout_ms;
+		usource->due_us = 0;
+	} else {
+		usource->timeout_us = -1;
+		usource->due_us = INT64_MAX;
+	}
+	usource->session = session;
+	usource->usb_ctx = usb_ctx;
+	usource->pollfds = g_ptr_array_new_full(8, &usb_source_free_pollfd);
+
+	for (upfd = upollfds; *upfd != NULL; upfd++)
+		usb_pollfd_added((*upfd)->fd, (*upfd)->events, usource);
+
+#if (LIBUSB_API_VERSION >= 0x01000104)
+	libusb_free_pollfds(upollfds);
+#else
+	free(upollfds);
+#endif
+	libusb_set_pollfd_notifiers(usb_ctx,
+		&usb_pollfd_added, &usb_pollfd_removed, usource);
+
+	return source;
+}
+
 /**
  * Find USB devices according to a connection string.
  *
@@ -188,86 +446,27 @@ SR_PRIV void sr_usb_close(struct sr_usb_dev_inst *usb)
 	sr_dbg("Closed USB device %d.%d.", usb->bus, usb->address);
 }
 
-static LIBUSB_CALL void usb_pollfd_added(libusb_os_handle fd,
-		short events, void *user_data)
-{
-	struct sr_session *session;
-	gintptr tag;
-
-	session = user_data;
-	tag = (gintptr)session->ctx->libusb_ctx;
-#ifdef G_OS_WIN32
-	events = G_IO_IN;
-#endif
-	sr_session_source_poll_add(session, tag, (gintptr)fd, events);
-}
-
-static LIBUSB_CALL void usb_pollfd_removed(libusb_os_handle fd, void *user_data)
-{
-	struct sr_session *session;
-	gintptr tag;
-
-	session = user_data;
-	tag = (gintptr)session->ctx->libusb_ctx;
-
-	sr_session_source_poll_remove(session, tag, (gintptr)fd);
-}
-
 SR_PRIV int usb_source_add(struct sr_session *session, struct sr_context *ctx,
 		int timeout, sr_receive_data_callback cb, void *cb_data)
 {
-	const struct libusb_pollfd **pollfds;
-	gintptr tag;
-	int i;
+	GSource *source;
 	int ret;
-	int events;
 
-	if (ctx->usb_source_present) {
-		sr_err("A USB event source is already present.");
-		return SR_ERR;
-	}
-	pollfds = libusb_get_pollfds(ctx->libusb_ctx);
-	if (!pollfds) {
-		sr_err("Failed to get libusb file descriptors.");
+	source = usb_source_new(session, ctx->libusb_ctx, timeout);
+	if (!source)
 		return SR_ERR;
-	}
-	tag = (gintptr)ctx->libusb_ctx;
-	ret = sr_session_source_add_internal(session,
-			timeout, cb, cb_data, tag);
 
-	ctx->usb_source_present = (ret == SR_OK);
+	g_source_set_callback(source, (GSourceFunc)cb, cb_data, NULL);
 
-	for (i = 0; ret == SR_OK && pollfds[i]; ++i) {
-#ifdef G_OS_WIN32
-		events = G_IO_IN;
-#else
-		events = pollfds[i]->events;
-#endif
-		ret = sr_session_source_poll_add(session, tag,
-				(gintptr)pollfds[i]->fd, events);
-	}
-#if (LIBUSB_API_VERSION >= 0x01000104)
-	libusb_free_pollfds(pollfds);
-#else
-	free(pollfds);
-#endif
-	if (ret != SR_OK)
-		return ret;
+	ret = sr_session_source_add_internal(session, ctx->libusb_ctx, source);
+	g_source_unref(source);
 
-	libusb_set_pollfd_notifiers(ctx->libusb_ctx,
-		&usb_pollfd_added, &usb_pollfd_removed, session);
-
-	return SR_OK;
+	return ret;
 }
 
 SR_PRIV int usb_source_remove(struct sr_session *session, struct sr_context *ctx)
 {
-	ctx->usb_source_present = FALSE;
-
-	libusb_set_pollfd_notifiers(ctx->libusb_ctx, NULL, NULL, NULL);
-
-	return sr_session_source_remove_internal(session,
-			(gintptr)ctx->libusb_ctx);
+	return sr_session_source_remove_internal(session, ctx->libusb_ctx);
 }
 
 SR_PRIV int usb_get_port_path(libusb_device *dev, char *path, int path_len)