source: Sophya/trunk/SophyaProg/Tests/zthr.cc@ 3225

#include "sopnamsp.h"
#include "zthread.h"
#include "resusage.h"

#include <iostream>
#include <vector>

#include "tmatrix.h"
#include "tvector.h"
#include "tarrinit.h"

#include <stdlib.h>
#include <stdio.h>

/* -------------------------------------------------
  Programme de test des classes de threads de SOPHYA
  SOPHYA::ZThread SOPHYA::ZMutex ...
  Exemples d'execution
  csh> time zthr mtx 2 500
  csh> time zthr arr 2 500
  csh> time zthr sync 4 1000
*/
 
#include <time.h>
#include <unistd.h>

#include "timing.h"
#include "ctimer.h"


// --- Structure d'argument pour fonction d'execution dans les threads de test
typedef struct  {
  int_4 thid, NTh;
  int_4 M, VSz;
} ztarg;

// --- fonction de test simple avec boucle de sleep
void funzt(void *arg)
{
time_t t0, t1;
int i;

ztarg * za = (ztarg *)arg;

t0 = time(NULL);
printf("+++++ funzt(ThId=%d) Entry to funzt (za.M=%d) +++++\n", za->thid, za->M);
int imax = za->M;
for(i=0; i<imax; i++)
  {
  sleep(3);
  t1 = time(NULL)-t0;
  printf("++funzt(ThId=%d)  Dt= %d \n", za->thid, (int)t1);
  }

return;
}

// --- fonction de test simple avec calcul matriciel
void mtx_funzt(void *arg)
{
  ztarg * za = (ztarg *)arg;
  cout << ">>>> mtx-funzt(ThId=" << za->thid << ") - Matrix size= " << za->M << endl;
  
  sa_size_t m = za->M;
  Matrix a1(m,m), a2(m,m), mxprod; 
  a1 = RandomSequence(RandomSequence::Gaussian, 0., 4.);
  a2 = RandomSequence(RandomSequence::Gaussian, 0., 3.);
  char buff[128];
  sprintf(buff, "mtx-funzt(ThId=%d) EndOfInit", za->thid); 
  PrtTim(buff);
  mxprod = a1*a2;
  sprintf(buff, "mtx-funzt(ThId=%d) EndOfMxProd", za->thid); 
  PrtTim(buff);
  return;
}
// --- fonction de test simple avec calcul matriciel
void arr_funzt(void *arg)
{
  ztarg * za = (ztarg *)arg;
  cout << ">>>> arr-funzt(ThId=" << za->thid << ") - Matrix size= " << za->M << endl;
  
  sa_size_t m = za->M;
  TMatrix<int_4> a1(m,m), a2(m,m), ares; 
  a1 = RegularSequence(1.,1.);
  a2 = RegularSequence(5.,3.);
  char buff[128];
  sprintf(buff, "arr-funzt(ThId=%d) EndOfInit", za->thid); 
  PrtTim(buff);
  ares = 4*a1*12*a2;
  sprintf(buff, "arr-funzt(ThId=%d) EndOfOper", za->thid); 
  PrtTim(buff);
  return;
}


// Structure de gestion utilisee par la classe MTVecPool
typedef struct {
  bool busy;
  int stat;
} St_VecPool;

