MCOL-3438. Made it more sensible. Ownership is still confusing even to

me, the author, but this makes the view of the SM-managed files consistent
for the sm* toolkit.

storage-manager/src/IOCoordinator.cpp

@@ -628,7 +628,7 @@ int IOCoordinator::open(const char *_filename, int openmode, struct stat *out)
 
 int IOCoordinator::listDirectory(const char *dirname, vector<string> *listing)
 {
-    bf::path p(metaPath / ownership.get(dirname));
+    bf::path p(metaPath / ownership.get(dirname, false));
     ++listingCount;
     
     listing->clear();
@@ -656,7 +656,7 @@ int IOCoordinator::listDirectory(const char *dirname, vector<string> *listing)
 
 int IOCoordinator::stat(const char *_path, struct stat *out)
 {
-    bf::path filename = ownership.get(_path);
+    bf::path filename = ownership.get(_path, false);
     
     if (bf::is_directory(metaPath/filename))
         return ::stat((metaPath/filename).string().c_str(), out);

storage-manager/src/Ownership.cpp

@@ -71,14 +71,14 @@ Ownership::~Ownership()
         releaseOwnership(it.first, true);
 }
 
-bf::path Ownership::get(const bf::path &p)
+bf::path Ownership::get(const bf::path &p, bool getOwnership)
 {
     bf::path ret, prefix, normalizedPath(p);
     bf::path::const_iterator pit;
     uint i, levels;
 
     normalizedPath.normalize();
-    //cerr << "Ownership::get() param = " << p.string() << endl;
+    //cerr << "Ownership::get() param = " << normalizedPath.string() << endl;
     if (prefixDepth > 0)
     {
         for (i = 0, pit = normalizedPath.begin(); i <= prefixDepth && pit != normalizedPath.end(); ++i, ++pit)
@@ -88,13 +88,11 @@ bf::path Ownership::get(const bf::path &p)
         //cerr << "prefix is " << prefix.string() << endl;
         for (levels = 0; pit != normalizedPath.end(); ++levels, ++pit)
             ret /= *pit;
-        if (ret.empty())
-        {
-            //cerr << "returning ''" << endl;
+        if (!getOwnership)
             return ret;
-        }
-        else if (levels == 1)
-            throw runtime_error("Ownership: given path " + p.string() + " does not have minimum number of directories");
+        if (levels <= 1)
+            throw runtime_error("Ownership: given path " + normalizedPath.string() + 
+                " does not have minimum number of directories");
     }
     else
     {
@@ -102,6 +100,9 @@ bf::path Ownership::get(const bf::path &p)
         prefix = *(normalizedPath.begin());
     }
     
+    if (!getOwnership)
+        return ret;
+    
     mutex.lock();
     if (ownedPrefixes.find(prefix) == ownedPrefixes.end())
     {

storage-manager/src/Ownership.h

@@ -37,8 +37,10 @@ class Ownership : public boost::noncopyable
 
         bool sharedFS();
         // returns the path "right shifted" by prefixDepth, and with ownership of that path.
-        // on error it returns an empty path.
-        boost::filesystem::path get(const boost::filesystem::path &);
+        // on error it throws a runtime exception
+        // setting getOwnership to false will return the modified path but not also take ownership
+        // of the returned prefix.
+        boost::filesystem::path get(const boost::filesystem::path &, bool getOwnership=true);
         
         
     private:

storage-manager/src/smcat.cpp

@@ -50,7 +50,7 @@ bool SMOnline()
     return false;
 }
 
-void catFileOffline(const char *filename)
+void catFileOffline(const char *filename, int prefixlen)
 {
     uint8_t data[8192];
     off_t offset = 0;
@@ -63,7 +63,7 @@ void catFileOffline(const char *filename)
         if (read_err < 0)
         {
             int l_errno = errno;
-            cerr << "Error reading " << filename << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+            cerr << "Error reading " << &filename[prefixlen] << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
         }
 
         while (count < read_err)
@@ -81,7 +81,7 @@ void catFileOffline(const char *filename)
     } while (read_err > 0);
 }
 
-void catFileOnline(const char *filename)
+void catFileOnline(const char *filename, int prefixlen)
 {
     uint8_t data[8192];
     off_t offset = 0;
@@ -94,7 +94,7 @@ void catFileOnline(const char *filename)
         if (read_err < 0)
         {
             int l_errno = errno;
-            cerr << "Error reading " << filename << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+            cerr << "Error reading " << &filename[prefixlen] << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
         }
 
         while (count < read_err)
@@ -112,6 +112,28 @@ void catFileOnline(const char *filename)
     } while (read_err > 0);
 }
 
