text:
#include "ModeCheckFS.h"

#include <common/toolkit/DisposalCleaner.h>
#include <common/toolkit/ListTk.h>
#include <common/toolkit/UnitTk.h>
#include <common/toolkit/UiTk.h>
#include <common/toolkit/ZipIterator.h>
#include <components/DataFetcher.h>
#include <components/worker/RetrieveChunksWork.h>
#include <net/msghelpers/MsgHelperRepair.h>
#include <toolkit/DatabaseTk.h>
#include <toolkit/FsckTkEx.h>

#include <program/Program.h>

#include <ftw.h>

#include <boost/lexical_cast.hpp>

template<unsigned Actions>
UserPrompter::UserPrompter(const FsckRepairAction (&possibleActions)[Actions],
   FsckRepairAction defaultRepairAction)
   : askForAction(true), possibleActions(possibleActions, possibleActions + Actions),
     repairAction(FsckRepairAction_UNDEFINED)
{
   if(Program::getApp()->getConfig()->getReadOnly() )
      askForAction = false;
   else
   if(Program::getApp()->getConfig()->getAutomatic() )
   {
      askForAction = false;
      repairAction = defaultRepairAction;
   }
}

FsckRepairAction UserPrompter::chooseAction(const std::string& prompt)
{
   if(askForAction)
      FsckTkEx::fsckOutput("> " + prompt, OutputOptions_LINEBREAK | OutputOptions_NOLOG);

   FsckTkEx::fsckOutput("> " + prompt, OutputOptions_NOSTDOUT);

   while(askForAction)
   {
      for(size_t i = 0; i < possibleActions.size(); i++)
      {
         FsckTkEx::fsckOutput(
            "   " + StringTk::uintToStr(i + 1) + ") "
               + FsckTkEx::getRepairActionDesc(possibleActions[i]), OutputOptions_NOLOG |
               OutputOptions_LINEBREAK);
      }

      for(size_t i = 0; i < possibleActions.size(); i++)
      {
         FsckTkEx::fsckOutput(
            "   " + StringTk::uintToStr(i + possibleActions.size() + 1) + ") "
               + FsckTkEx::getRepairActionDesc(possibleActions[i]) + " (apply for all)",
               OutputOptions_NOLOG | OutputOptions_LINEBREAK);
      }

      std::string inputStr;
      getline(std::cin, inputStr);

      unsigned input = StringTk::strToUInt(inputStr);

      if( (input > 0) && (input <= possibleActions.size() ) )
      {
         // user chose for this error
         repairAction = possibleActions[input - 1];
         break;
      }

      if( (input > possibleActions.size() ) && (input <= possibleActions.size() * 2) )
      {
         // user chose for all errors => do not ask again
         askForAction = false;
         repairAction = possibleActions[input - possibleActions.size() - 1];
         break;
      }
   }

   FsckTkEx::fsckOutput(" - [ra: " + FsckTkEx::getRepairActionDesc(repairAction, true) + "]",
      OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);

   return repairAction;
}

static FhgfsOpsErr handleDisposalItem(Node& owner, const std::string& entryID,
   const bool isMirrored)
{
   FhgfsOpsErr err = DisposalCleaner::unlinkFile(owner, entryID, isMirrored);
   if (err == FhgfsOpsErr_INUSE)
      return FhgfsOpsErr_SUCCESS;
   else
      return err;
}

ModeCheckFS::ModeCheckFS()
 : log("ModeCheckFS")
{
}

ModeCheckFS::~ModeCheckFS()
{
}

void ModeCheckFS::printHelp()
{
   std::cout <<
      "MODE ARGUMENTS:\n"
      " Optional:\n"
      "  --readOnly             Check only, skip repairs.\n"
      "  --runOffline           Run in offline mode, which disables the modification\n"
      "                         logging. Use this only if you are sure there is no user\n"
      "                         access to the file system while the check is running.\n"
      "  --automatic            Do not prompt for repair actions and automatically use\n"
      "                         reasonable default actions.\n"
      "  --noFetch              Do not build a new database from the servers and\n"
      "                         instead work on an existing database (e.g. from a\n"
      "                         previous read-only run).\n"
      "  --quotaEnabled         Enable checks for quota support.\n"
      "  --databasePath=<path>  Path to store for the database files. On systems with \n"
      "                         many files, the database can grow to a size of several \n"
      "                         100 GB.\n"
      "                         (Default: " CONFIG_DEFAULT_DBPATH ")\n"
      "  --overwriteDbFile      Overwrite an existing database file without prompt.\n"
      "  --ignoreDBDiskSpace    Ignore free disk space check for database file.\n"
      "  --logOutFile=<path>    Path to the fsck output file, which contains a copy of\n"
      "                         the console output.\n"
      "                         (Default: " CONFIG_DEFAULT_OUTFILE ")\n"
      "  --logStdFile=<path>    Path to the program error log file, which contains e.g.\n"
      "                         network error messages.\n"
      "                         (Default: " CONFIG_DEFAULT_LOGFILE ")\n"
      "  --forceRestart         Restart check even though another instance seems to be running.\n"
      "                         Use only if a previous run was aborted, and make sure no other\n"
      "                         fsck is running on the same file system at the same time.\n"
      "\n"
      "USAGE:\n"
      " This mode performs a full check and optional repair of a BeeGFS file system\n"
      " instance by building a database of the current file system contents on the\n"
      " local machine.\n"
      "\n"
      " The fsck gathers information from all BeeGFS server daemons in parallel through\n"
      " their configured network interfaces. All server components of the file\n"
      " system have to be running to start a check.\n"
      "\n"
      " If the fsck is running with the \"--runoffline\" argument, users may not\n"
      " access the file system during a run (otherwise false errors might be reported).\n"
      "\n"
      " Example: Check for errors, but skip repairs\n"
      "  $ beegfs-fsck --checkfs --readonly\n";

   std::cout << std::flush;
}

int ModeCheckFS::execute()
{
   App* app = Program::getApp();
   Config *cfg = app->getConfig();
   std::string databasePath = cfg->getDatabasePath();

   if ( this->checkInvalidArgs(cfg->getUnknownConfigArgs()) )
      return APPCODE_INVALID_CONFIG;

   FsckTkEx::printVersionHeader(false);
   printHeaderInformation();

   if (cfg->getNoFetch() && !Program::getApp()->getConfig()->getReadOnly()) {
      FsckTkEx::fsckOutput(
         "           WARNING:  RISK OF DATA LOSS!\n"
         "        NEVER USE THE SAME DATABASE TWICE!\n\n"

         "After running fsck, any database obtained previously no longer matches the state\n"
         "of the filesystem. A new database MUST be obtained before running fsck again.\n\n"

         "You are seeing this warning because you are using the option --nofetch.",
         OutputOptions_DOUBLELINEBREAK | OutputOptions_ADDLINEBREAKBEFORE);

      if (!uitk::userYNQuestion("Do you wish to continue?"))
         return APPCODE_USER_ABORTED;
   }

   if ( !FsckTkEx::checkReachability() )
      return APPCODE_COMMUNICATION_ERROR;

   if (!FsckTkEx::checkConsistencyStates())
   {
      FsckTkEx::fsckOutput("At least one meta or storage target is in NEEDS_RESYNC state. "
            "To prevent interference between fsck and resync operations, fsck will abort now. "
            "Please make sure that no resyncs are currently pending or running.",
            OutputOptions_DOUBLELINEBREAK | OutputOptions_ADDLINEBREAKBEFORE);
      return APPCODE_RUNTIME_ERROR;
   }

   if(cfg->getNoFetch() )
   {
      try {
         this->database.reset(new FsckDB(databasePath + "/fsckdb", cfg->getTuneDbFragmentSize(),
            cfg->getTuneDentryCacheSize(), false) );
      } catch (const FragmentDoesNotExist& e) {
         std::string err = "Database was found to be incomplete in path " + databasePath;
         log.logErr(err);
         FsckTkEx::fsckOutput(err);
         return APPCODE_RUNTIME_ERROR;
      }
   }
   else
   {
      int initDBRes = initDatabase();
      if (initDBRes)
         return initDBRes;

      disposeUnusedFiles();

      boost::scoped_ptr<ModificationEventHandler> modificationEventHandler;

      if (!cfg->getRunOffline())
      {
         modificationEventHandler.reset(
            new ModificationEventHandler(*this->database->getModificationEventsTable() ) );
         modificationEventHandler->start();
         Program::getApp()->setModificationEventHandler(modificationEventHandler.get() );

         // start modification logging
         auto startLogRes = FsckTkEx::startModificationLogging(app->getMetaNodes(),
               app->getLocalNode(), cfg->getForceRestart());

         if (startLogRes != FhgfsOpsErr_SUCCESS)
         {
            std::string errStr;
            switch (startLogRes)
            {
               case FhgfsOpsErr_INUSE:
                  errStr = "Another instance of beegfs-fsck is still running or was aborted. "
                     "Cannot start a new one unless --forceRestart command line switch is used. "
                     "Do this only after making sure there is no other instance of beefs-fsck "
                     "running.";

                  // stop the modification event handler
                  Program::getApp()->setModificationEventHandler(NULL);
                  modificationEventHandler->selfTerminate();
                  modificationEventHandler->join();
                  break;

               default:
                  errStr = "Unable to start file system modification logging. "
                     "Fsck cannot proceed.";
            }

            log.logErr(errStr);
            FsckTkEx::fsckOutput(errStr);

            return APPCODE_RUNTIME_ERROR;
         }
      }

      FhgfsOpsErr gatherDataRes = gatherData(cfg->getForceRestart());

      if (!cfg->getRunOffline())
      {
         // stop modification logging
         bool eventLoggingOK = FsckTkEx::stopModificationLogging(app->getMetaNodes());
         // stop mod event handler (to make it flush for the last time
         Program::getApp()->setModificationEventHandler(NULL);
         modificationEventHandler->selfTerminate();
         modificationEventHandler->join();

         // if event logging is not OK (i.e. that not all events might have been processed), go
         // into read-only mode
         if ( !eventLoggingOK )
         {
            Program::getApp()->getConfig()->disableAutomaticRepairMode();
            Program::getApp()->getConfig()->setReadOnly();

            FsckTkEx::fsckOutput("-----",
               OutputOptions_ADDLINEBREAKBEFORE | OutputOptions_FLUSH | OutputOptions_LINEBREAK);
            FsckTkEx::fsckOutput(
               "WARNING: Fsck did not get all modification events from metadata servers.",
               OutputOptions_FLUSH | OutputOptions_LINEBREAK);
            FsckTkEx::fsckOutput(
               "For instance, this might have happened due to network timeouts.",
               OutputOptions_FLUSH | OutputOptions_LINEBREAK);
            FsckTkEx::fsckOutput(
               "This means beegfs-fsck is not aware of all changes in the filesystem and it is "
               "not safe to execute repair actions. "
               "In the worst case executing repair actions can result in data loss.",
               OutputOptions_FLUSH | OutputOptions_DOUBLELINEBREAK);
            FsckTkEx::fsckOutput(
               "Thus, read-only mode was automatically enabled. "
               "You can still force repair by running beegfs-fsck again on the existing "
               "database with the -noFetch option.",
               OutputOptions_FLUSH | OutputOptions_DOUBLELINEBREAK);
            FsckTkEx::fsckOutput("Please press any key to continue.",
               OutputOptions_FLUSH | OutputOptions_LINEBREAK);
            FsckTkEx::fsckOutput("-----", OutputOptions_FLUSH | OutputOptions_DOUBLELINEBREAK);

            std::cin.get();
         }

      }

      if (gatherDataRes != FhgfsOpsErr_SUCCESS)
      {
         std::string errStr;
         switch (gatherDataRes)
         {
            case FhgfsOpsErr_INUSE:
               errStr = "Another instance of beegfs-fsck is still running or was aborted. "
                  "Cannot start a new one unless --forceRestart command line switch is used. "
                  "Do this only after making sure there is no other instance of beefs-fsck "
                  "running.";
               break;
            default:
               errStr = "An error occured while fetching data from servers. Fsck cannot proceed. "
                  "Please see log file for more information.";
         }

         log.logErr(errStr);
         FsckTkEx::fsckOutput(errStr);

         return APPCODE_RUNTIME_ERROR;
      }
   }

   checkAndRepair();

   return APPCODE_NO_ERROR;
}

