Archiver.cpp

#include "Archiver.h"
#include "ArchivingWorkerGroup.h"
#include "Sct/Archive/IOManagerArchiveFile.h" 
#include "Sct/SctNames.h"
#include "Sct/IpcObjectException.h"
#include "Sct/UnsupportedOperationError.h"
#include "Sct/StdExceptionWrapper.h"
#include "is/info.h"
#include <pmg/pmg_initSync.h>

#include "unistd.h"
#include <memory>

#include "TransferCommand.h"
#include "IsGetCommand.h"
#include "IsPutCommand.h"
#include "ArchiveGetCommand.h"
#include "ArchivePutCommand.h"
#include "Sct/Archive/IOParamsArchive.h"
#include "Sct/OmniMarshalling.h"
#include "Sct/ApplicationStartupDebugTools.h"

#include "ipc/core.h"

using namespace std;

using namespace Sct;
using namespace Sct::Archive;
using namespace Sct::IS;

using boost::recursive_mutex;

/* Obsolete:
void pmgSynch(void *) {
    pmg_initSync();
}
*/

int main(int argc, char** argv) {
    using namespace SctArchiving;
    setExceptionHandlers(argv[0]);

    IPCCore::init(argc,argv);
    Sct::ApplicationStartupDebugTools::announceStartOfMain(argc, argv, __FILE__, __LINE__);
 
    // renice this job!
    nice(10);
    
    ArchiverArguments args(argc, argv);
    args.print(std::cout);

    try {
      Archiver& s=Archiver::initialize(args);
      if (!s.publish()){
    throw IpcObjectException("ArchivingService failed to publish", __FILE__, __LINE__);
      }
      s.go(1);
      pmg_initSync();
      s.getArchiverServer().run();
    } catch (Throwable& e) {
      e.sendToMrs(MRS_FATAL);
      terminate();
    }
}

