DiskWriter.cc

#include <errno.h>
#include <iomanip>
#include <memory>
#include <regex>
#include <stdio.h>
#include <sstream>
#include <string>
#include <sys/stat.h>
#include <sys/types.h>

#include "evb/BU.h"
#include "evb/bu/DiskWriter.h"
#include "evb/bu/ResourceManager.h"
#include "evb/bu/RUproxy.h"
#include "evb/Exception.h"
#include "toolbox/task/WorkLoopFactory.h"
#include "xdata/String.h"
#include "xdata/Vector.h"


evb::bu::DiskWriter::DiskWriter
(
  BU* bu,
  std::shared_ptr<ResourceManager> resourceManager
) :
  bu_(bu),
  resourceManager_(resourceManager),
  configuration_(bu->getConfiguration()),
  buInstance_(bu->getApplicationDescriptor()->getInstance()),
  tmpFileIndex_(0),
  eventFIFO_(bu,"eventFIFO"),
  fileInfoFIFO_(bu,"fileInfoFIFO"),
  doProcessing_(false),
  active_(false)
{
  resetMonitoringCounters();
  startFileAccounting();
  umask(0);
}


evb::bu::DiskWriter::~DiskWriter()
{
  streamHandlers_.clear();

  if ( fileAccountingWorkLoop_ && fileAccountingWorkLoop_->isActive() )
    fileAccountingWorkLoop_->cancel();
}


void evb::bu::DiskWriter::handleEvent(EventPtr&& event)
{
  if ( configuration_->dropEventData )
  {
    resourceManager_->discardEvent(event);
  }
  else
  {
    std::lock_guard<std::mutex> guard(eventFIFOmutex_);

    eventFIFO_.enqWait(std::move(event));
  }
}


bool evb::bu::DiskWriter::getNextFileInfo(FileInfoPtr& fileInfo)
{
  return ( getReadyFileInfo(fileInfo) || populateFileInfo(fileInfo) );
}


bool evb::bu::DiskWriter::getReadyFileInfo(FileInfoPtr& fileInfo)
{
    std::lock_guard<std::mutex> guard(fileInfoFIFOmutex_);

    return fileInfoFIFO_.deq(fileInfo);
}


bool evb::bu::DiskWriter::populateFileInfo(FileInfoPtr& fileInfo)
{
  std::lock_guard<std::mutex> guard(fileInfoMapMutex_);

  bool haveCompleteFileInfo = false;
  EventPtr event;

  while ( !haveCompleteFileInfo && eventFIFO_.deq(event) )
  {
    const uint32_t lumiSection = event->getEvBid().lumiSection();
    FileInfoMap::iterator pos = fileInfoMap_.find(lumiSection);

    if ( pos == fileInfoMap_.end() )
    {
      pos = fileInfoMap_.emplace_hint(pos,
                                      lumiSection,
                                      std::make_unique<FileInfo>(
                                        (runRawDataDir_ / std::to_string(tmpFileIndex_++)).string()
                                      ));
      pos->second->events.reserve(configuration_->maxEventsPerFile);
      pos->second->header.lumiSection = lumiSection;
#ifdef EVB_USE_LOCS_VECTOR
      iovec fileHeader;
      fileHeader.iov_base = &(pos->second->header);
      fileHeader.iov_len = sizeof(FileHeader);
      pos->second->dataLocations.emplace_back(std::move(fileHeader));
#endif
    }

    pos->second->header.eventCount++;
    pos->second->header.fileSize += sizeof(EventInfo) + event->getEventInfo()->eventSize();
#ifdef EVB_USE_LOCS_VECTOR
    iovec eventHeader;
    eventHeader.iov_base = event->getEventInfo().get();
    eventHeader.iov_len = sizeof(EventInfo);
    pos->second->dataLocations.emplace_back(std::move(eventHeader));

    const DataLocations& frags = event->getDataLocations();
    pos->second->dataLocations.insert(pos->second->dataLocations.end(),frags.begin(),frags.end());
#endif
    pos->second->events.emplace_back(std::move(event));

    if ( pos->second->header.eventCount == configuration_->maxEventsPerFile )
    {
      fileInfo = std::move(pos->second);
      fileInfoMap_.erase(pos);
      haveCompleteFileInfo = true;
    }
  }

  return haveCompleteFileInfo;
}


