FileFetcher class for async. queuing of local and remote files

Author: shahor02

Common/Utils/CMakeLists.txt

@@ -18,6 +18,7 @@ o2_add_library(CommonUtils
                        src/KeyValParam.cxx
                        src/FileSystemUtils.cxx
                        src/FIFO.cxx
+                       src/FileFetcher.cxx
                PUBLIC_LINK_LIBRARIES ROOT::Hist ROOT::Tree Boost::iostreams O2::CommonDataFormat O2::Headers
                                      FairLogger::FairLogger)
 
@@ -36,6 +37,7 @@ o2_target_root_dictionary(CommonUtils
                                   include/CommonUtils/ConfigurationMacroHelper.h
                                   include/CommonUtils/RootSerializableKeyValueStore.h
                                   include/CommonUtils/KeyValParam.h
+                                  include/CommonUtils/FileFetcher.h
         )
 
 o2_add_test(TreeStream

Common/Utils/include/CommonUtils/FileFetcher.h

@@ -0,0 +1,115 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+/// \author ruben.shahoyan@cern.ch
+/// Code mostly based on DataDistribution:SubTimeFrameFileSource by Gvozden Nescovic
+
+#ifndef ALICEO2_FILEFETCHER_H_
+#define ALICEO2_FILEFETCHER_H_
+
+#include "CommonUtils/FIFO.h"
+#include <unordered_map>
+#include <string>
+#include <thread>
+#include <Rtypes.h>
+#include <mutex>
+#include <regex>
+
+namespace o2
+{
+namespace utils
+{
+
+class FileFetcher
+{
+ public:
+  struct FileRef {
+    std::string origName{};
+    std::string localName{}; // local alias for for remote files
+    bool remote = false;
+    bool copied = false;
+
+    const auto& getLocalName() const { return remote ? localName : origName; }
+    const auto& getOrigName() const { return origName; }
+  };
+
+  /*
+  * Create file fetcher with predefined cache and async copy of remote files
+  *
+  * @param input:     comma-separated list of input data files and/or files with list of data files and/or directories
+  * @param selRegex:  regex expression to select files needed for input
+  * @param remRegex:  optional regex expression to recognize remote files
+  * @param copyCmd:   optional command to copy remote files in format "<operation> ?src ?dst"
+  * @param copyCmd:   base directory for copied remote files
+  */
+  FileFetcher(const std::string& input,
+              const std::string& selRegex = "",
+              const std::string& remRegex = "",
+              const std::string& copyCmd = "",
+              const std::string& copyDir = "/tmp");
+  ~FileFetcher();
+
+  const auto& getFileRef(size_t i) const { return mInputFiles[i]; }
+
+  void setMaxFilesInQueue(size_t s) { mMaxInQueue = s > 0 ? s : 1; }
+  void setMaxLoops(size_t v) { mMaxLoops = v; }
+  bool isRunning() const { return mRunning; }
+  void start();
+  void stop();
+  void cleanup();
+  size_t getNLoops() const { return mNLoops; }
+  size_t getNFilesProc() const { return mNFilesProc; }
+  size_t getNFilesProcOK() const { return mNFilesProcOK; }
+  size_t getMaxFilesInQueue() const { return mMaxInQueue; }
+  size_t getNRemoteFiles() const { return mNRemote; }
+  size_t getNFiles() const { return mInputFiles.size(); }
+  size_t popFromQueue(bool discard = false);
+  size_t getQueueSize() const { return mQueue.size(); }
+  std::string getNextFileInQueue() const;
+  void discardFile(const std::string& fname);
+
+ private:
+  size_t nextInQueue() const;
+  void processInput(const std::string& input);
+  void processInput(const std::vector<std::string>& input);
+  void processDirectory(const std::string& name);
+  bool addInputFile(const std::string& fname);
+  std::string createCopyName(const std::string& fname) const;
+  bool copyFile(size_t id);
+  bool isRemote(const std::string& fname) const;
+  void fetcher();
+
+ private:
+  FIFO<size_t> mQueue{};
+  std::string mCopyDirName{"/tmp"};
+  std::string mCopyCmdLogFile{};
+  std::string mCopyCmd{};
+  std::unique_ptr<std::regex> mSelRegex;
+  std::unique_ptr<std::regex> mRemRegex;
+  std::unordered_map<std::string, size_t> mCopied{};
+  std::vector<FileRef> mInputFiles{};
+  size_t mNRemote{0};
+  size_t mMaxInQueue{5};
+  bool mRunning = false;
+  size_t mMaxLoops = 0;
+  size_t mNLoops = 0;
+  size_t mNFilesProc = 0;
+  size_t mNFilesProcOK = 0;
+  mutable std::mutex mMtx;
+  std::thread mFetcherThread{};
+
+  ClassDefNV(FileFetcher, 1);
+};
+
+} // namespace utils
+} // namespace o2
+
+#endif

Common/Utils/include/CommonUtils/StringUtils.h

@@ -93,10 +93,14 @@ struct Str {
   {
     std::string ss = s;
     rtrim(ss);
-    ltrim(ss);
     return ss;
   }
 
+  static inline bool endsWith(const std::string& s, const std::string& ending)
+  {
+    return (ending.size() > s.size()) ? false : std::equal(ending.rbegin(), ending.rend(), s.rbegin());
+  }
+
   // return vector of tokens from the string with provided delimiter. If requested, trim the spaces from tokens
   static std::vector<std::string> tokenize(const std::string& src, char delim, bool trimToken = true);
 

Common/Utils/src/CommonUtilsLinkDef.h

@@ -22,6 +22,8 @@
 #pragma link C++ class o2::utils::MemFileHelper + ;
 #pragma link C++ class o2::utils::RootSerializableKeyValueStore::SerializedInfo + ;
 #pragma link C++ class o2::utils::Str + ;
+#pragma link C++ class o2::utils::FileFetcher + ;
+
 #pragma link C++ class pair < string, o2::utils::RootSerializableKeyValueStore::SerializedInfo> + ;
 #pragma link C++ class map < string, o2::utils::RootSerializableKeyValueStore::SerializedInfo> + ;
 #pragma link C++ class o2::utils::RootSerializableKeyValueStore + ;