+int makePathPrefix(char *target, int targetlen)
+{
+    // MCOL-3438 -> add bogus directories to the front of each param
+    Config *config = Config::get();
+    int prefixDepth = stoi(config->getValue("ObjectStorage", "common_prefix_depth"));
+    target[0] = '/';
+    target[1] = 0;
+    int bufpos = 1;
+    
+    for (int i = 0; i < prefixDepth; i++)
+    {
+        if (bufpos + 3 >= targetlen)
+        {
+            cerr << "invalid prefix depth in ObjectStorage/common_prefix_depth";
+            exit(1);
+        }
+        memcpy(&target[bufpos], "x/\0", 3);
+        bufpos += 2;
+    }
+    return bufpos;
+}
+
 int main(int argc, char **argv)
 {
     if (argc < 2)
@@ -119,13 +141,18 @@ int main(int argc, char **argv)
         usage(argv[0]);
         return 1;
     }
+    
+    char prefix[8192];
+    int prefixlen = makePathPrefix(prefix, 8192);
 
     for (int i = 1; i < argc; i++)
     {
+        strncat(&prefix[prefixlen], argv[i], 8192 - prefixlen);
+    
         if (SMOnline())
-            catFileOnline(argv[i]);
+            catFileOnline(prefix, prefixlen);
         else
-            catFileOffline(argv[i]);
+            catFileOffline(prefix, prefixlen);
     }
     
     return 0;

storage-manager/src/smls.cpp

@@ -126,6 +126,28 @@ void lsOnline(const char *path)
     }
 }
 
+int makePathPrefix(char *target, int targetlen)
+{
+    // MCOL-3438 -> add bogus directories to the front of each param
+    Config *config = Config::get();
+    int prefixDepth = stoi(config->getValue("ObjectStorage", "common_prefix_depth"));
+    target[0] = '/';
+    target[1] = 0;
+    int bufpos = 1;
+    
+    for (int i = 0; i < prefixDepth; i++)
+    {
+        if (bufpos + 3 >= targetlen)
+        {
+            cerr << "invalid prefix depth in ObjectStorage/common_prefix_depth";
+            exit(1);
+        }
+        memcpy(&target[bufpos], "x/\0", 3);
+        bufpos += 2;
+    }
+    return bufpos;
+}
+
 int main(int argc, char **argv)
 {
     if (argc != 2)
@@ -134,13 +156,13 @@ int main(int argc, char **argv)
         return 1;
     }
     
-    // todo.  need to sanitize the input.  Ownership will remove X directories from the front
-    // of whatever is passed to it, possibly giving a nonsensical answer.
+    char prefix[8192];
+    makePathPrefix(prefix, 8192);
     
     if (SMOnline())
-        lsOnline(argv[1]);
+        lsOnline(strncat(prefix, argv[1], 8192));
     else
-        lsOffline(argv[1]);
+        lsOffline(strncat(prefix, argv[1], 8192));
     
     return 0;
 }

storage-manager/src/smput.cpp

@@ -52,7 +52,7 @@ bool SMOnline()
     return false;
 }
 
-void putOffline(const char *fname)
+void putOffline(const char *fname, int prefixlen)
 {
     uint8_t data[8192];
     int read_err, write_err;
@@ -64,7 +64,8 @@ void putOffline(const char *fname)
     if (read_err < 0)
     {
         int l_errno = errno;
-        cerr << "Failed to open/create " << fname << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+        cerr << "Failed to open/create " << &fname[prefixlen] << ": " << 
+            strerror_r(l_errno, (char *) data, 8192) << endl;
         exit(1);
     }
     
@@ -84,7 +85,8 @@ void putOffline(const char *fname)
             if (write_err < 0)
             {
                 int l_errno = errno;
-                cerr << "Error writing to " << fname << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+                cerr << "Error writing to " << &fname[prefixlen] << ": " << 
+                    strerror_r(l_errno, (char *) data, 8192) << endl;
                 exit(1);
             }
             count += write_err;
@@ -93,7 +95,7 @@ void putOffline(const char *fname)
     } while (read_err > 0);
 }
 
