Added instanciation of the threading machinery in the MountSession.

We currently only cover the Recalls, and we still need to start the threads.

castor/tape/tapeserver/daemon/MountSession.cpp

@@ -35,6 +35,7 @@
 #include "h/serrno.h"
 #include "castor/tape/tapeserver/SCSI/Device.hpp"
 #include "castor/tape/tapeserver/drive/Drive.hpp"
+#include "RecallTaskInjector.hpp"
 
 using namespace castor::tape;
 using namespace castor::log;
@@ -42,9 +43,10 @@ using namespace castor::log;
 castor::tape::tapeserver::daemon::MountSession::MountSession(
     const legacymsg::RtcpJobRqstMsgBody& clientRequest, 
     castor::log::Logger& logger, System::virtualWrapper & sysWrapper,
-    const utils::TpconfigLines & tpConfig): 
+    const utils::TpconfigLines & tpConfig,
+    const CastorConf & castorConf): 
     m_request(clientRequest), m_logger(logger), m_clientProxy(clientRequest),
-    m_sysWrapper(sysWrapper), m_tpConfig(tpConfig) {}
+    m_sysWrapper(sysWrapper), m_tpConfig(tpConfig), m_castorConf(castorConf) {}
 
 void castor::tape::tapeserver::daemon::MountSession::execute()
 throw (castor::tape::Exception) {
@@ -117,7 +119,7 @@ void castor::tape::tapeserver::daemon::MountSession::executeRead(LogContext & lc
   // We are ready to start the session. In case of read there is no interest in
   // creating the machinery before getting the tape mounted, so do it now.
   // 1) Get hold of the drive and check it.
-  utils::TpconfigLines::iterator configLine;
+  utils::TpconfigLines::const_iterator configLine;
   for (configLine = m_tpConfig.begin(); configLine != m_tpConfig.end(); configLine++) {
     if (configLine->unitName == m_request.driveUnit && configLine->density == m_volInfo.density) {
       break;
@@ -237,7 +239,18 @@ void castor::tape::tapeserver::daemon::MountSession::executeRead(LogContext & lc
     lc.log(LOG_ERR, "Notified client of end session with error");
     return;
   }
-  // TBC...
+  // We can now start instantiating all the components of the data path
+  {
+    castor::tape::tapeserver::daemon::MemoryManager mm(m_castorConf.rtcopydNbBufs, 
+        m_castorConf.rtcopydBufsz);
+    castor::tape::tapeserver::daemon::TapeReadSingleThread trst(*drive);
+    castor::tape::tapeserver::daemon::DiskWriteThreadPool dwtp(
+        m_castorConf.tapeserverdDiskThreads,
+        m_castorConf.tapebridgeBulkRequestRecallMaxFiles,
+        m_castorConf.tapebridgeBulkRequestRecallMaxBytes);
+    castor::tape::tapeserver::daemon::RecallTaskInjector rti(mm, trst, 
+        dwtp, m_clientProxy, lc);
+  }
   {
     // Temporary end of session, to please the ClientSimulator
     ClientProxy::RequestReport reqReport;

castor/tape/tapeserver/daemon/MountSession.hpp

@@ -30,6 +30,7 @@
 #include "castor/tape/tapeserver/system/Wrapper.hpp"
 #include "castor/tape/utils/utils.hpp"
 #include "ClientProxy.hpp"
+#include "TapeSingleThreadInterface.hpp"
 
 using namespace castor::tape;
 using namespace castor::log;
@@ -45,10 +46,51 @@ namespace daemon {
    */
   class MountSession {
   public:
+    /** Subclass holding all the contents from castor.conf file. The pre-
+     * extraction of the contents by the caller instead of direct getconfent
+     * calls allows unit testing */
+    class CastorConf {
+    public:
+      CastorConf(): rfioConnRetry(0), rfioConnRetryInt(0), rfioIOBufSize(0),
+        rtcopydBufsz(0), rtcopydNbBufs(0), tapeConfirmDriveFreeInterval(0),
+        tapeserverdDiskThreads(0) {}
+      uint32_t rfioConnRetry;
+      uint32_t rfioConnRetryInt;
+      uint32_t rfioIOBufSize;
+      uint32_t rtcopydBufsz;
+      uint32_t rtcopydNbBufs;
+      std::string tapeBadMIRHandlingRepair;
+      std::string tapeConfirmDriveFree;
+      uint32_t tapeConfirmDriveFreeInterval;
+      uint32_t tapebridgeBulkRequestMigrationMaxBytes;
+      uint32_t tapebridgeBulkRequestMigrationMaxFiles;
+      uint32_t tapebridgeBulkRequestRecallMaxBytes;
+      uint32_t tapebridgeBulkRequestRecallMaxFiles;
+// Other values found on production tape servers
+//      TAPE CRASHED_RLS_HANDLING RETRY
+//      TAPE CRASHED_RLS_HANDLING_RETRIES 3
+//      TAPE CRASHED_RLS_HANDLING_RETRY_DELAY 180
+//      TAPE DOWN_ON_TPALERT NO
+//      TAPE DOWN_ON_UNLOAD_FAILURE NO
+//      TAPEBRIDGE BULKREQUESTMIGRATIONMAXBYTES 5000000000
+//      TAPEBRIDGE BULKREQUESTMIGRATIONMAXFILES 500
+//      TAPEBRIDGE BULKREQUESTRECALLMAXBYTES 5000000000
+//      TAPEBRIDGE BULKREQUESTRECALLMAXFILES 500
+//      TAPEBRIDGE MAXBYTESBEFOREFLUSH 32000000000
+//      TAPEBRIDGE MAXFILESBEFOREFLUSH 500
+//      TAPEBRIDGE TAPEFLUSHMODE ONE_FLUSH_PER_N_FILES
+//      UPV HOST castorcupv
+//      VDQM HOST castorvdqm1
+//      VMGR HOST castorvmgr
+
+      // Additions for tapeserverd
+      uint32_t tapeserverdDiskThreads;
+    };
     /** Constructor */
     MountSession(const legacymsg::RtcpJobRqstMsgBody & clientRequest, 
             castor::log::Logger & logger, System::virtualWrapper & sysWrapper,
-            const utils::TpconfigLines & tpConfig);
+            const utils::TpconfigLines & tpConfig,
+            const CastorConf & castorConf);
     /** The only method. It will execute (like a task, that it is) */
     void execute() throw (Exception);
     /** Temporary method used for debugging while building the session class */
@@ -59,7 +101,8 @@ namespace daemon {
     ClientProxy m_clientProxy;
     ClientProxy::VolumeInfo m_volInfo;
     System::virtualWrapper & m_sysWrapper;
-    utils::TpconfigLines m_tpConfig;
+    const utils::TpconfigLines & m_tpConfig;
+    const CastorConf & m_castorConf;
     /** sub-part of execute for the read sessions */
     void executeRead(LogContext & lc);
     /** sub-part of execute for a write session */

castor/tape/tapeserver/daemon/tapeserverdTest.cpp

@@ -87,7 +87,10 @@ TEST(tapeServer, MountSessionGoodday) {
   "/dev/tape_T10D6116", "8000GC", "down", "acs0,1,1,6", "T10000"));
   tpConfig.push_back(utils::TpconfigLine("T10D6116", "T10KD6", 
   "/dev/tape_T10D6116", "5000GC", "down", "acs0,1,1,6", "T10000"));
-  MountSession sess(VDQMjob, logger, mockSys, tpConfig);
+  MountSession::CastorConf castorConf;
+  castorConf.rtcopydBufsz = 1024;
+  castorConf.rtcopydNbBufs = 10;
+  MountSession sess(VDQMjob, logger, mockSys, tpConfig, castorConf);
   sess.execute();
   simRun.wait();
   std::string temp = logger.getLog();
@@ -131,7 +134,10 @@ TEST(tapeServer, MountSessionNoSuchDrive) {
   "/dev/noSuchTape", "8000GC", "down", "acs0,1,1,6", "T10000"));
   tpConfig.push_back(utils::TpconfigLine("T10D6116", "T10KD6", 
   "/dev/noSuchTape", "5000GC", "down", "acs0,1,1,6", "T10000"));
-  MountSession sess(VDQMjob, logger, mockSys, tpConfig);
+  MountSession::CastorConf castorConf;
+  castorConf.rtcopydBufsz = 1024;
+  castorConf.rtcopydNbBufs = 10;
+  MountSession sess(VDQMjob, logger, mockSys, tpConfig, castorConf);
   sess.execute();
   simRun.wait();
   std::string temp = logger.getLog();