Commit 3077b50d authored by David COME's avatar David COME
Browse files

Added RecallReportPacker& to DiskWriteThreadPool

Changed the constructor to take that into account and its calls in several files
parent 1403fc12
......@@ -32,6 +32,7 @@
#include "DiskThreadPoolInterface.hpp"
#include "castor/log/LogContext.hpp"
#include "castor/tape/tapeserver/utils/suppressUnusedVariable.hpp"
#include "castor/tape/tapeserver/daemon/RecallReportPacker.hpp"
#include <vector>
#define __STDC_FORMAT_MACROS
#include <inttypes.h>
......@@ -43,7 +44,8 @@ namespace daemon {
class DiskWriteThreadPool : public DiskThreadPoolInterface<DiskWriteTaskInterface> {
public:
DiskWriteThreadPool(int nbThread, int maxFilesReq, int maxBlocksReq,castor::log::LogContext lc);
DiskWriteThreadPool(int nbThread, int maxFilesReq, int maxBlocksReq,
RecallReportPacker& report,castor::log::LogContext lc);
~DiskWriteThreadPool();
void startThreads();
......@@ -89,6 +91,7 @@ private:
uint32_t m_maxFilesReq;
uint64_t m_maxBytesReq;
RecallReportPacker& m_reporter;
castor::log::LogContext m_lc;
};
......
......@@ -246,13 +246,13 @@ void castor::tape::tapeserver::daemon::MountSession::executeRead(LogContext & lc
// to refer them to each other)
MemoryManager mm(m_castorConf.rtcopydNbBufs, m_castorConf.rtcopydBufsz);
TapeReadSingleThread trst(*drive);
DiskWriteThreadPool dwtp(m_castorConf.tapeserverdDiskThreads,
m_castorConf.tapebridgeBulkRequestRecallMaxFiles,
m_castorConf.tapebridgeBulkRequestRecallMaxBytes,lc);
RecallReportPacker rrp(m_clientProxy,
m_castorConf.tapebridgeBulkRequestMigrationMaxFiles,
lc);
RecallTaskInjector rti(mm, trst, dwtp, m_clientProxy, rrp, lc);
DiskWriteThreadPool dwtp(m_castorConf.tapeserverdDiskThreads,
m_castorConf.tapebridgeBulkRequestRecallMaxFiles,
m_castorConf.tapebridgeBulkRequestRecallMaxBytes,rrp,lc);
RecallTaskInjector rti(mm, trst, dwtp, m_clientProxy, lc);
dwtp.setJobInjector(&rti);
// We are now ready to put everything in motion. First step is to check
......
......@@ -9,12 +9,6 @@
using castor::log::LogContext;
using castor::log::Param;
namespace{
uint32_t blockID(const castor::tape::tapegateway::FileToRecallStruct& ftr)
{
return (ftr.blockId0() << 24) | (ftr.blockId1() << 16) | (ftr.blockId2() << 8) | ftr.blockId3();
}
}
namespace castor{
namespace tape{
......@@ -24,10 +18,10 @@ namespace daemon {
RecallTaskInjector::RecallTaskInjector(MemoryManager & mm,
TapeSingleThreadInterface<TapeReadTask> & tapeReader,
DiskThreadPoolInterface<DiskWriteTaskInterface> & diskWriter,
client::ClientInterface& client,RecallReportPacker& report,castor::log::LogContext lc) :
client::ClientInterface& client,castor::log::LogContext lc) :
m_thread(*this),m_memManager(mm),
m_tapeReader(tapeReader),m_diskWriter(diskWriter),
m_client(client),m_lc(lc),m_report(report)
m_client(client),m_lc(lc)
{}
void RecallTaskInjector::requestInjection(int maxFiles, int byteSizeThreshold, bool lastCall) {
......@@ -59,7 +53,7 @@ void RecallTaskInjector::injectBulkRecalls(const std::vector<castor::tape::tapeg
m_lc.log(LOG_INFO, "Logged file to recall");
DiskWriteTask * dwt = new DiskWriteTask((*it)->id(), blockID(**it),(*it)->path() ,m_memManager,m_report);
DiskWriteTask * dwt = new DiskWriteTask(**it ,m_memManager);
TapeReadFileTask * trt = new TapeReadFileTask(*dwt, (*it)->fseq(), blockID(**it));
m_diskWriter.push(dwt);
......
......@@ -51,7 +51,7 @@ public:
RecallTaskInjector(MemoryManager & mm,
TapeSingleThreadInterface<TapeReadTask> & tapeReader,
DiskThreadPoolInterface<DiskWriteTaskInterface> & diskWriter,client::ClientInterface& client,
RecallReportPacker& report,castor::log::LogContext lc);
castor::log::LogContext lc);
/**
......@@ -138,7 +138,6 @@ private:
castor::tape::threading::Mutex m_producerProtection;
castor::tape::threading::BlockingQueue<Request> m_queue;
RecallReportPacker& m_report;
};
} //end namespace daemon
......
......@@ -66,7 +66,7 @@ TEST(castor_tape_tapeserver_daemon, RecallTaskInjectorNominal) {
FakeDiskWriteThreadPool diskWrite;
FakeSingleTapeReadThread tapeRead(drive);
tapeserver::daemon::RecallReportPacker rrp(client,2,lc);
tapeserver::daemon::RecallTaskInjector rti(mm,tapeRead,diskWrite,client,rrp,lc);
tapeserver::daemon::RecallTaskInjector rti(mm,tapeRead,diskWrite,client,lc);
ASSERT_EQ(true,rti.synchronousInjection(6,blockSize));
ASSERT_EQ(nbFile,diskWrite.m_tasks.size());
......@@ -92,6 +92,7 @@ TEST(castor_tape_tapeserver_daemon, RecallTaskInjectorNominal) {
DiskWriteTaskInterface* diskWriteTask=diskWrite.m_tasks.pop();
TapeReadTask* tapeReadTask=tapeRead.m_tasks.pop();
//static_cast is needed otherwise compilation fails on SL5 with a raw NULL
ASSERT_EQ(static_cast<DiskWriteTaskInterface*>(NULL),diskWriteTask);
ASSERT_EQ(true,tapeReadTask->endOfWork());
delete diskWriteTask;
......@@ -109,7 +110,7 @@ TEST(castor_tape_tapeserver_daemon, RecallTaskInjectorNoFiles) {
FakeSingleTapeReadThread tapeRead(drive);
tapeserver::daemon::RecallReportPacker rrp(client,2,lc);
tapeserver::daemon::RecallTaskInjector rti(mm,tapeRead,diskWrite,client,rrp,lc);
tapeserver::daemon::RecallTaskInjector rti(mm,tapeRead,diskWrite,client,lc);
ASSERT_EQ(false,rti.synchronousInjection(6,blockSize));
ASSERT_EQ(0U,diskWrite.m_tasks.size());
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment