Disk prefix in PoolDriver.

diff --git a/src/Vfs.cpp b/src/Vfs.cpp
index 0550ef3..2478b5b 100644 (file)
--- a/src/Vfs.cpp
+++ b/src/Vfs.cpp
@@ -73,6 +73,9 @@ void VfsFactory::configure(const std::string& key, const std::string& value) thr
   else if (key == "NSPrefix") {
     this->nsPrefix_ = value;
   }
+  else if (key == "DiskPrefix") {
+    this->diskPrefix_ = value;
+  }
   else
     throw DmException(DMLITE_CFGERR(DMLITE_UNKNOWN_KEY),
                       "Unrecognised option " + key);
@@ -108,7 +111,7 @@ std::string VfsFactory::implementedPool() throw ()
 PoolDriver* VfsFactory::createPoolDriver() throw (DmException)
 {
   syslog(LOG_DEBUG, "%s", __func__);
-  return new VfsPoolDriver(this->hostName_,
+  return new VfsPoolDriver(this->hostName_, this->diskPrefix_,
                            this->tokenPasswd_, this->tokenUseIp_,
                            this->tokenLife_);
 }

diff --git a/src/Vfs.h b/src/Vfs.h
index a87511a..f43d56f 100644 (file)
--- a/src/Vfs.h
+++ b/src/Vfs.h
@@ -56,6 +56,7 @@ namespace dmlite {
   protected:  
     std::string hostName_;
     std::string nsPrefix_;
+    std::string diskPrefix_;
     
     std::string tokenPasswd_;
     bool        tokenUseIp_;

diff --git a/src/VfsDriver.cpp b/src/VfsDriver.cpp
index 56bd514..d3a48fd 100644 (file)
--- a/src/VfsDriver.cpp
+++ b/src/VfsDriver.cpp
@@ -18,12 +18,13 @@ using namespace dmlite;
 
 
 VfsPoolDriver::VfsPoolDriver(const std::string& host,
+                             const std::string& prefix,
                              const std::string& passwd, bool useIp,
                              unsigned life):
     hostName_(host),
     tokenPasswd_(passwd), tokenUseIp_(useIp), tokenLife_(life)
 {
-  // Nothing
+  this->prefix_ = prefix.empty() ? "" : prefix;
 }
 
 
@@ -135,7 +136,7 @@ uint64_t VfsPoolHandler::getTotalSpace(void) throw (DmException)
 {
   struct statfs fs;
   
-  wrapCall(statfs("/", &fs));
+  wrapCall(statfs(getLocalPath("/").c_str(), &fs));
   
   return fs.f_bsize * fs.f_blocks;
 }
@@ -144,9 +145,9 @@ uint64_t VfsPoolHandler::getTotalSpace(void) throw (DmException)
 
 uint64_t VfsPoolHandler::getFreeSpace(void) throw (DmException)
 {
-struct statfs fs;
+  struct statfs fs;
   
-  wrapCall(statfs("/", &fs));
+  wrapCall(statfs(getLocalPath("/").c_str(), &fs));
   
   return fs.f_bsize * fs.f_bfree;
 }
@@ -164,7 +165,7 @@ bool VfsPoolHandler::replicaIsAvailable(const Replica& replica) throw (DmExcepti
 {
   struct stat fstat;
   
-  return stat(replica.rfn.c_str(), &fstat) == 0;
+  return stat(getLocalPath(replica.rfn).c_str(), &fstat) == 0;
 }
 
 
@@ -172,10 +173,14 @@ bool VfsPoolHandler::replicaIsAvailable(const Replica& replica) throw (DmExcepti
 Location VfsPoolHandler::whereToRead(const Replica& replica) throw (DmException)
 {
   struct stat fstat;
+  std::string path;
   
-  wrapCall(stat(replica.rfn.c_str(), &fstat));
+  if (replica.rfn.empty() || replica.rfn[0] != '/') path = "/" + replica.rfn;
+  else path = replica.rfn;
+
+  wrapCall(stat(getLocalPath(path).c_str(), &fstat), "could not find '%s'", path.c_str());
   
-  Url rloc(replica.rfn);
+  Url rloc(path);
   
   Chunk single;
   
@@ -205,10 +210,13 @@ Location VfsPoolHandler::whereToWrite(const std::string& sfn) throw (DmException
   Chunk single;
   Url wloc;
   std::vector<std::string> components;
-  std::string curdir, parentdir;
+  std::string path, curdir, parentdir;
   struct stat fstat;
 
-  components = Url::splitPath(sfn);
+  if (sfn.empty() || sfn[0] != '/') path = "/" + sfn;
+  else path = sfn;
+
+  components = Url::splitPath(path);
   components.pop_back();
   parentdir = Url::joinPath(components);
 
@@ -218,25 +226,25 @@ Location VfsPoolHandler::whereToWrite(const std::string& sfn) throw (DmException
   // Catalog checks permissions in all parent directories, so with disk is in
   // sync with namespace, the creating of subtree should not be needed.
   //
-  if (stat(parentdir.c_str(), &fstat) == -1) {
-    components = Url::splitPath(sfn);
+  if (stat(getLocalPath(parentdir).c_str(), &fstat) == -1) {
+    components = Url::splitPath(path);
     curdir = "";
     for (unsigned int i = 0; i < components.size() - 1; i++) {
       // the first component always '/', let's use it as separator
       if (i < 2) curdir += components[i];
       else curdir = curdir + "/" + components[i];
       // ignore all errors - we stupidly go from '/' toward deeper levels
-      mkdir(curdir.c_str(), 0755);
+      mkdir(getLocalPath(curdir).c_str(), 0755);
     }
 
     // detailed error from stat() only after the attempt for the subtree
-    wrapCall(stat(curdir.c_str(), &fstat), "parent directory '%s' not available", curdir.c_str());
+    wrapCall(stat(getLocalPath(curdir).c_str(), &fstat), "parent directory '%s' not available", curdir.c_str());
   }
 
   if (!S_ISDIR(fstat.st_mode))
     vfsThrow(ENOTDIR, "'%s' is not directory", curdir.c_str());
 
-  wloc = Url(sfn);
+  wloc = Url(path);
 
   single.url.domain   = this->driver_->hostName_;
   single.url.path   = "/vfs" + wloc.path;
@@ -255,3 +263,15 @@ Location VfsPoolHandler::whereToWrite(const std::string& sfn) throw (DmException
 void VfsPoolHandler::doneWriting(const Location& loc) throw (DmException)
 {
 }
+
+
+
+const std::string VfsPoolHandler::getLocalPath(const std::string &path) {
+  if (path.empty())
+    return this->driver_->prefix_;
+
+  if (path[0] != '/')
+    return this->driver_->prefix_ + "/" + path;
+
+  return this->driver_->prefix_ + path;
+}

diff --git a/src/VfsDriver.h b/src/VfsDriver.h
index 049d8f2..606d745 100644 (file)
--- a/src/VfsDriver.h
+++ b/src/VfsDriver.h
@@ -11,7 +11,7 @@ namespace dmlite {
   /// Filesystem driver.
   class VfsPoolDriver: public PoolDriver {
   public:
-    VfsPoolDriver(const std::string&, const std::string&, bool, unsigned);
+    VfsPoolDriver(const std::string&, const std::string&, const std::string&, bool, unsigned);
     ~VfsPoolDriver();
 
     std::string getImplId() const throw();
@@ -32,6 +32,7 @@ namespace dmlite {
     StackInstance*         si_;
 
     std::string hostName_;
+    std::string prefix_;
     
     std::string tokenPasswd_;
     bool        tokenUseIp_;
@@ -63,6 +64,7 @@ namespace dmlite {
     std::string    poolName_;
 
     std::string getImplId() const throw();
+    const std::string getLocalPath(const std::string &path);
   };
 
 };

diff --git a/src/VfsPool.cpp b/src/VfsPool.cpp
index 15bd9e5..6a4d3e8 100644 (file)
--- a/src/VfsPool.cpp
+++ b/src/VfsPool.cpp
@@ -121,24 +121,29 @@ void VfsPoolManager::deletePool(const Pool&) throw (DmException)
 
 Location VfsPoolManager::whereToRead(const std::string& path) throw (DmException)
 {
-  struct stat fstat;
-  
-  wrapCall(stat(path.c_str(), &fstat));
-  
-  Url rloc(path);
-  
-  Chunk single;
-  
-  single.url.domain   = this->hostName_;
-  single.url.path   = "/vfs" + rloc.path;
-  single.offset = 0;
-  single.size   = fstat.st_size;
-  
-  single.url.query["token"] = dmlite::generateToken(this->userId_, single.url.path,
-                                                    this->tokenPasswd_,
-                                                    this->tokenLife_);
-  
-  return Location(1, single);
+  PoolHandler* handler;
+  Replica replica;
+  Location location;
+
+  // TODO: Calling vfs pool code directly is just simplification (it is better
+  //       to compute path in PoolDriver instead to repeating it here).
+  //
+  // Proper way should be through replicas and pick the random available pool:
+  //
+  // std::vector<Replica> replicas = this->stack_->getCatalog()->getReplicas(path);
+  // Pool pool = this->getPool(replicas[i].getString("pool"));
+  // PoolHandler* handler = this->stack_->getPoolDriver(pool.type)->createPoolHandler(pool.name);
+  // if (handler->replicaIsAvailable(replicas[i])) ...
+  // ...
+  //
+  // We create a stub replica instead.
+  //
+  handler = this->stack_->getPoolDriver("vfs")->createPoolHandler("vfs");
+  replica.rfn = path;
+  location = handler->whereToRead(replica);
+  delete handler;
+
+  return location;
 }