diff options
Diffstat (limited to 'src/ufo-roof-read-task.c')
| -rw-r--r-- | src/ufo-roof-read-task.c | 289 |
1 files changed, 289 insertions, 0 deletions
diff --git a/src/ufo-roof-read-task.c b/src/ufo-roof-read-task.c new file mode 100644 index 0000000..ebff9de --- /dev/null +++ b/src/ufo-roof-read-task.c @@ -0,0 +1,289 @@ +/* + * Copyright (C) 2011-2015 Karlsruhe Institute of Technology + * + * This file is part of Ufo. + * + * This library is free software: you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation, either + * version 3 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library. If not, see <http://www.gnu.org/licenses/>. + */ + +#define _GNU_SOURCE + +#include <stdio.h> + +#ifdef __APPLE__ +#include <OpenCL/cl.h> +#else +#include <CL/cl.h> +#endif + +#include "ufo-roof.h" +#include "ufo-roof-read-socket.h" +#include "ufo-roof-read-file.h" +#include "ufo-roof-read-task.h" + +struct _UfoRoofReadTaskPrivate { + gchar *config; // ROOF configuration file name + UfoRoofConfig *cfg; // Parsed ROOF parameters + UfoRoofReadInterface *reader; + + guint id; // Reader ID (defince sequential port number) + gboolean stop; // Flag requiring termination +}; + +static void ufo_task_interface_init (UfoTaskIface *iface); + +G_DEFINE_TYPE_WITH_CODE (UfoRoofReadTask, ufo_roof_read_task, UFO_TYPE_TASK_NODE, + G_IMPLEMENT_INTERFACE (UFO_TYPE_TASK, + ufo_task_interface_init)) + +#define UFO_ROOF_READ_TASK_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), UFO_TYPE_ROOF_READ_TASK, UfoRoofReadTaskPrivate)) + +enum { + PROP_0, + PROP_ID, + PROP_STOP, + PROP_CONFIG, + N_PROPERTIES +}; + +static GParamSpec *properties[N_PROPERTIES] = { NULL, }; + +UfoNode * +ufo_roof_read_task_new (void) +{ + return UFO_NODE (g_object_new (UFO_TYPE_ROOF_READ_TASK, NULL)); +} + +static void +ufo_roof_read_task_setup (UfoTask *task, + UfoResources *resources, + GError **error) +{ + GError *gerr = NULL; + + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task); + + if (!priv->config) + roof_setup_error(error, "ROOF configuration is not specified"); + + priv->cfg = ufo_roof_config_new(priv->config, &gerr); + if (!priv->cfg) roof_propagate_error(error, gerr, "roof_config_new: "); + + // Consistency checks + if (priv->id >= priv->cfg->n_streams) + roof_setup_error(error, "Specified Stream ID is %u, but only %u data streams is configured", priv->id, priv->cfg->n_streams); + + // Start actual reader + if (priv->cfg->path) + priv->reader = ufo_roof_read_file_new(priv->cfg, priv->id, &gerr); + else + priv->reader = ufo_roof_read_socket_new(priv->cfg, priv->id, &gerr); + + if (!priv->reader) + roof_propagate_error(error, gerr, "roof_read_new: "); + +} + + +static void +ufo_roof_read_task_finalize (GObject *object) +{ + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object); + + if (priv->reader) { + priv->reader->close(priv->reader); + } + + if (priv->cfg) { + ufo_roof_config_free(priv->cfg); + priv->cfg = NULL; + } + + if (priv->config) { + g_free(priv->config); + priv->config = NULL; + } + + G_OBJECT_CLASS (ufo_roof_read_task_parent_class)->finalize (object); +} + +static void +ufo_roof_read_task_get_requisition (UfoTask *task, + UfoBuffer **inputs, + UfoRequisition *requisition, + GError **error) +{ + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task); + UfoRoofConfig *cfg = priv->cfg; + + guint bytes = cfg->max_packets * cfg->max_packet_size + sizeof(UfoRoofPacketBlockHeader); + + // FIXME: Can this be made more elegant? + requisition->n_dims = 1; + requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0); +} + +static guint +ufo_roof_read_task_get_num_inputs (UfoTask *task) +{ + return 0; +} + +static guint +ufo_roof_read_task_get_num_dimensions (UfoTask *task, + guint input) +{ + return 0; +} + +static UfoTaskMode +ufo_roof_read_task_get_mode (UfoTask *task) +{ + return UFO_TASK_MODE_CPU | UFO_TASK_MODE_GENERATOR; +} + + +static gboolean +ufo_roof_read_task_generate (UfoTask *task, + UfoBuffer *output, + UfoRequisition *requisition) +{ + GError *gerr = NULL; + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (task); + UfoRoofConfig *cfg = priv->cfg; + + void *output_buffer = ufo_buffer_get_host_array(output, NULL); + UfoRoofPacketBlockHeader *header = UFO_ROOF_PACKET_BLOCK_HEADER(output_buffer, cfg); + + if (priv->stop) + return FALSE; + + guint packets = priv->reader->read(priv->reader, output_buffer, &gerr); + if (gerr) { + g_warning("Error reciving data: %s", gerr->message); + return FALSE; + } + + // FIXME: End of data (shall we restart in the network case?) + if (!packets) + return FALSE; + + // Shall I use UFO metadata (ufo_buffer_set_metadata) insead? + header->channel_id = priv->id; + header->n_packets = packets; + + return TRUE; +} + +static void +ufo_roof_read_task_set_property (GObject *object, + guint property_id, + const GValue *value, + GParamSpec *pspec) +{ + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object); + + switch (property_id) { + case PROP_CONFIG: + if (priv->config) g_free(priv->config); + priv->config = g_value_dup_string(value); + break; + case PROP_ID: + priv->id = g_value_get_uint (value); + break; + case PROP_STOP: + priv->stop = g_value_get_boolean (value); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec); + break; + } +} + +static void +ufo_roof_read_task_get_property (GObject *object, + guint property_id, + GValue *value, + GParamSpec *pspec) +{ + UfoRoofReadTaskPrivate *priv = UFO_ROOF_READ_TASK_GET_PRIVATE (object); + + switch (property_id) { + case PROP_CONFIG: + g_value_set_string(value, priv->config); + break; + case PROP_ID: + g_value_set_uint (value, priv->id); + break; + case PROP_STOP: + g_value_set_boolean (value, priv->stop); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec); + break; + } +} + +static void +ufo_task_interface_init (UfoTaskIface *iface) +{ + iface->setup = ufo_roof_read_task_setup; + iface->get_num_inputs = ufo_roof_read_task_get_num_inputs; + iface->get_num_dimensions = ufo_roof_read_task_get_num_dimensions; + iface->get_mode = ufo_roof_read_task_get_mode; + iface->get_requisition = ufo_roof_read_task_get_requisition; + iface->generate = ufo_roof_read_task_generate; +} + +static void +ufo_roof_read_task_class_init (UfoRoofReadTaskClass *klass) +{ + GObjectClass *oclass = G_OBJECT_CLASS (klass); + + oclass->set_property = ufo_roof_read_task_set_property; + oclass->get_property = ufo_roof_read_task_get_property; + oclass->finalize = ufo_roof_read_task_finalize; + + properties[PROP_CONFIG] = + g_param_spec_string ("config", + "ROOF configuration", + "Path to ROOF configuration file", + "", + G_PARAM_READWRITE); + + properties[PROP_ID] = + g_param_spec_uint ("id", + "Reader ID", + "ID for multi-port servers", + 0, G_MAXUINT, 1, + G_PARAM_READWRITE); + + properties[PROP_STOP] = + g_param_spec_boolean ("stop", + "Stop flag", + "Stop socket servers and terminates filter execution", + FALSE, + G_PARAM_READWRITE); + + for (guint i = PROP_0 + 1; i < N_PROPERTIES; i++) + g_object_class_install_property (oclass, i, properties[i]); + + g_type_class_add_private (oclass, sizeof(UfoRoofReadTaskPrivate)); +} + +static void +ufo_roof_read_task_init(UfoRoofReadTask *self) +{ + self->priv = UFO_ROOF_READ_TASK_GET_PRIVATE(self); +} |