cluster.cpp

Sector is a system infrastructure software that provides functionality for distributed data storage,

#include "dcclient.h"
#include <iostream>

using namespace std;
using namespace cb;

int main(int argc, char** argv)
{
   Sector::init(argv[1], atoi(argv[2]));


   vector<string> files;
   files.insert(files.end(), "ClusterData.1");
   files.insert(files.end(), "ClusterData.2");

   Stream input;
   if (input.init(files) < 0)
      return -1;

   Stream output;
   if (output.init(0) < 0)
      return -1;


   Process* myproc = Sector::createJob();

   const int cnum = 5;
   int* cx = new int[cnum];
   int* cy = new int[cnum];
   int64_t ad = 0;
   int64_t newad = 0;

   int64_t* acx = new int64_t[cnum];
   int64_t* acy = new int64_t[cnum];
   int* nx = new int[cnum];
   int* ny = new int[cnum];

   int* param = new int[2 * cnum + 1];
   param[0] = cnum;


   for (int i = 0; i < cnum; ++ i)
   {
      ad = acx[i] = acy[i] = nx[i] = ny[i] = 0;
      cx[i] = 100 * i;
      cy[i] = 100 * i;
   }

   do
   {
      for (int i = 0; i < cnum; ++ i)
      {
         param[i * 2 + 1] = cx[i];
         param[i * 2 + 2] = cy[i];
      }
      ad = newad;
      newad = 0;

      if (myproc->run(input, output, "myCluster", -1, (char*)param, (cnum * 2 + 1) * 4) < 0)
      {
         cout << "failed to find any computing resources." << endl;
         return -1;
      }

      while (true)
      {
         Result* res = NULL;

         if (-1 == myproc->read(res))
         {
            if (myproc->checkProgress() == -1)
            {
               cerr << "all SPEs failed\n";
               break;
            }

            if (myproc->checkProgress() == 100)
               break;

            continue;
         }

         for (int i = 0; i < cnum; ++ i)
         {
            acx[i] += *(int64_t*)(res->m_pcData + 8 * i);
            acy[i] += *(int64_t*)(res->m_pcData + 8 * cnum + 8 * i);
            nx[i] += *(int*)(res->m_pcData + 16 * cnum + 4 * i);
            ny[i] += *(int*)(res->m_pcData + 20 * cnum + 4 * i);
         }
         newad += *(int64_t*)(res->m_pcData + 24 * cnum);

         delete [] res->m_pcData;
         delete res;
         res = NULL;
      }

      for (int i = 0; i < cnum; ++ i)
      {
         if (nx[i] != 0)
            cx[i] = acx[i] / nx[i];
         if (ny[i] != 0)
            cy[i] = acy[i] / ny[i];
      }

      cout << "result " << newad << " ";
      for (int i = 0; i < cnum; ++i)
         cout << cx[i] << " " << cy[i] << " ";
      cout << endl;
   } while (abs(ad - newad) > 1);

   cout << "SPE COMPLETED " << endl;

   myproc->close();
   Sector::releaseJob(myproc);

   Sector::close();

   return 0;
}