Sync fork, prepare to push

src/flag_beamform_thread.c

@@ -8,6 +8,7 @@
 #include <stdlib.h>
 #include <pthread.h>
 #include <string.h>
+#include <sys/time.h>
 
 #include "cublas_beamformer.h"
 #include "hashpipe.h"
@@ -209,19 +210,22 @@ static void * run(hashpipe_thread_args_t * args) {
 		    }
 		   
 		    // Run the beamformer
-            // clock_t start, end;
-            // start = clock();
+            struct timeval tval_before, tval_after, tval_result;
+            gettimeofday(&tval_before, NULL);
 		    run_beamformer((signed char *)&db_in->block[curblock_in].data, (float *)&db_out->block[curblock_out].data);
-            // end = clock();
-            // double timeTaken = 0;
-            // timeTaken = ((double) (end - start))/CLOCKS_PER_SEC;
-            //printf("RTBF: %fs\n", timeTaken);
+            gettimeofday(&tval_after, NULL);
+            timersub(&tval_after, &tval_before, &tval_result);
+            if ((float) tval_result.tv_usec/1000 > 13) {
+                printf("RTBF: Warning!!!!!!!!! Time = %f ms\n", (float) tval_result.tv_usec/1000);
+            }
+
 		    check_count++;
 		   // if(check_count == 1000){
 		   // }
 			// Get block's starting mcnt for output block
 		    db_out->block[curblock_out].header.mcnt = tmp_header.mcnt;
 		    db_out->block[curblock_out].header.good_data = good_data;
+            //printf("BF: good_data = %lld\n", (long long int)good_data);
 		        
 		    // Mark output block as full and advance
 		    #if VERBOSE==1

src/flag_databuf.h