void evb::bu::DiskWriter::startProcessing(const uint32_t runNumber)
{
  if ( configuration_->closeOldRuns )
    closeAnyOldRuns();

  resetMonitoringCounters();
  runNumber_ = runNumber;
  tmpFileIndex_ = 0;

  std::ostringstream runDir;
  runDir << "run" << std::setfill('0') << std::setw(6) << runNumber_;

  boost::filesystem::path rawRunDir( configuration_->rawDataDir.value_ );
  rawRunDir /= runDir.str();
  runRawDataDir_ = rawRunDir / "open";

  runMetaDataDir_ = configuration_->metaDataDir.value_;
  runMetaDataDir_ /= runDir.str();

  fileInfoMap_.clear();
  streamHandlers_.clear();
  lumiStatistics_.clear();

  if ( ! configuration_->dropEventData )
  {
    createDir(runRawDataDir_);
    createDir(runMetaDataDir_);

    populateHltdDirectory(rawRunDir);
    createLockFile(rawRunDir);

    const boost::filesystem::path jsdDir = runMetaDataDir_ / configuration_->jsdDirName.value_;
    createDir(jsdDir);
    defineEoLS(jsdDir);
    defineEoR(jsdDir);

    for (uint16_t i=0; i < configuration_->numberOfDiskWriters; ++i)
    {
      streamHandlers_.emplace_back(std::make_unique<StreamHandler>(
                                     bu_,
                                     this,
                                     resourceManager_,
                                     std::to_string(i)));
    }
  }

  doProcessing_ = true;
  fileAccountingWorkLoop_->submit(fileAccountingAction_);
}


bool evb::bu::DiskWriter::idle() const
{
  if ( !eventFIFO_.empty() || !fileInfoFIFO_.empty() ) return false;

  for (auto const& handler : streamHandlers_)
  {
    if ( !handler->idle() ) return false;
  }

  return true;
}


void evb::bu::DiskWriter::drain()
{
  while ( !idle() || active_ ) ::usleep(1000);
}