/*
 * initializes the database, this is done here and not inside the main app, because we do not want
 * it to effect the help modes; DB file shall only be created if user really runs a check
 *
 * @return integer value symbolizing an appcode
 */
int ModeCheckFS::initDatabase()
{
   Config* cfg = Program::getApp()->getConfig();

   // create the database path
   Path dbPath(cfg->getDatabasePath() + "/fsckdb");

   if ( !StorageTk::createPathOnDisk(dbPath, false) )
   {
      FsckTkEx::fsckOutput("Could not create path for database files: " +
         dbPath.str());
      return APPCODE_INITIALIZATION_ERROR;
   }

   // check disk space
   if ( !FsckTkEx::checkDiskSpace(dbPath) )
      return APPCODE_INITIALIZATION_ERROR;

   // check if DB file path already exists and is not empty
   if ( StorageTk::pathExists(dbPath.str())
      && StorageTk::pathHasChildren(dbPath.str()) && (!cfg->getOverwriteDbFile()) )
   {
      FsckTkEx::fsckOutput("The database path already exists: " + dbPath.str());
      FsckTkEx::fsckOutput("If you continue now any existing database files in that path will be "
         "deleted.");

      std::string input;

      while (input.size() != 1 || !strchr("YyNn", input[0]))
      {
         FsckTkEx::fsckOutput("Do you want to continue? (Y/N)");
         std::getline(std::cin, input);
      }

      switch (input[0])
      {
         case 'Y':
         case 'y':
            break; // just do nothing and go ahead
         case 'N':
         case 'n':
         default:
            // abort here
            return APPCODE_USER_ABORTED;
      }
   }

   struct ops
   {
      static int visit(const char* path, const struct stat*, int type, struct FTW* state)
      {
         if(state->level == 0)
            return 0;
         else
         if(type == FTW_F || type == FTW_SL)
            return ::unlink(path);
         else
            return ::rmdir(path);
      }
   };

   int ftwRes = ::nftw(dbPath.str().c_str(), ops::visit, 10, FTW_DEPTH | FTW_PHYS);
   if(ftwRes)
   {
      FsckTkEx::fsckOutput("Could not empty path for database files: " + dbPath.str() );
      return APPCODE_INITIALIZATION_ERROR;
   }

   try {
      this->database.reset(
         new FsckDB(dbPath.str(), cfg->getTuneDbFragmentSize(),
            cfg->getTuneDentryCacheSize(), true) );
   } catch (const std::runtime_error& e) {
      FsckTkEx::fsckOutput("Database " + dbPath.str() + " is corrupt");
      return APPCODE_RUNTIME_ERROR;
   }

   return 0;
}

void ModeCheckFS::printHeaderInformation()
{
   Config* cfg = Program::getApp()->getConfig();

   // get the current time and create some nice output, so the user can see at which time the run
   // was started (especially nice to find older runs in the log file)
   time_t t;
   time(&t);
   std::string timeStr = std::string(ctime(&t));
   FsckTkEx::fsckOutput(
      "Started BeeGFS fsck in forward check mode [" + timeStr.substr(0, timeStr.length() - 1)
         + "]\nLog will be written to " + cfg->getLogStdFile() + "\nDatabase will be saved in "
         + cfg->getDatabasePath(), OutputOptions_LINEBREAK | OutputOptions_HEADLINE);

   if (Program::getApp()->getMetaMirrorBuddyGroupMapper()->getSize() > 0)
   {
      FsckTkEx::fsckOutput(
            "IMPORTANT NOTICE: The initiation of a repair action on a metadata mirror group "
            "will temporarily set the consistency state of the secondary node to bad to prevent "
            "interaction during the repair. After fsck finishes, it will set the states "
            "of all affected nodes to needs-resync, so the repaired data will be synced from their "
            "primary. \n\nIf beegfs-fsck is aborted prematurely, a resync needs to be initiated "
            "manually so all targets are consistent (good) again." ,
            OutputOptions_DOUBLELINEBREAK | OutputOptions_HEADLINE);
   }
}

void ModeCheckFS::disposeUnusedFiles()
{
   Config* cfg = Program::getApp()->getConfig();

   if (cfg->getReadOnly() || cfg->getNoFetch())
      return;

   FsckTkEx::fsckOutput("Step 2: Delete unused files from disposal: ", OutputOptions_NONE);

   using namespace std::placeholders;

   uint64_t errors = 0;

   DisposalCleaner dc(*Program::getApp()->getMetaMirrorBuddyGroupMapper());
   dc.run(Program::getApp()->getMetaNodes()->referenceAllNodes(),
         handleDisposalItem,
         [&] (const auto&, const auto&) { errors += 1; });

   if (errors > 0)
      FsckTkEx::fsckOutput("Some files could not be deleted.");
   else
      FsckTkEx::fsckOutput("Finished.");
}

FhgfsOpsErr ModeCheckFS::gatherData(bool forceRestart)
{
   FsckTkEx::fsckOutput("Step 3: Gather data from nodes: ", OutputOptions_DOUBLELINEBREAK);

   DataFetcher dataFetcher(*this->database, forceRestart);
   const FhgfsOpsErr retVal = dataFetcher.execute();

   FsckTkEx::fsckOutput("", OutputOptions_LINEBREAK);

   return retVal;
}

template<typename Obj, typename State>
int64_t ModeCheckFS::checkAndRepairGeneric(Cursor<Obj> cursor,
   void (ModeCheckFS::*repair)(Obj&, State&), State& state)
{
   int64_t errorCount = 0;

   while(cursor.step() )
   {
      Obj* entry = cursor.get();

      (this->*repair)(*entry, state);
      errorCount++;
   }

   if(errorCount)
   {
      database->getDentryTable()->commitChanges();
      database->getFileInodesTable()->commitChanges();
      database->getDirInodesTable()->commitChanges();
      database->getChunksTable()->commitChanges();
      database->getContDirsTable()->commitChanges();
      database->getFsIDsTable()->commitChanges();

      FsckTkEx::fsckOutput(">>> Found " + StringTk::int64ToStr(errorCount)
         + " errors. Detailed information can also be found in "
         + Program::getApp()->getConfig()->getLogOutFile() + ".",
      OutputOptions_DOUBLELINEBREAK);

   }
   else
   {
      FsckTkEx::fsckOutput(">>> None found", OutputOptions_FLUSH | OutputOptions_LINEBREAK);
   }

   return errorCount;
}

int64_t ModeCheckFS::checkAndRepairDanglingDentry()
{
   FsckRepairAction fileActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_DELETEDENTRY,
   };

   FsckRepairAction dirActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_DELETEDENTRY,
      FsckRepairAction_CREATEDEFAULTDIRINODE,
   };

   UserPrompter forFiles(fileActions, FsckRepairAction_DELETEDENTRY);
   UserPrompter forDirs(dirActions, FsckRepairAction_CREATEDEFAULTDIRINODE);

   std::pair<UserPrompter*, UserPrompter*> prompt(&forFiles, &forDirs);

   FsckTkEx::fsckOutput("* Checking: Dangling directory entry ...",
      OutputOptions_FLUSH |OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findDanglingDirEntries(),
      &ModeCheckFS::repairDanglingDirEntry, prompt);
}

int64_t ModeCheckFS::checkAndRepairWrongInodeOwner()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_CORRECTOWNER,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_CORRECTOWNER);

   FsckTkEx::fsckOutput("* Checking: Wrong owner node saved in inode ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findInodesWithWrongOwner(),
      &ModeCheckFS::repairWrongInodeOwner, prompt);
}

int64_t ModeCheckFS::checkAndRepairWrongOwnerInDentry()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_CORRECTOWNER,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_CORRECTOWNER);

   FsckTkEx::fsckOutput("* Checking: Dentry points to inode on wrong node ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findDirEntriesWithWrongOwner(),
      &ModeCheckFS::repairWrongInodeOwnerInDentry, prompt);
}

int64_t ModeCheckFS::checkAndRepairOrphanedContDir()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_CREATEDEFAULTDIRINODE,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_CREATEDEFAULTDIRINODE);

   FsckTkEx::fsckOutput("* Checking: Content directory without an inode ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findOrphanedContDirs(),
      &ModeCheckFS::repairOrphanedContDir, prompt);
}

int64_t ModeCheckFS::checkAndRepairOrphanedDirInode()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_LOSTANDFOUND,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_LOSTANDFOUND);

   FsckTkEx::fsckOutput("* Checking: Dir inode without a dentry pointing to it (orphaned inode) ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   bool result = checkAndRepairGeneric(this->database->findOrphanedDirInodes(),
      &ModeCheckFS::repairOrphanedDirInode, prompt);

   releaseLostAndFound();
   return result;
}

int64_t ModeCheckFS::checkAndRepairOrphanedFileInode()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_DELETEINODE,
      //FsckRepairAction_LOSTANDFOUND,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_DELETEINODE);

   FsckTkEx::fsckOutput("* Checking: File inode without a dentry pointing to it (orphaned inode) ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findOrphanedFileInodes(),
      &ModeCheckFS::repairOrphanedFileInode, prompt);
}

int64_t ModeCheckFS::checkAndRepairOrphanedChunk()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_NOTHING);
   RepairChunkState state = { &prompt, "", FsckRepairAction_UNDEFINED };

   FsckTkEx::fsckOutput("* Checking: Chunk without an inode pointing to it (orphaned chunk) ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findOrphanedChunks(),
      &ModeCheckFS::repairOrphanedChunk, state);
}

int64_t ModeCheckFS::checkAndRepairMissingContDir()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_CREATECONTDIR,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_CREATECONTDIR);

   FsckTkEx::fsckOutput("* Checking: Directory inode without a content directory ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findInodesWithoutContDir(),
      &ModeCheckFS::repairMissingContDir, prompt);
}

int64_t ModeCheckFS::checkAndRepairWrongFileAttribs()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_UPDATEATTRIBS,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_UPDATEATTRIBS);

   FsckTkEx::fsckOutput("* Checking: Attributes of file inode are wrong ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findWrongInodeFileAttribs(),
      &ModeCheckFS::repairWrongFileAttribs, prompt);
}

