/* * 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 . */ #include #ifdef __APPLE__ #include #else #include #endif #include "ufo-roof.h" #include "ufo-roof-buffer.h" #include "ufo-roof-build-task.h" struct _UfoRoofBuildTaskPrivate { gchar *config; // ROOF configuration file name UfoRoofConfig *cfg; // Parsed ROOF parameters UfoRoofBuffer *buf; // Ring buffer for incomming UDP packet guint number; // Number of datasets to read gboolean stop; // Stop flag }; static void ufo_task_interface_init (UfoTaskIface *iface); G_DEFINE_TYPE_WITH_CODE (UfoRoofBuildTask, ufo_roof_build_task, UFO_TYPE_TASK_NODE, G_IMPLEMENT_INTERFACE (UFO_TYPE_TASK, ufo_task_interface_init)) #define UFO_ROOF_BUILD_TASK_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), UFO_TYPE_ROOF_BUILD_TASK, UfoRoofBuildTaskPrivate)) enum { PROP_0, PROP_STOP, PROP_NUMBER, PROP_CONFIG, N_PROPERTIES }; static GParamSpec *properties[N_PROPERTIES] = { NULL, }; UfoNode * ufo_roof_build_task_new (void) { return UFO_NODE (g_object_new (UFO_TYPE_ROOF_BUILD_TASK, NULL)); } static void ufo_roof_build_task_setup (UfoTask *task, UfoResources *resources, GError **error) { GError *gerr = NULL; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_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-build-setup: "); priv->buf = ufo_roof_buffer_new(priv->cfg, &gerr); if (!priv->buf) roof_propagate_error(error, gerr, "roof-build-setup: "); } static void ufo_roof_build_task_finalize (GObject *object) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (object); if (priv->buf) { ufo_roof_buffer_free(priv->buf); priv->buf = NULL; } 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_build_task_parent_class)->finalize (object); } static void ufo_roof_build_task_get_requisition (UfoTask *task, UfoBuffer **inputs, UfoRequisition *requisition, GError **error) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); guint bytes = priv->cfg->dataset_size; // 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_build_task_get_num_inputs (UfoTask *task) { return 1; } static guint ufo_roof_build_task_get_num_dimensions (UfoTask *task, guint input) { return 1; } static UfoTaskMode ufo_roof_build_task_get_mode (UfoTask *task) { return UFO_TASK_MODE_CPU | UFO_TASK_MODE_REDUCTOR; } static gboolean ufo_roof_build_task_process (UfoTask *task, UfoBuffer **inputs, UfoBuffer *output, UfoRequisition *requisition) { GError *gerr = NULL; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); UfoRoofConfig *cfg = priv->cfg; UfoRoofBuffer *buf = priv->buf; gboolean ready = FALSE; // UfoRequisition in_req; // ufo_buffer_get_requisition (inputs[0], &in_req); uint8_t *data = (uint8_t*)ufo_buffer_get_host_array(inputs[0], NULL); UfoRoofPacketBlockHeader *header = UFO_ROOF_PACKET_BLOCK_HEADER(data, cfg); if (priv->stop) return FALSE; for (guint i = 0; i < header->n_packets; i++) { guint packet_id = 0; // Otherwise considered consecutive and handled by the buffer if (cfg->header_size >= sizeof(UfoRoofPacketHeader)) { UfoRoofPacketHeader *pheader = UFO_ROOF_PACKET_HEADER(data); packet_id = pheader->packet_id + 1; } ready |= ufo_roof_buffer_set_fragment(buf, header->channel_id, packet_id, data, &gerr); if (gerr) roof_print_error(gerr); data += cfg->max_packet_size; } // FIXME: if 2nd dataset is ready (2nd and 3rd?), skip the first one? // printf("proc (%s) - channel: %i, packets: %i\n", ready?"yes":" no", header->channel_id, header->n_packets); return !ready; } static gboolean ufo_roof_build_task_generate (UfoTask *task, UfoBuffer *output, UfoRequisition *requisition) { gboolean ready = FALSE; GError *gerr = NULL; UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); // UfoRoofConfig *cfg = priv->cfg; UfoRoofBuffer *buf = priv->buf; void *output_buffer = ufo_buffer_get_host_array(output, NULL); if (priv->stop) return FALSE; ready = ufo_roof_buffer_get_dataset(buf, output_buffer, &gerr); if (gerr) roof_print_error(gerr); if ((priv->number)&&(buf->current_id >= priv->number)) { priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } // printf("gen(%s) %i\n", ready?"yes":" no", buf->current_id); return ready; } static void ufo_roof_build_task_set_property (GObject *object, guint property_id, const GValue *value, GParamSpec *pspec) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_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_STOP: priv->stop = g_value_get_boolean (value); break; case PROP_NUMBER: priv->number = g_value_get_uint (value); break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec); break; } } static void ufo_roof_build_task_get_property (GObject *object, guint property_id, GValue *value, GParamSpec *pspec) { UfoRoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (object); switch (property_id) { case PROP_CONFIG: g_value_set_string(value, priv->config); break; case PROP_STOP: g_value_set_boolean (value, priv->stop); break; case PROP_NUMBER: g_value_set_uint (value, priv->number); 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_build_task_setup; iface->get_num_inputs = ufo_roof_build_task_get_num_inputs; iface->get_num_dimensions = ufo_roof_build_task_get_num_dimensions; iface->get_mode = ufo_roof_build_task_get_mode; iface->get_requisition = ufo_roof_build_task_get_requisition; iface->process = ufo_roof_build_task_process; iface->generate = ufo_roof_build_task_generate; } static void ufo_roof_build_task_class_init (UfoRoofBuildTaskClass *klass) { GObjectClass *oclass = G_OBJECT_CLASS (klass); oclass->set_property = ufo_roof_build_task_set_property; oclass->get_property = ufo_roof_build_task_get_property; oclass->finalize = ufo_roof_build_task_finalize; properties[PROP_CONFIG] = g_param_spec_string ("config", "ROOF configuration", "Path to ROOF configuration file", "", G_PARAM_READWRITE); properties[PROP_STOP] = g_param_spec_boolean ("stop", "Stop flag", "Stop socket servers and terminates filter execution", FALSE, G_PARAM_READWRITE); properties[PROP_NUMBER] = g_param_spec_uint("number", "Number of datasets to receive", "Number of datasets to receive", 0, G_MAXUINT, 0, 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(UfoRoofBuildTaskPrivate)); } static void ufo_roof_build_task_init(UfoRoofBuildTask *self) { self->priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE(self); }