void evb::bu::DiskWriter::stopProcessing()
{
  doProcessing_ = false;

  while ( active_ || !eventFIFO_.empty() ) ::usleep(1000);

  processLumiSections(false);

  if ( ! fileInfoMap_.empty() )
  {
    std::ostringstream msg;
    msg << "There are unaccounted events for the following lumi sections:";
    for (auto const& fileInfo : fileInfoMap_)
    {
      msg << " " << fileInfo.first;
    }
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  while ( ! fileInfoFIFO_.empty() ) ::usleep(1000);

  for (auto const& handler : streamHandlers_)
  {
    handler->stopProcessing();
  }

  moveFiles();
  closeLumiSections();

  if ( ! lumiStatistics_.empty() )
  {
    std::ostringstream msg;
    msg << "There are unaccounted files for the following lumi sections:";
    for (auto const& stat : lumiStatistics_)
    {
      msg << " " << stat.first;
    }
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  streamHandlers_.clear();

  if ( !configuration_->dropEventData )
  {
    writeEoR();
    removeDir(runRawDataDir_);

    if ( configuration_->deleteRawDataFiles )
    {
      removeDir(runRawDataDir_.parent_path());
      removeDir(runMetaDataDir_);
    }
  }
}


void evb::bu::DiskWriter::startFileAccounting()
{
  try
  {
    fileAccountingWorkLoop_ =
      toolbox::task::getWorkLoopFactory()->getWorkLoop(bu_->getIdentifier("fileAccounting"), "waiting");

    if ( !fileAccountingWorkLoop_->isActive() )
      fileAccountingWorkLoop_->activate();

    fileAccountingAction_ =
      toolbox::task::bind(this,
                          &evb::bu::DiskWriter::fileAccounting,
                          bu_->getIdentifier("fileAccountingAction"));
  }
  catch(xcept::Exception& e)
  {
    std::string msg = "Failed to start lumi accounting workloop";
    XCEPT_RETHROW(exception::WorkLoop, msg, e);
  }
}


bool evb::bu::DiskWriter::fileAccounting(toolbox::task::WorkLoop* wl)
{
  if ( ! doProcessing_ ) return false;

  active_ = true;

  try
  {
    while (
      processLumiSections(true) ||
      closeStaleFiles() ||
      moveFiles() ||
      closeLumiSections()
    ) {};
  }
  catch(xcept::Exception& e)
  {
    active_ = false;
    bu_->getStateMachine()->processFSMEvent( Fail(e) );
  }
  catch(std::exception& e)
  {
    active_ = false;
    XCEPT_DECLARE(exception::DiskWriting,
                  sentinelException, e.what());
    bu_->getStateMachine()->processFSMEvent( Fail(sentinelException) );
  }
  catch(...)
  {
    active_ = false;
    XCEPT_DECLARE(exception::DiskWriting,
                  sentinelException, "unkown exception");
    bu_->getStateMachine()->processFSMEvent( Fail(sentinelException) );
  }

  active_ = false;

  ::usleep(1000);

  return doProcessing_;
}


bool evb::bu::DiskWriter::processLumiSections(const bool completeLumiSectionsOnly)
{
  bool workDone = false;
  ResourceManager::LumiSectionAccountPtr lumiSectionAccount;

  while ( resourceManager_->getNextLumiSectionAccount(lumiSectionAccount,completeLumiSectionsOnly) )
  {
    workDone = true;

    if ( configuration_->dropEventData ) continue;

    LumiStatistics::iterator lumiStatistics = getLumiStatistics(lumiSectionAccount->lumiSection);
    if ( lumiStatistics->second->nbEvents > 0 )
    {
      std::ostringstream msg;
      msg << "Got a duplicated account for lumi section " << lumiSectionAccount->lumiSection;
      XCEPT_RAISE(exception::EventOrder, msg.str());
    }
    if ( lumiSectionAccount->nbEvents == 0 )
    {
      lumiStatistics->second->isEmpty = true;
    }
    else
    {
      lumiStatistics->second->nbEvents = lumiSectionAccount->nbEvents;
      lumiStatistics->second->nbIncompleteEvents = lumiSectionAccount->nbIncompleteEvents;
    }
    lumiStatistics->second->totalEvents = bu_->getTotalEventsInLumiSection(lumiSectionAccount->lumiSection);

    {
      std::lock_guard<std::mutex> guard(fileInfoMapMutex_);

      FileInfoMap::iterator fileInfo = fileInfoMap_.find(lumiSectionAccount->lumiSection);
      if ( fileInfo != fileInfoMap_.end() )
      {
        fileInfoFIFO_.enqWait(std::move(fileInfo->second));
        fileInfoMap_.erase(fileInfo);
      }
    }
  }

  return workDone;
}


bool evb::bu::DiskWriter::closeStaleFiles()
{
  bool workDone = false;

  std::lock_guard<std::mutex> guard(fileInfoMapMutex_);

  FileInfoMap::iterator fileInfo = fileInfoMap_.begin();

  while ( fileInfo != fileInfoMap_.end() )
  {
    if ( fileInfo->second->getFileAge() > configuration_->maxFileAgeSeconds )
    {
      fileInfoFIFO_.enqWait(std::move(fileInfo->second));
      fileInfoMap_.erase(fileInfo++);
      workDone = true;
    }
    else
    {
      ++fileInfo;
    }
  }

  return workDone;
}


bool evb::bu::DiskWriter::closeLumiSections()
{
  bool workDone = false;

  LumiStatistics::iterator it = lumiStatistics_.begin();
  while ( it != lumiStatistics_.end() )
  {
    if ( it->second->isComplete() )
    {
      workDone = true;
      writeEoLS(it->second);

      {
        std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);

        if ( it->second->nbEventsWritten > 0 )
          diskWriterMonitoring_.nbLumiSections++;
        if ( diskWriterMonitoring_.currentLumiSection < it->second->lumiSection )
          diskWriterMonitoring_.currentLumiSection = it->second->lumiSection;
      }

      lumiStatistics_.erase(it++);
    }
    else
    {
      ++it;
    }
  }

  return workDone;
}


bool evb::bu::DiskWriter::moveFiles()
{
  FileInfoPtr fileInfo;
  bool workDone = false;

  for (auto const& handler : streamHandlers_)
  {
    while ( handler->getNextClosedFileInfo(fileInfo) )
    {
      workDone = true;
      handleRawDataFile(fileInfo);
    }
  }

  return workDone;
}


void evb::bu::DiskWriter::handleRawDataFile(const FileInfoPtr& fileInfo)
{
  const uint32_t lumiSection = fileInfo->header.lumiSection;
  const LumiStatistics::iterator lumiStatistics = getLumiStatistics(lumiSection);

  if ( configuration_->deleteRawDataFiles )
  {
    boost::filesystem::remove(fileInfo->fileName);
  }
  else
  {
    std::ostringstream fileNameStream;
    fileNameStream << std::setfill('0') <<
      "run"<< std::setw(6) << runNumber_ <<
      "_ls" << std::setw(4) << lumiSection <<
      "_index" << std::setw(6) << lumiStatistics->second->index++ <<
      ".raw";

    boost::filesystem::rename(fileInfo->fileName,runRawDataDir_.parent_path() / fileNameStream.str());
  }

  {
    std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);

    diskWriterMonitoring_.nbFiles++;
    diskWriterMonitoring_.nbEventsWritten += fileInfo->header.eventCount;
    if ( diskWriterMonitoring_.currentLumiSection < lumiSection)
      diskWriterMonitoring_.currentLumiSection = lumiSection;
    if ( diskWriterMonitoring_.lastLumiSection < lumiSection )
      diskWriterMonitoring_.lastLumiSection = lumiSection;
  }

  lumiStatistics->second->fileCount++;
  lumiStatistics->second->nbEventsWritten += fileInfo->header.eventCount;
  lumiStatistics->second->nbBytesWritten += fileInfo->header.fileSize;
}


