Differenze

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

--- magistraleinformaticanetworking:spm:samplefastflowcode [19/05/2011 alle 14:27 (14 anni fa)] – Marco Danelutto
+++ magistraleinformaticanetworking:spm:samplefastflowcode [19/05/2011 alle 14:33 (14 anni fa)] (versione attuale) – Marco Danelutto
@@ Linea 2: / Linea 2: @@
 This is the code used during the FastFlow lesson.
+All code should be compiled with a command such as
+<code> g++ -Idir-where-ff-dir-has-been-saved -lpthread -o file file.cpp</code>
 <code c++ pipe.cpp>
@@ Linea 154: / Linea 156: @@
 }
 </code>
+In case we want to parallelize the increase integers stage with a farm, we can keep the same code and use the following main code:
+<code c++ altmain.cpp>
+int main(int argc, char * argv[]) {
+  int tasks = 10;  // dummy, should be taken from the input line
+  // allocator must be initialized before using it
+  allocator.init();
+  // declare a pipeline object
+  ff_pipeline pipe_a;
+  // add stages in order:
+  //    the first stage added is the first pipeline stage
+  //    the i-th stage added is the i-th pipeline stage
+  pipe_a.add_stage(new GenerateStream(tasks));
+  // we create a farm to process increments in the second stage
+  // declare a ff_farm object
+  ff_farm<> farm;
+  // default collector
+  farm.add_collector(NULL);
+  // declare worker vector (string)
+  std::vector<ff_node *> workers;
+  // two workers
+  workers.push_back(new IntIncStage);
+  workers.push_back(new IntIncStage);
+  // add workers to farm
+  farm.add_workers(workers);
+  // second pipeline stage is the farm
+  pipe_a.add_stage(&farm);
+  // the third one is the print stream
+  pipe_a.add_stage(new DrainStage);
+  // now we ca run the application:
+  std::cout << "Starting application ..." << std::endl;
+  pipe_a.run();
+  std::cout << "Application started" << std::endl;
+  // here more (unrelated) work can be performed ...
+  // when results are needed we should wait for application termination
+  pipe_a.wait();
+  // alternatively, if we have nothing else to do, we can issue a single
+  // call such as:     pipe_a.run_and_wait_end();
+  // in this case the pipeline is started and its termination awaited
+  // synchronously
+  std::cout << "Application terminated" << std::endl;
+  return(0);
+}
+</code>
+Sample farm code (taken from FF tests/ directory):
+<code c++ farm.cpp>
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/* ***************************************************************************
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License version 2 as
+ *  published by the Free Software Foundation.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+ *
+ *  As a special exception, you may use this file as part of a free software
+ *  library without restriction.  Specifically, if other files instantiate
+ *  templates or use macros or inline functions from this file, or you compile
+ *  this file and link it with other files to produce an executable, this
+ *  file does not by itself cause the resulting executable to be covered by
+ *  the GNU General Public License.  This exception does not however
+ *  invalidate any other reasons why the executable file might be covered by
+ *  the GNU General Public License.
+ *
+ ****************************************************************************
+ */
+/**
+ Very basic test for the FastFlow farm.
+*/
+#include <vector>
+#include <iostream>
+#include <ff/farm.hpp>
+using namespace ff;
+// generic worker
+class Worker: public ff_node {
+public:
+    void * svc(void * task) {
+        int * t = (int *)task;
+        std::cout << "Worker " << ff_node::get_my_id()
+                  << " received task " << *t << "\n";
+        return task;
+    }
+    // I don't need the following functions for this test
+    //int   svc_init() { return 0; }
+    //void  svc_end() {}
+};
+// the gatherer filter
+class Collector: public ff_node {
+public:
+    void * svc(void * task) {
+        int * t = (int *)task;
+        if (*t == -1) return NULL;
+        return task;
+    }
+};
+// the load-balancer filter
+class Emitter: public ff_node {
+public:
+    Emitter(int max_task):ntask(max_task) {};
+    void * svc(void *) {
+        int * task = new int(ntask);
+        --ntask;
+        if (ntask<0) return NULL;
+        return task;
+    }
+private:
+    int ntask;
+};
+int main(int argc, char * argv[]) {
+    if (argc<3) {
+        std::cerr << "use: "
+                  << argv[0]
+                  << " nworkers streamlen\n";
+        return -1;
+    }
+    int nworkers=atoi(argv[1]);
+    int streamlen=atoi(argv[2]);
+    if (!nworkers || !streamlen) {
+        std::cerr << "Wrong parameters values\n";
+        return -1;
+    }
+    ff_farm<> farm; // farm object
+    Emitter E(streamlen);
+    farm.add_emitter(&E);
+    std::vector<ff_node *> w;
+    for(int i=0;i<nworkers;++i) w.push_back(new Worker);
+    farm.add_workers(w); // add all workers to the farm
+    Collector C;
+    farm.add_collector(&C);
+    if (farm.run_and_wait_end()<0) {
+        error("running farm\n");
+        return -1;
+    }
+    std::cerr << "DONE, time= " << farm.ffTime() << " (ms)\n";
+    farm.ffStats(std::cerr);
+    return 0;
+}
+</code>
+Sample accelerator code:
+<code c++ accelerator.cpp>
+#include <iostream>
+#include <math.h>
+#include <ff/farm.hpp>
+#include <ff/node.hpp>
+using namespace std;
+using namespace ff;
+// should be global to be accessible from workers
+#define MAX 1024
+double x[MAX];
+double y[MAX];
+#define ITERNO 100000
+int iterno = ITERNO;
+class Worker: public ff_node {
+  int svc_init() {
+    cout << "Worker initialized" << endl;
+    return 0;
+  }
+  void * svc(void * in) {
+    int i = * ((int *) in);
+    // cout << "Computing " << i << endl;
+    for(int j=0; j<iterno; j++)
+       x[i] = sin(x[i]);
+    y[i] += x[i];
+    return in;
+  }
+};
+#define NW 2
+int main(int argc, char * argv[])
+{
+  ffTime(START_TIME);
+  cout << "init " << argc  << endl;
+  int nw = (argc==1 ? NW : atoi(argv[1]));
+  iterno = (argc<=2 ? ITERNO : atoi(argv[2]));
+  cout << "using " << nw << " workers iterating " << iterno << " times "<< endl;
+  // init input (fake)
+  for(int i=0; i<MAX; i++) {
+    x[i] = (double)(i*10);
+    y[i] = (double)(-i*13);
+  }
+  cout << "Setting up farm" << endl;
+  ff_farm<> farm(true,nw);    // true for offloading, NW for the number of workers ...
+  std::vector<ff_node *> w;   // prepare workers
+  for(int i=0;i<nw;++i)
+    w.push_back(new Worker);
+  farm.add_workers(w);        // add them to the farm
+  farm.run_then_freeze();     // run farm asynchronously
+  cout << "Sending tasks ..." << endl;
+  int tasks[MAX];
+  for(int i=0; i<MAX; i++) {
+     tasks[i]=i;
+     farm.offload((void *) &tasks[i]);
+  }
+  farm.offload((void *) FF_EOS);
+  cout << "Waiting termination" << endl;
+  farm.wait();
+  cout << "Farm terminated after computing for " << farm.ffTime() << endl;
+  ffTime(STOP_TIME);
+  cout << "Spent overall " << ffTime(GET_TIME) << endl;
+}
+</code>