namespace SctArchiving {

Archiver* Archiver::archiver = 0;

Archiver& Archiver::instance() {
  if(!archiver) throw IllegalStateError("Attempt to use uninitialised Archiver", __FILE__, __LINE__);
  return *archiver;
}

Archiver& Archiver::initialize(ArchiverArguments args){
  archiver = new Archiver(args);
  return *archiver;
}

Archiver::Archiver(ArchiverArguments args) : 
    IPCNamedObject<POA_ArchivingServiceI::ArchivingServiceInterface>(SctNames::getPartition(), args.instanceName()),
    nArchived(),     nRetrieved(0), 
    nValidated(0),   isTimeTaken(0.), 
    fileTimeTaken(0.), m_args(args) {
  
  auto_ptr<ISInfoReceiver> ir(new ISInfoReceiver(SctNames::getPartition()));
  if (!ir.get()) throw ConfigurationException("Archiver::initialize can't make infoReceiver ", __FILE__, __LINE__) ;
  infoReceiver = ir;
  workergroup = new ArchivingWorkerGroup();
  workergroup->reportTo(args.getISStatusName());
  ostringstream oss;
  oss << "Started with persistent directory=" << m_args.getPersistentDirectory()
      << " and retrieval server = " << m_args.getOutputISServer();

  SctNames::Mrs() << "ARCHIVING_SERVICE" << MRS_TEXT(oss.str()) 
          << MRS_INFORMATION << ENDM;
  m_archive_manager = &IOManagerArchiveFile::instance();
  m_retrieval_is_server=Sct::SctNames::getRetrievedDataName();
  m_suspend_callback=false;
  m_archive_manager->setCompressionLevel(m_args.getCompressionLevel());
}

Archiver::~Archiver() throw() {
  while (workergroup->busy() || workergroup->queueSize() ) {
    sleep(1);
  }
  workergroup->join();
}

void Archiver::addCommand(boost::shared_ptr<ArchivingCommand> command)const {
  workergroup->push(command);
}

IPCServer& Archiver::getArchiverServer() throw() {
  static IPCServer archiverServer;
  return archiverServer;
}

inline bool isControl(std::string name){
  return ( name.find("TestData")!=string::npos || name.find("ControlData")!=string::npos);
};

void Archiver::archive_callback(ISCallbackInfo *isc){
  try{
    bool over_write=isControl(isc->name());
    if (isc->reason() == is::Updated && !over_write ) {
      throw IoException(string("That is odd ... ") + isc->name() +  " has been modified! I'm not going to overwrite the data."
            , __FILE__, __LINE__);
    }
    if (isc->reason() != is::Created && isc->reason() != is::Updated) return;
    if (Archiver::instance().m_suspend_callback){
        return;
    }

    shared_ptr<IONameIS> name ( new IONameIS(isc->name()));
    shared_ptr<GetCommand> get (new IsGetCommand(name));
    shared_ptr<PutCommand> put (new ArchivePutCommand());

    if (over_write){
      shared_ptr<Sct::Archive::IOParamsArchive> params;
      params = shared_ptr<Sct::Archive::IOParamsArchive>(new IOParamsArchive());
      params->setOverWrite();
      put->setParams(params);
    }

    shared_ptr<TransferCommand> command( new TransferCommand(get, put));
    Archiver::instance().workergroup->push( command );
    
  } catch (Sct::Throwable& e){
    e.sendToMrs();
  }catch(std::exception& e){
    StdExceptionWrapper(e).sendToMrs();
  }
}

IOManagerArchive& Archiver::getIOManagerArchive() const {
  return *m_archive_manager;
}

string Archiver::internal_getRetrieveIsServer() const throw(){
  return  m_retrieval_is_server;
}

  
char * Archiver::getRetrieveIsServer() throw() {
  return copyStringToCorba(internal_getRetrieveIsServer());
};

void  Archiver::setRetrieveIsServer (const char * serverName){
  m_retrieval_is_server=serverName;
}

void Archiver::setPersistentDirectory(const char* newDirectory) {
  SctNames::setPersistentDir(newDirectory);
}

void Archiver::suspendCallbacks (CORBA::Boolean doSuspend){
   if (doSuspend) SctNames::Mrs() << MRS_TEXT("ArchivingService being suspended from callbacks") 
                  << "ARCHIVE_SUSPEND" << MRS_INFORMATION << ENDM;
   m_suspend_callback=doSuspend;
}


void Archiver::subscribe(const string& servername, const string& regexp, ISCallbackInfo::Callback callback){
  if (servername==m_args.getOutputISServer()){
    ostringstream oss;
    oss << "ArchivingService is not allowed to subscribe to [" << servername 
    << "] since it is publishing there and this could set up a loop";
    IoException(oss.str(), __FILE__, __LINE__).sendToMrs();
  }
  ISInfo::Status s=infoReceiver->subscribe(servername.c_str(), regexp.c_str(), callback);
  if (s!=ISInfo::Success) {
    ostringstream os;
    os <<"Archiving service could not subscribe to [" << servername
       << "] to retrieve [" << regexp << "]";
    IsException(s, os.str(), __FILE__, __LINE__).sendToMrs();
  } else {
    ostringstream oss;
    oss << "Archiving service subscribed to IS server [" << servername
    << "] to retrieve [" << regexp << "]";
    SctNames::Mrs() << "ARCHIVE_SUBSCRIBE" << MRS_TEXT(oss.str())
            << MRS_INFORMATION << ENDM;
  }
}
  
  void Archiver::go(unsigned nWorker) throw(IsException) {
    const list<SctService::Arguments::Subscription> theList = m_args.getInputISServers();
    for (list<SctService::Arguments::Subscription>::const_iterator i=theList.begin(); 
     i!= theList.end(); ++i){
      subscribe((*i).server, (*i).regexp, archive_callback);
    }
    workergroup->go(m_args.getNWorkers());
    if (m_args.recoveryMode()) recover();
  }

  void Archiver::incrimentNArchived() throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++nArchived;
  }
  
  void Archiver::incrimentNRetrieved() throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++nRetrieved;
  }
  
  void Archiver::incrimentNValidated() throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    ++nValidated;
  }
  
  void Archiver::addISTime(double time) throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    isTimeTaken += time;
  }

  void Archiver::addFileTime(double time) throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    fileTimeTaken += time;
  }

