/* * 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 #include #include #ifdef __APPLE__ #include #else #include #endif #include "hw_sched.h" #include "ufo-roof.h" #include "ufo-roof-read.h" #include "ufo-roof-buffer.h" #include "ufo-roof-read-socket.h" #include "ufo-roof-read-file.h" #include "ufo-roof-build-task.h" typedef enum { BUILD_AUTO = 0, BUILD_RAW, BUILD_SINO, BUILD_UFO } BuildType; struct _RoofBuildTaskPrivate { gchar *config; // ROOF configuration file name RoofConfig *cfg; // Parsed ROOF parameters // RoofBuffer *buf; // Ring buffer for incomming UDP packet RoofReadInterface **rdi; // Reader interfaces, one per socket (no threading) // RoofRead *rd; // Threading interface HWSched sched; gchar *path; // UFO file path for simulation mode guint first_file_number; // Number of a first simulated file (0 or 1) BuildType build; // What dataset do we build: ROOF sinogram or raw network data guint number; // Number of datasets to read gboolean stop; // Stop flag gboolean simulate; // Indicates if we are running in network or simulation modes guint64 announced; // For debugging guint64 generated; // Total number for control guint n_threads; // Number of schedulled threads struct timespec last_fragment_timestamp; }; static void ufo_task_interface_init (UfoTaskIface *iface); G_DEFINE_TYPE_WITH_CODE (RoofBuildTask, 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, RoofBuildTaskPrivate)) static GEnumValue build_values[] = { { BUILD_AUTO, "BUILD_AUTO", "auto" }, { BUILD_RAW, "BUILD_RAW", "raw" }, { BUILD_SINO, "BUILD_SINO", "sino" }, { BUILD_UFO, "BUILD_UFO", "ufo" }, { 0, NULL, NULL} }; enum { PROP_0, PROP_STOP, PROP_SIMULATE, PROP_PATH, PROP_FIRST, PROP_NUMBER, PROP_BUILD, 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) { guint i; GError *gerr = NULL; RoofBuildTaskPrivate *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, priv->simulate?UFO_ROOF_CONFIG_SIMULATION:UFO_ROOF_CONFIG_DEFAULT, &gerr); if (!priv->cfg) roof_propagate_error(error, gerr, "roof-build-setup: "); priv->rdi = (RoofReadInterface**)calloc(priv->cfg->n_streams, sizeof(RoofReadInterface*)); if (!priv->rdi) roof_setup_error(error, "Failed to allocate memory for RoofReadInterface array"); for (i = 0; i < priv->cfg->n_streams; i++) { if (priv->simulate) { if (!priv->path) roof_setup_error(error, "Path to simulated data should be specified"); priv->rdi[i] = ufo_roof_read_file_new(priv->cfg, priv->path, priv->first_file_number + i, &gerr); } else priv->rdi[i] = ufo_roof_read_socket_new(priv->cfg, i, &gerr); if (!priv->rdi[i]) roof_propagate_error(error, gerr, "roof_read_interface_new: "); priv->rdc[i] = ufo_roof_read_context_new(priv->cfg, priv->rdi[i], &gerr); if (!priv->rdc[i]) roof_propagate_error(error, gerr, "roof_read_context_new: "); } // We try to distribute sockets uniformly respecting sockets_per_thread as maximum limit priv->n_threads = priv->cfg->n_streams / priv->cfg->sockets_per_thread; if (priv->cfg->n_streams % priv->cfg->sockets_per_thread) priv->n_threads++; guint extra = 0, sockets_per_thread = priv->cfg->n_streams / priv->n_threads; if (priv->cfg->n_streams % priv->n_threads) extra = priv->cfg->n_streams - priv->n_threads * sockets_per_thread; guint from, to; for (i = 0; i < priv->n_threads; i++) { guint to = from + sockets_per_thread; if (i < extra) to++; ctx->thr[i]= ufo_roof_thread_new(priv->cfg, priv->rdc, from, to, &gerr); if (!ctx->thr[i]) roof_propagate_error(error, gerr, "ufo_roof_thread_new (%i): ", i); } if (priv->build == BUILD_AUTO) { if (priv->cfg->roof_mode) priv->build = BUILD_SINO; else priv->build = BUILD_RAW; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_BUILD]); } /* priv->buf = ufo_roof_buffer_new(priv->cfg, (priv->build == BUILD_RAW)?1:2, priv->number, &gerr); if ((gerr)||(!priv->buf)) roof_propagate_error(error, gerr, "roof_buffer_new: "); priv->rd = ufo_roof_read_new(priv->cfg, priv->rdi, priv->buf, &gerr); if (gerr) roof_propagate_error(error, gerr, "roof_read_new: "); */ priv->sched = hw_sched_create(priv->cfg->n_read_threads); if (!priv->sched) roof_setup_error(error, "Failed to schedule builder threads"); clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp); } static void ufo_roof_build_task_finalize (GObject *object) { RoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (object); if (priv->sched) { hw_sched_destroy(priv->sched); priv->sched = NULL; } /* if (priv->rd) { ufo_roof_read_free(priv->rd); priv->rd = NULL; } if (priv->buf) { ufo_roof_buffer_free(priv->buf); priv->buf = NULL; } */ if (priv->rdi) { guint i; for (i = 0; i < priv->cfg->n_streams; i++) { if (priv->rdi[i]) priv->rdi[i]->close(priv->rdi[i]); } free(priv->rdi); priv->rdi = NULL; } if (priv->cfg) { ufo_roof_config_free(priv->cfg); priv->cfg = NULL; } if (priv->config) { g_free(priv->config); priv->config = NULL; } if (priv->path) { g_free(priv->path); priv->path = 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) { RoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); // FIXME: Can we handle data types more elegant? // FIXME: Kill BUILD_RAW ? if (priv->build == BUILD_RAW) { guint bytes = priv->cfg->dataset_size; requisition->n_dims = 1; requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0); } else if (priv->build == BUILD_SINO) { guint bytes = priv->cfg->fan_bins * priv->cfg->bit_depth / 8; requisition->n_dims = 2; requisition->dims[0] = bytes / sizeof(float) + ((bytes%sizeof(float))?1:0); requisition->dims[1] = priv->cfg->fan_projections; } else if (priv->build == BUILD_UFO) { requisition->n_dims = 2; requisition->dims[0] = priv->cfg->fan_bins; requisition->dims[1] = priv->cfg->fan_projections; } } 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_GENERATOR; } static gboolean ufo_roof_build_task_generate (UfoTask *task, UfoBuffer *output, UfoRequisition *requisition) { gboolean ready = FALSE; gulong seqid; GError *gerr = NULL; GValue ival = G_VALUE_INIT; GValue lval = G_VALUE_INIT; RoofBuildTaskPrivate *priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE (task); RoofConfig *cfg = priv->cfg; RoofBuffer *buf = priv->buf; void *output_buffer = ufo_buffer_get_host_array(output, NULL); if (priv->stop) return FALSE; priv->current_dataset; priv->current_buffer = output_buffer; err = hw_sched_schedule_thread_task(sched, (void*)&tnv_ctx, ufo_roof_build_task_read); if (!err) err = hw_sched_wait_task(sched); if (err) { fprintf(stderr, "Error %i scheduling init threads", err); exit(-1); } /* // FIXME: Wait or break. Not both. do { ready = ufo_roof_buffer_wait_dataset(buf, output_buffer, &seqid, cfg->network_timeout, &gerr); if (gerr) roof_print_error(gerr); if (!ready) { ready = ufo_roof_buffer_skip_to_ready(buf); if (!ready) { priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); return FALSE; } } } while (!ready); */ // FIXME: integrate fastwriter somewhere here? if (priv->build == BUILD_UFO) { switch (cfg->bit_depth) { case 8: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_8U); break; case 16: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_16U); break; case 32: ufo_buffer_convert(output, UFO_BUFFER_DEPTH_32U); break; default: printf("Usupported bit-depth %u\n", cfg->bit_depth); } } // Metadata: plane and sequential number within the plane g_value_init (&ival, G_TYPE_UINT); g_value_init (&lval, G_TYPE_ULONG); if (priv->build != BUILD_UFO) { g_value_set_uint (&ival, cfg->bit_depth); ufo_buffer_set_metadata (output, "bpp", &ival); } g_value_set_uint (&ival, 1 + seqid % cfg->n_planes); ufo_buffer_set_metadata (output, "plane", &ival); g_value_set_ulong (&lval, seqid / cfg->n_planes); ufo_buffer_set_metadata (output, "plane_id", &lval); g_value_set_ulong (&lval, seqid); ufo_buffer_set_metadata (output, "seqid", &lval); g_value_unset(&lval); g_value_unset(&ival); // FIXME: Or shall we start from counting from the ID of the first registerd dataset if ((priv->number)&&(buf->current_id >= priv->number)) { // printf("%lu datasets processed, stopping\n", buf->current_id); priv->stop = TRUE; g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]); } if (ready) priv->generated++; if (((priv->number > 0)&&(priv->number <= 100))||((buf->current_id - priv->announced) > 1000)) { if (ready) printf("Processing dataset %li (ready ), next : %u out of %u\n", buf->current_id, buf->n_fragments[buf->current_id%buf->ring_size], buf->fragments_per_dataset); else printf("Skipping dataset %li (timeout), acquired: %u out of %u\n", buf->current_id + 1, 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) { RoofBuildTaskPrivate *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_SIMULATE: priv->simulate = g_value_get_boolean (value); break; case PROP_PATH: if (priv->path) g_free(priv->path); priv->path = g_value_dup_string(value); break; case PROP_FIRST: priv->first_file_number = g_value_get_uint (value); break; case PROP_NUMBER: priv->number = g_value_get_uint (value); break; case PROP_BUILD: priv->build = g_value_get_enum (value); if ((priv->build == BUILD_AUTO)&&(priv->cfg)) { if (priv->cfg->roof_mode) priv->build = BUILD_SINO; else priv->build = BUILD_RAW; } 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) { RoofBuildTaskPrivate *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_SIMULATE: g_value_set_boolean (value, priv->simulate); break; case PROP_PATH: g_value_set_string(value, priv->path?priv->path:""); break; case PROP_FIRST: g_value_set_uint (value, priv->first_file_number); break; case PROP_NUMBER: g_value_set_uint (value, priv->number); break; case PROP_BUILD: g_value_set_enum (value, priv->build); break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec); break; } } static void ufo_task_interface_init (UfoTaskIface *iface) { roof_init(); 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->generate = ufo_roof_build_task_generate; } static void ufo_roof_build_task_class_init (RoofBuildTaskClass *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_SIMULATE] = g_param_spec_boolean ("simulate", "Simulation mode", "Read data from the specified files instead of network", FALSE, G_PARAM_READWRITE); properties[PROP_PATH] = g_param_spec_string ("path", "Input files for simulation mode", "Optional path to input files for simulation mode (parameter from configuration file is used if not specified)", "", G_PARAM_READWRITE); properties[PROP_FIRST] = g_param_spec_uint ("first_file_number", "Offset to the first read file", "Offset to the first read file", 0, G_MAXUINT, 0, 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); properties[PROP_BUILD] = g_param_spec_enum ("build", "Build type (\"raw\", \"sino\", \"ufo\")", "Build type (\"raw\" - raw data, \"sino\" - arrange in sinogram, \"ufo\" - arrange in sinogram and convert UFO floating-point format)", g_enum_register_static ("build", build_values), 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(RoofBuildTaskPrivate)); } static void ufo_roof_build_task_init(RoofBuildTask *self) { self->priv = UFO_ROOF_BUILD_TASK_GET_PRIVATE(self); }