evb::bu::DiskWriter::LumiStatistics::iterator evb::bu::DiskWriter::getLumiStatistics(const uint32_t lumiSection)
{
  std::lock_guard<std::mutex> guard(lumiStatisticsMutex_);

  return lumiStatistics_.emplace(lumiSection,std::make_unique<LumiInfo>(lumiSection)).first;
}


void evb::bu::DiskWriter::appendMonitoringItems(InfoSpaceItems& items)
{
  nbFilesWritten_ = 0;
  nbLumiSections_ = 0;
  currentLumiSection_ = 0;

  items.add("nbFilesWritten", &nbFilesWritten_);
  items.add("nbLumiSections", &nbLumiSections_);
  items.add("currentLumiSection", &currentLumiSection_);
}


void evb::bu::DiskWriter::updateMonitoringItems()
{
  std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);

  nbFilesWritten_ = diskWriterMonitoring_.nbFiles;
  nbLumiSections_ = diskWriterMonitoring_.nbLumiSections;
  currentLumiSection_ = diskWriterMonitoring_.currentLumiSection;
}


void evb::bu::DiskWriter::resetMonitoringCounters()
{
  std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);

  diskWriterMonitoring_.nbFiles = 0;
  diskWriterMonitoring_.nbEventsWritten = 0;
  diskWriterMonitoring_.nbLumiSections = 0;
  diskWriterMonitoring_.currentLumiSection = 0;
  diskWriterMonitoring_.lastLumiSection = 0;
}


uint32_t evb::bu::DiskWriter::getNbLumiSections() const
{
  std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);
  return diskWriterMonitoring_.nbLumiSections;
}


void evb::bu::DiskWriter::configure()
{
  eventFIFO_.clear();
  fileInfoFIFO_.clear();
  streamHandlers_.clear();
  lumiStatistics_.clear();

  if ( ! configuration_->dropEventData )
  {
    eventFIFO_.resize(configuration_->maxEvtsUnderConstruction);
    fileInfoFIFO_.resize(configuration_->fileInfoFIFOCapacity);
    createDir(configuration_->rawDataDir.value_);
    createDir(configuration_->metaDataDir.value_);
  }
}


void evb::bu::DiskWriter::createDir(const boost::filesystem::path& path) const
{
  if ( ! boost::filesystem::exists(path) &&
       ( ! boost::filesystem::create_directories(path) ) )
  {
    std::ostringstream msg;
    msg << "Failed to create directory " << path.string();
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }
}


void evb::bu::DiskWriter::removeDir(const boost::filesystem::path& path) const
{
  if ( boost::filesystem::exists(path) )
  {
    try
    {
      boost::filesystem::remove_all(path);
    }
    catch(boost::filesystem::filesystem_error& e)
    {
      std::ostringstream msg;
      msg << "Failed to remove directory " << path.string();
      msg << ": " << e.what();
      XCEPT_RAISE(exception::DiskWriting, msg.str());
    }
  }
}


