Update on code!

executablecode/Makefile

@@ -20,7 +20,7 @@ OBJDIR = lib
 MODULE = run
 SRCS_generated_DIR = generated_src/
 include Makefile.src
-SRCS_base = src/transactions.c src/syncchannel.c src/timers.c src/myerrors.c src/debug.c
+SRCS_base = src/transactions.c src/syncchannel.c src/timers.c src/myerrors.c src/debug.c src/storeevents.c
 SRCS_base_DIR = .
 SRCS_base_DIRSRC = src/ 
 OBJS_executor = $(SRCS_base:%.c=lib/%.o)

executablecode/generated_src/main.c

 #include <stdio.h>
 #include <pthread.h>
 #include <unistd.h>
+#include <stdlib.h>
 
 #include "transactions.h"
 #include "syncchannel.h"
@@ -13,9 +14,13 @@
 void *send(void *arg) {
   int x = 2;
   int y = 3;
+  
+  int index;
 
   int *p[2];
 
+  synccell *requests[2];
+
   debugMsg("Setting timer 12 ...");
   setTimerMs(12, 10000);
   debugMsg("Setting timer 12 done");
@@ -36,6 +41,32 @@ void *send(void *arg) {
   sendSyncParams(1, p, 2);
   debugMsg("Send OK");
 
+  
+  // Testing multirequest
+  // Trying to send with a timer of 10 seconds
+  debugMsg("Setting timer 12 ...");
+  setTimerMs(12, 10000);
+  debugMsg("Setting timer 12 done");
+
+  requests[0] = (synccell *)(malloc(sizeof(synccell) + 2*sizeof(int *)));
+  requests[0]->ID = 1;
+  requests[0]->type = SENDING;
+  requests[0]->nParams = 2;
+  requests[0]->params[0] = &x;
+  requests[0]->params[1] = &y;
+  
+  requests[1] = (synccell *)(malloc(sizeof(synccell)));
+  requests[1]->ID = 12;
+  requests[1]->type = TIMER_EXPIRATION;
+  
+  debugMsg(" -------------- Making requests");
+
+  x = 31;
+  y = 51;
+  index = makeRequests(requests, 2);
+
+  debugInt("---------------- Request completed", index);
+
   return NULL;
 }
 
@@ -44,6 +75,7 @@ void *receive(void *arg) {
   int x;
   int y;
   int *p[2];
+  int sleepTime;
 
   debugInt("Setting timer ", 13 + myid);
   setTimerMs(13+myid, 1000);
@@ -54,6 +86,24 @@ void *receive(void *arg) {
   p[1] = &y;
   receiveSyncParams(1, p ,2);
   debugThreeInts("Receive OK", x, y, myid);
+  
+  resetTimer(13+myid);
+  // random wait between 5 and 15 seconds
+  // Trying to receive
+
+  sleepTime = 5 + (rand() % 10);
+  debugTwoInts("---------- Waiting for seconds:", sleepTime, myid);
+  setTimerMs(13+myid, sleepTime * 1000);
+
+  debugInt("-------------- Waiting for timerExpiration", myid);
+  waitForTimerExpiration(13+myid);
+  debugInt("-------------- Timer has expired", myid);
+  
+  
+  receiveSyncParams(1, p ,2);
+  debugThreeInts("--------------- Second receive OK", x, y, myid);
+  
+
 
   return NULL;
 }
@@ -72,13 +122,13 @@ int main(int argc, char * argv[]) {
   debugMsg("Creating threads");
   pthread_create(&sender, NULL, send, NULL);
   pthread_create(&receiver0, NULL, receive, (void *)1);
-  pthread_create(&receiver1, NULL, receive, (void *)2);
+  //pthread_create(&receiver1, NULL, receive, (void *)2);
   //pthread_create(&sender, NULL, send, NULL);
 
   debugMsg("Starting threads");
   pthread_join(sender, NULL);
   pthread_join(receiver0, NULL);
-  pthread_join(receiver1, NULL);
+  //pthread_join(receiver1, NULL);
   
   debugMsg("All done");
 

executablecode/src/myerrors.c

@@ -3,6 +3,18 @@
 
 #include "myerrors.h"
 
+
+
+
+void criticalErrorInt(char *msg, int value) {
+  if (msg != NULL) {
+    printf("\nCritical error: %s, %d\n", msg, value);
+  }
+
+  exit(-1);
+}
+
+
 void criticalError(char *msg) {
   if (msg != NULL) {
     printf("\nCritical error: %s\n", msg);

executablecode/src/myerrors.h

@@ -3,6 +3,7 @@
 #ifndef MY_ERRORS_H
 #define MY_ERRORS_H
 
+void criticalErrorInt(char *msg, int value);
 void criticalError(char *msg);
 
 #endif

executablecode/src/storeevents.c

+
+
+
+#include <time.h>
+#include <sys/time.h>
+
+#include "storeevents.h"
+#include "transactions.h"
+
+
+FILE *file;
+hrtime_t startTime;
+
+void initStoreEvents() {
+  file = fopen("events.txt", "w");
+  startTime = gethrtime();
+}
+
+void addEvent(synccell *cell) {
+  fprintf(file, "task %d, transaction %d @t0+%lldms\n", cell->taskID, cell->ID, (cell->completionTime - startTime)/1000000); 
+}
+
+

executablecode/src/storeevents.h

+
+
+#ifndef STORE_EVENTS_H
+#define STORE_EVENTS_H
+
+#include <stdio.h>
+#include <time.h>
+#include <sys/time.h>
+
+#include "transactions.h"
+
+
+void initStoreEvents();
+void addEvent(synccell *cell);
+
+
+#endif
+
+

executablecode/src/syncchannel.h

@@ -3,6 +3,10 @@
 #ifndef SYNC_CHANNEL_H
 #define SYNC_CHANNEL_H
 
+#include "transactions.h"
+
+synccell * addSyncRequest(int channel_id, int *params[], int nParams, int type);
+
 void sendSync(int channel_id);
 void sendSyncParams(int channel_id, int *param[], int nParams);
 

executablecode/src/timers.c

@@ -66,6 +66,7 @@ synccell *makeTimerCell(int timer_id, long long value) {
   synccell *cell = getTimerCell(timer_id);
 
   if (cell == NULL) {
+    debugMsg("*** Making a new timer cell");
     cell = (synccell *)(malloc(sizeof(synccell)));
 
     if (cell == NULL) {
@@ -80,6 +81,7 @@ synccell *makeTimerCell(int timer_id, long long value) {
 
     addCell(cell);
   } else {
+    debugMsg("*** Reusing a timer cell");
     // If timer is expired
     cell->transactionDone = DEFINED;
     // If timer is set but not yet expired -> must stop the timer!
@@ -207,7 +209,7 @@ void waitForTimerExpiration(int timer_id) {
   cell = getTimerCell(timer_id);
 
   if (cell == NULL) {
-    criticalError("Unknown Timer");
+    criticalErrorInt("Unknown Timer", timer_id);
   }
 
   while (cell->transactionDone != DONE) {
@@ -226,7 +228,7 @@ void resetTimer(int timer_id) {
   cell = getTimerCell(timer_id);
 
   if (cell == NULL) {
-    criticalError("Unknown Timer");
+    criticalErrorInt("Unknown Timer", timer_id);
   }
 
   cell->transactionDone = CANCELLED;

executablecode/src/timers.h

@@ -7,6 +7,8 @@
 
 void initTimerManagement();
 
+synccell *getTimerCell(int timer_id);
+void setTimer(int timer_id, long long value); // in nanoseconds
 void setTimerMs(int timer_id, long long value);
 void waitForTimerExpiration(int timer_id);
 void resetTimer(int timer_id);

executablecode/src/transactions.c

@@ -6,6 +6,8 @@
 #include <stdio.h>
 
 #include "transactions.h"
+#include "syncchannel.h"
+#include "timers.h"
 #include "myerrors.h"
 #include "debug.h"
 
@@ -112,9 +114,116 @@ void removeRequest(synccell *cell) {
 }
 
 
+int RequestsDone(synccell *cells[], int nbOfRequests) {
+  int i;
 
-void makeRequests(synccell cells[]) {
+  for(i=0; i<nbOfRequests; i++) {
+    if (cells[i] != NULL) {
+      if (cells[i]->transactionDone  == DONE)  {
+	return i;
+      }
+    }
+  }
+  
+  return -1;
+}
+
+
+// Sending, receiving, timer_expiration
+int WaitAndStoreRequests(synccell *cells[], int nbOfRequests) {
+  int i;
+  synccell *newcells[nbOfRequests];
+  int index;
+
+  for(i=0; i<nbOfRequests; i++) {
+    if ((cells[i]->type == SENDING) || (cells[i]->type == RECEIVING)) {
+      newcells[i] = addSyncRequest(cells[i]->ID, cells[i]->params, cells[i]->nParams, cells[i]->type);
+    } else if (cells[i]->type == TIMER_EXPIRATION){
+      newcells[i] = getTimerCell(cells[i]->ID);
+    } else {
+      newcells[i] = NULL;
+    }
+  }
+
+  
+  while((index = RequestsDone(newcells, nbOfRequests)) == -1) {
+    pthread_cond_wait(&multiType, &syncmutex);
+  }
+
+  return index;
 }
 
 
+// Returns the completed request
+int makeRequests(synccell *cells[], int nbOfRequests) {
+  int i;
+  int completed = -1;
+  synccell *cell = NULL;
+  int random_integer = rand() % nbOfRequests;
+  int index;
+
+  pthread_mutex_lock(&syncmutex);
+
+  // See whether a request can be immediatly completed
+  for(i=0; i<nbOfRequests; i++) {
+    index = (i + random_integer) % nbOfRequests;
+    
+    if (cells[index]->type == SENDING) {
+      cell = getPending(cells[index]->ID, RECEIVING);
+      if (cell != NULL) {
+	completed = index;
+	break;
+      }
+    } 
+
+    if (cells[index]->type == RECEIVING) {
+       cell = getPending(cells[index]->ID, RECEIVING);
+      if (cell != NULL) {
+	completed = index;
+	break;
+      }
+    }
+
+    if (cells[index]->type == TIMER) {
+      if (cells[index]->timerValue < MIN_TIMER_VALUE) {
+	completed = index;
+	break;
+      } else {
+	// Can set the timer
+	setTimer(cells[index]->ID, cells[index]->timerValue);
+      }
+    }
+
+    if (cells[index]->type == TIMER_EXPIRATION) {
+      cell = getTimerCell(cells[index]->ID);
+      
+      if (cell == NULL) {
+	criticalErrorInt("Unknown Timer", cells[index]->ID);
+      }
+      
+      if (cell->transactionDone == DONE) {
+	completed = index;
+	break;
+      }
+      
+    }
+
+    if (cells[index]->type == TIMER_RESET) {
+      resetTimer(cells[index]->ID);
+      completed = index;
+      break;
+    }
+  }
+
+  if (completed == -1) {
+    // Requests must be stored, and we wait for a request to be served
+    completed = WaitAndStoreRequests(cells, nbOfRequests);
+  }  
+
+  pthread_mutex_unlock(&syncmutex);
+  
+  return completed;
+  }
+
+
 

executablecode/src/transactions.h

@@ -11,6 +11,8 @@
 #define RECEIVING 1
 #define SENDING 0
 #define TIMER 2
+#define TIMER_EXPIRATION 3
+#define TIMER_RESET 4
 
 // For transaction done
 #define DEFINED 0
@@ -18,13 +20,16 @@
 #define DONE 2
 #define CANCELLED 3
 
+#define MIN_TIMER_VALUE 1000 // in nanoseconds
+
 struct synccell{
   struct synccell* next;
   int ID;
+  int taskID;
   int type; /* RECEIVING, SENDING, TIMER */
   char transactionDone;
-  long long timerValue;
-  hrtime_t startTime;
+  long long timerValue; // in nanoseconds
+  hrtime_t completionTime;
   timer_t timer;
   int nParams;
   int *params[];
@@ -45,10 +50,7 @@ synccell * getRandomCell();
 synccell * getPending(int channel_id, int type);
 void removeRequest(synccell *cell);
 
-
-
-void makeRequests(synccell cells[]);
-
+int makeRequests(synccell *cells[], int nbOfRequests);
 
 #endif