Added TThread::Lock when new task is assinging

Martin Vala · Martin Vala · commit f68394349a5a · 2012-09-17T10:00:49.000+02:00
diff --git a/ParallelTasks/TTaskManager.cxx b/ParallelTasks/TTaskManager.cxx
@@ -25,13 +25,13 @@ ClassImp(TTaskManager)
 
 //_________________________________________________________________________________________________
 TTaskManager::TTaskManager(const char *name, const char *title) :
-   TTaskParallel(name, title),
-   fTaskThreadPools(0),
-   fIsAllAssigned(kFALSE),
-   fDepCondition(),
-   fUseMonitoring(kFALSE),
-   fMonThreadPool(0),
-   fTaskMon("monServ", "Task Monitor Serv")
+TTaskParallel(name, title),
+fTaskThreadPools(0),
+fIsAllAssigned(kFALSE),
+fDepCondition(),
+fUseMonitoring(kFALSE),
+fMonThreadPool(0),
+fTaskMon("monServ", "Task Monitor Serv")
 {
    // Std constructor
 
@@ -54,12 +54,13 @@ TTaskManager::~TTaskManager() {
 void TTaskManager::Exec(Option_t *option) {
    // Exec of manager task
 
-//   Printf("TTaskManager::Exec START ...");
+   //   Printf("TTaskManager::Exec START ...");
 
    if (!fParent && !fTaskThreadPools) {
+      //      TLockGuard lock(&fMutex);
 
       Printf("TTaskManager::SHouldBeInit ...");
-      TLockGuard lock(&fMutex);
+
       TTaskPoolManager *tpm;
       fTaskThreadPools = new TObjArray();
 
@@ -73,50 +74,61 @@ void TTaskManager::Exec(Option_t *option) {
       for (i = 0; i < kAllTypes; i++) {
          tpm = new TTaskPoolManager(fNumOfThreads[i]);
          fTaskThreadPools->Add(tpm);
-//         tpm->Print();
+         //         tpm->Print();
       }
-
    }
 
    // Loops until not all tasks are assigned
-//   Int_t counter = 0;
+   //   Int_t counter = 0;
    while (1) {
-      TLockGuard lock(&fMutex);
-//      Printf("Manager loop %d", counter++);
-      SetAllAssigned();
+
+      TThread::Lock();
+      //      TLockGuard lock(&fMutex);
+      //      Printf("Manager loop %d", counter++);
+//      {
+//         TLockGuard lock(&fMutex);
+         fIsAllAssigned = kTRUE;
+//      }
       RunTask(option);
-      if (fIsAllAssigned) break;
+      if (fIsAllAssigned) {TThread::UnLock();break;}
+      TThread::UnLock();
       fDepCondition.Wait();
    }
 
-   TIter next(fTaskThreadPools);
-   TTaskPoolManager *pool;
-   while ((pool = (TTaskPoolManager *)next())) {
-      pool->Stop(kTRUE,kTRUE);
+   {
+      TLockGuard lock(&fMutex);
+
+      TIter next(fTaskThreadPools);
+      TTaskPoolManager *pool;
+      while ((pool = (TTaskPoolManager *)next())) {
+         pool->Stop(kTRUE,kTRUE);
+      }
    }
    //
    //////      TODO Serving tasks
    ////      FinishServingTasks();
    ////
    // wait untill serving task are are finished
-//   next.Reset();
-//   while ((pool = (TTaskPoolManager *)next())) {
-//      pool->Stop(kTRUE);
-//   }
+   //   next.Reset();
+   //   while ((pool = (TTaskPoolManager *)next())) {
+   //      pool->Stop(kTRUE);
+   //   }
    //
    //   // sync monitoring
    //   if (fUseMonitoring) fMonThreadPool->Stop(kTRUE);
    //
    if (!fParent) RestoreManager();
 
-//   Printf("Done");
+   //   Printf("Done");
 }
 
 //_________________________________________________________________________________________________
 void TTaskManager::PushTask(TTaskParallel *t) {
 
-   TTaskPoolManager *taskPoolMgr = (TTaskPoolManager *)fTaskThreadPools->At(t->GetType());
-   taskPoolMgr->PushTask(t);
+   {
+      TTaskPoolManager *taskPoolMgr = (TTaskPoolManager *)fTaskThreadPools->At(t->GetType());
+      taskPoolMgr->PushTask(t);
+   }
 }
 
 //_________________________________________________________________________________________________
@@ -162,12 +174,12 @@ void TTaskManager::RestoreManager() {
    TThread::Lock();
    SetStatusType(TTaskParallel::kWaiting, kTRUE);
    TThread::UnLock();
-//   TIter next(fTaskThreadPools);
-//   TTaskPoolManager *pool;
-//   Printf("Restore Manager Pools %lld %s", fTaskThreadPools->GetEntries(), GetName());
-//   while ((pool = (TTaskPoolManager *)next())) {
-//      pool->SetStop(kFALSE);
-//   }
+   //   TIter next(fTaskThreadPools);
+   //   TTaskPoolManager *pool;
+   //   Printf("Restore Manager Pools %lld %s", fTaskThreadPools->GetEntries(), GetName());
+   //   while ((pool = (TTaskPoolManager *)next())) {
+   //      pool->SetStop(kFALSE);
+   //   }
 
 
    //   Printf("Main Task Manager sleeping ");
diff --git a/ParallelTasks/TTaskManager.h b/ParallelTasks/TTaskManager.h
@@ -44,7 +44,6 @@ class TTaskManager : public TTaskParallel {
 private:
 
    static TTaskManager *fgTaskManager;              // global manager
-   TMutex               fMutex;                     // mutex for manager
    TObjArray           *fTaskThreadPools;           // list of thread pools
    Int_t                fNumOfThreads[kAllTypes];   // number of threads reserved for different type
    Bool_t               fIsAllAssigned;             // flag if all tasks are done
diff --git a/ParallelTasks/TTaskParallel.cxx b/ParallelTasks/TTaskParallel.cxx
@@ -18,11 +18,12 @@ ClassImp(TTaskParallel)
 
 //_________________________________________________________________________________________________
 TTaskParallel::TTaskParallel(const char *name, const char *title) :
-   TTask(name, title),
-   fTaskStatusType(kWaiting),
-   fTaskType(kCpu),
-   fParent(0),
-   fListDeps(0)
+TTask(name, title),
+fTaskStatusType(kWaiting),
+fTaskType(kCpu),
+fParent(0),
+fListDeps(0),
+fMutex()
 {
    // Std constructor
 
@@ -37,11 +38,11 @@ TTaskParallel::~TTaskParallel() {
 
 //_________________________________________________________________________________________________
 TTaskParallel::TTaskParallel(const TTaskParallel &obj) :
-   TTask(obj),
-   fTaskStatusType(obj.fTaskStatusType),
-   fTaskType(obj.fTaskType),
-   fParent(obj.fParent),
-   fListDeps(obj.fListDeps)
+         TTask(obj),
+         fTaskStatusType(obj.fTaskStatusType),
+         fTaskType(obj.fTaskType),
+         fParent(obj.fParent),
+         fListDeps(obj.fListDeps)
 
 {
    //
@@ -109,24 +110,31 @@ void TTaskParallel::RunTask(Option_t *option, TTaskParallel::ETaskType /*type*/)
    if (!fTasks) return;
 
    TIter next(fTasks);
-   TTask *task;
+//   TTask *task;
    TTaskParallel *t;
    TTaskManager *taskMgr = TTaskManager::GetTaskManager();
-   while ((task = (TTask *) next())) {
-      if (!task->IsActive()) continue;
-      t = (TTaskParallel *) task;
-//      Printf("Testing %s [%s] %p", t->GetName(), t->GetStatusTypeName(), t->GetParent());
+   while ((t = (TTaskParallel *) next())) {
+      if (!t->IsActive()) continue;
+//      t = (TTaskParallel *) task;
+      //      Printf("Testing %s [%s] %p", t->GetName(), t->GetStatusTypeName(), t->GetParent());
       if (t->GetStatusType() == TTaskParallel::kWaiting) {
          if (t->HasDependency()) {
             // task was not assigned
-            taskMgr->SetAllAssigned(kFALSE);
+            {
+//               TLockGuard lock(taskMgr->GetMutex());
+               taskMgr->SetAllAssigned(kFALSE);
+            }
          } else {
-            t->SetStatusType(TTaskParallel::kAssigned);
-//            Printf("Pushing task %s [%s]", t->GetName(), t->GetStatusTypeName());
+            {
+//               TLockGuard lock2(taskMgr->GetMutex());
+//               TLockGuard lock(t->GetMutex());
+               t->SetStatusType(TTaskParallel::kAssigned);
+            }
+            //            Printf("Pushing task %s [%s]", t->GetName(), t->GetStatusTypeName());
             taskMgr->PushTask(t);
          }
       }
-//      gSystem->Sleep(100);
+      //      gSystem->Sleep(100);
       t->RunTask(option);
    }
 
@@ -145,11 +153,15 @@ Bool_t TTaskParallel::HasDependency() {
 
 //_________________________________________________________________________________________________
 void TTaskParallel::SetStatusType(ETaskStatusType t, Bool_t recursivly) {
+
+//   TLockGuard lock2(GetMutex());
+
    fTaskStatusType = t;
    if (recursivly) {
       TIter next(fTasks);
       TTaskParallel *task;
       while ((task = (TTaskParallel *)next())) {
+//         TLockGuard lock(task->GetMutex());
          task->SetStatusType(t, recursivly);
       }
    }
@@ -159,18 +171,18 @@ void TTaskParallel::SetStatusType(ETaskStatusType t, Bool_t recursivly) {
 const char *TTaskParallel::GetStatusTypeName(ETaskStatusType t) {
 
    switch (t) {
-      case kWaiting:
-         return "W";
-      case kAssigned:
-         return "A";
-      case kRunning:
-         return "R";
-      case kDoneServing:
-         return "DS";
-      case kDone:
-         return "D";
-      default:
-         return "";
+   case kWaiting:
+      return "W";
+   case kAssigned:
+      return "A";
+   case kRunning:
+      return "R";
+   case kDoneServing:
+      return "DS";
+   case kDone:
+      return "D";
+   default:
+      return "";
    }
    return "";
 }
@@ -179,14 +191,14 @@ const char *TTaskParallel::GetStatusTypeName(ETaskStatusType t) {
 const char *TTaskParallel::GetTypeName(ETaskType t) {
 
    switch (t) {
-      case kCpu:
-         return "CPU";
-      case kIO:
-         return "IO";
-      case kFake:
-         return "FAKE";
-      default:
-         return "";
+   case kCpu:
+      return "CPU";
+   case kIO:
+      return "IO";
+   case kFake:
+      return "FAKE";
+   default:
+      return "";
    }
    return "";
 }
diff --git a/ParallelTasks/TTaskParallel.h b/ParallelTasks/TTaskParallel.h
@@ -48,13 +48,16 @@ class TTaskParallel : public TTask {
    void RunTask(Option_t *option, TTaskParallel::ETaskType type = TTaskParallel::kCpu);
 
    Bool_t HasDependency();
+   TMutex *GetMutex() { return &fMutex;}
+
 
 protected:
 
-   ETaskStatusType        fTaskStatusType;// Status tyoe
-   ETaskType              fTaskType;      // task type
-   TTask                 *fParent;        // partent taks
-   TList                 *fListDeps;      // dependency list
+   ETaskStatusType      fTaskStatusType;     // Status tyoe
+   ETaskType            fTaskType;           // task type
+   TTask               *fParent;             // partent taks
+   TList               *fListDeps;           // dependency list
+   TMutex               fMutex;              // mutex for manager
 
    ClassDef(TTaskParallel, 1)
 
diff --git a/ParallelTasks/TTaskStress.cxx b/ParallelTasks/TTaskStress.cxx
@@ -63,6 +63,10 @@ void TTaskStress::Exec(Option_t *option) {
    // Exec of manager task
    //
 
+//   Printf("Running task %s",GetName());
+//   gSystem->Sleep(1);
+   return;
+
    TStopwatch timer;
    timer.Start();
 
@@ -83,12 +87,15 @@ void TTaskStress::Exec(Option_t *option) {
 
    if (fType == kSleep) {
 //      TThread::Lock();
-      UInt_t time = (UInt_t) r.Gaus(5000, 1000);
-      time = 1;
+
+      Int_t multi=1000;
+      multi = 10;
+      UInt_t time = (UInt_t) r.Gaus(5*multi, 1*multi);
+//      time = 1;
 //      TThread::UnLock();
-//      Printf("[%s]%s%s S (%ld)", option, prefix.Data(), GetName(), time);
+//      Printf("[%s]%s%s S (%lld)", option, prefix.Data(), GetName(), time);
       gSystem->Sleep(time);
-//      Printf("[%s]%s%s D (%ld)", option, prefix.Data(), GetName(), time);
+//      Printf("[%s]%s%s D (%lld)", option, prefix.Data(), GetName(), time);
    } else if (fType == kCpu) {
       TH1D h(TString::Format("myTaskStress_%s", GetName()).Data(), "My Stress task hist", 100, -10, 10);
       h.SetDirectory(0);
diff --git a/macros/ParallelTasks/TestTaskManagerSimple.C b/macros/ParallelTasks/TestTaskManagerSimple.C