-void putOnline(const char *fname)
+void putOnline(const char *fname, int prefixlen)
 {
     uint8_t data[8192];
     int read_err, write_err;
@@ -106,7 +108,8 @@ void putOnline(const char *fname)
     if (!df)
     {
         int l_errno = errno;
-        cerr << "Failed to open/create " << fname << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+        cerr << "Failed to open/create " << &fname[prefixlen] << ": " << 
+            strerror_r(l_errno, (char *) data, 8192) << endl;
         exit(1);
     }
     
@@ -126,7 +129,8 @@ void putOnline(const char *fname)
             if (write_err < 0)
             {
                 int l_errno = errno;
-                cerr << "Error writing to " << fname << ": " << strerror_r(l_errno, (char *) data, 8192) << endl;
+                cerr << "Error writing to " << &fname[prefixlen] << ": " << 
+                    strerror_r(l_errno, (char *) data, 8192) << endl;
                 exit(1);
             }
             count += write_err;
@@ -134,6 +138,28 @@ void putOnline(const char *fname)
     } while (read_err > 0);
 }
 
+int makePathPrefix(char *target, int targetlen)
+{
+    // MCOL-3438 -> add bogus directories to the front of each param
+    Config *config = Config::get();
+    int prefixDepth = stoi(config->getValue("ObjectStorage", "common_prefix_depth"));
+    target[0] = '/';
+    target[1] = 0;
+    int bufpos = 1;
+    
+    for (int i = 0; i < prefixDepth; i++)
+    {
+        if (bufpos + 3 >= targetlen)
+        {
+            cerr << "invalid prefix depth in ObjectStorage/common_prefix_depth";
+            exit(1);
+        }
+        memcpy(&target[bufpos], "x/\0", 3);
+        bufpos += 2;
+    }
+    return bufpos;
+}
+
 int main(int argc, char **argv)
 {
     if (argc != 2)
@@ -142,10 +168,13 @@ int main(int argc, char **argv)
         return 1;
     }
     
+    char prefix[8192];
+    int prefixlen = makePathPrefix(prefix, 8192);
+    
     if (SMOnline())
-        putOnline(argv[1]);
+        putOnline(strncat(prefix, argv[1], 8192), prefixlen);
     else
-        putOffline(argv[1]);
+        putOffline(strncat(prefix, argv[1], 8192), prefixlen);
     return 0;
 }
     

storage-manager/src/smrm.cpp

@@ -50,19 +50,54 @@ bool SMOnline()
     return false;
 }
 
-void rmOffline(int argCount, const char **args)
+#define min(x, y) (x < y ? x : y)
+
+void rmOffline(int argCount, const char **args, const char *prefix, uint prefixlen)
 {
     boost::scoped_ptr<IOCoordinator> ioc(IOCoordinator::get());
-    for (int i = 1; i < argCount; i++)
-        ioc->unlink(args[i]);
-}
-
-void rmOnline(int argCount, const char **args)
-{
-    idbdatafile::SMFileSystem fs;
+    char buf[16384];
+    strncpy(buf, prefix, prefixlen);
     
     for (int i = 1; i < argCount; i++)
-        fs.remove(args[i]);
+    {
+        memcpy(&buf[prefixlen], args[i], min(16383 - prefixlen, strlen(args[i])) + 1);
+        ioc->unlink(buf);
+    }
+}
+
+void rmOnline(int argCount, const char **args, const char *prefix, uint prefixlen)
+{
+    idbdatafile::SMFileSystem fs;
+    char buf[16384];
+    strncpy(buf, prefix, prefixlen);
+    
+    for (int i = 1; i < argCount; i++)
+    {
+        memcpy(&buf[prefixlen], args[i], min(16383 - prefixlen, strlen(args[i])) + 1);
+        fs.remove((char *) memcpy(&buf[prefixlen], args[i], min(16383 - prefixlen, strlen(args[i])) + 1));
+    }
+}
+
+int makePathPrefix(char *target, int targetlen)
+{
+    // MCOL-3438 -> add bogus directories to the front of each param
+    Config *config = Config::get();
+    int prefixDepth = stoi(config->getValue("ObjectStorage", "common_prefix_depth"));
+    target[0] = '/';
+    target[1] = 0;
+    int bufpos = 1;
+    
+    for (int i = 0; i < prefixDepth; i++)
+    {
+        if (bufpos + 3 >= targetlen)
+        {
+            cerr << "invalid prefix depth in ObjectStorage/common_prefix_depth";
+            exit(1);
+        }
+        memcpy(&target[bufpos], "x/\0", 3);
+        bufpos += 2;
+    }
+    return bufpos;
 }
 
 int main(int argc, const char **argv)
@@ -73,9 +108,12 @@ int main(int argc, const char **argv)
         return 1;
     }
     
+    char prefix[8192];
+    uint prefixlen = makePathPrefix(prefix, 8192);
+    
     if (SMOnline())
-        rmOnline(argc, argv);
+        rmOnline(argc, argv, prefix, prefixlen);
     else
-        rmOffline(argc, argv);
+        rmOffline(argc, argv, prefix, prefixlen);
     return 0;
 }