cgicc::div evb::bu::DiskWriter::getHtmlSnipped() const
{
  using namespace cgicc;

  cgicc::div div;
  div.add(p("DiskWriter"));

  {
    table table;
    table.set("title","Statistics for files generated. This numbers are updated only then a file is closed. Thus, they lag a bit behind. If 'dropEventData' is true, these counters stay at 0.");

    std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);

    table.add(tr()
              .add(td("# files written"))
              .add(td(std::to_string(diskWriterMonitoring_.nbFiles))));
    table.add(tr()
              .add(td("# events written"))
              .add(td(std::to_string(diskWriterMonitoring_.nbEventsWritten))));
    table.add(tr()
              .add(td("# finished lumi sections with files"))
              .add(td(std::to_string(diskWriterMonitoring_.nbLumiSections))));
    table.add(tr()
              .add(td("last lumi section with files"))
              .add(td(std::to_string(diskWriterMonitoring_.lastLumiSection))));
    table.add(tr()
              .add(td("current lumi section"))
              .add(td(std::to_string(diskWriterMonitoring_.currentLumiSection))));
    div.add(table);
  }

  div.add(eventFIFO_.getHtmlSnipped());

  div.add(fileInfoFIFO_.getHtmlSnipped());

  {
    cgicc::table table;
    table.set("title","List of writer threads. Each thread writes events independently.");

    table.add(tr()
              .add(th("Event writers").set("colspan","5")));
    table.add(tr()
              .add(td("writer"))
              .add(td("active"))
              .add(td("#files ready"))
              .add(td("#events"))
              .add(td("MB written")));

    for (auto const& handler : streamHandlers_)
    {
      table.add(handler->getWriterTableRow());
    }

    div.add(table);
  }

  return div;
}


void evb::bu::DiskWriter::closeAnyOldRuns() const
{
  const boost::filesystem::path rawDataDir( configuration_->rawDataDir.value_ );

  if ( ! boost::filesystem::exists(rawDataDir) ) return;

  boost::filesystem::directory_iterator dirIter(rawDataDir);

  while ( dirIter != boost::filesystem::directory_iterator() )
  {
    const std::string fileName = dirIter->path().string();
    size_t pos = fileName.rfind("run");
    if ( pos != std::string::npos )
    {
      boost::filesystem::path eorPath = *dirIter;
      eorPath /= boost::filesystem::path( "run" + fileName.substr(pos+3) + "_ls0000_EoR.jsn" );
      if ( ! boost::filesystem::exists(eorPath) )
      {
        try
        {
          std::ofstream json(eorPath.string().c_str());
          json.close();
        }
        catch (...) {} // Ignore any failures in case that the run directory is removed while we try to write
      }
    }
    ++dirIter;
  }
}


void evb::bu::DiskWriter::populateHltdDirectory(const boost::filesystem::path& runDir) const
{
  const boost::filesystem::path tmpPath = runDir / "tmp";
  createDir(tmpPath);
  getHLTmenu(tmpPath);
  writeHLTinfo(tmpPath);
  writeBlacklist(tmpPath);
  writeWhitelist(tmpPath);

  const boost::filesystem::path hltPath( runDir / configuration_->hltDirName.value_  );
  boost::filesystem::rename(tmpPath,hltPath);
}


void evb::bu::DiskWriter::getHLTmenu(const boost::filesystem::path& tmpDir) const
{
  if ( configuration_->hltParameterSetURL.value_.empty() ) return;

  std::string url(configuration_->hltParameterSetURL.value_);

  CURL* curl = curl_easy_init();
  if ( ! curl )
  {
    XCEPT_RAISE(exception::DiskWriting, "Could not initialize curl for retrieving the HLT menu");
  }

  try
  {
    char lastChar = *url.rbegin();
    if ( lastChar != '/' ) url += '/';

    for (auto const& hltFile : configuration_->hltFiles)
    {
      const std::string fileName = hltFile.toString();
      retrieveFromURL(curl, url+fileName, tmpDir/fileName);
    }
  }
  catch(xcept::Exception& e)
  {
    curl_easy_cleanup(curl);
    throw(e);
  }

  curl_easy_cleanup(curl);
}


