Fix 64-bit ids, ROOF ids are big-endian, skip incomplete datasets at the end and if with specified latency, try to reconstruct from desynchronized streams (harmful, debugging only)

1 files changed, 14 insertions, 7 deletions

diff --git a/src/ufo-roof-build-task.c b/src/ufo-roof-build-task.c
index 9d55d38..a39ed11 100644
--- a/src/ufo-roof-build-task.c
+++ b/src/ufo-roof-build-task.c
@@ -18,6 +18,7 @@
  */
 
 #include <stdio.h>
+#include <endian.h>
 
 #ifdef __APPLE__
 #include <OpenCL/cl.h>
@@ -46,7 +47,7 @@ struct _UfoRoofBuildTaskPrivate {
     gboolean                    stop;                                   // Stop flag
     gboolean                    simulate;                               // Indicates if we are running in network or simulation modes
 
-    guint 			announced;				// For debugging
+    guint64 			announced;				// For debugging
 
     struct timespec		last_fragment_timestamp;
 };
@@ -210,12 +211,12 @@ ufo_roof_build_task_process (UfoTask *task,
 
     const uint8_t *fragment = data;
     for (guint i = 0; i < header->n_packets; i++) {
-        guint packet_id = 0;
+        guint64 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;
+            packet_id = be64toh(pheader->packet_id) + 1;
         }
 
 	    // FIXME: Can we kill here the dataset finished during the previous step of iteration
@@ -235,8 +236,14 @@ ufo_roof_build_task_process (UfoTask *task,
 
 	    // 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]);
+            ready = ufo_roof_buffer_skip_to_ready(buf);
+            if (ready) {
+                    // FIXME: shall we really reset timer here?
+	        clock_gettime(CLOCK_REALTIME, &priv->last_fragment_timestamp);
+            } else {
+	        priv->stop = TRUE;
+	        g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]);
+            }
 	}
     }
 
@@ -306,13 +313,13 @@ ufo_roof_build_task_generate (UfoTask *task,
 
 	// 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);
+//	printf("%lu datasets processed, stopping\n", buf->current_id);
         priv->stop = TRUE;
         g_object_notify_by_pspec (G_OBJECT(task), properties[PROP_STOP]);
     }
 
     if (((priv->number > 0)&&(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);
+	printf("Processing dataset %li (%s), next: %u out of %u\n", buf->current_id + (ready?0:1), (ready?"ready     ":"timeout   "), buf->n_fragments[buf->current_id%buf->ring_size], buf->fragments_per_dataset);
 	priv->announced = buf->current_id;
     }