// -------------------------------------------------------------------
// Structure de gestion de zones memoire partagee (des vecteurs) entre
// threads - qui doivent operer successivement sur les vecteurs
// -------------------------------------------------------------------
class MTVecPool {
public:
  MTVecPool(uint_4 nth, uint_4 vsz, uint_4 nvec)
  {
    if (nth > 60) throw ParmError("MTVecPool::MTVecPool() nth > 60");
    if ((nth < 1) || (vsz < 2))
        throw ParmError("MTVecPool::MTVecPool() nth<1 OR vsz<2 ");
    _vmx.SetSize(vsz, nvec);
    _nth = nth;
    _vsz = vsz; 
    TVector<int_8> xx(2);
    for(int k=0; k<nth; k++) _vecp.push_back(xx);
    cout << "-- MTVecPool(nth=" << nth << ")" << endl;
    _vmx.Show();
  }
  ~MTVecPool()   { }
  // Renvoie un pointeur de vecteur pour thread tid
  TVector<int_8>*  GetVecP(uint_4 tid, uint_4& idx)
  {
    if (tid >= _nth) ParmError("MTVecPool::GetVecP() tid > _nth");
    //DBG  cout << "DBG-GetVecP(tid= " << tid << ")" << endl;
    if (tid == 0) {
      mex.lock();
      St_VecPool stv;
      idx = _vecs.size();
      _vecp[tid].Share(_vmx.Column(idx));
      stv.busy = true;
      stv.stat = 0;
      _vecs.push_back(stv);
      mex.unlock();
      //DBG cout << "DBG-GetVecP(tid= " << tid << ") -> Idx=" << idx << " VecSz=" << &(_vecs[idx].vv) << endl;
      return (&(_vecp[tid]));
    }
    else {
      mex.lock();
      bool found = false;
      while (!found) {
        for(uint_4 k=0; k<_vecs.size(); k++) {
          if ( (_vecs[k].stat == tid) && (! _vecs[k].busy) ) {
            found = true;  idx = k;
            _vecs[k].stat = tid; _vecs[k].busy = true; 
            break;
          }
        }
        if (found) {
          _vecp[tid].Share(_vmx.Column(idx));
          mex.unlock();
          //DBG  cout << "DBG-GetVecP(tid= " << tid << ") -> nv=" << hex << rv << dec << endl;
          return (&(_vecp[tid]));
        }
        else { 
          mex.broadcast();
          mex.wait();
        }
      }
    }
  }
  // Renvoie un vecteur pour thread tid
  TVector<int_8>  GetVec(uint_4 tid, uint_4& idx)
  {
    if (tid >= _nth) ParmError("MTVecPool::GetVec() tid > _nth");
    //DBG  cout << "DBG-GetVec(tid= " << tid << ")" << endl;
    if (tid == 0) {
      mex.lock();
      St_VecPool stv;
      idx = _vecs.size();
      stv.busy = true;
      stv.stat = 0;
      _vecs.push_back(stv);
      mex.unlock();
      //DBG cout << "DBG-GetVec(tid= " << tid << ") -> Idx=" << idx << " VecSz=" << &(_vecs[idx].vv) << endl;
      return (_vmx.Column(idx));
    }
    else {
      mex.lock();
      bool found = false;
      while (!found) {
        for(uint_4 k=0; k<_vecs.size(); k++) {
          if ( (_vecs[k].stat == tid) && (! _vecs[k].busy) ) {
            found = true;  idx = k;
            _vecs[k].stat = tid; _vecs[k].busy = true; 
            break;
          }
        }
        if (found) {
          mex.unlock();
          //DBG  cout << "DBG-GetVec(tid= " << tid << ") -> nv=" << hex << rv << dec << endl;
          return (_vmx.Column(idx));
        }
        else { 
          mex.broadcast();
          mex.wait();
        }
      }
    }
  }

  // Retourne l'index du vecteur au gestionnaire, qui le marque comme disponible
  void RetVec(uint_4 idx)
  {
    //DBG cout << "DBG-RetVec(idx= " << idx << ")" << endl;
    ZSync zs(mex, 2);
    _vecs[idx].busy = false; _vecs[idx].stat++; 
    zs.NOp();
  }

