DidaWiki

We implement a three stage FastFlow pipeline where:

• the first stage produces a stream of float Vectors
• the second stage maps a dummy (time consuming) function on the Vector items
• the third stage just prints results

The code is the wollowing one:

#include <iostream>
#include <ff/pipeline.hpp>
#include <math.h>
 
// in case you want to target a CUDA GPU (local) 
// you have to uncomment the following define and rename the file to .cu
 
//#define SKEPU_CUDA
 
#include <skepu/vector.h>
#include <skepu/map.h>
 
using namespace ff;
 
int N = 10; 
 
//
// This is the stage generatin a stream of streamlen Vector items 
// directed to the node output channel 
//
class Source: public ff_node {
public:
  Source(unsigned int streamlen):streamlen(streamlen)  {}
 
  void * svc(void * task) {
    if(streamlen != 0) {
      // declaring (and allocating) a new Vector
      skepu::Vector<float> * v = new skepu::Vector<float>(N,(float)streamlen);
      // decrease number of items to deliver on the input stream
      streamlen--;
      task = (void *) v;
#ifdef DEBUG
      std::cout << "Source delivering:" << *v << std::endl;
#endif
    } else {
      // if we reached the limit, just output and End Of Stream
      task = (void *) FF_EOS; 
    }
    return task;
  }
private:
  unsigned int streamlen;
};
 
// 
// this is the node simply printing the Vector data items appearing on the input stream
//
class Drain: public ff_node {
  void * svc(void * task) {
    // cast input task to Vector *
    skepu::Vector<float> * v = (skepu::Vector<float> *) task;
#ifdef DEBUG
    // print the Vector
    std::cout << "Drain got " << *v << std::endl;
#endif
    // free it up
    free(task);
    return(GO_ON);
  }
}; 
 
// 
// this is the function to be mapped
// take a float a and compute ITERNO times "a=sin(a)"
// useless. Only there to consume some 
//
#define ITERNO 800000
 
UNARY_FUNC(iters, float, a,
           for(int _i=0; _i<ITERNO; _i++) a = sin(a); return(a);
           )
 
// hosts the lenght of the vectors
int NN = 10; 
 
// 
// this is the middle pipeline stage 
// gets a vector and maps iters over it 
// (it uses SkePU map within the node that from the 
//  outside looks like as a normal FastFlow ff_node)
//
class MapStage:public ff_node {
  void *svc(void * task) {
    // cast the input task 
    skepu::Vector<float> * v = (skepu::Vector<float> *) task;
#ifdef DEBUG
    std::cout << "MapStage got: " << *v << std::endl; 
#endif
    // declare the output vector
    skepu::Vector<float> * r = new skepu::Vector<float>(NN);
    // then declare the mapper
    skepu::Map<iters> skepumap(new iters);
    // and eventually invoke the mapper
    skepumap(*v, *r);
#ifdef DEBUG
    std::cout << "MapStage delivering: " << *r << std::endl; 
#endif
    free(task);
    return((void *) r);
  }
};
 
 
int main(int argc, char * argv[]) {
  if (argc!=3) {
    std::cerr << "use: "  << argv[0] << " streamlen veclen\n";
    return -1;
  }
  N = atoi(argv[2]);
 
  // bild a 3-stage pipeline
  ff_pipeline pipe;
  pipe.add_stage(new Source(atoi(argv[1])));
  pipe.add_stage(new MapStage());
  pipe.add_stage(new Drain());
 
 
  // and execut it taking execution time(s)
  ffTime(START_TIME);
  if (pipe.run_and_wait_end()<0) {
    error("running pipeline\n");
    return -1;
  }
  ffTime(STOP_TIME);
 
  // and print results
  std::cerr << "DONE, pipe  time= " << pipe.ffTime() << " (ms)\n";
  std::cerr << "DONE, total time= " << ffTime(GET_TIME) << " (ms)\n";
  pipe.ffStats(std::cerr);
  return 0;
}