int64_t ModeCheckFS::checkAndRepairWrongDirAttribs()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_UPDATEATTRIBS,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_UPDATEATTRIBS);

   FsckTkEx::fsckOutput("* Checking: Attributes of dir inode are wrong ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findWrongInodeDirAttribs(),
      &ModeCheckFS::repairWrongDirAttribs, prompt);
}

int64_t ModeCheckFS::checkAndRepairFilesWithMissingTargets()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_DELETEFILE,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_NOTHING);

   FsckTkEx::fsckOutput("* Checking: File has a missing target in stripe pattern ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(
      this->database->findFilesWithMissingStripeTargets(
         Program::getApp()->getTargetMapper(), Program::getApp()->getMirrorBuddyGroupMapper() ),
      &ModeCheckFS::repairFileWithMissingTargets, prompt);
}

int64_t ModeCheckFS::checkAndRepairDirEntriesWithBrokeByIDFile()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_RECREATEFSID,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_RECREATEFSID);

   FsckTkEx::fsckOutput("* Checking: Dentry-by-ID file is broken or missing ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findDirEntriesWithBrokenByIDFile(),
      &ModeCheckFS::repairDirEntryWithBrokenByIDFile, prompt);
}

int64_t ModeCheckFS::checkAndRepairOrphanedDentryByIDFiles()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_RECREATEDENTRY,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_RECREATEDENTRY);

   FsckTkEx::fsckOutput("* Checking: Dentry-by-ID file is present, but no corresponding dentry ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findOrphanedFsIDFiles(),
      &ModeCheckFS::repairOrphanedDentryByIDFile, prompt);
}

int64_t ModeCheckFS::checkAndRepairChunksWithWrongPermissions()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_FIXPERMISSIONS,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_FIXPERMISSIONS);

   FsckTkEx::fsckOutput("* Checking: Chunk has wrong permissions ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findChunksWithWrongPermissions(),
      &ModeCheckFS::repairChunkWithWrongPermissions, prompt);
}

// no repair at the moment
int64_t ModeCheckFS::checkAndRepairChunksInWrongPath()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_MOVECHUNK,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_MOVECHUNK);

   FsckTkEx::fsckOutput("* Checking: Chunk is saved in wrong path ...",
      OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(this->database->findChunksInWrongPath(),
      &ModeCheckFS::repairWrongChunkPath, prompt);
}

int64_t ModeCheckFS::checkDuplicateInodeIDs()
{
   FsckTkEx::fsckOutput("* Checking: Duplicated inode IDs ... ",
      OutputOptions_FLUSH |  OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findDuplicateInodeIDs(),
      &ModeCheckFS::logDuplicateInodeID, dummy);
}