void evb::bu::DiskWriter::writeHLTinfo(const boost::filesystem::path& tmpDir) const
{
  const boost::filesystem::path hltInfoPath( tmpDir / configuration_->hltinfoName.value_ );

  std::ofstream json(hltInfoPath.string().c_str());

  json << "{"                                                                     << std::endl;
  json << "   \"isGlobalRun\": " << configuration_->isGlobalRun.toString() << "," << std::endl;
  json << "   \"daqSystem\": \"" << configuration_->daqSystem.value_ << "\","     << std::endl;
  json << "   \"daqInstance\": \"" << configuration_->daqInstance.value_ << "\"," << std::endl;
  json << "   \"fuGroup\": \"" << configuration_->fuGroup.value_ << "\""          << std::endl;
  json << "}"                                                                     << std::endl;
  json.close();
}


void evb::bu::DiskWriter::writeBlacklist(const boost::filesystem::path& tmpDir) const
{
  const boost::filesystem::path blacklistPath( tmpDir / configuration_->blacklistName.value_ );
  const uint16_t count = writeHostList(blacklistPath, configuration_->fuBlacklist);
  resourceManager_->setBlacklistedFUcount(count);
}


void evb::bu::DiskWriter::writeWhitelist(const boost::filesystem::path& tmpDir) const
{
  const boost::filesystem::path whitelistPath( tmpDir / configuration_->whitelistName.value_ );
  const uint16_t count = writeHostList(whitelistPath, configuration_->fuWhitelist);
  resourceManager_->setActiveFUcount(count);
}


uint16_t evb::bu::DiskWriter::writeHostList(const boost::filesystem::path& path, const xdata::String& hosts) const
{
  uint16_t count = 0;

  std::ofstream hostlist(path.string().c_str());

  const std::regex regex("[a-zA-Z0-9]+-[a-zA-Z0-9-]+");
  std::sregex_iterator res(hosts.value_.begin(), hosts.value_.end(), regex);

  const std::sregex_iterator end;
  if ( res == end )
  {
    hostlist << "[]" << std::endl;
  }
  else
  {
    hostlist << "[\"" << res->str()  << "\"";
    count = 1;
    while (++res != end)
    {
      hostlist << ", \"" << res->str() << "\"";
      ++count;
    }
    hostlist << "]" << std::endl;
  }
  hostlist.close();

  return count;
}