const char* Archiver::getStatus() const throw(){
  recursive_mutex::scoped_lock scoped_lock(counterMutex);
  ostringstream os;
  os << "\t=== Archiving Service ===" << endl
     << "nWorkers=" << workergroup->nWorkers() << endl
     << "\t(Busy=" << workergroup->busy() << ")" << endl
     << "Queue = " << workergroup->queueSize();
  if (workergroup->isFifo()) {
    os << " [FIFO]\n";
  }else{
    os << " [FILO]\n";
  }
  os<< "# Archived = " << nArchived << endl
     << "# Retrieved = " << nRetrieved  <<endl
     << "# Validated = " << nValidated << endl
     << " Approx average time / command / thread :" << endl 
     << "\tArchive = " << getAverageTime(fileTimeTaken) << endl
     << "\tIS/ISProxy = " << getAverageTime(isTimeTaken) << endl;
  os << "Compression level = " << getIOManagerArchive().getCompressionLevel() << endl;
  os << "IOManagerArchive Information: \n"
     << getIOManagerArchive().status() << endl;
  return os.str().c_str();
}

  double Archiver::getAverageTime(double time) const throw(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    long ncommands =  nArchived + nRetrieved + nValidated;
    return (ncommands>0) ? time/ncommands : 0.;
  }

  char* Archiver::status(){
    recursive_mutex::scoped_lock scoped_lock(counterMutex);
    std::string status=getStatus();
    char *statusMessage = new char[status.size()+1];
    strcpy(statusMessage, status.c_str());
    return statusMessage;
  }
  
  void Archiver::archiveISName(const char* ioNameIS){
    try {
      shared_ptr<IONameIS> name( new IONameIS(ioNameIS));
      shared_ptr<GetCommand> get (new IsGetCommand(name));
      shared_ptr<PutCommand> put (new ArchivePutCommand());

      if (isControl(ioNameIS)){
    shared_ptr<Sct::Archive::IOParamsArchive> params;
    params = shared_ptr<Sct::Archive::IOParamsArchive>(new IOParamsArchive());
    params->setOverWrite();
    put->setParams(params);
      }

      shared_ptr<TransferCommand> command( new TransferCommand(get, put));
      workergroup->push( command );
    }catch(Throwable& e){
      e.sendToMrs();
    }
  }

  void Archiver::retrieveISName (const char* ioNameIS){
    try{ 
      IONameIS nameis(ioNameIS);
      IONameArchiveFile archfile( nameis.getUniqueID(), nameis.getClassName());
      char* blah = const_cast<char*> (archfile.getIOName().c_str());
      retrieveArchName (blah);
    }catch(Throwable& e){
      e.sendToMrs();
    }
  }

  void Archiver::retrieve(const char* runNumber, const char* scanNumber, const char* className, const char* specifier){
    try{ 
      std::ostringstream oss;
      oss << m_args.getPersistentDirectory() << "/" << className 
      << "." << runNumber << "." << scanNumber 
      << "." << specifier << ".gz";
      IONameArchiveFile arch(oss.str());
      char* blah = const_cast<char*>(arch.getIOName().c_str());
      retrieveArchName(blah);
    }catch(Throwable& e){
      e.sendToMrs();
    }
  }

  void Archiver::retrieveArchName (const char* archivingName){
    try{
      shared_ptr<IONameArchiveFile> name(new IONameArchiveFile(archivingName));
      shared_ptr<GetCommand> get (new ArchiveGetCommand(name));
      shared_ptr<IsPutCommand> put (new IsPutCommand());
      put->setServer(internal_getRetrieveIsServer());
      shared_ptr<TransferCommand> command( new TransferCommand(get, put));
      workergroup->push( command );
    }catch(Throwable& e){
      e.sendToMrs();
    }
  }

  
  CORBA::Short Archiver::busy(){
    return workergroup->busy();
  }

  CORBA::Short Archiver::queueLength(){
    return workergroup->queueSize();
  }

  CORBA::Short Archiver::getCompressionLevel () {
    return getIOManagerArchive().getCompressionLevel();
  }
  
  bool Archiver::isFifo(){
    return workergroup->isFifo();
  }

  void Archiver::setFifo(bool val){
    workergroup->setFifo(val);
  }

  void Archiver::setCompressionLevel (CORBA::Short level) {
    getIOManagerArchive().setCompressionLevel(level);
  }
  void Archiver::recover(){
    SctNames::Mrs() << MRS_TEXT("Sorry - recovery not yet implimented") 
            << "ANALYSIS_RECOVER" << MRS_DIAGNOSTIC << ENDM;
  }
} // end of namespace SctArchiving

---