/* * 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 guint announced; // For debugging struct timespec last_fragment_timestamp; }; 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, priv->number, &gerr); if (!priv->buf) roof_propagate_error(error, gerr, "roof-build-setup: "); clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp); } 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); const 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; const uint8_t *fragment = data; 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(fragment); packet_id = pheader->packet_id + 1; } // FIXME: Can we kill here the dataset finished during the previous step of iteration ready |= ufo_roof_buffer_set_fragment(buf, header->channel_id, packet_id, fragment + cfg->header_size, &gerr); if (gerr) roof_print_error(gerr); fragment += cfg->max_packet_size; } // FIXME: if 2nd dataset is ready (2nd and 3rd?), skip the first one? if (ready) { clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp); } else { struct timespec current_time; clock_gettime(CLOCK_REALTIME, ¤t_time); // No new accepted events within timeout if (((current_time.tv_sec - priv->last_fragment_timestamp.tv_sec) * 1000000 + (current_time.tv_nsec - priv->last_fragment_timestamp.tv_nsec) / 1000) > cfg->network_timeout) { priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } } /* printf("proc (%s - %u of %u) - channel: %i, packets: %i, first dataset: %i\n", ready?"yes":" no", buf->n_fragments[(buf->current_id)%buf->ring_size], buf->fragments_per_dataset, header->channel_id, header->n_packets, (cfg->header_size >= sizeof(UfoRoofPacketHeader))?UFO_ROOF_PACKET_HEADER(data)->packet_id / (cfg->dataset_size / cfg->payload_size / cfg->n_streams):0); */ 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); // FIXME: Or shall we start from counting from the ID of the first registerd dataset if ((priv->number)&&(buf->current_id >= priv->number)) { // printf("%u datasets processed, stopping\n", buf->current_id); priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } if ((priv->number < 100)||((buf->current_id - priv->announced) > 1000)) { printf("Generating dataset %i (%s), next: %u out of %u)\n", buf->current_id, ready?"yes":" no", buf->n_fragments[buf->current_id%buf->ring_size], buf->fragments_per_dataset); priv->announced = 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); }