void evb::bu::DiskWriter::retrieveFromURL(CURL* curl, const std::string& url, const boost::filesystem::path& output) const
{
  const char* path = output.string().c_str();
  FILE* file = fopen(path,"w");
  if ( ! file )
  {
    std::ostringstream msg;
    msg << "Failed to open file " << path
      << ": " << strerror(errno);
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  curl_easy_setopt(curl, CURLOPT_WRITEDATA, file);
  curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
  curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); //allow libcurl to follow redirection
  curl_easy_setopt(curl, CURLOPT_FAILONERROR, true);

  const CURLcode result = curl_easy_perform(curl);

  if ( result != CURLE_OK )
  {
    fclose(file);

    std::ostringstream msg;
    msg << "Failed to retrieve the HLT information from " << url
      << ": " << curl_easy_strerror(result);
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  fclose(file);
}


void evb::bu::DiskWriter::createLockFile(const boost::filesystem::path& runDir) const
{
  const boost::filesystem::path fulockPath( runDir / configuration_->fuLockName.value_ );
  const char* path = fulockPath.string().c_str();
  std::ofstream fulock(path);
  fulock << "1 0";
  fulock.close();
}


void evb::bu::DiskWriter::writeEoLS(const LumiInfoPtr& lumiInfo) const
{
  std::ostringstream fileNameStream;
  fileNameStream << std::setfill('0') <<
    "run" << std::setw(6) << runNumber_ <<
    "_ls"  << std::setw(4) << lumiInfo->lumiSection <<
    "_EoLS.jsn";
  const boost::filesystem::path jsonFile = runMetaDataDir_ / fileNameStream.str();

  if ( boost::filesystem::exists(jsonFile) )
  {
    std::ostringstream msg;
    msg << "The JSON file " << jsonFile.string() << " already exists";
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  const std::string path = jsonFile.string() + ".tmp";
  std::ofstream json(path.c_str());
  json << "{"                                                              << std::endl;
  json << "   \"data\" : [ \""
    << lumiInfo->nbEventsWritten << "\", \""
    << lumiInfo->fileCount << "\", \""
    << lumiInfo->totalEvents << "\", \""
    << lumiInfo->nbIncompleteEvents << "\", \""
    << lumiInfo->nbBytesWritten << "\" ],"                                 << std::endl;
  json << "   \"definition\" : \"" << eolsDefFile_.string() << "\","       << std::endl;
  json << "   \"source\" : \"BU-"  << buInstance_ << "\""                  << std::endl;
  json << "}"                                                              << std::endl;
  json.close();

  boost::filesystem::rename(path,jsonFile);
}


void evb::bu::DiskWriter::writeEoR() const
{
  std::ostringstream fileNameStream;
  fileNameStream << std::setfill('0') <<
    "run" << std::setw(6) << runNumber_ <<
    "_ls0000_EoR.jsn";
  const boost::filesystem::path jsonFile = runMetaDataDir_ / fileNameStream.str();

  if ( boost::filesystem::exists(jsonFile) )
  {
    std::ostringstream msg;
    msg << "The JSON file " << jsonFile.string() << " already exists";
    XCEPT_RAISE(exception::DiskWriting, msg.str());
  }

  std::lock_guard<std::mutex> guard(diskWriterMonitoringMutex_);
  const std::string path = jsonFile.string() + ".tmp";
  std::ofstream json(path.c_str());
  json << "{"                                                                           << std::endl;
  json << "   \"data\" : [ \""
    << diskWriterMonitoring_.nbEventsWritten << "\", \""
    << diskWriterMonitoring_.nbFiles         << "\", \""
    << diskWriterMonitoring_.nbLumiSections  << "\", \""
    << diskWriterMonitoring_.lastLumiSection << "\" ],"                                 << std::endl;
  json << "   \"definition\" : \"" << eorDefFile_.string() << "\","                     << std::endl;
  json << "   \"source\" : \"BU-"  << buInstance_   << "\""                             << std::endl;
  json << "}"                                                                           << std::endl;
  json.close();

  boost::filesystem::rename(path,jsonFile);
}


void evb::bu::DiskWriter::defineEoLS(const boost::filesystem::path& jsdDir)
{
  eolsDefFile_ = jsdDir / "EoLS.jsd";

  const std::string path = eolsDefFile_.string() + ".tmp";
  std::ofstream json(path.c_str());
  json << "{"                                                 << std::endl;
  json << "   \"legend\" : ["                                 << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NEvents\","                  << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NFiles\","                   << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"TotalEvents\","              << std::endl;
  json << "         \"operation\" : \"max\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NLostEvents\","              << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NBytes\","                   << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      }"                                           << std::endl;
  json << "   ]"                                              << std::endl;
  json << "}"                                                 << std::endl;
  json.close();

  boost::filesystem::rename(path,eolsDefFile_);
}


void evb::bu::DiskWriter::defineEoR(const boost::filesystem::path& jsdDir)
{
  eorDefFile_ = jsdDir / "EoR.jsd";

  const std::string path = eorDefFile_.string() + ".tmp";
  std::ofstream json(path.c_str());
  json << "{"                                                 << std::endl;
  json << "   \"legend\" : ["                                 << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NEvents\","                  << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NFiles\","                   << std::endl;
  json << "         \"operation\" : \"sum\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"NLumis\","                   << std::endl;
  json << "         \"operation\" : \"max\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      },"                                          << std::endl;
  json << "      {"                                           << std::endl;
  json << "         \"name\" : \"LastLumi\","                 << std::endl;
  json << "         \"operation\" : \"max\","                 << std::endl;
  json << "         \"type\" : \"integer\""                   << std::endl;
  json << "      }"                                           << std::endl;
  json << "   ]"                                              << std::endl;
  json << "}"                                                 << std::endl;
  json.close();

  boost::filesystem::rename(path,eorDefFile_);
}


/// emacs configuration
/// Local Variables: -
/// mode: c++ -
/// c-basic-offset: 2 -
/// indent-tabs-mode: nil -
/// End: -