X-Git-Url: https://sigrok.org/gitweb/?p=libsigrok.git;a=blobdiff_plain;f=src%2Fsession.c;h=8d3f66966a4f156c632d185d4f788274b15e58be;hp=e000dd0ef7efb7be12e2d4ab4a9887c92fcce46f;hb=64f26f744bf3185eddb07e8bfe9855e3c7bfac19;hpb=62d7945f8059ccbf56dfa2e5eb60671dd5bc959b diff --git a/src/session.c b/src/session.c index e000dd0e..8d3f6696 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 + * Copyright (C) 2015 Daniel Elstner * * 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 @@ -17,6 +18,7 @@ * along with this program. If not, see . */ +#include #include #include #include @@ -44,25 +46,159 @@ * @{ */ -struct source { - int64_t due; - 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. + * @see https://developer.gnome.org/glib/stable/glib-The-Main-Event-Loop.html + */ +struct fd_source { + GSource base; - int timeout; - gboolean is_usb; -}; + int64_t timeout_us; + int64_t due_us; -struct datafeed_callback { - sr_datafeed_callback cb; - void *cb_data; + /* Meta-data needed to keep track of installed sources */ + struct sr_session *session; + void *key; + + GPollFD pollfd; }; +/** FD event source prepare() method. + * This is called immediately before poll(). + */ +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. + * This is called after poll() returns to check whether an event fired. + */ +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. + * This is called if either prepare() or check() returned TRUE. + */ +static gboolean fd_source_dispatch(GSource *source, + GSourceFunc callback, void *user_data) +{ + struct fd_source *fsource; + unsigned int revents; + gboolean keep; + + fsource = (struct fd_source *)source; + revents = fsource->pollfd.revents; + + 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 events 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 *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. * @@ -87,9 +223,13 @@ SR_API int sr_session_new(struct sr_context *ctx, session = g_malloc0(sizeof(struct sr_session)); session->ctx = ctx; - session->running = FALSE; - session->abort_session = FALSE; - g_mutex_init(&session->stop_mutex); + + g_mutex_init(&session->main_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; @@ -115,12 +255,14 @@ SR_API int sr_session_destroy(struct sr_session *session) } sr_session_dev_remove_all(session); - g_mutex_clear(&session->stop_mutex); - if (session->trigger) - sr_trigger_free(session->trigger); - g_slist_free_full(session->owned_devs, (GDestroyNotify)sr_dev_inst_free); + sr_session_datafeed_callback_remove_all(session); + + g_hash_table_unref(session->event_sources); + + g_mutex_clear(&session->main_mutex); + g_free(session); return SR_OK; @@ -198,7 +340,7 @@ SR_API int sr_session_dev_add(struct sr_session *session, /* If sdi->driver is NULL, this is a virtual device. */ if (!sdi->driver) { /* Just add the device, don't run dev_open(). */ - session->devs = g_slist_append(session->devs, (gpointer)sdi); + session->devs = g_slist_append(session->devs, sdi); sdi->session = session; return SR_OK; } @@ -209,9 +351,12 @@ SR_API int sr_session_dev_add(struct sr_session *session, return SR_ERR_BUG; } - session->devs = g_slist_append(session->devs, (gpointer)sdi); + session->devs = g_slist_append(session->devs, sdi); sdi->session = session; + /* TODO: This is invalid if the session runs in a different thread. + * The usage semantics and restrictions need to be documented. + */ if (session->running) { /* Adding a device to a running session. Commit settings * and start acquisition on that device now. */ @@ -221,8 +366,7 @@ SR_API int sr_session_dev_add(struct sr_session *session, sr_strerror(ret)); return ret; } - if ((ret = sdi->driver->dev_acquisition_start(sdi, - (void *)sdi)) != SR_OK) { + if ((ret = sr_dev_acquisition_start(sdi)) != SR_OK) { sr_err("Failed to start acquisition of device in " "running session (%s)", sr_strerror(ret)); return ret; @@ -261,6 +405,45 @@ SR_API int sr_session_dev_list(struct sr_session *session, GSList **devlist) return SR_OK; } +/** + * Remove a device instance from a session. + * + * @param session The session to remove from. Must not be NULL. + * @param sdi The device instance to remove from a session. Must not + * be NULL. Also, sdi->driver and sdi->driver->dev_open must + * not be NULL. + * + * @retval SR_OK Success. + * @retval SR_ERR_ARG Invalid argument. + * + * @since 0.4.0 + */ +SR_API int sr_session_dev_remove(struct sr_session *session, + struct sr_dev_inst *sdi) +{ + if (!sdi) { + sr_err("%s: sdi was NULL", __func__); + return SR_ERR_ARG; + } + + if (!session) { + sr_err("%s: session was NULL", __func__); + return SR_ERR_ARG; + } + + /* If sdi->session is not session, the device is not in this + * session. */ + if (sdi->session != session) { + sr_err("%s: not assigned to this session", __func__); + return SR_ERR_ARG; + } + + session->devs = g_slist_remove(session->devs, sdi); + sdi->session = NULL; + + return SR_OK; +} + /** * Remove all datafeed callbacks in a session. * @@ -361,143 +544,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; -} - -/** - * Call every device in the current session's callback. - * - * For sessions not driven by select loops such as sr_session_run(), - * but driven by another scheduler, this can be used to poll the devices - * from within that scheduler. - * - * @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; - int64_t stop_time; - int64_t min_due; - int64_t due; - unsigned int i; - int ret, timeout; - int revents; - gboolean stop_checked; - gboolean stopped; - struct source *source; - GPollFD *pollfd; - gintptr poll_object; -#ifdef HAVE_LIBUSB_1_0 - int64_t usb_due; - struct timeval tv; -#endif - if (session->num_sources <= 0) { - sr_session_check_aborted(session); - return SR_OK; - } - start_time = g_get_monotonic_time(); - min_due = INT64_MAX; - - for (i = 0; i < session->num_sources; ++i) { - if (session->sources[i].due < min_due) - min_due = session->sources[i].due; - } -#ifdef 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_due = start_time + tv.tv_usec - + (int64_t)tv.tv_sec * G_USEC_PER_SEC; - if (usb_due < min_due) - min_due = usb_due; - } - } -#endif - if (min_due > start_time) - timeout = (min_due == INT64_MAX) ? -1 - : MIN((min_due - start_time + 999) / 1000, INT_MAX); - else - timeout = 0; - - ret = g_poll(session->pollfds, session->num_sources, timeout); -#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(); - stop_checked = FALSE; - stopped = FALSE; - - for (i = 0; i < session->num_sources; i++) { - source = &session->sources[i]; - pollfd = &session->pollfds[i]; - revents = (ret > 0) ? pollfd->revents : 0; - due = source->due; -#ifdef HAVE_LIBUSB_1_0 - if (source->is_usb && usb_due < due) - 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. - */ - poll_object = source->poll_object; - if (source->timeout > 0) - source->due = stop_time + INT64_C(1000) * source->timeout; - pollfd->revents = 0; - /* - * Invoke the source's callback on an event or timeout. - */ - if (!source->cb(pollfd->fd, revents, source->cb_data)) - sr_session_source_remove(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); - stop_checked = TRUE; - } - /* Restart loop as the sources list may have changed. */ - i = 0; - } - if (!stop_checked) - sr_session_check_aborted(session); - - return SR_OK; -} - static int verify_trigger(struct sr_trigger *trigger) { struct sr_trigger_stage *stage; @@ -534,13 +580,173 @@ 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 *main_context; + + g_mutex_lock(&session->main_mutex); + + /* May happen if sr_session_start() is called a second time + * while the session is still running. + */ + if (session->main_context) { + sr_err("Main context already set."); + + g_mutex_unlock(&session->main_mutex); + return SR_ERR; + } + main_context = g_main_context_ref_thread_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(main_context)) { + g_main_context_release(main_context); + + sr_dbg("Using thread-default main context."); + } else { + g_main_context_unref(main_context); + + sr_dbg("Creating our own main context."); + main_context = g_main_context_new(); + } + session->main_context = main_context; + + 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) { + 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; +} + +static unsigned int session_source_attach(struct sr_session *session, + GSource *source) +{ + unsigned int id = 0; + + g_mutex_lock(&session->main_mutex); + + if (session->main_context) + id = g_source_attach(source, session->main_context); + else + sr_err("Cannot add event source without main context."); + + g_mutex_unlock(&session->main_mutex); + + return id; +} + +/* Idle handler; invoked when the number of registered event sources + * for a running session drops to zero. + */ +static gboolean delayed_stop_check(void *data) +{ + struct sr_session *session; + + session = data; + session->stop_check_id = 0; + + /* Session already ended? */ + if (!session->running) + return G_SOURCE_REMOVE; + + /* New event sources may have been installed in the meantime. */ + if (g_hash_table_size(session->event_sources) != 0) + return G_SOURCE_REMOVE; + + session->running = FALSE; + unset_main_context(session); + + sr_info("Stopped."); + + /* This indicates a bug in user code, since it is not valid to + * restart or destroy a session while it may still be running. + */ + if (!session->main_loop && !session->stopped_callback) { + sr_err("BUG: Session stop left unhandled."); + return G_SOURCE_REMOVE; + } + if (session->main_loop) + g_main_loop_quit(session->main_loop); + + if (session->stopped_callback) + (*session->stopped_callback)(session->stopped_cb_data); + + return G_SOURCE_REMOVE; +} + +static int stop_check_later(struct sr_session *session) +{ + GSource *source; + unsigned int source_id; + + if (session->stop_check_id != 0) + return SR_OK; /* idle handler already installed */ + + source = g_idle_source_new(); + g_source_set_callback(source, &delayed_stop_check, session, NULL); + + source_id = session_source_attach(session, source); + session->stop_check_id = source_id; + + g_source_unref(source); + + return (source_id != 0) ? SR_OK : SR_ERR; +} + /** * Start a session. * + * When this function returns with a status code indicating success, the + * session is running. Use sr_session_stopped_callback_set() to receive + * notification upon completion, or call sr_session_run() to block until + * the session stops. + * + * Session events will be processed in the context of the current thread. + * If a thread-default GLib main context has been set, and is not owned by + * any other thread, it will be used. Otherwise, libsigrok will create its + * own main context for the current thread. + * * @param session The session to use. Must not be NULL. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. + * @retval SR_ERR Other error. * * @since 0.4.0 */ @@ -548,8 +754,8 @@ SR_API int sr_session_start(struct sr_session *session) { struct sr_dev_inst *sdi; struct sr_channel *ch; - GSList *l, *c; - int enabled_channels, ret; + GSList *l, *c, *lend; + int ret; if (!session) { sr_err("%s: session was NULL", __func__); @@ -562,53 +768,100 @@ SR_API int sr_session_start(struct sr_session *session) return SR_ERR_ARG; } - if (session->trigger && verify_trigger(session->trigger) != SR_OK) + if (session->running) { + sr_err("Cannot (re-)start session while it is still running."); return SR_ERR; + } - sr_info("Starting."); + if (session->trigger) { + ret = verify_trigger(session->trigger); + if (ret != SR_OK) + return ret; + } - ret = SR_OK; + /* Check enabled channels and commit settings of all devices. */ for (l = session->devs; l; l = l->next) { sdi = l->data; - enabled_channels = 0; for (c = sdi->channels; c; c = c->next) { ch = c->data; - if (ch->enabled) { - enabled_channels++; + if (ch->enabled) break; - } } - if (enabled_channels == 0) { - ret = SR_ERR; - sr_err("%s using connection %s has no enabled channels!", - sdi->driver->name, sdi->connection_id); - break; + if (!c) { + sr_err("%s device %s has no enabled channels.", + sdi->driver->name, sdi->connection_id); + return SR_ERR; } - if ((ret = sr_config_commit(sdi)) != SR_OK) { - sr_err("Failed to commit device settings before " - "starting acquisition (%s)", sr_strerror(ret)); + ret = sr_config_commit(sdi); + if (ret != SR_OK) { + sr_err("Failed to commit %s device %s settings " + "before starting acquisition.", + sdi->driver->name, sdi->connection_id); + return ret; + } + } + + ret = set_main_context(session); + if (ret != SR_OK) + return ret; + + sr_info("Starting."); + + session->running = TRUE; + + /* Have all devices start acquisition. */ + for (l = session->devs; l; l = l->next) { + if (!(sdi = l->data)) { + sr_err("Device sdi was NULL, can't start session."); + ret = SR_ERR; break; } - if ((ret = sdi->driver->dev_acquisition_start(sdi, sdi)) != SR_OK) { - sr_err("%s: could not start an acquisition " - "(%s)", __func__, sr_strerror(ret)); + ret = sr_dev_acquisition_start(sdi); + if (ret != SR_OK) { + sr_err("Could not start %s device %s acquisition.", + sdi->driver->name, sdi->connection_id); break; } } - /* TODO: What if there are multiple devices? Which return code? */ + if (ret != SR_OK) { + /* If there are multiple devices, some of them may already have + * started successfully. Stop them now before returning. */ + lend = l->next; + for (l = session->devs; l != lend; l = l->next) { + sdi = l->data; + sr_dev_acquisition_stop(sdi); + } + /* TODO: Handle delayed stops. Need to iterate the event + * sources... */ + session->running = FALSE; - return ret; + unset_main_context(session); + return ret; + } + + if (g_hash_table_size(session->event_sources) == 0) + stop_check_later(session); + + return SR_OK; } /** - * Run a session. + * Block until the running session stops. + * + * This is a convenience function which creates a GLib main loop and runs + * it to process session events until the session stops. + * + * Instead of using this function, applications may run their own GLib main + * loop, and use sr_session_stopped_callback_set() to receive notification + * when the session finished running. * * @param session The session to use. Must not be NULL. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. + * @retval SR_ERR Other error. * * @since 0.4.0 */ @@ -618,99 +871,157 @@ SR_API int sr_session_run(struct sr_session *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; + if (!session->running) { + sr_err("No session running."); + return SR_ERR; + } + if (session->main_loop) { + sr_err("Main loop already created."); + return SR_ERR; } - session->running = TRUE; - sr_info("Running."); + g_mutex_lock(&session->main_mutex); - /* Do we have real sources? */ - if (session->num_sources == 1 && session->pollfds[0].fd == -1) { - /* Dummy source, freewheel over it. */ - while (session->num_sources) - session->sources[0].cb(-1, 0, session->sources[0].cb_data); - } else { - /* Real sources, use g_poll() main loop. */ - while (session->num_sources) - sr_session_iteration(session); + if (!session->main_context) { + sr_err("Cannot run without main context."); + g_mutex_unlock(&session->main_mutex); + return SR_ERR; } + 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; } +static gboolean session_stop_sync(void *user_data) +{ + struct sr_session *session; + struct sr_dev_inst *sdi; + GSList *node; + + session = user_data; + + if (!session->running) + return G_SOURCE_REMOVE; + + sr_info("Stopping."); + + for (node = session->devs; node; node = node->next) { + sdi = node->data; + sr_dev_acquisition_stop(sdi); + } + + return G_SOURCE_REMOVE; +} + /** * Stop a session. * - * The session is stopped immediately, with all acquisition sessions stopped - * and hardware drivers cleaned up. + * This requests the drivers of each device participating in the session to + * abort the acquisition as soon as possible. Even after this function returns, + * event processing still continues until all devices have actually stopped. + * + * Use sr_session_stopped_callback_set() to receive notification when the event + * processing finished. * - * This must be called from within the session thread, to prevent freeing - * resources that the session thread will try to use. + * This function is reentrant. That is, it may be called from a different + * thread than the one executing the session, as long as it can be ensured + * that the session object is valid. + * + * If the session is not running, sr_session_stop() silently does nothing. * * @param session The session to use. Must not be NULL. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. * - * @private + * @since 0.4.0 */ -SR_PRIV int sr_session_stop_sync(struct sr_session *session) +SR_API int sr_session_stop(struct sr_session *session) { - struct sr_dev_inst *sdi; - GSList *l; + GMainContext *main_context; if (!session) { sr_err("%s: session was NULL", __func__); return SR_ERR_ARG; } - sr_info("Stopping."); + g_mutex_lock(&session->main_mutex); - 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); - } + main_context = (session->main_context) + ? g_main_context_ref(session->main_context) + : NULL; + + g_mutex_unlock(&session->main_mutex); + + if (!main_context) { + sr_dbg("No main context set; already stopped?"); + /* Not an error; as it would be racy. */ + return SR_OK; } - session->running = FALSE; + g_main_context_invoke(main_context, &session_stop_sync, session); + g_main_context_unref(main_context); return SR_OK; } /** - * Stop a session. + * Return whether the session is currently running. + * + * Note that this function should be called from the same thread + * the session was started in. * - * The session is stopped immediately, with all acquisition sessions being - * stopped and hardware drivers cleaned up. + * @param session The session to use. Must not be NULL. + * + * @retval TRUE Session is running. + * @retval FALSE Session is not running. + * @retval SR_ERR_ARG Invalid session passed. * - * If the session is run in a separate thread, this function will not block - * until the session is finished executing. It is the caller's responsibility - * to wait for the session thread to return before assuming that the session is - * completely decommissioned. + * @since 0.4.0 + */ +SR_API int sr_session_is_running(struct sr_session *session) +{ + if (!session) { + sr_err("%s: session was NULL", __func__); + return SR_ERR_ARG; + } + return session->running; +} + +/** + * Set the callback to be invoked after a session stopped running. + * + * Install a callback to receive notification when a session run stopped. + * This can be used to integrate session execution with an existing main + * loop, without having to block in sr_session_run(). + * + * Note that the callback will be invoked in the context of the thread + * that calls sr_session_start(). * * @param session The session to use. Must not be NULL. + * @param cb The callback to invoke on session stop. May be NULL to unset. + * @param cb_data User data pointer to be passed to the callback. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. * * @since 0.4.0 */ -SR_API int sr_session_stop(struct sr_session *session) +SR_API int sr_session_stopped_callback_set(struct sr_session *session, + sr_session_stopped_callback cb, void *cb_data) { if (!session) { sr_err("%s: session was NULL", __func__); - return SR_ERR_BUG; + return SR_ERR_ARG; } - - g_mutex_lock(&session->stop_mutex); - session->abort_session = TRUE; - g_mutex_unlock(&session->stop_mutex); + session->stopped_callback = cb; + session->stopped_cb_data = cb_data; return SR_OK; } @@ -724,7 +1035,6 @@ static void datafeed_dump(const struct sr_datafeed_packet *packet) { const struct sr_datafeed_logic *logic; const struct sr_datafeed_analog *analog; - const struct sr_datafeed_analog2 *analog2; /* Please use the same order as in libsigrok.h. */ switch (packet->type) { @@ -745,21 +1055,16 @@ static void datafeed_dump(const struct sr_datafeed_packet *packet) sr_dbg("bus: Received SR_DF_LOGIC packet (%" PRIu64 " bytes, " "unitsize = %d).", logic->length, logic->unitsize); break; - case SR_DF_ANALOG: - analog = packet->payload; - sr_dbg("bus: Received SR_DF_ANALOG packet (%d samples).", - analog->num_samples); - break; case SR_DF_FRAME_BEGIN: sr_dbg("bus: Received SR_DF_FRAME_BEGIN packet."); break; case SR_DF_FRAME_END: sr_dbg("bus: Received SR_DF_FRAME_END packet."); break; - case SR_DF_ANALOG2: - analog2 = packet->payload; - sr_dbg("bus: Received SR_DF_ANALOG2 packet (%d samples).", - analog2->num_samples); + case SR_DF_ANALOG: + analog = packet->payload; + sr_dbg("bus: Received SR_DF_ANALOG packet (%d samples).", + analog->num_samples); break; default: sr_dbg("bus: Received unknown packet type: %d.", packet->type); @@ -767,6 +1072,42 @@ static void datafeed_dump(const struct sr_datafeed_packet *packet) } } +/** + * Helper to send a meta datafeed package (SR_DF_META) to the session bus. + * + * @param sdi The device instance to send the package from. Must not be NULL. + * @param key The config key to send to the session bus. + * @param var The value to send to the session bus. + * + * @retval SR_OK Success. + * @retval SR_ERR_ARG Invalid argument. + * + * @private + */ +SR_PRIV int sr_session_send_meta(const struct sr_dev_inst *sdi, + uint32_t key, GVariant *var) +{ + struct sr_config *cfg; + struct sr_datafeed_packet packet; + struct sr_datafeed_meta meta; + int ret; + + cfg = sr_config_new(key, var); + + memset(&meta, 0, sizeof(meta)); + + packet.type = SR_DF_META; + packet.payload = &meta; + + meta.config = g_slist_append(NULL, cfg); + + ret = sr_session_send(sdi, &packet); + g_slist_free(meta.config); + sr_config_free(cfg); + + return ret; +} + /** * Send a packet to whatever is listening on the datafeed bus. * @@ -853,60 +1194,65 @@ 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 pollfd The GPollFD. - * @param[in] timeout Max time to wait before the callback is called, - * ignored if 0. - * @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 is_usb TRUE for a libusb polling source + * @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_BUG Event source with @a key already installed. + * @retval SR_ERR Other error. + * + * @private */ SR_PRIV int sr_session_source_add_internal(struct sr_session *session, - GPollFD *pollfd, int timeout, sr_receive_data_callback cb, - void *cb_data, gintptr poll_object, gboolean is_usb) + void *key, GSource *source) { - struct source *new_sources, *s; - GPollFD *new_pollfds; - - if (!cb) { - sr_err("%s: cb was NULL", __func__); - return SR_ERR_ARG; + /* + * 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.) + */ + if (g_hash_table_contains(session->event_sources, key)) { + sr_err("Event source with key %p already exists.", key); + return SR_ERR_BUG; } + g_hash_table_insert(session->event_sources, key, source); - /* Note: cb_data can be NULL, that's not a bug. */ + if (session_source_attach(session, source) == 0) + return SR_ERR; - new_pollfds = g_realloc(session->pollfds, - sizeof(GPollFD) * (session->num_sources + 1)); - new_sources = g_realloc(session->sources, sizeof(struct source) * - (session->num_sources + 1)); + return SR_OK; +} - new_pollfds[session->num_sources] = *pollfd; - s = &new_sources[session->num_sources++]; - if (timeout > 0) - s->due = g_get_monotonic_time() + INT64_C(1000) * timeout; - else - s->due = INT64_MAX; - s->cb = cb; - s->cb_data = cb_data; - s->poll_object = poll_object; - s->timeout = timeout; - s->is_usb = is_usb; - session->pollfds = new_pollfds; - session->sources = new_sources; +/** @private */ +SR_PRIV int sr_session_fd_source_add(struct sr_session *session, + void *key, gintptr fd, int events, int timeout, + sr_receive_data_callback cb, void *cb_data) +{ + GSource *source; + int ret; - return SR_OK; + source = fd_source_new(session, key, fd, events, timeout); + if (!source) + return SR_ERR; + + g_source_set_callback(source, G_SOURCE_FUNC(cb), cb_data, NULL); + + ret = sr_session_source_add_internal(session, key, source); + g_source_unref(source); + + return ret; } /** * Add an event source for a file descriptor. * * @param session The session to use. Must not be NULL. - * @param fd The file descriptor. + * @param fd The file descriptor, or a negative value to create a timer source. * @param events Events to check for. - * @param timeout Max time to wait before the callback is called, ignored if 0. + * @param 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. * @@ -914,26 +1260,26 @@ SR_PRIV int sr_session_source_add_internal(struct sr_session *session, * @retval SR_ERR_ARG Invalid argument. * * @since 0.3.0 + * @private */ -SR_API int sr_session_source_add(struct sr_session *session, int fd, +SR_PRIV int sr_session_source_add(struct sr_session *session, int fd, int events, int timeout, sr_receive_data_callback cb, void *cb_data) { - GPollFD p; - - p.fd = fd; - p.events = events; - p.revents = 0; - - return sr_session_source_add_internal(session, &p, timeout, - cb, cb_data, fd, FALSE); + if (fd < 0 && timeout < 0) { + sr_err("Cannot create timer source without timeout."); + return SR_ERR_ARG; + } + return sr_session_fd_source_add(session, GINT_TO_POINTER(fd), + fd, events, timeout, cb, cb_data); } /** * Add an event source for a GPollFD. * * @param session The session to use. Must not be NULL. - * @param pollfd The GPollFD. - * @param timeout Max time to wait before the callback is called, ignored if 0. + * @param pollfd The GPollFD. Must not be NULL. + * @param 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. * @@ -941,13 +1287,18 @@ SR_API int sr_session_source_add(struct sr_session *session, int fd, * @retval SR_ERR_ARG Invalid argument. * * @since 0.3.0 + * @private */ -SR_API int sr_session_source_add_pollfd(struct sr_session *session, +SR_PRIV int sr_session_source_add_pollfd(struct sr_session *session, GPollFD *pollfd, int timeout, sr_receive_data_callback cb, void *cb_data) { - return sr_session_source_add_internal(session, pollfd, timeout, - cb, cb_data, (gintptr)pollfd, FALSE); + if (!pollfd) { + sr_err("%s: pollfd was NULL", __func__); + return SR_ERR_ARG; + } + return sr_session_fd_source_add(session, pollfd, pollfd->fd, + pollfd->events, timeout, cb, cb_data); } /** @@ -956,7 +1307,8 @@ SR_API int sr_session_source_add_pollfd(struct sr_session *session, * @param session The session to use. Must not be NULL. * @param channel The GIOChannel. * @param events Events to poll on. - * @param timeout Max time to wait before the callback is called, ignored if 0. + * @param 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. * @@ -964,63 +1316,57 @@ SR_API int sr_session_source_add_pollfd(struct sr_session *session, * @retval SR_ERR_ARG Invalid argument. * * @since 0.3.0 + * @private */ -SR_API int sr_session_source_add_channel(struct sr_session *session, +SR_PRIV int sr_session_source_add_channel(struct sr_session *session, GIOChannel *channel, int events, int timeout, sr_receive_data_callback cb, void *cb_data) { - GPollFD p; + GPollFD pollfd; + 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 _WIN32 - g_io_channel_win32_make_pollfd(channel, events, &p); + g_io_channel_win32_make_pollfd(channel, events, &pollfd); #else - p.fd = g_io_channel_unix_get_fd(channel); - p.events = events; - p.revents = 0; + pollfd.fd = g_io_channel_unix_get_fd(channel); + pollfd.events = events; #endif - return sr_session_source_add_internal(session, &p, timeout, cb, - cb_data, (gintptr)channel, FALSE); + return sr_session_fd_source_add(session, channel, pollfd.fd, + pollfd.events, timeout, cb, cb_data); } /** - * Remove the source belonging to the specified channel. + * 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_ARG Invalid arguments - * @retval SR_ERR_BUG Internal error + * @retval SR_ERR_BUG No event source for poll_object found. + * + * @private */ -static int _sr_session_source_remove(struct sr_session *session, gintptr poll_object) +SR_PRIV int sr_session_source_remove_internal(struct sr_session *session, + void *key) { - unsigned int old; + GSource *source; - if (!session->sources || !session->num_sources) { - sr_err("%s: sources was NULL", __func__); + 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 (!source) { + sr_warn("Cannot remove non-existing event source %p.", key); return SR_ERR_BUG; } - - for (old = 0; old < session->num_sources; old++) { - if (session->sources[old].poll_object == poll_object) - break; - } - - /* fd not found, nothing to do */ - if (old == session->num_sources) - return SR_OK; - - session->num_sources--; - - if (old != session->num_sources) { - memmove(&session->pollfds[old], &session->pollfds[old + 1], - (session->num_sources - old) * sizeof(GPollFD)); - memmove(&session->sources[old], &session->sources[old + 1], - (session->num_sources - old) * sizeof(struct source)); - } - - session->pollfds = g_realloc(session->pollfds, sizeof(GPollFD) * session->num_sources); - session->sources = g_realloc(session->sources, sizeof(struct source) * session->num_sources); + g_source_destroy(source); return SR_OK; } @@ -1036,10 +1382,11 @@ static int _sr_session_source_remove(struct sr_session *session, gintptr poll_ob * @retval SR_ERR_BUG Internal error. * * @since 0.3.0 + * @private */ -SR_API int sr_session_source_remove(struct sr_session *session, int fd) +SR_PRIV int sr_session_source_remove(struct sr_session *session, int fd) { - return _sr_session_source_remove(session, (gintptr)fd); + return sr_session_source_remove_internal(session, GINT_TO_POINTER(fd)); } /** @@ -1047,17 +1394,22 @@ SR_API int sr_session_source_remove(struct sr_session *session, int fd) * * @param session The session to use. Must not be NULL. * @param pollfd The poll descriptor for which the source should be removed. - * + * Must not be NULL. * @return SR_OK upon success, SR_ERR_ARG upon invalid arguments, or * SR_ERR_MALLOC upon memory allocation errors, SR_ERR_BUG upon * internal errors. * * @since 0.2.0 + * @private */ -SR_API int sr_session_source_remove_pollfd(struct sr_session *session, +SR_PRIV int sr_session_source_remove_pollfd(struct sr_session *session, GPollFD *pollfd) { - return _sr_session_source_remove(session, (gintptr)pollfd); + if (!pollfd) { + sr_err("%s: pollfd was NULL", __func__); + return SR_ERR_ARG; + } + return sr_session_source_remove_internal(session, pollfd); } /** @@ -1065,27 +1417,84 @@ SR_API int sr_session_source_remove_pollfd(struct sr_session *session, * * @param session The session to use. Must not be NULL. * @param channel The channel for which the source should be removed. - * + * Must not be NULL. * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid argument. * @return SR_ERR_BUG Internal error. * * @since 0.2.0 + * @private */ -SR_API int sr_session_source_remove_channel(struct sr_session *session, +SR_PRIV int sr_session_source_remove_channel(struct sr_session *session, GIOChannel *channel) { - return _sr_session_source_remove(session, (gintptr)channel); + if (!channel) { + sr_err("%s: channel was NULL", __func__); + return SR_ERR_ARG; + } + 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. + * @retval SR_ERR Other error. + * + * @private + */ +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); + + if (g_hash_table_size(session->event_sources) > 0) + return SR_OK; + + /* If no event sources are left, consider the acquisition finished. + * This is pretty crude, as it requires all event sources to be + * registered via the libsigrok API. + */ + return stop_check_later(session); } static void copy_src(struct sr_config *src, struct sr_datafeed_meta *meta_copy) { + struct sr_config *item; + +#if GLIB_CHECK_VERSION(2, 67, 3) + item = g_memdup2(src, sizeof(*src)); +#else + item = g_memdup(src, sizeof(*src)); +#endif + g_variant_ref(src->data); - meta_copy->config = g_slist_append(meta_copy->config, - g_memdup(src, sizeof(struct sr_config))); + meta_copy->config = g_slist_append(meta_copy->config, item); } -SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet, +SR_API int sr_packet_copy(const struct sr_datafeed_packet *packet, struct sr_datafeed_packet **copy) { const struct sr_datafeed_meta *meta; @@ -1094,6 +1503,9 @@ SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet, struct sr_datafeed_logic *logic_copy; const struct sr_datafeed_analog *analog; struct sr_datafeed_analog *analog_copy; + struct sr_analog_encoding *encoding_copy; + struct sr_analog_meaning *meaning_copy; + struct sr_analog_spec *spec_copy; uint8_t *payload; *copy = g_malloc0(sizeof(struct sr_datafeed_packet)); @@ -1117,22 +1529,41 @@ SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet, break; case SR_DF_LOGIC: logic = packet->payload; - logic_copy = g_malloc(sizeof(logic)); + logic_copy = g_malloc(sizeof(*logic_copy)); + if (!logic_copy) + return SR_ERR; logic_copy->length = logic->length; logic_copy->unitsize = logic->unitsize; + logic_copy->data = g_malloc(logic->length * logic->unitsize); + if (!logic_copy->data) { + g_free(logic_copy); + return SR_ERR; + } memcpy(logic_copy->data, logic->data, logic->length * logic->unitsize); (*copy)->payload = logic_copy; break; case SR_DF_ANALOG: analog = packet->payload; - analog_copy = g_malloc(sizeof(analog)); - analog_copy->channels = g_slist_copy(analog->channels); - analog_copy->num_samples = analog->num_samples; - analog_copy->mq = analog->mq; - analog_copy->unit = analog->unit; - analog_copy->mqflags = analog->mqflags; + analog_copy = g_malloc(sizeof(*analog_copy)); + analog_copy->data = g_malloc( + analog->encoding->unitsize * analog->num_samples); memcpy(analog_copy->data, analog->data, - analog->num_samples * sizeof(float)); + analog->encoding->unitsize * analog->num_samples); + analog_copy->num_samples = analog->num_samples; +#if GLIB_CHECK_VERSION(2, 67, 3) + encoding_copy = g_memdup2(analog->encoding, sizeof(*analog->encoding)); + meaning_copy = g_memdup2(analog->meaning, sizeof(*analog->meaning)); + spec_copy = g_memdup2(analog->spec, sizeof(*analog->spec)); +#else + encoding_copy = g_memdup(analog->encoding, sizeof(*analog->encoding)); + meaning_copy = g_memdup(analog->meaning, sizeof(*analog->meaning)); + spec_copy = g_memdup(analog->spec, sizeof(*analog->spec)); +#endif + analog_copy->encoding = encoding_copy; + analog_copy->meaning = meaning_copy; + analog_copy->meaning->channels = g_slist_copy( + analog->meaning->channels); + analog_copy->spec = spec_copy; (*copy)->payload = analog_copy; break; default: @@ -1143,7 +1574,7 @@ SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet, return SR_OK; } -void sr_packet_free(struct sr_datafeed_packet *packet) +SR_API void sr_packet_free(struct sr_datafeed_packet *packet) { const struct sr_datafeed_meta *meta; const struct sr_datafeed_logic *logic; @@ -1177,15 +1608,17 @@ void sr_packet_free(struct sr_datafeed_packet *packet) break; case SR_DF_ANALOG: analog = packet->payload; - g_slist_free(analog->channels); g_free(analog->data); + g_free(analog->encoding); + g_slist_free(analog->meaning->channels); + g_free(analog->meaning); + g_free(analog->spec); g_free((void *)packet->payload); break; default: sr_err("Unknown packet type %d", packet->type); } g_free(packet); - } /** @} */