  // Verifie l'etat memoire de tous les vecteurs et fait des print
  int Check()
  {
    cout << "MTVecPool::Check() NVec=" << _vecs.size() << " VSz=" 
         << _vsz << " NThreads=" << _nth << endl;
    int nerr = 0;
    int_8 sum = 0;
    int_8 p2 = 1;
    int_8 min,max;
    for(int i=0; i<_nth; i++) { sum += p2; p2 *= 2; }
    for(uint_4 k=0; k<_vecs.size(); k++) {
      if ( (_vecs[k].busy) || (_vecs[k].stat != _nth) ) {
        cout << " Check()/Pb Busy Or Stat  for k=" << k << endl;
        nerr++;
      }
      _vmx.Column(k) -= sum;
      _vmx.Column(k).MinMax(min, max);
      if ((min!=0) || (max!=0)) {
        cout << " Check()/Pb vec[k=" << k << "] != (sum=" << sum << ")" << endl;
        nerr++;
      }
    }
    if (nerr == 0) cout << "MTVecPool::Check()  - OK (NErr=0)" << endl;
    else cout << "MTVecPool::Check() PB NErr=" << nerr << endl;
    return nerr;
  }

  // ... variables membres
  ZMutex mex;
  uint_4 _vsz;
  uint_4 _nth;
  TMatrix<int_8> _vmx;
  vector< St_VecPool> _vecs;
  vector< TVector<int_8> > _vecp;
};


static MTVecPool* mtvp = NULL;

// --- fonction de test avec synchronisation entre threads en utilisant pointeur de vecteurs
void syncp_funzt(void *arg)
{
  ztarg * za = (ztarg *)arg;
  cout << ">>>> syncp_funzt(ThId=" << za->thid << ") - NVec/NLoop= " << za->M << endl;
  
  if (mtvp == NULL) 
    throw NullPtrError("syncp_funzt: MTVecPool* mtvp = NULL");

  int_4 L = za->M;
  int_4 VS = za->VSz;
  int_8 p2 = 1;
  uint_4 k, ii, tid;
  tid = za->thid;
  for(k=0; k<tid; k++) p2 *= 2;

  char buff[128];
  sprintf(buff, "syncp_funzt(ThId=%d) StarOfLoop", za->thid); 
  PrtTim(buff);
  uint_4 idx;
  for(k=0; k<L; k++) {
    *(mtvp->GetVecP(tid, idx)) += p2;
    //DBG cout << "DBG-syncp_funzt(tid=" << tid << ", idx=" << idx << endl;
      mtvp->RetVec(idx);
  }
  sprintf(buff, "syncp_funzt(ThId=%d) EndOfLoop", za->thid); 
  PrtTim(buff);
  return;
}
// --- fonction de test avec synchronisation entre threads
void sync_funzt(void *arg)
{
  ztarg * za = (ztarg *)arg;
  cout << ">>>> sync_funzt(ThId=" << za->thid << ") - NVec/NLoop= " << za->M << endl;
  
  if (mtvp == NULL) 
    throw NullPtrError("sync_funzt: MTVecPool* mtvp = NULL");

  int_4 L = za->M;
  int_4 VS = za->VSz;
  int_8 p2 = 1;
  uint_4 k, ii, tid;
  tid = za->thid;
  for(k=0; k<tid; k++) p2 *= 2;

  char buff[128];
  sprintf(buff, "sync_funzt(ThId=%d) StarOfLoop", za->thid); 
  PrtTim(buff);
  uint_4 idx;
  for(k=0; k<L; k++) {
    mtvp->GetVec(tid, idx) += p2;
    //DBG cout << "DBG-sync_funzt(tid=" << tid << ", idx=" << idx << endl;
    mtvp->RetVec(idx);
  }
  sprintf(buff, "sync_funzt(ThId=%d) EndOfLoop", za->thid); 
  PrtTim(buff);
  return;
}

class CountLock : public ZMutex {
  int count;
public:
  CountLock() { count = 0; }
  inline int Count() { lock(); int rc = ++count; unlock(); return(rc); 
  }
};


static int N = 1;
static int M = 5;
static int VSZ = 32;

int main(int narg, char *arg[])

