Fixed missing conditional finish for last thread in DiskReadThreadPool

Renamed member variables to canonycal m_
Renamed TaskInjector->end to TaskInjector->finish() to match other threads conventions.
Fixed missing finish() in RecallTaskInjectorTest
Added missing finish() call in TapeReadTSingleThread.

castor/tape/tapeserver/daemon/DiskReadThreadPool.cpp

@@ -94,8 +94,11 @@ namespace daemon {
       }
     } //end of while(1)
     // We now acknowledge to the task injector that read reached the end. There
-    // will hence be no more requests for more.
-    m_parent.m_injector->finish();
+    // will hence be no more requests for more. (last thread turns off the light)
+    if (0 == --m_nbActiveThread) {
+      m_parent.m_injector->finish();
+      m_lc.log(LOG_DEBUG, "Signaled to task injector the end of disk read threads");
+    }
     m_lc.log(LOG_DEBUG, "Finishing of DiskReadWorkerThread");
   }
   

castor/tape/tapeserver/daemon/DiskReadThreadPool.hpp

@@ -55,17 +55,17 @@ private:
   class DiskReadWorkerThread: private castor::tape::threading::Thread {
   public:
     DiskReadWorkerThread(DiskReadThreadPool & manager):
-    threadID(m_nbActiveThread++),m_parent(manager),lc(m_parent.m_lc) {
-       log::LogContext::ScopedParam param(lc, log::Param("threadID", threadID));
-       lc.log(LOG_INFO,"DiskWrite Thread created");
+    m_threadID(m_nbActiveThread++),m_parent(manager),m_lc(m_parent.m_lc) {
+       log::LogContext::ScopedParam param(m_lc, log::Param("threadID", m_threadID));
+       m_lc.log(LOG_INFO,"DiskWrite Thread created");
     }
     void startThreads() { start(); }
     void waitThreads() { wait(); }
   private:
     static tape::threading::AtomicCounter<int> m_nbActiveThread;
-    const int threadID;
+    const int m_threadID;
     DiskReadThreadPool & m_parent;
-    castor::log::LogContext lc;
+    castor::log::LogContext m_lc;
     
     virtual void run();
   };

castor/tape/tapeserver/daemon/RecallTaskInjector.cpp

@@ -148,7 +148,7 @@ void RecallTaskInjector::WorkerThread::run()
   bool stillReading =true;
   while(stillReading) {
     Request req = _this.m_queue.pop();
-    stillReading = req.end;
+    if (req.end) stillReading = false;
     LogContext::ScopedParam sp(_this.m_lc, Param("lastCall", req.lastCall));
     _this.m_lc.log(LOG_INFO,"In RecallJobInjector::WorkerThread::run(): popping extra request");
   }

castor/tape/tapeserver/daemon/RecallTaskInjector.hpp

@@ -86,7 +86,7 @@ public:
    */
   void startThreads();
   
-  void end();
+  void finish();
 private:
   
   /**

castor/tape/tapeserver/daemon/RecallTaskInjectorTest.cpp

@@ -76,6 +76,7 @@ TEST(castor_tape_tapeserver_daemon, RecallTaskInjectorNominal) {
   rti.startThreads();
   rti.requestInjection(6,blockSize,false);
   rti.requestInjection(6,blockSize,true);
+  rti.finish();
   rti.waitThreads();
   
   //pushed nbFile*2 files + 1 end of Work task

castor/tape/tapeserver/daemon/TapeReadSingleThread.hpp

@@ -82,14 +82,19 @@ private:
     while(1) {
       // NULL indicated the end of work
       TapeReadTask * task = popAndRequestMoreJobs();
+      m_logContext.log(LOG_DEBUG, "TapeReadThread: just got one more job");
       if (task) {
         task->execute(*rs, m_logContext);
         delete task;
         m_filesProcessed++;
       } else {
-        return;
+        break;
       }
     }
+    // We now acknowledge to the task injector that read reached the end. There
+    // will hence be no more requests for more. (last thread turns off the light)
+    m_taskInjector->finish();
+    m_logContext.log(LOG_DEBUG, "Finishing Tape Read Thread. Just signalled task injector of the end");
   }
   
   uint64_t m_maxFilesRequest;

castor/tape/tapeserver/daemon/TaskInjector.hpp

@@ -32,7 +32,7 @@ namespace daemon {
 class TaskInjector{
 public:
   virtual void requestInjection(int maxFiles, int maxBlocks, bool lastCall) = 0;
-  virtual void end() = 0;
+  virtual void finish() = 0;
   virtual ~TaskInjector() {}
 };