@@ -194,7 +194,7 @@ typedef struct flag_input_databuf {
  * It is the output buffer of the flag_transpose_thread.
  * It is the input buffer of the flag_correlator_thread.
  */
-#define N_GPU_INPUT_BLOCKS 2
+#define N_GPU_INPUT_BLOCKS 20
 
 // A typedef for a GPU input block header
 typedef struct flag_gpu_input_header {
@@ -260,7 +260,7 @@ typedef struct flag_pfb_gpu_input_databuf {
 /*
  * GPU OUTPUT BUFFER STRUCTURES
  */
-#define N_GPU_OUT_BLOCKS 2
+#define N_GPU_OUT_BLOCKS 20
 
 // A typedef for a correlator output block header
 typedef struct flag_gpu_output_header {

src/flag_net_thread.c

@@ -169,11 +169,22 @@ static inline void cleanup_blocks(flag_input_databuf_t * db) {
     hashpipe_status_unlock_safe(st_p);
 
     int i;
+    int rv;
     for (i = 0; i < N_INPUT_BLOCKS; i++) {
         #if VERBOSE==1
         printf("NET: Waiting for block %d to be free...\n", i);
         #endif
-        flag_input_databuf_wait_free(db, i);
+
+        while ((rv = flag_input_databuf_wait_free(db, i)) != HASHPIPE_OK) {
+	    if (rv == HASHPIPE_TIMEOUT) {
+		continue;
+	    }
+	    else {
+		hashpipe_error(__FUNCTION__, "error waiting for free databuf");
+		pthread_exit(NULL);
+		break;
+	    }
+	}
         #if VERBOSE==1
         printf("NET: Initializing block %d\n", i);
         #endif
@@ -185,8 +196,21 @@ static inline void cleanup_blocks(flag_input_databuf_t * db) {
 // Method to mark the block as filled
 static void set_block_filled(flag_input_databuf_t * db, block_info_t * binfo) {
 
+    //struct timeval tval_before, tval_after, tval_result;
+    //gettimeofday(&tval_before, NULL);
+
     uint32_t block_idx = get_block_idx(binfo->mcnt_start);
-    flag_input_databuf_wait_free(db, block_idx);
+    int rv;
+    while ((rv = flag_input_databuf_wait_free(db, block_idx)) != HASHPIPE_OK) {
+	if (rv == HASHPIPE_TIMEOUT) {
+	    continue;
+	}
+	else {
+	    hashpipe_error(__FUNCTION__, "error waiting for free databuf");
+	    pthread_exit(NULL);
+	    break;
+	}
+    }
  
     // Validate that we're filling blocks in the proper sequence
     int next_filled = (last_filled + 1)% N_INPUT_BLOCKS;
@@ -222,6 +246,9 @@ static void set_block_filled(flag_input_databuf_t * db, block_info_t * binfo) {
     //hashpipe_status_lock_safe(st_p);
     //hgeti4(st_p->buf, "XID", &binfo->self_xid);
     //hashpipe_status_unlock_safe(st_p);
+    //gettimeofday(&tval_after, NULL);
+    //timersub(&tval_after, &tval_before, &tval_result);
+    //printf("NET: Time = %f\n", (float)tval_result.tv_usec/1000);
 }
 
 #define WINDOW_SIZE 50
@@ -259,8 +286,20 @@ static inline int64_t process_packet(flag_input_databuf_t * db, struct hashpipe_
 
         // Initialize next block
         uint64_t  b;
+	int rv;
+
         for (b = (N_INPUT_BLOCKS - WINDOW_SIZE + 1)*Nm + cur_mcnt; b < (N_INPUT_BLOCKS + 1)*Nm + cur_mcnt; b++) {
-            flag_input_databuf_wait_free(db, get_block_idx(b) );
+
+	    while ((rv = flag_input_databuf_wait_free(db, get_block_idx(b))) != HASHPIPE_OK) {
+		if (rv == HASHPIPE_TIMEOUT) {
+		    continue;
+		}
+		else {
+		    hashpipe_error(__FUNCTION__, "error waiting for databuf free");
+		    pthread_exit(NULL);
+		    break;
+		}
+	    }
             initialize_block(db, b);
         }
 
@@ -319,8 +358,18 @@ static inline int64_t process_packet(flag_input_databuf_t * db, struct hashpipe_
     }
 
     // Calculate starting points for writing packet payload into buffer
-    // POSSIBLE RACE CONDITION!!!! Need to lock db->block access with semaphore
-    flag_input_databuf_wait_free(db, dest_block_idx);    
+    // POSSIBLE RACE CONDITION!!!! Need to lock db->block access with semaphor
+    int rv;
+    while ((rv = flag_input_databuf_wait_free(db, dest_block_idx)) != HASHPIPE_OK) {
+        if (rv == HASHPIPE_TIMEOUT) {
+	    continue;
+	}
+	else {
+	    hashpipe_error(__FUNCTION__, "error waiting for databuf free");
+	    pthread_exit(NULL);
+	    break;
+	}
+    }
     uint64_t * dest_p  = db->block[dest_block_idx].data + flag_input_databuf_idx(binfo.m, binfo.f, 0, 0);
     const uint64_t * payload_p = (uint64_t *)(p->data+8); // Ignore header
 
@@ -467,7 +516,10 @@ static void *run(hashpipe_thread_args_t * args) {
     int i;
     for (i = 0; i < N_INPUT_BLOCKS-1; i++) {
         // Wait until block semaphore is free
-        if (flag_input_databuf_wait_free(db, i) != HASHPIPE_OK) {
+        while ((rv = flag_input_databuf_wait_free(db, i)) != HASHPIPE_OK) {
+	    if (rv == HASHPIPE_TIMEOUT) {
+		continue;
+	    }
             if (errno == EINTR) { // Interrupt occurred
                 hashpipe_error(__FUNCTION__, "waiting for free block interrupted\n");
                 pthread_exit(NULL);

src/flag_transpose_thread.c

@@ -9,6 +9,7 @@
 #include <pthread.h>
 #include <string.h>
 #include <unistd.h>
+#include <sys/time.h>
 
 #include <xgpu.h>
 #include "hashpipe.h"
@@ -109,9 +110,9 @@ static void * run(hashpipe_thread_args_t * args) {
                 /**********************************************
                  * Perform transpose
                  **********************************************/
+                struct timeval tval_before, tval_after, tval_result;
 
-                // clock_t start, end;
-                // start = clock();
+                gettimeofday(&tval_before, NULL);
 
                 int m; int f;
                 int t; int c;
@@ -131,10 +132,13 @@ static void * run(hashpipe_thread_args_t * args) {
                     }
                 }
 
-                // end = clock();
-                // double timeTaken = 0;
-                // timeTaken = ((double) (end - start))/CLOCKS_PER_SEC;
-                // printf("TRA: %fs\n", timeTaken);
+                gettimeofday(&tval_after, NULL);
+
+                timersub(&tval_after, &tval_before, &tval_result);
+                if ((float)tval_result.tv_usec/1000 > 10) {
+                    printf("TRA: WARNING!!!!!!!!! - Time = %f ms\n", (float)tval_result.tv_usec/1000);
+                }
+
                 /***********************************************
                  * Add header information to output block
                  ***********************************************/