#include "sopnamsp.h" #include "zthread.h" #include "resusage.h" #include #include #include "tmatrix.h" #include "tvector.h" #include "tarrinit.h" #include #include /* ------------------------------------------------- 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 #include #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; ithid, (int)t1); } return; } // --- fonction de test simple avec calcul matriciel (produit de 2 matrices double) 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 sur tableaux 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 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; correction le 23/05/2007 - * (prod.mtx par erreur) 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 xx(2); for(int k=0; k* 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 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 _vmx; vector< St_VecPool> _vecs; vector< TVector > _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; kthid); PrtTim(buff); uint_4 idx; for(k=0; kGetVecP(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; kthid); PrtTim(buff); uint_4 idx; for(k=0; kGetVec(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 vza; vector vzth; for(int i=0; ithid = 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; istart(); sleep(1); cout << "***ResourceUsage before thr[i].join()" << endl; cout << res; cout << "***zthr Joining Threads ..." << endl; for(int i=0; ijoin(); 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; iCheck(); tm.Nop(); delete mtvp; } } else { PrtTim("BeginOfCount"); CountLock clk; int kk; for(kk=0; kk