Differenze

Queste sono le differenze tra la revisione selezionata e la versione attuale della pagina.

--- magistraleinformaticanetworking:spm:sources14ottspm [14/10/2014 alle 17:33 (11 anni fa)] – Marco Danelutto
+++ magistraleinformaticanetworking:spm:sources14ottspm [15/10/2014 alle 15:13 (11 anni fa)] (versione attuale) – Marco Danelutto
@@ Linea 1: / Linea 1: @@
 ===== Sample sources =====
-C version with timings
+C version with timings (to be compiled with **gcc -pthread -lm**)
 <code c++ farm.c>
 #include <stdio.h>
@@ Linea 165: / Linea 165: @@
-C++ version with timings (to be compiled with <strong>g++ -std=c++0x -pthread -lm -rt</strong>
+C++ version with timings (to be compiled with **g++ -std=c++0x -pthread -lm -rt**)
 <code c++ Farm.cpp>
 #include <iostream>
@@ Linea 273: / Linea 273: @@
   cout << "All workers terminated " << endl;
   cout << "Elapsed time " <<   (t1.tv_sec-t0.tv_sec)*1000.0 + ((float) (t1.tv_usec - t0.tv_usec))/1000.0 << " msecs " << endl;
+  return(0);
+}
+</code>
+Version C++/Phtread with synchronizations wrapped into a new Queue<T> type:
+<code c++ FarmQueue.cpp>
+#include <iostream>
+#include <queue>
+#include <pthread.h>
+using namespace std;
+class Task {
+private:
+  int n;
+  float * v;
+public:
+  Task(int n, float * v):n(n),v(v) {}
+  float * get() {
+    return v;
+  }
+  int length() {
+    return n;
+  }
+};
+template <class Task> class Queue {
+private:
+  std::queue<Task> tasks;
+  pthread_mutex_t lock;
+public:
+  Queue() {
+    lock = PTHREAD_MUTEX_INITIALIZER;
+  }
+  ~Queue() {
+    // TBD
+  }
+  void send(Task m) {
+    pthread_mutex_lock(&lock);
+    tasks.push(m);
+    pthread_mutex_unlock(&lock);
+    return;
+  }
+  Task receive() {
+    pthread_mutex_lock(&lock);
+    Task t = (Task) tasks.front();
+    tasks.pop();
+    pthread_mutex_unlock(&lock);
+    return t;
+  }
+};
+typedef struct __comms {
+  Queue<Task> * in;
+  Queue<Task> * out;
+} COMMS;
+Task f(Task x) {
+  return x;
+}
+void * body(void * x) {
+  COMMS * q = (COMMS *) x;
+  int * i = new int();
+  while(true) {
+    Task t = (q->in)->receive();
+    if(t.length() < 0)
+      break; // EOS
+    Task r = f(t);
+    (*i)++;
+    q->out->send(r);
+  }
+  return ((void *) i);
+}
+int main(int argc, char * argv []) {
+  Queue<Task> tasks;
+  Queue<Task> ress;
+  COMMS c;
+  c.in = &tasks;
+  c.out = &ress;
+  int nw = atoi(argv[1]);
+  pthread_t * tid = new pthread_t[nw];
+  for(int i=0; i<nw; i++) {
+    int ret = pthread_create(&tid[i], NULL, body, (void *) &c);
+    if(ret != 0) {
+      // TBD
+      cerr << "error creating thread " << i << endl;
+    }
+  }
   return(0);
 }
 </code>