void ModeCheckFS::logDuplicateInodeID(checks::DuplicatedInode& dups, int&)
{
   FsckTkEx::fsckOutput(">>> Found duplicated ID " + dups.first.str(),
      OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   for(const auto& it : dups.second)
   {
      FsckTkEx::fsckOutput("   * Found on " + std::string(it.second ? "buddy group " : "node ")
            + StringTk::uintToStr(it.first), OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   }
}

int64_t ModeCheckFS::checkDuplicateChunks()
{
   FsckTkEx::fsckOutput("* Checking: Duplicated chunks ...", OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findDuplicateChunks(),
      &ModeCheckFS::logDuplicateChunk, dummy);
}

void ModeCheckFS::logDuplicateChunk(std::list<FsckChunk>& dups, int&)
{
   FsckTkEx::fsckOutput(">>> Found duplicated Chunks for ID " + dups.begin()->getID(),
      OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   for(std::list<FsckChunk>::iterator it = dups.begin(), end = dups.end();
         it != end; ++it)
   {
      FsckTkEx::fsckOutput("   * Found on target " + StringTk::uintToStr(it->getTargetID() )
         + (it->getBuddyGroupID()
               ? ", buddy group " + StringTk::uintToStr(it->getBuddyGroupID() )
               : "")
         + " in path " + it->getSavedPath()->str(),
         OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   }
}

int64_t ModeCheckFS::checkDuplicateContDirs()
{
   FsckTkEx::fsckOutput("* Checking: Duplicated content directores ...",
         OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findDuplicateContDirs(),
      &ModeCheckFS::logDuplicateContDir, dummy);
}

void ModeCheckFS::logDuplicateContDir(std::list<db::ContDir>& dups, int&)
{
   FsckTkEx::fsckOutput(">>> Found duplicated content directories for ID " + dups.front().id.str(),
      OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   for (auto it = dups.begin(), end = dups.end(); it != end; ++it)
   {
      FsckTkEx::fsckOutput("   * Found on " +
            std::string(it->isBuddyMirrored ? "buddy group " : "node ") +
            StringTk::uintToStr(it->saveNodeID),
         OutputOptions_LINEBREAK | OutputOptions_NOSTDOUT);
   }
}

int64_t ModeCheckFS::checkMismirroredDentries()
{
   FsckTkEx::fsckOutput("* Checking: Bad target mirror information in dentry ...",
         OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findMismirroredDentries(),
      &ModeCheckFS::logMismirroredDentry, dummy);
}

void ModeCheckFS::logMismirroredDentry(db::DirEntry& entry, int&)
{
   FsckTkEx::fsckOutput(">>> Found mismirrored dentry " +
         database->getDentryTable()->getNameOf(entry) + " on " +
         (entry.isBuddyMirrored ? "buddy group " : "node ") +
         StringTk::uintToStr(entry.saveNodeID) + " " + StringTk::uintToStr(entry.entryOwnerNodeID));
}

int64_t ModeCheckFS::checkMismirroredDirectories()
{
   FsckTkEx::fsckOutput("* Checking: Bad content mirror information in dir inode ...",
         OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findMismirroredDirectories(),
      &ModeCheckFS::logMismirroredDirectory, dummy);
}

void ModeCheckFS::logMismirroredDirectory(db::DirInode& dir, int&)
{
   FsckTkEx::fsckOutput(">>> Found mismirrored directory " + dir.id.str() + " on " +
         (dir.isBuddyMirrored ? "buddy group " : "node ") +
         StringTk::uintToStr(dir.saveNodeID));
;
}

int64_t ModeCheckFS::checkMismirroredFiles()
{
   FsckTkEx::fsckOutput("* Checking: Bad content mirror information in file inode ...",
         OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   int dummy = 0;
   return checkAndRepairGeneric(this->database->findMismirroredFiles(),
      &ModeCheckFS::logMismirroredFile, dummy);
}

void ModeCheckFS::logMismirroredFile(db::FileInode& file, int&)
{
   FsckTkEx::fsckOutput(">>> Found mismirrored file " + file.id.str() + " on " +
         std::string(file.isBuddyMirrored ? "buddy group " : "node ") +
         StringTk::uintToStr(file.saveNodeID));
}

int64_t ModeCheckFS::checkAndRepairMalformedChunk()
{
   FsckRepairAction possibleActions[] = {
      FsckRepairAction_NOTHING,
      FsckRepairAction_DELETECHUNK,
   };

   UserPrompter prompt(possibleActions, FsckRepairAction_DELETECHUNK);

   FsckTkEx::fsckOutput("* Checking: Malformed chunk ...", OutputOptions_FLUSH | OutputOptions_LINEBREAK);

   return checkAndRepairGeneric(Cursor<FsckChunk>(database->getMalformedChunksList()->cursor()),
      &ModeCheckFS::repairMalformedChunk, prompt);
}

void ModeCheckFS::repairMalformedChunk(FsckChunk& chunk, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Chunk ID: " + chunk.getID() + " on " +
         (chunk.getBuddyGroupID()
            ? "buddy group " + StringTk::uintToStr(chunk.getBuddyGroupID())
            : "target " + StringTk::uintToStr(chunk.getTargetID())));

   switch (action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_DELETECHUNK: {
      FhgfsOpsErr refErr;
      NodeStore* storageNodeStore = Program::getApp()->getStorageNodes();
      auto node = storageNodeStore->referenceNodeByTargetID(chunk.getTargetID(),
         Program::getApp()->getTargetMapper(), &refErr);

      if(!node)
      {
         FsckTkEx::fsckOutput("could not get storage target "
            + StringTk::uintToStr(chunk.getTargetID() ), OutputOptions_LINEBREAK);
         return;
      }

      FsckChunkList chunks(1, chunk);
      FsckChunkList failed;

      MsgHelperRepair::deleteChunks(*node, &chunks, &failed);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::checkAndRepair()
{
   FsckTkEx::fsckOutput("Step 4: Check for errors... ", OutputOptions_DOUBLELINEBREAK);

   Config* cfg = Program::getApp()->getConfig();

   int64_t errorCount = 0;

   errorCount += checkDuplicateInodeIDs();
   errorCount += checkDuplicateChunks();
   errorCount += checkDuplicateContDirs();
   errorCount += checkMismirroredDentries();
   errorCount += checkMismirroredDirectories();
   errorCount += checkMismirroredFiles();

   if(errorCount)
   {
      FsckTkEx::fsckOutput("Found errors beegfs-fsck cannot fix. Please consult the log for "
         "more information.", OutputOptions_LINEBREAK);
      return;
   }

   errorCount += checkAndRepairMalformedChunk();
   errorCount += checkAndRepairFilesWithMissingTargets();
   errorCount += checkAndRepairOrphanedDentryByIDFiles();
   errorCount += checkAndRepairDirEntriesWithBrokeByIDFile();
   errorCount += checkAndRepairOrphanedChunk();
   errorCount += checkAndRepairChunksInWrongPath();
   errorCount += checkAndRepairWrongInodeOwner();
   errorCount += checkAndRepairWrongOwnerInDentry();
   errorCount += checkAndRepairOrphanedContDir();
   errorCount += checkAndRepairOrphanedDirInode();
   errorCount += checkAndRepairOrphanedFileInode();
   errorCount += checkAndRepairDanglingDentry();
   errorCount += checkAndRepairMissingContDir();
   errorCount += checkAndRepairWrongFileAttribs();
   errorCount += checkAndRepairWrongDirAttribs();

   if ( cfg->getQuotaEnabled())
   {
      errorCount += checkAndRepairChunksWithWrongPermissions();
   }

   if ( cfg->getReadOnly() )
   {
      FsckTkEx::fsckOutput(
         "Found " + StringTk::int64ToStr(errorCount)
            + " errors. Detailed information can also be found in "
            + Program::getApp()->getConfig()->getLogOutFile() + ".",
         OutputOptions_ADDLINEBREAKBEFORE | OutputOptions_LINEBREAK);
      return;
   }

   for (auto it = secondariesSetBad.begin(); it != secondariesSetBad.end(); ++it)
   {
      auto secondary = *it;

      FsckTkEx::fsckOutput(">>> Setting metadata node " + StringTk::intToStr(secondary.val())
            + " to needs-resync", OutputOptions_LINEBREAK);

      auto setRes = MsgHelperRepair::setNodeState(secondary, TargetConsistencyState_NEEDS_RESYNC);
      if (setRes != FhgfsOpsErr_SUCCESS)
         FsckTkEx::fsckOutput("Failed: " + boost::lexical_cast<std::string>(setRes),
               OutputOptions_LINEBREAK);
   }

   if(errorCount > 0)
      FsckTkEx::fsckOutput(">>> Found " + StringTk::int64ToStr(errorCount) + " errors <<< ",
         OutputOptions_ADDLINEBREAKBEFORE | OutputOptions_LINEBREAK);
}

//////////////////////////
// internals      ///////
/////////////////////////
void ModeCheckFS::repairDanglingDirEntry(db::DirEntry& entry,
   std::pair<UserPrompter*, UserPrompter*>& prompt)
{
   FsckDirEntry fsckEntry = entry;

   FsckRepairAction action;
   std::string promptText = "Entry ID: " + fsckEntry.getID() + "; Path: " +
         this->database->getDentryTable()->getPathOf(entry) + "; " +
         (entry.isBuddyMirrored ? "Buddy group: " : "Node: ") +
         StringTk::uintToStr(entry.saveNodeID);

   if(fsckEntry.getEntryType() == FsckDirEntryType_DIRECTORY)
      action = prompt.second->chooseAction(promptText);
   else
      action = prompt.first->chooseAction(promptText);

   fsckEntry.setName(this->database->getDentryTable()->getNameOf(entry) );

   FsckDirEntryList entries(1, fsckEntry);
   FsckDirEntryList failedEntries;

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_DELETEDENTRY: {
      MsgHelperRepair::deleteDanglingDirEntries(fsckEntry.getSaveNodeID(),
            fsckEntry.getIsBuddyMirrored(), &entries, &failedEntries,
            secondariesSetBad);

      if(failedEntries.empty() )
      {
         this->database->getDentryTable()->remove(entries);
         this->deleteFsIDsFromDB(entries);
      }

      break;
   }

   case FsckRepairAction_CREATEDEFAULTDIRINODE: {
      FsckDirInodeList createdInodes;

      // create mirrored inodes iff the dentry was mirrored. if a contdir with the same id exists,
      // a previous check will have created an inode for it, leaving this dentry not dangling.
      MsgHelperRepair::createDefDirInodes(fsckEntry.getSaveNodeID(), fsckEntry.getIsBuddyMirrored(),
            {std::make_tuple(fsckEntry.getID(), fsckEntry.getIsBuddyMirrored())}, &createdInodes,
            secondariesSetBad);

      this->database->getDirInodesTable()->insert(createdInodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairWrongInodeOwner(FsckDirInode& inode, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Entry ID: " + inode.getID()
      + "; Path: "
      + this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(inode.getID() ) ) );

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_CORRECTOWNER: {
      inode.setOwnerNodeID(inode.getSaveNodeID() );

      FsckDirInodeList inodes(1, inode);
      FsckDirInodeList failed;

      MsgHelperRepair::correctInodeOwners(inode.getSaveNodeID(), inode.getIsBuddyMirrored(),
            &inodes, &failed, secondariesSetBad);

      if(failed.empty() )
         this->database->getDirInodesTable()->update(inodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairWrongInodeOwnerInDentry(std::pair<db::DirEntry, NumNodeID>& error,
   UserPrompter& prompt)
{
   FsckDirEntry fsckEntry = error.first;

   FsckRepairAction action = prompt.chooseAction("File ID: " + fsckEntry.getID()
      + "; Path: " + this->database->getDentryTable()->getPathOf(error.first) );

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_CORRECTOWNER: {
      fsckEntry.setName(this->database->getDentryTable()->getNameOf(error.first) );

      FsckDirEntryList dentries(1, fsckEntry);
      NumNodeIDList owners(1, NumNodeID(error.second) );
      FsckDirEntryList failed;

      MsgHelperRepair::correctInodeOwnersInDentry(fsckEntry.getSaveNodeID(),
            fsckEntry.getIsBuddyMirrored(), &dentries, &owners, &failed,
            secondariesSetBad);

      if(failed.empty() )
         this->database->getDentryTable()->updateFieldsExceptParent(dentries);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairOrphanedDirInode(FsckDirInode& inode, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Directory ID: " + inode.getID() + "; " +
         (inode.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") + inode.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_LOSTANDFOUND: {
      if(!ensureLostAndFoundExists() )
      {
         log.logErr("Orphaned dir inodes could not be linked to lost+found, because lost+found "
            "directory could not be created");
         return;
      }

      FsckDirInodeList inodes(1, inode);
      FsckDirEntryList created;
      FsckDirInodeList failed;

      MsgHelperRepair::linkToLostAndFound(*this->lostAndFoundNode, &this->lostAndFoundInfo, &inodes,
         &failed, &created, secondariesSetBad);

      if(failed.empty() )
         this->database->getDentryTable()->insert(created);

      // on server side, each dentry also created a dentry-by-ID file
      FsckFsIDList createdFsIDs;
      for(FsckDirEntryListIter iter = created.begin(); iter != created.end(); iter++)
      {
         FsckFsID fsID(iter->getID(), iter->getParentDirID(), iter->getSaveNodeID(),
            iter->getSaveDevice(), iter->getSaveInode(), iter->getIsBuddyMirrored());
         createdFsIDs.push_back(fsID);
      }

      this->database->getFsIDsTable()->insert(createdFsIDs);
      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairOrphanedFileInode(FsckFileInode& inode, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("File ID: " + inode.getID() + "; " +
         (inode.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") + inode.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_DELETEINODE: {
      FsckFileInodeList inodes(1, inode);
      StringList failed;

      MsgHelperRepair::deleteFileInodes(inode.getSaveNodeID(), inode.getIsBuddyMirrored(), inodes,
            failed, secondariesSetBad);

      if(failed.empty() )
         this->database->getFileInodesTable()->remove(inodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairOrphanedChunk(FsckChunk& chunk, RepairChunkState& state)
{
   // ask for repair action only once per chunk id, not once per chunk id and node
   if(state.lastID != chunk.getID() )
      state.lastChunkAction = state.prompt->chooseAction("Chunk ID: " + chunk.getID() + " on " +
            (chunk.getBuddyGroupID()
               ? "buddy group " + StringTk::uintToStr(chunk.getBuddyGroupID())
               : "target " + StringTk::uintToStr(chunk.getTargetID())));

   state.lastID = chunk.getID();

   switch(state.lastChunkAction)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_DELETECHUNK: {
      FhgfsOpsErr refErr;
      NodeStore* storageNodeStore = Program::getApp()->getStorageNodes();
      auto node = storageNodeStore->referenceNodeByTargetID(chunk.getTargetID(),
         Program::getApp()->getTargetMapper(), &refErr);

      if(!node)
      {
         FsckTkEx::fsckOutput("could not get storage target "
            + StringTk::uintToStr(chunk.getTargetID() ), OutputOptions_LINEBREAK);
         return;
      }

      FsckChunkList chunks(1, chunk);
      FsckChunkList failed;

      MsgHelperRepair::deleteChunks(*node, &chunks, &failed);

      if(failed.empty() )
         this->database->getChunksTable()->remove(
               {db::EntryID::fromStr(chunk.getID()), chunk.getTargetID(), chunk.getBuddyGroupID()});

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairMissingContDir(FsckDirInode& inode, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Directory ID: " + inode.getID() +
      "; Path: " +
      this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(inode.getID())) + "; " +
      (inode.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") + inode.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_CREATECONTDIR: {
      FsckDirInodeList inodes(1, inode);
      StringList failed;

      MsgHelperRepair::createContDirs(inode.getSaveNodeID(), inode.getIsBuddyMirrored(), &inodes,
            &failed, secondariesSetBad);

      if(failed.empty() )
      {
         // create a list of cont dirs from dir inode
         FsckContDirList contDirs(1,
            FsckContDir(inode.getID(), inode.getSaveNodeID(), inode.getIsBuddyMirrored()));

         this->database->getContDirsTable()->insert(contDirs);
      }

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairOrphanedContDir(FsckContDir& dir, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Directory ID: " + dir.getID() + "; " +
         (dir.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") + dir.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_CREATEDEFAULTDIRINODE: {
      FsckDirInodeList createdInodes;
      MsgHelperRepair::createDefDirInodes(dir.getSaveNodeID(), dir.getIsBuddyMirrored(),
            {std::make_tuple(dir.getID(), dir.getIsBuddyMirrored())}, &createdInodes,
            secondariesSetBad);

      this->database->getDirInodesTable()->insert(createdInodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairWrongFileAttribs(std::pair<FsckFileInode, checks::InodeAttribs>& error,
   UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("File ID: " + error.first.getID() + "; Path: " +
         this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(error.first.getID())) +
         "; " + (error.first.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") +
         error.first.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_UPDATEATTRIBS: {
      error.first.setFileSize(error.second.size);
      error.first.setNumHardLinks(error.second.nlinks);

      FsckFileInodeList inodes(1, error.first);
      FsckFileInodeList failed;

      MsgHelperRepair::updateFileAttribs(error.first.getSaveNodeID(),
            error.first.getIsBuddyMirrored(), &inodes, &failed, secondariesSetBad);

      if(failed.empty() )
         this->database->getFileInodesTable()->update(inodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairWrongDirAttribs(std::pair<FsckDirInode, checks::InodeAttribs>& error,
   UserPrompter& prompt)
{
   std::string filePath = this->database->getDentryTable()->getPathOf(
      db::EntryID::fromStr(error.first.getID() ) );
   FsckRepairAction action = prompt.chooseAction("Directory ID: " + error.first.getID() +
         "; Path: " + filePath + "; " +
         (error.first.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") +
         error.first.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_UPDATEATTRIBS: {
      FsckDirInodeList inodes(1, error.first);
      FsckDirInodeList failed;

      // update the file attribs in the inode objects (even though they may not be used
      // on the server side, we need updated values here to update the DB
      error.first.setSize(error.second.size);
      error.first.setNumHardLinks(error.second.nlinks);

      MsgHelperRepair::updateDirAttribs(error.first.getSaveNodeID(),
            error.first.getIsBuddyMirrored(), &inodes, &failed, secondariesSetBad);

      if(failed.empty() )
         this->database->getDirInodesTable()->update(inodes);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairFileWithMissingTargets(db::DirEntry& entry, UserPrompter& prompt)
{
   FsckDirEntry fsckEntry = entry;

   FsckRepairAction action = prompt.chooseAction("Entry ID: " + fsckEntry.getID() +
         "; Path: " + this->database->getDentryTable()->getPathOf(entry) + "; " +
         (entry.isBuddyMirrored ? "Buddy group: " : "Node: ") +
         StringTk::uintToStr(entry.entryOwnerNodeID));

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      return;

   case FsckRepairAction_DELETEFILE: {
      fsckEntry.setName(this->database->getDentryTable()->getNameOf(entry) );

      FsckDirEntryList dentries(1, fsckEntry);
      FsckDirEntryList failed;

      MsgHelperRepair::deleteFiles(fsckEntry.getSaveNodeID(), fsckEntry.getIsBuddyMirrored(),
            &dentries, &failed);

      if(failed.empty() )
         this->deleteFilesFromDB(dentries);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairDirEntryWithBrokenByIDFile(db::DirEntry& entry, UserPrompter& prompt)
{
   FsckDirEntry fsckEntry = entry;

   FsckRepairAction action = prompt.chooseAction("Entry ID: " + fsckEntry.getID() +
         "; Path: " + this->database->getDentryTable()->getPathOf(entry) + "; " +
         (entry.isBuddyMirrored ? "Buddy group: " : "Node: ") +
         StringTk::uintToStr(entry.entryOwnerNodeID));

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_RECREATEFSID: {
      fsckEntry.setName(this->database->getDentryTable()->getNameOf(entry) );

      FsckDirEntryList dentries(1, fsckEntry);
      FsckDirEntryList failed;

      MsgHelperRepair::recreateFsIDs(fsckEntry.getSaveNodeID(), fsckEntry.getIsBuddyMirrored(),
            &dentries, &failed, secondariesSetBad);

      if(failed.empty() )
      {
         // create a FsID list from the dentry
         FsckFsIDList idList(1,
            FsckFsID(fsckEntry.getID(), fsckEntry.getParentDirID(), fsckEntry.getSaveNodeID(),
               fsckEntry.getSaveDevice(), fsckEntry.getSaveInode(),
               fsckEntry.getIsBuddyMirrored()));

         this->database->getFsIDsTable()->insert(idList);
      }

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairOrphanedDentryByIDFile(FsckFsID& id, UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction(
         "Entry ID: " + id.getID() +
         "; Path: " +
         this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(id.getID())) + "; " +
         (id.getIsBuddyMirrored() ? "Buddy group: " : "Node: ") + id.getSaveNodeID().str());

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_RECREATEDENTRY: {
      FsckFsIDList fsIDs(1, id);
      FsckFsIDList failed;
      FsckDirEntryList createdDentries;
      FsckFileInodeList createdInodes;

      MsgHelperRepair::recreateDentries(id.getSaveNodeID(), id.getIsBuddyMirrored(), &fsIDs,
            &failed, &createdDentries, &createdInodes, secondariesSetBad);

      if(failed.empty() )
      {
         this->database->getDentryTable()->insert(createdDentries);
         this->database->getFileInodesTable()->insert(createdInodes);
      }

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairChunkWithWrongPermissions(std::pair<FsckChunk, FsckFileInode>& error,
   UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction(
         "Chunk ID: " + error.first.getID() + "; "
         + (error.first.getBuddyGroupID()
         ? "Buddy group: " + StringTk::uintToStr(error.first.getBuddyGroupID())
         : "Target: " + StringTk::uintToStr(error.first.getTargetID()))
         + "; File path: "
         + this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(error.first.getID())));

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_FIXPERMISSIONS: {
      NodeStore* nodeStore = Program::getApp()->getStorageNodes();
      TargetMapper* targetMapper = Program::getApp()->getTargetMapper();

      // we will need the PathInfo later to send the SetAttr message and we don't have it in the
      // chunk
      PathInfoList pathInfoList(1, *error.second.getPathInfo() );

      // set correct permissions
      error.first.setUserID(error.second.getUserID() );
      error.first.setGroupID(error.second.getGroupID() );

      FsckChunkList chunkList(1, error.first);
      FsckChunkList failed;

      auto storageNode = nodeStore->referenceNode(
         targetMapper->getNodeID(error.first.getTargetID() ) );
      MsgHelperRepair::fixChunkPermissions(*storageNode, chunkList, pathInfoList, failed);

      if(failed.empty() )
         this->database->getChunksTable()->update(chunkList);

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::repairWrongChunkPath(std::pair<FsckChunk, FsckFileInode>& error,
   UserPrompter& prompt)
{
   FsckRepairAction action = prompt.chooseAction("Entry ID: " + error.first.getID()
      + "; " + (error.first.getBuddyGroupID()
         ? "Group: " + StringTk::uintToStr(error.first.getBuddyGroupID())
         : "Target: " + StringTk::uintToStr(error.first.getTargetID()))
      + "; Chunk path: " + error.first.getSavedPath()->str()
      + "; File path: "
      + this->database->getDentryTable()->getPathOf(db::EntryID::fromStr(error.first.getID() ) ) );

   switch(action)
   {
   case FsckRepairAction_UNDEFINED:
   case FsckRepairAction_NOTHING:
      break;

   case FsckRepairAction_MOVECHUNK: {
      NodeStore* nodeStore = Program::getApp()->getStorageNodes();
      TargetMapper* targetMapper = Program::getApp()->getTargetMapper();

      auto storageNode = nodeStore->referenceNode(
         targetMapper->getNodeID(error.first.getTargetID() ) );

      std::string moveToPath = DatabaseTk::calculateExpectedChunkPath(error.first.getID(),
         error.second.getPathInfo()->getOrigUID(),
         error.second.getPathInfo()->getOrigParentEntryID(),
         error.second.getPathInfo()->getFlags() );

      if(MsgHelperRepair::moveChunk(*storageNode, error.first, moveToPath, false) )
      {
         FsckChunkList chunks(1, error.first);
         this->database->getChunksTable()->update(chunks);
         break;
      }

      // chunk file exists at the correct location
      FsckTkEx::fsckOutput("Chunk file for " + error.first.getID()
         + " already exists at the correct location. ", OutputOptions_NONE);

      if(Program::getApp()->getConfig()->getAutomatic() )
      {
         FsckTkEx::fsckOutput("Will not attempt automatic repair.", OutputOptions_LINEBREAK);
         break;
      }

      char chosen = 0;

      while(chosen != 'y' && chosen != 'n')
      {
         FsckTkEx::fsckOutput("Move anyway? (y/n) ", OutputOptions_NONE);

         std::string line;
         std::getline(std::cin, line);

         if(line.size() == 1)
            chosen = line[0];
      }

      if(chosen != 'y')
         break;

      if(MsgHelperRepair::moveChunk(*storageNode, error.first, moveToPath, true) )
      {
         FsckChunkList chunks(1, error.first);
         this->database->getChunksTable()->update(chunks);
      }
      else
      {
         FsckTkEx::fsckOutput("Repair failed, see log file for more details.",
            OutputOptions_LINEBREAK);
      }

      break;
   }

   default:
      throw std::runtime_error("bad repair action");
   }
}

void ModeCheckFS::deleteFsIDsFromDB(FsckDirEntryList& dentries)
{
   // create the fsID list
   FsckFsIDList fsIDs;

   for ( FsckDirEntryListIter iter = dentries.begin(); iter != dentries.end(); iter++ )
   {
      FsckFsID fsID(iter->getID(), iter->getParentDirID(), iter->getSaveNodeID(),
         iter->getSaveDevice(), iter->getSaveInode(), iter->getIsBuddyMirrored());
      fsIDs.push_back(fsID);
   }

   this->database->getFsIDsTable()->remove(fsIDs);
}

void ModeCheckFS::deleteFilesFromDB(FsckDirEntryList& dentries)
{
   for ( FsckDirEntryListIter dentryIter = dentries.begin(); dentryIter != dentries.end();
      dentryIter++ )
   {
      std::pair<bool, db::FileInode> inode = this->database->getFileInodesTable()->get(
         dentryIter->getID() );

      if(inode.first)
      {
         this->database->getFileInodesTable()->remove(inode.second.id);
         this->database->getChunksTable()->remove(inode.second.id);
      }
   }

   this->database->getDentryTable()->remove(dentries);

   // delete the dentry-by-id files
   this->deleteFsIDsFromDB(dentries);
}

bool ModeCheckFS::ensureLostAndFoundExists()
{
   this->lostAndFoundNode = MsgHelperRepair::referenceLostAndFoundOwner(&this->lostAndFoundInfo);

   if(!this->lostAndFoundNode)
   {
      if(!MsgHelperRepair::createLostAndFound(this->lostAndFoundNode, this->lostAndFoundInfo) )
         return false;
   }

   std::pair<bool, FsckDirInode> dbLaF = this->database->getDirInodesTable()->get(
      this->lostAndFoundInfo.getEntryID() );

   if(dbLaF.first)
      this->lostAndFoundInode.reset(new FsckDirInode(dbLaF.second) );

   return true;
}

void ModeCheckFS::releaseLostAndFound()
{
   if(!this->lostAndFoundNode)
      return;

   if(this->lostAndFoundInode)
   {
      FsckDirInodeList updates(1, *this->lostAndFoundInode);
      this->database->getDirInodesTable()->update(updates);
   }

   this->lostAndFoundInode.reset();
}

max_stars_repo_name: TomatoYoung/beegfs

max_issues_repo_name: TomatoYoung/beegfs

max_forks_repo_name: TomatoYoung/beegfs


text:
/**
 * Copyright (c) 2009 Carnegie Mellon University.
 *     All rights reserved.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an "AS
 *  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 *  express or implied.  See the License for the specific language
 *  governing permissions and limitations under the License.
 *
 * For more about this software visit:
 *
 *      http://www.graphlab.ml.cmu.edu
 *
 */

#ifndef GRAPHLAB_DC_HPP
#define GRAPHLAB_DC_HPP
#include <iostream>
#include <boost/iostreams/stream.hpp>
#include <graphlab/parallel/pthread_tools.hpp>
#include <graphlab/util/resizing_array_sink.hpp>
#include <graphlab/util/blocking_queue.hpp>
#include <graphlab/util/multi_blocking_queue.hpp>
#include <graphlab/util/dense_bitset.hpp>
#include <graphlab/serialization/serialization_includes.hpp>
#include <graphlab/metrics/metrics.hpp>

#include <graphlab/rpc/dc_types.hpp>
#include <graphlab/rpc/dc_internal_types.hpp>

#include <graphlab/rpc/dc_receive.hpp>
#include <graphlab/rpc/dc_send.hpp>
#include <graphlab/rpc/dc_comm_base.hpp>
#include <graphlab/rpc/dc_dist_object_base.hpp>

#include <graphlab/rpc/is_rpc_call.hpp>
#include <graphlab/rpc/portable_dispatch.hpp>
#include <graphlab/rpc/portable_issue.hpp>
#include <graphlab/rpc/function_call_issue.hpp>
#include <graphlab/rpc/request_issue.hpp>
#include <graphlab/rpc/reply_increment_counter.hpp>
#include <graphlab/rpc/function_ret_type.hpp>

#include <boost/preprocessor.hpp>
#include <graphlab/rpc/function_arg_types_def.hpp>

namespace graphlab {

/**
 * \ingroup rpc
Distributed control constructor parameters.
*/
struct dc_init_param{
  /** A vector containing a list of hostnames/ipaddresses and port numbers
  * of all machines participating in this RPC program.
  * for instance:
  * \code
  * machines.push_back("127.0.0.1:10000");
  * machines.push_back("127.0.0.1:10001");
  * \endcode
  */
  std::vector<std::string> machines;

  /** Additional construction options of the form
    "key1=value1,key2=value2".
    Available options are:

    \li \b compressed=yes Use ZLib compressed communication
    \li \b buffered_send=yes Put an circular buffer on outgoing transmission
    \li \b buffered_queued_send=yes Put a queue buffer on outgoing transmission
    \li \b buffered_queued_send_single=yes Like buffered_queued but use only one sending thread
    \li \b buffered_recv=yes Put a buffer on incoming transmissions
                             (not recommended. Tends to decrease performance)

    Internal options which should not be used
    \li \b __socket__=NUMBER Forces TCP comm to use this socket number for its
                             listening socket instead of creating a new one.
                             The socket must already be bound to the listening port.
  */
  std::string initstring;

  /** The index of this machine into the machines vector */
  procid_t curmachineid;
  /** Number of background RPC handling threads to create */
  size_t numhandlerthreads;
  /** The communication method. Must be TCP_COMM */
  dc_comm_type commtype;
};

#define DEFAULT_NUMHANDLERTHREADS 8
#define DEFAULT_COMMTYPE TCP_COMM
#define MAX_N_PROCS 64

// forward declaration for dc services
class dc_services;
class distributed_control;

/**
 * \ingroup rpc
graphlab::distributed_control is the primary distributed RPC object. This class initializes distributed
communication, as well as provide basic RPC routines and collective operations.

In addition to the documented functions, the following RPC routines are provided.

\par void distributed_control::remote_call(procid_t targetmachine, function, ...)
 remote_call performs a non-blocking RPC call to the target machine to
 run the provided function pointer. All arguments are transmitted by value
 and must be serializable.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

\par void distributed_control::fast_remote_call(procid_t targetmachine, function, ...)
 fast_remote_call is the same as remote_call, but the receiver completes the function
 call using the receiving thread instead of a thread pool. This should only be used if
 the function is short and does not block.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

\par void distributed_control::control_call(procid_t targetmachine, function, ...)
 Same as remote_call, but calls performed using the control_call do not contribute
 to the call counter and has no effect on graphlab::async_consensus.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

\par RetType distributed_control::remote_request(procid_t targetmachine, function, ...)
 remote_request performs a blocking RPC call to the target machine to
 run the provided function pointer. All arguments are transmitted by value
 and must be serializable. This call only returns when the target machine
 completes the function call. The return value of the call is serialized and
 returned.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

\par RetType distributed_control::fast_remote_request(procid_t targetmachine, function, ...)
 fast_remote_request is the same as remote_request, but the receiver completes the function
 call using the receiving thread instead of a thread pool. This should only be used if
 the function is short and does not block.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

\par void distributed_control::control_request(procid_t targetmachine, function, ...)
 Same as remote_request, but calls performed using the control_request do not contribute
 to the call counter and has no effect on graphlab::async_consensus.
 \li \b targetmachine: The ID of the machine to run the function on
 \li \b function: The function to run on the target machine

*/
class distributed_control{
  public:
        /**  Each element of the function call queue is a data/len pair */
    struct function_call_block{
      function_call_block() {}
      function_call_block(procid_t source, const dc_impl::packet_hdr& hdr,
                          char* data, size_t len):
                          source(source), hdr(hdr),
                          data(data), len(len) {}
      procid_t source;
      dc_impl::packet_hdr hdr;
      char* data;
      size_t len;
    };
  private:
   /// initialize receiver threads. private form of the constructor
   void init(const std::vector<std::string> &machines,
             const std::string &initstring,
             procid_t curmachineid,
             size_t numhandlerthreads,
             dc_comm_type commtype = DEFAULT_COMMTYPE);

  /// a pointer to the communications subsystem
  dc_impl::dc_comm_base* comm;

  /// senders and receivers to all machines
  std::vector<dc_impl::dc_receive*> receivers;
  std::vector<dc_impl::dc_send*> senders;

  /// A thread group of function call handlers
  thread_group fcallhandlers;

  /// a queue of functions to be executed
  std::vector<blocking_queue<function_call_block> > fcallqueue;

  /// A map of function name to dispatch function. Used for "portable" calls
  dc_impl::dispatch_map_type portable_dispatch_call_map;
  dc_impl::dispatch_map_type portable_dispatch_request_map;

  /// object registrations;
  std::vector<void*> registered_objects;
  std::vector<dc_impl::dc_dist_object_base*> registered_rmi_instance;

  /// For convenience, we provide a instance of dc_services
  dc_services* distributed_services;

  /// ID of the local machine
  procid_t localprocid;

  /// Number of machines
  procid_t localnumprocs;

  std::vector<atomic<size_t> > global_calls_sent;
  std::vector<atomic<size_t> > global_calls_received;

  bool single_sender;

  /// the callback given to the comms class. Called when data is inbound
  friend void dc_recv_callback(void* tag, procid_t src, const char* buf, size_t len);

  /// the callback given to the comms class. Called when data is inbound
  template <typename T> friend class dc_dist_object;

  /// disable the operator= by placing it in private
  distributed_control& operator=(const distributed_control& dc) { return *this; }

  std::map<std::string, std::string> parse_options(std::string initstring);

  volatile inline size_t num_registered_objects() {
    return registered_objects.size();
  }

  void compute_master_ranks();
  procid_t masterid;

  metrics rpc_metrics;

 public:

  distributed_control(dc_init_param initparam) {
    init(initparam.machines,
         initparam.initstring,
         initparam.curmachineid,
         initparam.numhandlerthreads,
         initparam.commtype);
  }

  distributed_control(const std::vector<std::string> &machines,
                      const std::string &initstring,
                      procid_t curmachineid,
                      size_t numhandlerthreads = DEFAULT_NUMHANDLERTHREADS,
                      dc_comm_type commtype = DEFAULT_COMMTYPE) {
    init(machines, initstring, curmachineid, numhandlerthreads, commtype);
  }

  ~distributed_control();

  /// returns the id of the current processor
  inline procid_t procid() const {
    return localprocid;
  }

  /// returns the number of processors in total.
  inline procid_t numprocs() const {
    return localnumprocs;
  }

  /**
  Sets the sequentialization key to the new value, returning the previous value.
  When the key is set to an arbitrary non-zero value, all
  remote calls/remote requests made by the current thread will be
  sequentially processed by the remote machines.

  All RPC calls made using the same key value will sequentialize.

  User should
  oldval = set_sequentialization_key(newval)
  ...
  ... do stuff
  ...
  set_sequentialization_key(oldval)
  */
  static unsigned char set_sequentialization_key(unsigned char newkey);

  /**
  Creates a new sequentialization key, returning the old value.
  All remote calls/remote requests made by the current thread will be
  sequentially processed by the remote machines.

  Essentially all RPC calls made using the same key value will sequentialize.
  However, since new_sequentialization_key() uses a very naive key selection system,
  we recommend the use of set_sequentialization_key() especially in the case of
  multi-threaded code.

  User should
  oldval = new_sequentialization_key()
  ...
  ... do stuff
  ...
  set_sequentialization_key(oldval)
  All RPC calls in while the key is set will be sequentialized on the receiving
  machine.
  */
  static unsigned char new_sequentialization_key();

  /// gets the current sequentialization key. This function is not generally useful.
  static unsigned char get_sequentialization_key();

  /*
  This generates the interface functions for the standard calls, basic calls, and fast calls
  The generated code looks like this:

  template<typename F , typename T0> void remote_call (procid_t target, F remote_function , const T0 &i0 )
  {
    ASSERT_LT(target, senders.size());
    dc_impl::remote_call_issue1 <F , T0> ::exec(senders[target],
                                                STANDARD_CALL,
                                                target,
                                                remote_function ,
                                                i0 );
  }
  The arguments passed to the RPC_INTERFACE_GENERATOR ARE: (interface name, issue processor name, flags)

  */
  #define GENARGS(Z,N,_)  BOOST_PP_CAT(const T, N) BOOST_PP_CAT(&i, N)
  #define GENI(Z,N,_) BOOST_PP_CAT(i, N)
  #define GENT(Z,N,_) BOOST_PP_CAT(T, N)
  #define GENARC(Z,N,_) arc << BOOST_PP_CAT(i, N);

  #define RPC_INTERFACE_GENERATOR(Z,N,FNAME_AND_CALL) \
  template<typename F BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM_PARAMS(N, typename T)> \
  void  BOOST_PP_TUPLE_ELEM(3,0,FNAME_AND_CALL) (procid_t target, F remote_function BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM(N,GENARGS ,_) ) {  \
    ASSERT_LT(target, senders.size()); \
    BOOST_PP_CAT( BOOST_PP_TUPLE_ELEM(3,1,FNAME_AND_CALL),N) \
        <F BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM_PARAMS(N, T)> \
          ::exec(senders[target],  BOOST_PP_TUPLE_ELEM(3,2,FNAME_AND_CALL), target, remote_function BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM(N,GENI ,_) ); \
  }   \

  /*
  Generates the interface functions. 3rd argument is a tuple (interface name, issue name, flags)
  */
  BOOST_PP_REPEAT(6, RPC_INTERFACE_GENERATOR, (remote_call, dc_impl::remote_call_issue, STANDARD_CALL) )
  BOOST_PP_REPEAT(6, RPC_INTERFACE_GENERATOR, (fast_remote_call,dc_impl::remote_call_issue, FAST_CALL) )
  BOOST_PP_REPEAT(6, RPC_INTERFACE_GENERATOR, (control_call, dc_impl::remote_call_issue, (FAST_CALL | CONTROL_PACKET)) )

  #define REQUEST_INTERFACE_GENERATOR(Z,N,ARGS) \
  template<typename F BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM_PARAMS(N, typename T)> \
    BOOST_PP_TUPLE_ELEM(3,0,ARGS) (procid_t target, F remote_function BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM(N,GENARGS ,_) ) {  \
    ASSERT_LT(target, senders.size()); \
    return BOOST_PP_CAT( BOOST_PP_TUPLE_ELEM(3,1,ARGS),N) \
        <F BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM_PARAMS(N, T)> \
          ::exec(senders[target],  BOOST_PP_TUPLE_ELEM(3,2,ARGS), target, remote_function BOOST_PP_COMMA_IF(N) BOOST_PP_ENUM(N,GENI ,_) ); \
  }   \

  /*
  Generates the interface functions. 3rd argument is a tuple (interface name, issue name, flags)
  */
 // BOOST_PP_REPEAT(6, REQUEST_INTERFACE_GENERATOR, (typename dc_impl::function_ret_type<__GLRPC_FRESULT>::type remote_request, dc_impl::remote_request_issue, 0) )
   BOOST_PP_REPEAT(6, REQUEST_INTERFACE_GENERATOR, (typename dc_impl::function_ret_type<__GLRPC_FRESULT>::type remote_request, dc_impl::remote_request_issue, STANDARD_CALL) )

  BOOST_PP_REPEAT(6, REQUEST_INTERFACE_GENERATOR, (typename dc_impl::function_ret_type<__GLRPC_FRESULT>::type fast_remote_request, dc_impl::remote_request_issue, FAST_CALL) )
  BOOST_PP_REPEAT(6, REQUEST_INTERFACE_GENERATOR, (typename dc_impl::function_ret_type<__GLRPC_FRESULT>::type control_request, dc_impl::remote_request_issue, (FAST_CALL | CONTROL_PACKET)) )

  #undef RPC_INTERFACE_GENERATOR
  #undef REQUEST_INTERFACE_GENERATOR
  #undef GENARC
  #undef GENT
  #undef GENI
  #undef GENARGS

  /**
   * \cond DC_INTERNAL
  Immediately calls the function described by the data
  inside the buffer. This should not be called directly.
  */
  void exec_function_call(procid_t source, const dc_impl::packet_hdr& hdr, std::istream &istrm);

  /**
  Performs a deferred function call using the information
  inside the buffer. This function will take over ownership of
  the buffer and will free it when done
  */
  void deferred_function_call(procid_t source, const dc_impl::packet_hdr& hdr,
                              char* buf, size_t len);

  /**
  This is called by the function handler threads
  */
  void fcallhandler_loop(size_t id);

  inline void inc_calls_sent(procid_t procid) {
//    ASSERT_FALSE(full_barrier_in_effect);
    global_calls_sent[procid].inc();
  }

  inline void inc_calls_received(procid_t procid) {

    if (!full_barrier_in_effect) {
        global_calls_received[procid].inc();
        if (full_barrier_in_effect) {

      if (global_calls_received[procid].inc() == calls_to_receive[procid]) {
        // if it was me who set the bit
        if (procs_complete.set_bit(procid) == false) {
          // then decrement the incomplete count.
          // if it was me to decreased it to 0
          // lock and signal
          full_barrier_lock.lock();
          if (num_proc_recvs_incomplete.dec() == 0) {
            full_barrier_cond.signal();
          }
          full_barrier_lock.unlock();
        }
      }

        }
    }
    else {
      //check the proc I just incremented.
      // If I just exceeded the required size, I need
      // to decrement the full barrier counter
      if (global_calls_received[procid].inc() == calls_to_receive[procid]) {
        // if it was me who set the bit
        if (procs_complete.set_bit(procid) == false) {
          // then decrement the incomplete count.
          // if it was me to decreased it to 0
          // lock and signal
          full_barrier_lock.lock();
          if (num_proc_recvs_incomplete.dec() == 0) {
            full_barrier_cond.signal();
          }
          full_barrier_lock.unlock();
        }
      }
    }
  }
  /// \endcond

  inline size_t calls_sent() const {
    size_t ctr = 0;
    for (size_t i = 0;i < numprocs(); ++i) {
      ctr += global_calls_sent[i].value;
    }
    return ctr;
  }

  inline size_t calls_received() const {
    size_t ctr = 0;
    for (size_t i = 0;i < numprocs(); ++i) {
      ctr += global_calls_received[i].value;
    }
    return ctr;
  }

  inline size_t bytes_sent() const {
    if (single_sender) {
      return senders[0]->bytes_sent();
    }
    else {
      size_t ret = 0;
      for (size_t i = 0;i < senders.size(); ++i) ret += senders[i]->bytes_sent();
      return ret;
    }
  }

  inline size_t network_bytes_sent() const {
    return comm->network_bytes_sent();
  }

  inline size_t bytes_received() const {
    size_t ret = 0;
    for (size_t i = 0;i < receivers.size(); ++i) ret += receivers[i]->bytes_received();
    return ret;
  }

  /**
    Returns true if this is the process with the lowest ID
    currently running on this machine in this working directory
  */
  inline bool is_master_rank() const {
    return masterid == procid();
  }

  /**
    Returns the lowest ID of all the processes
    currently running on this machine in this working directory
  */
  inline procid_t master_rank() const {
    return masterid;
  }

  /**
    registers a portable RPC call.
  */
  template <typename F, F f>
  void register_rpc(std::string c) {
    portable_dispatch_request_map[c] = (dc_impl::dispatch_type)
              dc_impl::portable_detail::find_dispatcher<F,        // function type
                              __GLRPC_FRESULT,                            // result
                              boost::function_traits<
                                    typename boost::remove_pointer<F>::type
                                                    >::arity ,   // number of arguments
                              f,                                    // function itself
                              typename dc_impl::is_rpc_call<F>::type  // whether it is an RPC style call
                              >::dispatch_request_fn();

    portable_dispatch_call_map[c] = (dc_impl::dispatch_type)
              dc_impl::portable_detail::find_dispatcher<F,        // function type
                              __GLRPC_FRESULT,                            // result
                              boost::function_traits<
                                    typename boost::remove_pointer<F>::type
                                                    >::arity ,   // number of arguments
                              f,                                    // function itself
                              typename dc_impl::is_rpc_call<F>::type  // whether it is an RPC style call
                              >::dispatch_call_fn();
  }

  /// \cond DC_INTERNAL
  inline size_t register_object(void* v, dc_impl::dc_dist_object_base *rmiinstance) {
    ASSERT_NE(v, (void*)NULL);
    registered_objects.push_back(v);
    registered_rmi_instance.push_back(rmiinstance);
    return registered_objects.size() - 1;
  }

  inline void* get_registered_object(size_t id) {
    while(id >= num_registered_objects()) sched_yield();
    ASSERT_NE(registered_objects[id], (void*)NULL);
    return registered_objects[id];
  }

  inline dc_impl::dc_dist_object_base* get_rmi_instance(size_t id) {
    while(id >= num_registered_objects()) sched_yield();
    ASSERT_NE(registered_rmi_instance[id], (void*)NULL);
    return registered_rmi_instance[id];
  }
  inline void clear_registered_object(size_t id) {
    registered_objects[id] = (void*)NULL;
    registered_rmi_instance[id] = NULL;
  }

  /**
  This is depreated. Use the public functions. In particular
  services().full_barrier() may not work as expected
  */
  __attribute__((__deprecated__)) dc_services& services();

  /// \endcond

  /**
   This comm barrier is not a true "barrier" but is
   essentially a sequentialization point. It guarantees that
   all calls from this machine to the target machine performed
   before the comm_barrier() call are completed before any call
   sent after the comm barrier() call.
  */
  void comm_barrier(procid_t targetmachine);

  /**
    This is a convenience function which broadcasts a comm_barrier()
    \note having all machines call the comm barrier does not guarantee
    that all calls have been processed. Basically 'p' local barriers
    do not result in a global barrier.
  */
  void comm_barrier();

// Temp hack.
  long long int total_bytes_sent;
  long long int get_total_bytes_sent() {
     return total_bytes_sent;
  }

  /**
  This is a blocking send_to. It send an object T to the target
  machine, but waits for the target machine to call recv_from
  before returning. Functionally similar to MPI's matched sending/receiving
  */
  template <typename U>
  inline void send_to(procid_t target, U& t, bool control = false);

  /**
  A blocking recv_from. Must be matched with a send_to call from the
  target before both source and target resumes.
  */
  template <typename U>
  inline void recv_from(procid_t source, U& t, bool control = false);

  /**
    This function allows one machine to broadcasts a variable to all machines.

    The originator calls broadcast with data provided in
    in 'data' and originator set to true.
    All other callers call with originator set to false.

    The originator will then return 'data'. All other machines
    will receive the originator's transmission in the "data" parameter.

    This call is guaranteed to have barrier-like behavior. That is to say,
    this call will block until all machines enter the broadcast function.

    \note Behavior is undefined if more than one machine calls broadcast
    with originator set to true.

    \note Behavior is undefined if multiple threads on the same machine
    call broadcast simultaneously. If multiple-thread broadcast is necessary,
    each thread should use its own instance of the services class.
  */
  template <typename U>
  inline void broadcast(U& data, bool originator, bool control = false);

  /**
   * Collects information contributed by each machine onto
   * one machine.
   * data must be of length data[numprocs].
   * My data is stored in data[dc.procid()].
   * when function returns, machine sendto will have the complete vector
   * where data[i] is the data contributed by machine i.
   * All machines must have the same parameter for "sendto"
   */
  template <typename U>
  inline void gather(std::vector<U>& data, procid_t sendto, bool control = false);

  /**
   * Each machine creates a vector 'data' with size equivalent to the number of machines.
   * Each machine then fills the entry data[procid()] with information that it
   * wishes to communicate.
   * After calling all_gather(), all machines will return with identical
   * vectors 'data', where data[i] contains the information machine i stored.
   */
  template <typename U>
  inline void all_gather(std::vector<U>& data, bool control = false);

  /**
   * This function is takes a vector of local elements T which must
   * be comparable and constructs a vector of length numprocs where
   * each element is a subset of the local contribution from that
   * machine and the union of all elements in the union of all local
   * contributions and all entries are unique:
   *
   * Usage: Each process reads the files that are stored locally and
   * wants to know which subset of local files to read even when
   * multiple processes see the same files.
   */
  template <typename U>
  inline void gather_partition(const std::vector<U>& local_contribution,
                        std::vector< std::vector<U> >& ret_partition,
                        bool control = false);

/**
    A regular barrier equivalent to MPI_Barrier.
    A machine entering this barrier will wait until every machine
    reaches this barrier before continuing. Only one thread from each machine
    should call the barrier.

    \see full_barrier
    */
  void barrier();

 /*****************************************************************************
                      Implementation of Full Barrier
*****************************************************************************/
  /**
  Similar to the barrier(), but provides additional guarantees that
  all calls issued prior to this barrier are completed before
  returning.

  \note This function could return prematurely if
  other threads are still issuing function calls since we
  cannot differentiate between calls issued before the barrier
  and calls issued while the barrier is being evaluated.
  Therefore, when used in a multithreaded scenario, the user must ensure
  that all other threads which may perform operations using this object
  are stopped before the full barrier is initated.

  \see barrier
  */
  void full_barrier();
 private:
  mutex full_barrier_lock;
  conditional full_barrier_cond;
  std::vector<size_t> calls_to_receive;
  // used to inform the counter that the full barrier
  // is in effect and all modifications to the calls_recv
  // counter will need to lock and signal
  bool full_barrier_in_effect;

  /** number of 'source' processor counts which have
  not achieved the right recv count */
  atomic<size_t> num_proc_recvs_incomplete;

  /// Marked as 1 if the proc is complete
  dense_bitset procs_complete;

 /*****************************************************************************
                      Collection of Statistics
*****************************************************************************/

 private:
  struct collected_statistics {
    size_t callssent;
    size_t bytessent;
    size_t network_bytessent;
    collected_statistics(): callssent(0), bytessent(0), network_bytessent(0) { }
    void save(oarchive &oarc) const {
      oarc << callssent << bytessent << network_bytessent;
    }
    void load(iarchive &iarc) {
      iarc >> callssent >> bytessent >> network_bytessent;
    }
  };
 public:
  /** Gather RPC statistics. All machines must call
   this function at the same time. However, only proc 0 will
   return values */
  std::map<std::string, size_t> gather_statistics();

  /** Fills metrics information. All machines must call this
   * function simultaneously. Only proc 0 will have metrics
   */
  void fill_metrics();

  /** returns metrics information collected by fill_metrics
   *  Only proc 0 will have metrics
   */
  inline metrics get_metrics() {
    return rpc_metrics;
  }

  inline void reset_metrics() {
    logstream(LOG_WARNING) << "Metrics cannot be reset on distributed control" << std::endl;
  }

  /** Dumps the metric information to a reporter
   * Only proc 0 will have metrics
   */
  inline void report_metrics(imetrics_reporter &reporter) {
    rpc_metrics.report(reporter);
  }

};

} // namespace graphlab

#define REGISTER_RPC(dc, f) dc.register_rpc<typeof(f)*, f>(std::string(BOOST_PP_STRINGIZE(f)))

#include <graphlab/rpc/function_arg_types_undef.hpp>
#include <graphlab/rpc/function_call_dispatch.hpp>
#include <graphlab/rpc/request_dispatch.hpp>
#include <graphlab/rpc/dc_dist_object.hpp>
#include <graphlab/rpc/dc_services.hpp>

namespace graphlab {

template <typename U>
inline void distributed_control::send_to(procid_t target, U& t, bool control) {
  distributed_services->send_to(target, t, control);
}

template <typename U>
inline void distributed_control::recv_from(procid_t source, U& t, bool control) {
  distributed_services->recv_from(source, t, control);
}

template <typename U>
inline void distributed_control::broadcast(U& data, bool originator, bool control) {
  distributed_services->broadcast(data, originator, control);
}

template <typename U>
inline void distributed_control::gather(std::vector<U>& data, procid_t sendto, bool control) {
  distributed_services->gather(data, sendto, control);
}

template <typename U>
inline void distributed_control::all_gather(std::vector<U>& data, bool control) {
  distributed_services->all_gather(data, control);
}

template <typename U>
inline void distributed_control::gather_partition(const std::vector<U>& local_contribution,
                      std::vector< std::vector<U> >& ret_partition,
                      bool control) {
  distributed_services->gather_partition(local_contribution, ret_partition, control);
}

}
#endif

ext: hpp

max_stars_repo_name: iivek/graphlab-cmu-mirror

max_issues_repo_name: iivek/graphlab-cmu-mirror

max_forks_repo_name: iivek/graphlab-cmu-mirror


text:
#include "nodeid_service.hpp"
#include "common.hpp"
#include "sessions.hpp"
#include "message.tmpl.hpp"

#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/global_fun.hpp>
#include <boost/multi_index_container.hpp>

#include <utility>
#include <string>
#include <chrono>
#include <unordered_set>

using std::pair;
using std::vector;
using std::string;
using beltpp::ip_address;
namespace chrono = std::chrono;
using steady_clock = chrono::steady_clock;
using system_clock = chrono::system_clock;
using time_point = steady_clock::time_point;

namespace publiqpp
{

class nodeid_address_header
{
public:
    beltpp::ip_address address;
    string node_address;
    time_point checked_time_point;
};
class nodeid_address_unit
{
public:
    enum class verified_type {never, past, current};

    nodeid_address_unit();
    nodeid_address_unit(nodeid_address_unit&&);
    ~nodeid_address_unit();

    nodeid_address_unit& operator = (nodeid_address_unit&&);

    nodeid_address_header header;
    beltpp::ip_address ssl_address;
    std::unique_ptr<session_action_broadcast_address_info> ptr_action;
    verified_type verified = verified_type::never;
};

class nodeid_container
{
public:
    struct by_nodeid
    {
        inline static std::string extract(nodeid_address_unit const& ob)
        {
            return ob.header.node_address;
        }
    };

    struct by_nodeid_and_address
    {
        inline static pair<string, string> extract(nodeid_address_unit const& ob)
        {
            return std::make_pair(ob.header.node_address, ob.header.address.to_string());
        }
    };

    struct by_checked_time_point
    {
        inline static time_point extract(nodeid_address_unit const& ob)
        {
            return ob.header.checked_time_point;
        }
    };

    using type = ::boost::multi_index_container<nodeid_address_unit,
    boost::multi_index::indexed_by<
        boost::multi_index::hashed_non_unique<boost::multi_index::tag<struct by_nodeid>,
            boost::multi_index::global_fun<nodeid_address_unit const&, std::string, &by_nodeid::extract>>,
        boost::multi_index::hashed_unique<boost::multi_index::tag<struct by_nodeid_and_address>,
            boost::multi_index::global_fun<nodeid_address_unit const&, pair<string, string>, &by_nodeid_and_address::extract>>,
        boost::multi_index::ordered_non_unique<boost::multi_index::tag<struct by_checked_time_point>,
            boost::multi_index::global_fun<nodeid_address_unit const&, std::chrono::steady_clock::time_point, &by_checked_time_point::extract>>
    >>;
};

nodeid_address_unit::nodeid_address_unit() = default;
nodeid_address_unit::nodeid_address_unit(nodeid_address_unit&&) = default;
nodeid_address_unit::~nodeid_address_unit() = default;
nodeid_address_unit& nodeid_address_unit::operator = (nodeid_address_unit&&) = default;

namespace detail
{
class nodeid_service_impl
{
public:
    nodeid_container::type nodeids;
};
}

nodeid_service::nodeid_service()
    : m_pimpl(new detail::nodeid_service_impl())
{}
nodeid_service::~nodeid_service() = default;

void nodeid_service::add(std::string const& node_address,
                         beltpp::ip_address const& address,
                         beltpp::ip_address const& ssl_address,
                         std::unique_ptr<session_action_broadcast_address_info>&& ptr_action)
{
    nodeid_address_unit nodeid_item;
    nodeid_item.header.address = address;
    nodeid_item.header.node_address = node_address;
    nodeid_item.verified = nodeid_address_unit::verified_type::never;
    nodeid_item.ssl_address = ssl_address;

    auto it_find = m_pimpl->nodeids.find(nodeid_item.header.node_address);
    if (it_find != m_pimpl->nodeids.end() &&
        it_find->verified == nodeid_address_unit::verified_type::current)
        return;

    auto header_address = nodeid_item.header.address;
    auto header_node_address = nodeid_item.header.node_address;

    auto insert_result = m_pimpl->nodeids.insert(std::move(nodeid_item));
    auto it_nodeid = insert_result.first;

    // there was an old code checking if the item preventing the insert has the same header ip address and node address
    // but this pair is the only unique key for this container
    // just in case, check for logic errors. can remove the checks later, entirelly
    if (it_nodeid->header.address != header_address)
        throw std::logic_error("nodeid_service::add: it_nodeid->header.address != header_address");
    if (it_nodeid->header.node_address != header_node_address)
        throw std::logic_error("nodeid_service::add: it_nodeid->header.node_address != header_node_address");

    /*
    if (insert_result.second ||
        (
            it_nodeid->header.address == nodeid_item.header.address &&
            it_nodeid->header.node_address == nodeid_item.header.node_address
        ))
    */
    {
        bool modified;
        B_UNUSED(modified);
        modified = m_pimpl->nodeids.modify(it_nodeid, [&ptr_action, &ssl_address](nodeid_address_unit& item)
        {
            item.ptr_action = std::move(ptr_action);
            item.ssl_address = ssl_address;
        });
        assert(modified);
    }
}

void nodeid_service::keep_successful(std::string const& node_address,
                                     beltpp::ip_address const& address,
                                     bool verified)
{
    auto it = m_pimpl->nodeids.find(node_address);
    if (it == m_pimpl->nodeids.end())
    {
        assert(false);
        throw std::logic_error("nodeid_service::keep_successful "
            "cannot find the expected nodeid");
    }
    else
    {
        size_t erased_count = 0, kept_count = 0;
        while (it != m_pimpl->nodeids.end() &&
               it->header.node_address == node_address)
        {
            if (it->header.address != address)
            {
                it = m_pimpl->nodeids.erase(it);
                ++erased_count;
            }
            else
            {
                bool modified;
                B_UNUSED(modified);
                modified = m_pimpl->nodeids.modify(it, [verified](nodeid_address_unit& item)
                {
                    item.verified = verified ?
                                        nodeid_address_unit::verified_type::current :
                                        nodeid_address_unit::verified_type::past;
                });
                assert(modified);
                ++it;
                ++kept_count;
            }
        }

        assert(kept_count == 1);
        if (kept_count != 1)
            throw std::logic_error("nodeid_service keep_successful()");

        if (false == verified)
        {
            assert(erased_count == 0);
            if (erased_count != 0)
                throw std::logic_error("nodeid_service keep_successful(false)");
        }

        it = m_pimpl->nodeids.find(node_address);
        if (it == m_pimpl->nodeids.end())
        {
            assert(false);
            throw std::logic_error("this is not possible");
        }
        else
        {
            //  can I do this modify in the while loop above?
            //  maybe it's possible, because this affects different index
            bool modified;
            B_UNUSED(modified);
            modified = m_pimpl->nodeids.modify(it, [](nodeid_address_unit& item)
            {
                item.header.checked_time_point = steady_clock::now();
            });
            assert(modified);
        }
    }
}

void nodeid_service::erase_failed(std::string const& node_address,
                                  beltpp::ip_address const& address)
{
    auto it = m_pimpl->nodeids.find(node_address);
    if (it == m_pimpl->nodeids.end())
    {
        assert(false);
        throw std::logic_error("nodeid_service::erase_failed "
            "cannot find the expected entry");
    }
    else
    {
        size_t erased_count = 0, kept_count = 0;
        while (it != m_pimpl->nodeids.end() &&
               it->header.node_address == node_address)
        {
            if (it->header.address == address)
            {
                it = m_pimpl->nodeids.erase(it);
                ++erased_count;
            }
            else
            {
                ++it;
                ++kept_count;
            }
        }

        B_UNUSED(kept_count);
        B_UNUSED(erased_count);
    }
}

void nodeid_service::take_actions(std::function<void (std::string const&/* node_address*/,
                                                      beltpp::ip_address const&/* address*/,
                                                      std::unique_ptr<session_action_broadcast_address_info>&&/* ptr_action*/)> const& callback)
{
    std::unordered_set<string> prev_node_addresses;

    for (auto it = m_pimpl->nodeids.begin(); it != m_pimpl->nodeids.end(); ++it)
    {
        auto const& nodeid_item = *it;
        if (nodeid_item.ptr_action &&
            0 == prev_node_addresses.count(nodeid_item.header.node_address))
        {
            prev_node_addresses.insert(nodeid_item.header.node_address);

            std::unique_ptr<session_action_broadcast_address_info> ptr_action;

            bool modified;
            B_UNUSED(modified);
            modified = m_pimpl->nodeids.modify(it, [&ptr_action](nodeid_address_unit& item)
            {
                ptr_action = std::move(item.ptr_action);
            });
            assert(modified);

            callback(nodeid_item.header.node_address,
                     nodeid_item.header.address,
                     std::move(ptr_action));
        }
    }
}

BlockchainMessage::PublicAddressesInfo nodeid_service::get_addresses() const
{
    BlockchainMessage::PublicAddressesInfo result;
    BlockchainMessage::PublicAddressesInfo result_verified;

    auto& index_by_checked_time_point = m_pimpl->nodeids.template get<nodeid_container::by_checked_time_point>();

    for (auto it = index_by_checked_time_point.rbegin();
         it != index_by_checked_time_point.rend();
         ++it)
    {
        BlockchainMessage::PublicAddressInfo address_info;
        beltpp::assign(address_info.ip_address, it->header.address);
        address_info.node_address = it->header.node_address;

        beltpp::assign(address_info.ssl_ip_address, it->ssl_address);

        if (it->verified == nodeid_address_unit::verified_type::current)
        {
            address_info.seconds_since_checked = 0;

            result_verified.addresses_info.push_back(std::move(address_info));
        }
        else
        {
            chrono::seconds seconds_since;
            if (it->verified == nodeid_address_unit::verified_type::never)
                seconds_since = chrono::duration_cast<chrono::seconds>(system_clock::now() - system_clock::time_point());
            else
                seconds_since = chrono::duration_cast<chrono::seconds>(steady_clock::now() - it->header.checked_time_point);

            address_info.seconds_since_checked = uint64_t(seconds_since.count());

            result.addresses_info.push_back(std::move(address_info));
        }
    }

    result_verified.addresses_info.insert(result_verified.addresses_info.end(),
                                          result.addresses_info.begin(),
                                          result.addresses_info.end());

    result = std::move(result_verified);

    return result;
}

}// end of namespace publiqpp

ext: cpp