infrastructure to emit more event types

codes/codes-workload.h

@@ -65,7 +65,7 @@ struct codes_workload_op
         struct {
             int file_id;  /* file to operate on */
         } close;
-    };
+    }u;
 };
 
 /* load and initialize workload of of type "type" with parameters specified by

src/workload/test-workload-method.c

@@ -9,6 +9,8 @@
  * testing/validation purposes.
  */
 
+#include <assert.h>
+
 #include "ross.h"
 #include "codes/codes-workload.h"
 #include "codes-workload-method.h"
@@ -16,6 +18,18 @@
 int test_workload_load(const char* params, int rank);
 void test_workload_get_next(int rank, struct codes_workload_op *op);
 
+/* state information for each rank that is retrieving requests */
+struct wkload_stream_state
+{
+    int rank;
+    struct wkload_stream_state* next;
+    struct codes_workload_op op_array[16];
+    int op_array_len;
+    int op_array_index;
+};
+
+struct wkload_stream_state* wkload_streams = NULL;
+
 struct codes_workload_method test_workload_method = 
 {
     .method_name = "test",
@@ -28,14 +42,75 @@ int test_workload_load(const char* params, int rank)
     /* no params in this case; this example will work with any number of
      * ranks
      */
+    struct wkload_stream_state* new;
+
+    new = malloc(sizeof(*new));
+    if(!new)
+        return(-1);
+
+    new->rank = rank;
+
+    /* synthetic workload, just open file for now */
+    new->op_array_len = 1;
+    new->op_array_index = 0;
+    new->op_array[0].op_type = CODES_WK_OPEN;
+    new->op_array[0].u.open.file_id = 3;
+    new->op_array[0].u.open.create_flag = 1;
+
+    /* add to front of list of streams that we are tracking */
+    new->next = wkload_streams;
+    wkload_streams = new;
+
     return(0);
 }
 
 void test_workload_get_next(int rank, struct codes_workload_op *op)
 {
-    /* TODO: fill in a more complex example... */
+    struct wkload_stream_state* tmp = wkload_streams;
+    struct wkload_stream_state* tmp2 = wkload_streams;
+
+    /* find stream associated with this rank */
+    while(tmp)
+    {
+        if(tmp->rank == rank)
+            break;
+        tmp = tmp->next;
+    }
+
+    assert(tmp);
+    assert(tmp->rank == rank);
+
+    if(tmp->op_array_index < tmp->op_array_len)
+    {
+        *op = tmp->op_array[tmp->op_array_index];
+        tmp->op_array_index++;
+    }
+    else
+    {
+        /* no more operations */
+        op->op_type = CODES_WK_END;
+
+        /* destroy this instance */
+        if(wkload_streams == tmp)
+        {
+            wkload_streams = tmp->next;
+        }
+        else
+        {
+            while(tmp2)
+            {
+                if(tmp2->next == tmp)
+                {
+                    tmp2->next = tmp->next;
+                    break;
+                }
+                tmp2 = tmp2->next;
+            }
+        }
+
+        free(tmp);
+    }
 
-    op->op_type = CODES_WK_END;
     return;
 }