XrdApps::JCache: major extension to allow disconnected file serving -…

… enable via config async=1|true or XRD_JCACHE_ASYNC=1|true or via CGI ?xrd.jcache.async=1

- with async operation there is no source check done about the filesize and the assumption is, that the filesize and mtime is correct
XrdApps::JCache: add new counter showing the time 'lost' in waiting for Open

src/XrdApps.cmake

@@ -225,6 +225,10 @@ add_library(
   XrdApps/XrdClJCachePlugin/file/XrdClJCacheFile.cc
   XrdApps/XrdClJCachePlugin/vector/XrdClVectorCache.cc
   XrdApps/XrdClJCachePlugin/vector/XrdClVectorCache.hh
+  XrdApps/XrdClJCachePlugin/handler/XrdClJCacheOpenHandler.cc
+  XrdApps/XrdClJCachePlugin/handler/XrdClJCacheOpenHandler.hh
+  XrdApps/XrdClJCachePlugin/handler/XrdClJCacheReadHandler.hh
+  XrdApps/XrdClJCachePlugin/handler/XrdClJCacheReadVHandler.hh
   XrdApps/XrdClJCachePlugin/cache/Journal.cc
   XrdApps/XrdClJCachePlugin/cache/Journal.hh
   XrdApps/XrdClJCachePlugin/cache/IntervalTree.hh

src/XrdApps/XrdClJCachePlugin/cache/Journal.cc

@@ -89,18 +89,22 @@ void Journal::read_jheader() {
     return;
   }
   // TODO: understand why the mtime is +-1s
-  if ((abs(fheader.mtime - jheader.mtime) > 1) ||
-      (fheader.mtime_nsec != jheader.mtime_nsec) ||
-      (fheader.filesize != jheader.filesize)) {
-    if (fheader.mtime) {
+  if (jheader.mtime) {
+    if ((abs(fheader.mtime - jheader.mtime) > 1) ||
+        (fheader.mtime_nsec != jheader.mtime_nsec) ||
+        (jheader.filesize && (fheader.filesize != jheader.filesize))) {
       std::cerr << "warning: remote file change detected - purging path:"
                 << path << std::endl;
       std::cerr << fheader.mtime << ":" << jheader.mtime << " "
                 << fheader.mtime_nsec << ":" << jheader.mtime_nsec << " "
                 << fheader.filesize << ":" << jheader.filesize << std::endl;
+      reset();
+      return;
     }
-    reset();
-    return;
+  } else {
+    // we assume the contents referenced in the header is ok to allow disconnected ops
+    jheader.mtime = fheader.mtime;
+    jheader.filesize = fheader.filesize;
   }
 }
 
@@ -155,13 +159,27 @@ int Journal::read_journal() {
 //! Journal attach
 //------------------------------------------------------------------------------
 int Journal::attach(const std::string &lpath, uint64_t mtime,
-                    uint64_t mtime_nsec, uint64_t size) {
+                    uint64_t mtime_nsec, uint64_t size, bool ifexists) {
   std::lock_guard<std::mutex> guard(mtx);
   path = lpath;
-  jheader.mtime = mtime;
-  jheader.mtime_nsec = mtime_nsec;
-  jheader.filesize = size;
 
+  if (!ifexists) {
+    jheader.mtime = mtime;
+    jheader.mtime_nsec = mtime_nsec;
+    jheader.filesize = size;
+  }
+
+  if (ifexists) {
+    struct stat buf;
+    // check if there is already a journal for this file known
+    if (::stat(path.c_str(), &buf)) {
+      return -ENOENT;
+    } else {
+      if ((size_t)buf.st_size < sizeof(jheader_t)) {
+        return -EINVAL;
+      }
+    }
+  }
   if ((fd == -1)) {
     // need to open the file
     size_t tries = 0;

src/XrdApps/XrdClJCachePlugin/cache/Journal.hh

@@ -81,7 +81,7 @@ public:
 
   // base class interface
   int attach(const std::string &path, uint64_t mtime, uint64_t mtime_nsec,
-             uint64_t size);
+             uint64_t size, bool ifexists=false);
   int detach();
   int unlink();
 
@@ -100,6 +100,16 @@ public:
 
   std::string dump();
 
+  off_t getHeaderFileSize() {
+    std::lock_guard<std::mutex> guard(mtx);
+    return jheader.filesize;
+  }
+
+   off_t getHeaderMtime() {
+    std::lock_guard<std::mutex> guard(mtx);
+    return jheader.mtime;
+  }
+
 private:
   void process_intersection(interval_tree<uint64_t, const void *> &write,
                             interval_tree<uint64_t, uint64_t>::iterator acr,

src/XrdApps/XrdClJCachePlugin/file/CacheStats.hh

@@ -47,7 +47,7 @@ namespace JCache {
 struct CacheStats {
   CacheStats(bool doe = false)
       : bytesRead(0), bytesReadV(0), bytesCached(0), bytesCachedV(0),
-        readOps(0), readVOps(0), readVreadOps(0), nreadfiles(0),
+        readOps(0), readVOps(0), readVreadOps(0), nreadfiles(0), totaldatasize(0),
         dumponexit(doe), peakrate(0) {
     // Get the current real time
     struct timeval now;
@@ -219,6 +219,7 @@ struct CacheStats {
     outFile << "  \"readVreadOps\": " << readVreadOps.load() << ",\n";
     outFile << "  \"nreadfiles\": " << nreadfiles.load() << ",\n";
     outFile << "  \"totaldatasize\": " << totaldatasize.load() << ",\n";
+    outFile << "  \"opentime_ms\": " << opentime.load() << ",\n";
 
     std::lock_guard<std::mutex> lock(urlMutex);
     outFile << "  \"urls\": [";
@@ -315,6 +316,8 @@ struct CacheStats {
         << std::endl;
     oss << "# JCache : open unique f. read      : " << sStats.UniqueUrls()
         << std::endl;
+    oss << "# JCache : time to open files (s)   : " << std::setprecision(3) << sStats.opentime.load()
+        << std::endl;
     oss << "# "
            "-------------------------------------------------------------------"
            "---- #"
@@ -361,6 +364,8 @@ struct CacheStats {
   std::atomic<uint64_t> readVreadOps;
   std::atomic<uint64_t> nreadfiles;
   std::atomic<uint64_t> totaldatasize;
+  std::atomic<double>   opentime;
+
   std::atomic<bool> dumponexit;
   std::set<std::string> urls;
   std::mutex urlMutex;