X-Git-Url: https://sigrok.org/gitweb/?a=blobdiff_plain;f=src%2Fsession.c;h=cc5f09dccb607910d3cb4ef62789203114bf6aaf;hb=c2bf5506ee7864975917a7b6e7e93b78226887ce;hp=d8a3b00b5efabe157ab546498bee39139b249853;hpb=155b680da482cea2381becb73c51cfb838bff31e;p=libsigrok.git diff --git a/src/session.c b/src/session.c index d8a3b00b..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 + * 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,12 +18,13 @@ * along with this program. If not, see . */ +#include #include #include #include #include #include -#include "libsigrok.h" +#include #include "libsigrok-internal.h" /** @cond PRIVATE */ @@ -43,41 +45,194 @@ * @{ */ -struct source { - int timeout; - sr_receive_data_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; -}; - struct datafeed_callback { sr_datafeed_callback cb; void *cb_data; }; +/** Custom GLib event source for generic descriptor I/O. + * @internal + */ +struct fd_source { + GSource base; + + int64_t timeout_us; + int64_t due_us; + + /* Meta-data needed to keep track of installed sources */ + struct sr_session *session; + void *key; + + 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. - * Currently, there can be only one session at a time within the same process. + * + * @param ctx The context in which to create the new session. + * @param new_session This will contain a pointer to the newly created + * session if the return value is SR_OK, otherwise the value + * is undefined and should not be used. Must not be NULL. * * @retval SR_OK Success. - * @retval SR_ERR_BUG A session exists already. + * @retval SR_ERR_ARG Invalid argument. * * @since 0.4.0 */ -SR_API int sr_session_new(struct sr_session **new_session) +SR_API int sr_session_new(struct sr_context *ctx, + struct sr_session **new_session) { struct sr_session *session; + if (!new_session) + return SR_ERR_ARG; + session = g_malloc0(sizeof(struct sr_session)); - session->source_timeout = -1; - session->running = FALSE; - session->abort_session = FALSE; - g_mutex_init(&session->stop_mutex); + session->ctx = ctx; + + 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; @@ -88,6 +243,8 @@ SR_API int sr_session_new(struct sr_session **new_session) * Destroy a session. * This frees up all memory used by the session. * + * @param session The session to destroy. Must not be NULL. + * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. * @@ -101,9 +258,11 @@ 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); + + g_hash_table_unref(session->event_sources); + + g_mutex_clear(&session->main_mutex); g_free(session); @@ -116,6 +275,8 @@ SR_API int sr_session_destroy(struct sr_session *session) * The session itself (i.e., the struct sr_session) is not free'd and still * exists after this function returns. * + * @param session The session to use. Must not be NULL. + * * @retval SR_OK Success. * @retval SR_ERR_BUG Invalid session passed. * @@ -145,6 +306,7 @@ SR_API int sr_session_dev_remove_all(struct sr_session *session) /** * Add a device instance to a session. * + * @param session The session to add to. Must not be NULL. * @param sdi The device instance to add to a session. Must not * be NULL. Also, sdi->driver and sdi->driver->dev_open must * not be NULL. @@ -178,8 +340,6 @@ SR_API int sr_session_dev_add(struct sr_session *session, /* If sdi->driver is NULL, this is a virtual device. */ if (!sdi->driver) { - sr_dbg("%s: sdi->driver was NULL, this seems to be " - "a virtual device; continuing", __func__); /* Just add the device, don't run dev_open(). */ session->devs = g_slist_append(session->devs, (gpointer)sdi); sdi->session = session; @@ -218,6 +378,7 @@ SR_API int sr_session_dev_add(struct sr_session *session, /** * List all device instances attached to a session. * + * @param session The session to use. Must not be NULL. * @param devlist A pointer where the device instance list will be * stored on return. If no devices are in the session, * this will be NULL. Each element in the list points @@ -246,6 +407,8 @@ SR_API int sr_session_dev_list(struct sr_session *session, GSList **devlist) /** * Remove all datafeed callbacks in a session. * + * @param session The session to use. Must not be NULL. + * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. * @@ -267,6 +430,7 @@ SR_API int sr_session_datafeed_callback_remove_all(struct sr_session *session) /** * Add a datafeed callback to a session. * + * @param session The session to use. Must not be NULL. * @param cb Function to call when a chunk of data is received. * Must not be NULL. * @param cb_data Opaque pointer passed in by the caller. @@ -291,9 +455,7 @@ SR_API int sr_session_datafeed_callback_add(struct sr_session *session, return SR_ERR_ARG; } - if (!(cb_struct = g_try_malloc0(sizeof(struct datafeed_callback)))) - return SR_ERR_MALLOC; - + cb_struct = g_malloc0(sizeof(struct datafeed_callback)); cb_struct->cb = cb; cb_struct->cb_data = cb_data; @@ -303,75 +465,45 @@ SR_API int sr_session_datafeed_callback_add(struct sr_session *session, return SR_OK; } +/** + * Get the trigger assigned to this session. + * + * @param session The session to use. + * + * @retval NULL Invalid (NULL) session was passed to the function. + * @retval other The trigger assigned to this session (can be NULL). + * + * @since 0.4.0 + */ SR_API struct sr_trigger *sr_session_trigger_get(struct sr_session *session) { - return session->trigger; -} - -SR_API int sr_session_trigger_set(struct sr_session *session, struct sr_trigger *trig) -{ - session->trigger = trig; + if (!session) + return NULL; - return SR_OK; + return session->trigger; } /** - * 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. + * Set the trigger of this session. * - * @param block If TRUE, this call will wait for any of the session's - * sources to fire an event on the file descriptors, or - * any of their timeouts to activate. In other words, this - * can be used as a select loop. - * If FALSE, all sources have their callback run, regardless - * of file descriptor or timeout status. + * @param session The session to use. Must not be NULL. + * @param trig The trigger to assign to this session. Can be NULL. * * @retval SR_OK Success. - * @retval SR_ERR Error occured. + * @retval SR_ERR_ARG Invalid argument. + * + * @since 0.4.0 */ -static int sr_session_iteration(struct sr_session *session, gboolean block) +SR_API int sr_session_trigger_set(struct sr_session *session, struct sr_trigger *trig) { - unsigned int i; - int ret; + if (!session) + return SR_ERR_ARG; - ret = g_poll(session->pollfds, session->num_sources, - block ? session->source_timeout : 0); - for (i = 0; i < session->num_sources; i++) { - if (session->pollfds[i].revents > 0 || (ret == 0 - && session->source_timeout == session->sources[i].timeout)) { - /* - * Invoke the source's callback on an event, - * or if the poll timed out and this source - * asked for that timeout. - */ - if (!session->sources[i].cb(session->pollfds[i].fd, - session->pollfds[i].revents, - session->sources[i].cb_data)) - sr_session_source_remove(session, - session->sources[i].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. - */ - g_mutex_lock(&session->stop_mutex); - if (session->abort_session) { - sr_session_stop_sync(session); - /* But once is enough. */ - session->abort_session = FALSE; - } - g_mutex_unlock(&session->stop_mutex); - } + session->trigger = trig; return SR_OK; } - static int verify_trigger(struct sr_trigger *trigger) { struct sr_trigger_stage *stage; @@ -407,9 +539,91 @@ 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. * + * @param session The session to use. Must not be NULL. + * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. * @@ -418,8 +632,9 @@ static int verify_trigger(struct sr_trigger *trigger) SR_API int sr_session_start(struct sr_session *session) { struct sr_dev_inst *sdi; - GSList *l; - int ret; + struct sr_channel *ch; + GSList *l, *c; + int enabled_channels, ret; if (!session) { sr_err("%s: session was NULL", __func__); @@ -435,11 +650,31 @@ 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; + for (c = sdi->channels; c; c = c->next) { + ch = c->data; + if (ch->enabled) { + enabled_channels++; + 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 ((ret = sr_config_commit(sdi)) != SR_OK) { sr_err("Failed to commit device settings before " "starting acquisition (%s)", sr_strerror(ret)); @@ -452,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; @@ -460,8 +699,11 @@ SR_API int sr_session_start(struct sr_session *session) /** * Run a session. * + * @param session The session to use. Must not be NULL. + * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid session passed. + * @retval SR_ERR Error during event processing. * * @since 0.4.0 */ @@ -471,67 +713,61 @@ 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; } - 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."); - /* 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, TRUE); - } + 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. - * - * @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; } /** @@ -545,8 +781,11 @@ SR_PRIV int sr_session_stop_sync(struct sr_session *session) * to wait for the session thread to return before assuming that the session is * completely decommissioned. * + * @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 */ @@ -554,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); } /** @@ -573,17 +817,22 @@ 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) { case SR_DF_HEADER: sr_dbg("bus: Received SR_DF_HEADER packet."); break; - case SR_DF_TRIGGER: - sr_dbg("bus: Received SR_DF_TRIGGER packet."); + case SR_DF_END: + sr_dbg("bus: Received SR_DF_END packet."); break; case SR_DF_META: sr_dbg("bus: Received SR_DF_META packet."); break; + case SR_DF_TRIGGER: + sr_dbg("bus: Received SR_DF_TRIGGER packet."); + break; case SR_DF_LOGIC: logic = packet->payload; sr_dbg("bus: Received SR_DF_LOGIC packet (%" PRIu64 " bytes, " @@ -594,15 +843,17 @@ static void datafeed_dump(const struct sr_datafeed_packet *packet) sr_dbg("bus: Received SR_DF_ANALOG packet (%d samples).", analog->num_samples); break; - case SR_DF_END: - sr_dbg("bus: Received SR_DF_END packet."); - 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); + break; default: sr_dbg("bus: Received unknown packet type: %d.", packet->type); break; @@ -623,10 +874,13 @@ static void datafeed_dump(const struct sr_datafeed_packet *packet) * @private */ SR_PRIV int sr_session_send(const struct sr_dev_inst *sdi, - const struct sr_datafeed_packet *packet) + const struct sr_datafeed_packet *packet) { GSList *l; struct datafeed_callback *cb_struct; + struct sr_datafeed_packet *packet_in, *packet_out; + struct sr_transform *t; + int ret; if (!sdi) { sr_err("%s: sdi was NULL", __func__); @@ -638,6 +892,46 @@ SR_PRIV int sr_session_send(const struct sr_dev_inst *sdi, return SR_ERR_ARG; } + if (!sdi->session) { + sr_err("%s: session was NULL", __func__); + return SR_ERR_BUG; + } + + /* + * Pass the packet to the first transform module. If that returns + * another packet (instead of NULL), pass that packet to the next + * transform module in the list, and so on. + */ + packet_in = (struct sr_datafeed_packet *)packet; + for (l = sdi->session->transforms; l; l = l->next) { + t = l->data; + sr_spew("Running transform module '%s'.", t->module->id); + ret = t->module->receive(t, packet_in, &packet_out); + if (ret < 0) { + sr_err("Error while running transform module: %d.", ret); + return SR_ERR; + } + if (!packet_out) { + /* + * If any of the transforms don't return an output + * packet, abort. + */ + sr_spew("Transform module didn't return a packet, aborting."); + return SR_OK; + } else { + /* + * Use this transform module's output packet as input + * for the next transform module. + */ + packet_in = packet_out; + } + } + packet = packet_in; + + /* + * If the last transform did output a packet, pass it to all datafeed + * callbacks. + */ for (l = sdi->session->datafeed_callbacks; l; l = l->next) { if (sr_log_loglevel_get() >= SR_LOG_DBG) datafeed_dump(packet); @@ -651,99 +945,104 @@ SR_PRIV int sr_session_send(const struct sr_dev_inst *sdi, /** * Add an event source for a file descriptor. * - * @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 TODO. - * + * @param session The session to use. Must not be NULL. + * @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_MALLOC Memory allocation error. + * @retval SR_ERR_BUG Event source with @a key already installed. + * @retval SR_ERR Other error. */ -static int _sr_session_source_add(struct sr_session *session, GPollFD *pollfd, - int timeout, sr_receive_data_callback cb, void *cb_data, gintptr poll_object) +SR_PRIV int sr_session_source_add_internal(struct sr_session *session, + 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; + 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.) + */ + 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. */ + g_mutex_lock(&session->main_mutex); - new_pollfds = g_try_realloc(session->pollfds, - sizeof(GPollFD) * (session->num_sources + 1)); - if (!new_pollfds) { - sr_err("%s: new_pollfds malloc failed", __func__); - return SR_ERR_MALLOC; + if (session->main_context) { + if (g_source_attach(source, session->main_context) > 0) + ret = SR_OK; + else + ret = SR_ERR; + } else { + sr_err("Cannot add event source without main context."); + ret = SR_ERR; } + g_mutex_unlock(&session->main_mutex); - new_sources = g_try_realloc(session->sources, sizeof(struct source) * - (session->num_sources + 1)); - if (!new_sources) { - sr_err("%s: new_sources malloc failed", __func__); - return SR_ERR_MALLOC; - } + return ret; +} - new_pollfds[session->num_sources] = *pollfd; - s = &new_sources[session->num_sources++]; - s->timeout = timeout; - s->cb = cb; - s->cb_data = cb_data; - s->poll_object = poll_object; - session->pollfds = new_pollfds; - session->sources = new_sources; +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) +{ + GSource *source; + int ret; - if (timeout != session->source_timeout && timeout > 0 - && (session->source_timeout == -1 || timeout < session->source_timeout)) - session->source_timeout = timeout; + source = fd_source_new(session, key, fd, events, timeout); + if (!source) + return SR_ERR; - return SR_OK; + g_source_set_callback(source, (GSourceFunc)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 fd The file descriptor. + * @param session The session to use. Must not be NULL. + * @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. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid argument. - * @retval SR_ERR_MALLOC Memory allocation error. * * @since 0.3.0 */ 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) { - GPollFD p; - - (void) session; - - p.fd = fd; - p.events = events; - - return _sr_session_source_add(session, &p, timeout, cb, cb_data, (gintptr)fd); + if (fd < 0 && timeout < 0) { + sr_err("Cannot create timer source without timeout."); + return SR_ERR_ARG; + } + return attach_fd_source(session, GINT_TO_POINTER(fd), + fd, events, timeout, cb, cb_data); } /** * Add an event source for a GPollFD. * - * @param pollfd The GPollFD. - * @param timeout Max time to wait before the callback is called, ignored if 0. + * @param session The session to use. Must not be NULL. + * @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. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid argument. - * @retval SR_ERR_MALLOC Memory allocation error. * * @since 0.3.0 */ @@ -751,24 +1050,27 @@ SR_API int sr_session_source_add_pollfd(struct sr_session *session, GPollFD *pollfd, int timeout, sr_receive_data_callback cb, void *cb_data) { - (void) session; - - return _sr_session_source_add(session, pollfd, timeout, cb, - cb_data, (gintptr)pollfd); + if (!pollfd) { + sr_err("%s: pollfd was NULL", __func__); + return SR_ERR_ARG; + } + return attach_fd_source(session, pollfd, pollfd->fd, + pollfd->events, timeout, cb, cb_data); } /** * Add an event source for a GIOChannel. * + * @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. * * @retval SR_OK Success. * @retval SR_ERR_ARG Invalid argument. - * @retval SR_ERR_MALLOC Memory allocation error. * * @since 0.3.0 */ @@ -776,75 +1078,49 @@ 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) { - GPollFD p; - - (void) session; + GPollFD pollfd; -#ifdef _WIN32 - g_io_channel_win32_make_pollfd(channel, events, &p); + 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 + g_io_channel_win32_make_pollfd(channel, events, &pollfd); #else - p.fd = g_io_channel_unix_get_fd(channel); - p.events = events; + pollfd.fd = g_io_channel_unix_get_fd(channel); + pollfd.events = events; #endif - - return _sr_session_source_add(session, &p, timeout, cb, cb_data, (gintptr)channel); + return attach_fd_source(session, channel, pollfd.fd, + pollfd.events, timeout, cb, cb_data); } /** - * Remove the source belonging to the specified channel. - * - * @todo Add more error checks and logging. + * Remove the source identified by the specified poll object. * - * @param poll_object The channel for which the source should be removed. + * @param session The session to use. Must not be NULL. + * @param key The key by which the source is identified. * * @retval SR_OK Success - * @retval SR_ERR_ARG Invalid arguments - * @retval SR_ERR_MALLOC Memory allocation error - * @retval SR_ERR_BUG Internal error + * @retval SR_ERR_BUG No event source for poll_object found. */ -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) { - struct source *new_sources; - GPollFD *new_pollfds; - 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 -= 1; - - 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)); - } - - new_pollfds = g_try_realloc(session->pollfds, sizeof(GPollFD) * session->num_sources); - if (!new_pollfds && session->num_sources > 0) { - sr_err("%s: new_pollfds malloc failed", __func__); - return SR_ERR_MALLOC; - } - - new_sources = g_try_realloc(session->sources, sizeof(struct source) * session->num_sources); - if (!new_sources && session->num_sources > 0) { - sr_err("%s: new_sources malloc failed", __func__); - return SR_ERR_MALLOC; - } - - session->pollfds = new_pollfds; - session->sources = new_sources; + g_source_destroy(source); return SR_OK; } @@ -852,27 +1128,26 @@ static int _sr_session_source_remove(struct sr_session *session, gintptr poll_ob /** * Remove the source belonging to the specified file descriptor. * + * @param session The session to use. Must not be NULL. * @param fd The file descriptor for which the source should be removed. * * @retval SR_OK Success * @retval SR_ERR_ARG Invalid argument - * @retval SR_ERR_MALLOC Memory allocation error. * @retval SR_ERR_BUG Internal error. * * @since 0.3.0 */ SR_API int sr_session_source_remove(struct sr_session *session, int fd) { - (void) session; - - return _sr_session_source_remove(session, (gintptr)fd); + return sr_session_source_remove_internal(session, GINT_TO_POINTER(fd)); } /** * Remove the source belonging to the specified poll descriptor. * + * @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. @@ -882,19 +1157,21 @@ SR_API int sr_session_source_remove(struct sr_session *session, int fd) SR_API int sr_session_source_remove_pollfd(struct sr_session *session, GPollFD *pollfd) { - (void) session; - - 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); } /** * Remove the source belonging to the specified channel. * + * @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. - * @retval SR_ERR_MALLOC Memory allocation error. * @return SR_ERR_BUG Internal error. * * @since 0.2.0 @@ -902,9 +1179,166 @@ SR_API int sr_session_source_remove_pollfd(struct sr_session *session, SR_API int sr_session_source_remove_channel(struct sr_session *session, GIOChannel *channel) { - (void) session; + 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. + */ +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) +{ + g_variant_ref(src->data); + meta_copy->config = g_slist_append(meta_copy->config, + g_memdup(src, sizeof(struct sr_config))); +} + +SR_PRIV int sr_packet_copy(const struct sr_datafeed_packet *packet, + struct sr_datafeed_packet **copy) +{ + const struct sr_datafeed_meta *meta; + struct sr_datafeed_meta *meta_copy; + const struct sr_datafeed_logic *logic; + struct sr_datafeed_logic *logic_copy; + const struct sr_datafeed_analog *analog; + struct sr_datafeed_analog *analog_copy; + uint8_t *payload; + + *copy = g_malloc0(sizeof(struct sr_datafeed_packet)); + (*copy)->type = packet->type; + + switch (packet->type) { + case SR_DF_TRIGGER: + case SR_DF_END: + /* No payload. */ + break; + case SR_DF_HEADER: + payload = g_malloc(sizeof(struct sr_datafeed_header)); + memcpy(payload, packet->payload, sizeof(struct sr_datafeed_header)); + (*copy)->payload = payload; + break; + case SR_DF_META: + meta = packet->payload; + meta_copy = g_malloc0(sizeof(struct sr_datafeed_meta)); + g_slist_foreach(meta->config, (GFunc)copy_src, meta_copy->config); + (*copy)->payload = meta_copy; + break; + case SR_DF_LOGIC: + logic = packet->payload; + logic_copy = g_malloc(sizeof(logic)); + logic_copy->length = logic->length; + logic_copy->unitsize = logic->unitsize; + 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; + memcpy(analog_copy->data, analog->data, + analog->num_samples * sizeof(float)); + (*copy)->payload = analog_copy; + break; + default: + sr_err("Unknown packet type %d", packet->type); + return SR_ERR; + } + + return SR_OK; +} + +void sr_packet_free(struct sr_datafeed_packet *packet) +{ + const struct sr_datafeed_meta *meta; + const struct sr_datafeed_logic *logic; + const struct sr_datafeed_analog *analog; + struct sr_config *src; + GSList *l; + + switch (packet->type) { + case SR_DF_TRIGGER: + case SR_DF_END: + /* No payload. */ + break; + case SR_DF_HEADER: + /* Payload is a simple struct. */ + g_free((void *)packet->payload); + break; + case SR_DF_META: + meta = packet->payload; + for (l = meta->config; l; l = l->next) { + src = l->data; + g_variant_unref(src->data); + g_free(src); + } + g_slist_free(meta->config); + g_free((void *)packet->payload); + break; + case SR_DF_LOGIC: + logic = packet->payload; + g_free(logic->data); + g_free((void *)packet->payload); + break; + case SR_DF_ANALOG: + analog = packet->payload; + g_slist_free(analog->channels); + g_free(analog->data); + g_free((void *)packet->payload); + break; + default: + sr_err("Unknown packet type %d", packet->type); + } + g_free(packet); - return _sr_session_source_remove(session, (gintptr)channel); } /** @} */