{

  if (narg < 4) {
    cout << " Usage: zthr select N LM [Sz] " << endl;
    cout << "  select= sl -> simple loop with sleep " << endl;
    cout << "  select= mtx -> matrix init and multiply mx1*mx2" << endl;
    cout << "  select= arr -> array/matrix init and operation c1*a1+c2*a2 " << endl;
    cout << "  select= clk -> Mutex lock count  " << endl;
    cout << "  select= sync -> Thread synchronisation using ZMutex" << endl;
    cout << "  select= syncp -> Thread synchronisation using ZMutex , Vector pointers" << endl;
    cout << "  N= Number of threads (sl/mtx) or CountLock " << endl;
    cout << "  LM = Loop limit (sl/sync) or Matrix size (mtx) " << endl;
    cout << "  Sz = Vector size for select=sync,syncp (default=32) " << endl;
    return(1);
  }

  string sel = arg[1];
  if ((sel != "sl") && (sel != "mtx") && (sel != "arr") && 
      (sel != "sync") && (sel != "syncp") && (sel != "clk")) {
    cout << "zthr/erreur argument sel (!= sl / mtx / arr / clk) " << endl;
    return 2;
  }

  //--  Decodage arguments
  N = atoi(arg[2]);
  M = atoi(arg[3]);
  if (narg > 4) VSZ = atoi(arg[4]);
  cout << "zthr/Info: select=" << sel << " N=" << N << " M= " << M 
       << " VSz=" << VSZ << endl;
 
  
  InitTim();
  SophyaInit();

  int rc = 0;
  try {
    ResourceUsage res(ResourceUsage::RU_All);
    if ((sel == "mtx") || (sel == "arr") || (sel == "sl") || (sel == "sync") || (sel == "syncp")) {
      if ( (sel == "sync") || (sel == "syncp"))  mtvp = new MTVecPool(N,VSZ,M);
      vector<ztarg *> vza;
      vector<ZThread *> vzth;
      for(int i=0; i<N; i++) {
        cout << "*****zthr: Creating Thread " << i+1 << " /" << N << endl;
        ZThread * pzt = new ZThread();
        ztarg* zap = new ztarg;
        vzth.push_back(pzt);
        // ATTENTION : il faut que le thid = 0 ... N-1 (et pas 1)
        zap->thid = i;  zap->M = M;
        zap->NTh = N;   zap->VSz = VSZ;
        vza.push_back(zap);
        if (sel == "mtx")  pzt->setAction(mtx_funzt, vza[i]);
        else if (sel == "arr")  pzt->setAction(arr_funzt, vza[i]);
        else if (sel == "sync")  pzt->setAction(sync_funzt, vza[i]);
        else if (sel == "syncp")  pzt->setAction(syncp_funzt, vza[i]);
        else pzt->setAction(funzt, vza[i]);
      }
      cout << "***zthr: Starting threads ... " << endl;
      PrtTim("***zthr/StarThr");
      for(int i=0; i<N; i++) vzth[i]->start();
      sleep(1);
      cout << "***ResourceUsage before thr[i].join()" << endl;
      cout << res;
      cout << "***zthr Joining Threads ..." << endl;
      for(int i=0; i<N; i++) vzth[i]->join();
      cout << "***zthr Threads Z1 ... Z" << N << " Finished OK" << endl;
      cout << " zthr/Resusage: getDataSize() = " << res.getDataSize() << "  getStackSize()=" 
           << res.getStackSize() << endl;
      cout << res;
      for(int i=0; i<N; i++) {
        delete vzth[i];
        delete vza[i];
      }
      if (mtvp) {
        Timer tm("MTVecPool::Check()");
        mtvp->Check();
        tm.Nop();
        delete mtvp;
      }
    }
    else {
      PrtTim("BeginOfCount");
      CountLock clk;
      int kk;
      for(kk=0; kk<atoi(arg[3]); kk++) {
        clk.Count();
      }
      cout << " End CountLock-Test Count= " << clk.Count() << endl;
    }
  }
  catch (PThrowable exc) {
    cerr << "zthr: catched Exception " << exc.Msg() << endl;
    rc = 77;
  }  
  catch (...) {
    cerr << " catched unknown (...) exception (lpk.cc) " << endl; 
    rc = 78; 
  } 
  
  return(rc);
  
}