Changeset 873

Timestamp:

01/29/08 17:36:56

Author:

robert

Message:

Added MachinePool?::reportTimingStatus() to provide an task progress estimate.

Files:

• trunk/include/vpb/MachinePool (modified) (2 diffs)
• trunk/src/vpb/MachinePool.cpp (modified) (5 diffs)

trunk/include/vpb/MachinePool

@@ -130 +130 @@
-        void taskFailed(Task* task, int result);
-        
+        OpenThreads::Mutex& getRunningTasksMutex() const { return _runningTasksMutex; }
+        
-        RunningTasks& getRunningTasks() { return _runningTasks; }
-        
@@ -239 +241 @@
-        void updateMachinePool();
-
+        /** Generate a report of the task timing status, i.e. how long do we expect before completion.*/
+        void reportTimingStatus();
+
-        /** Generate a report of the task timing stats.*/
-        void reportTimingStats();

trunk/src/vpb/MachinePool.cpp

@@ -57 +57 @@
-            _task->write();
-            
+            // machine->log(osg::NOTICE,"machine=%s running task=%s",machine->getHostName().c_str(),_task->getFileName().c_str());
+
-            machine->startedTask(_task.get());
-
@@ -88 +90 @@
-            }
-            
-machine->log(osg::NOTICE,"%s  : completed in %f  seconds : %s result=%d",machine->getHostName().c_str(),duration,application.c_str()
+// machine->log(osg::NOTICE,"machine=%s completed task=%s in %f seconds, result=%d",machine->getHostName().c_str(),_task->getFileName().c_str(),duration
-        }
-
@@ -185 +187 @@
-    }
-
-NOTICE
+INFO
-
-    return system(executionString.c_str());
@@ -301 +303 @@
-    OpenThreads::ScopedLock<OpenThreads::Mutex> lock(_runningTasksMutex);
-    _runningTasks[task] = osg::Timer::instance()->time_s();
+
+    log(osg::NOTICE,"machine=%s running task=%s",getHostName().c_str(),task->getFileName().c_str());
-}
-
-void Machine::endedTask(Task* task)
-{
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-}
-
@@ -790 +803 @@
-}
-
+void MachinePool::reportTimingStatus()
+{
+    unsigned int numTasksPending = _operationQueue->getNumOperationsInQueue();
+
+    unsigned int numTasksCompleted = 0;
+    double totalComputeTime = 0.0;
+    unsigned int numCores = 0;
+    
+    for(Machines::iterator itr = _machines.begin();
+        itr != _machines.end();
+        ++itr)
+    {
+        Machine* machine = itr->get();
+        OpenThreads::ScopedLock<OpenThreads::Mutex> lock(machine->getRunningTasksMutex());
+
+        TaskStatsMap& taskStatsMap = machine->getTaskStatsMap();
+        for(TaskStatsMap::iterator titr = taskStatsMap.begin();
+            titr != taskStatsMap.end();
+            ++titr)
+        {
+            TaskStats& stats = titr->second;
+            
+            numTasksCompleted += stats.numTasks();
+            totalComputeTime += stats.totalTime();
+        }
+        numCores += getNumThreads();
+    }
+        
+    double averageTaskTime = (numTasksCompleted!=0) ? (totalComputeTime/ double(numTasksCompleted)) : 0;
+
+    double currentTime = osg::Timer::instance()->time_s();
+    double estimatedTimeOfLastCompletion = currentTime;
+    unsigned int numTasksRunning = 0;
+    for(Machines::iterator itr = _machines.begin();
+        itr != _machines.end();
+        ++itr)
+    {
+        Machine* machine = itr->get();
+        OpenThreads::ScopedLock<OpenThreads::Mutex> lock(machine->getRunningTasksMutex());
+        
+        Machine::RunningTasks& runningTasks = machine->getRunningTasks();
+        numTasksRunning = runningTasks.size();
+        for(Machine::RunningTasks::iterator ritr = runningTasks.begin();
+            ritr != runningTasks.end();
+            ++ritr)
+        {
+            double startTime = ritr->second;
+            double elapsedTime = currentTime - startTime;
+            double estimatedEndTime = (elapsedTime < averageTaskTime) ? startTime + averageTaskTime : currentTime+1.0;
+            if (estimatedTimeOfLastCompletion < estimatedEndTime) estimatedTimeOfLastCompletion = estimatedEndTime;
+        }
+    }
+    
+    double numTaskPendingAcrossAllCores = (numTasksPending>0) ? ceil(double(numTasksPending) / double(numCores)) : 0;
+    estimatedTimeOfLastCompletion += numTaskPendingAcrossAllCores*averageTaskTime;
+    double estimateTimeToCompletion = estimatedTimeOfLastCompletion-currentTime;
+    
+    log(osg::NOTICE,"Number of tasks completed %i, running %i, pending %i. Estimated time to completion %.1f seconds, %2.1f%% done.",numTasksCompleted, numTasksRunning, numTasksPending, estimateTimeToCompletion, 100.0*currentTime/estimatedTimeOfLastCompletion);
+
+}
+
-void MachinePool::reportTimingStats()
-{