DidaWiki

<code c++ masterworker_mapping.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. * ** */

/* Example of master-worker computation: * * ——————— ←- * | | | * | —> Worker– | * v | | * Scheduler — —> Worker—— * ^ | * | —> Worker– * | | * ——————— * * */

#include <cstdlib> #include <cstdio> #include <string> #include <ff/farm.hpp> #include <ff/mapping_utils.hpp>

using namespace ff;

class Scheduler: public ff_node { public:

  Scheduler(unsigned streamlen, ff_loadbalancer *lb=0):
      lb(lb), numtasks(streamlen), seed(111) {}

  int svc_init() {
      printf("Scheduler is on core %d\n", ff_getMyCpu());
      srandom(seed);
      return 0;
  }
  
  void * svc(void *task) {
      if (task==NULL) { // first time
          for(size_t i=0;i<numtasks;++i)
              ff_send_out(new long(random() % 10));
          return GO_ON;
      }
      const long t = *(static_cast<long*>(task));
      //printf("Scheduler: got %ld from %d\n", t, lb->get_channel_id());
      if ( t > 0 ) ff_send_out(task);
      else --numtasks;
      
      return (numtasks>0)?GO_ON:NULL;
  }

private:

  ff_loadbalancer *lb;
  unsigned numtasks;
  unsigned seed;

};

class Worker: public ff_node { public:

  int svc_init() {
      printf("Worker(%d) is on core %d\n", get_my_id(),ff_getMyCpu());
      return 0;
  }

  void *svc(void *task) {
      long *t = static_cast<long*>(task);
      printf("Worker(%d): got %ld\n", get_my_id(), *t);
      --*t;        
      return task;
  }

};

int main(int argc, char * argv[]) {

  if (argc!=3) {
      std::cerr << "use: "  << argv[0] << " streamlen nworkers\n";
      return -1;
  }
  long streamlen = atol(argv[1]);
  long nworkers  = atoi(argv[2]);
  
  ff_farm<> farm;
  farm.add_emitter(new Scheduler(streamlen)); //, farm.getlb()));

  std::vector<ff_node*> w;
  for(int i=0;i<nworkers;++i)
      w.push_back(new Worker);
  farm.add_workers(w);
  farm.wrap_around();

  // - try to comment this code to see where threads are pinned by default
  // - try to comment this code and to compile with -DNO_DEFAULT_MAPPING

#if 1

  // pinning threads to specific cores
  const int ncores = ff_numCores(); // gets the number of cores
  std::string s;
  for(int i=0;i<ncores;++i) {
      s = s+std::to_string(i)+",";
  }
  s=std::to_string(nworkers % ncores)+","+s.substr(0,s.find_last_of(","));
  printf("The mapping string is: %s\n", s.c_str());
  threadMapper::instance()->setMappingList(s.c_str());

#endif

  if (farm.run_and_wait_end()<0) {
      error("running farm\n");
      return -1;
  }
  std::cerr << "DONE, pipe  time= " << farm.ffTime() << " (ms)\n";
  return 0;

} <code>