fs/lustre: recognize soft-links on file open

the llapi_file_get_stripe() function does not accept as an input file a
soft-link. Check therefore whether the filename provided in File_open
is a soft-link, retrieve the real file name in case it is, make sure
it is also on a Lustre file system, and use the real filename in the
llapi_file_get_stripe() function call instead.

Fixes issue #12141

Signed-off-by: Edgar Gabriel <Edgar.Gabriel@amd.com>

Author: edgargabriel

ompi/mca/fs/base/base.h

@@ -37,6 +37,22 @@
 #include "ompi/mca/fs/fs.h"
 
 
+#ifdef HAVE_SYS_STATFS_H
+#include <sys/statfs.h> /* or <sys/vfs.h> */
+#endif
+#ifdef HAVE_SYS_PARAM_H
+#include <sys/param.h>
+#endif
+#ifdef HAVE_SYS_MOUNT_H
+#include <sys/mount.h>
+#endif
+#ifdef HAVE_SYS_STAT_H
+#include <sys/stat.h>
+#endif
+#ifdef HAVE_UNISTD_H
+#include <unistd.h>
+#endif
+
 BEGIN_C_DECLS
 
 OMPI_DECLSPEC int mca_fs_base_file_select(struct ompio_file_t *file,
@@ -62,6 +78,42 @@ OMPI_DECLSPEC int mca_fs_base_file_get_size (ompio_file_t *fh, OMPI_MPI_OFFSET_T
 OMPI_DECLSPEC int mca_fs_base_file_set_size (ompio_file_t *fh, OMPI_MPI_OFFSET_TYPE size);
 OMPI_DECLSPEC int mca_fs_base_file_close (ompio_file_t *fh);
 
+
+static inline bool mca_fs_base_is_link (const char *filename)
+{
+    int err;
+    bool ret = true;
+    struct stat statbuf;
+
+    err = lstat(filename, &statbuf);
+
+    if (err || (!S_ISLNK(statbuf.st_mode))) {
+        ret = false;
+    }
+
+    return ret;
+}
+
+static inline void mca_fs_base_get_real_filename (const char *filename, char **rfilename)
+{
+    int namelen;
+    char linkbuf[PATH_MAX+1];
+
+    namelen = readlink(filename, linkbuf, PATH_MAX);
+    if (namelen == -1) {
+        /* something strange has happened between the time that
+         * we determined that this was a link and the time that
+         * we attempted to read it; punt and use the old name.
+         */
+        *rfilename = strdup(filename);
+    }
+    else {
+        /* successfully read the link */
+        linkbuf[namelen] = '\0'; /* readlink doesn't null terminate */
+        *rfilename = strdup(linkbuf);
+    }
+}
+
 /*
  * Globals
  */

ompi/mca/fs/base/fs_base_get_parent_dir.c

@@ -35,36 +35,17 @@
 #include "ompi/mca/fs/base/base.h"
 #include "ompi/mca/common/ompio/common_ompio.h"
 
-#ifdef HAVE_SYS_STATFS_H
-#include <sys/statfs.h> /* or <sys/vfs.h> */
-#endif
-#ifdef HAVE_SYS_PARAM_H
-#include <sys/param.h>
-#endif
-#ifdef HAVE_SYS_MOUNT_H
-#include <sys/mount.h>
-#endif
-#ifdef HAVE_SYS_STAT_H
-#include <sys/stat.h>
-#endif
-#ifdef HAVE_UNISTD_H
-#include <unistd.h>
-#endif
 
 void mca_fs_base_get_parent_dir ( char *filename, char **dirnamep)
 {
-    int err;
     char *dir = NULL, *slash;
-    struct stat statbuf;
 
     if (strlen(filename) < 1) {
         opal_asprintf(dirnamep, ".%s", OPAL_PATH_SEP);
         return;
     }
 
-    err = lstat(filename, &statbuf);
-
-    if (err || (!S_ISLNK(statbuf.st_mode))) {
+    if (!mca_fs_base_is_link(filename)) {
 	/* no such file, or file is not a link; these are the "normal"
 	 * cases where we can just return the parent directory.
 	 */
@@ -76,22 +57,7 @@ void mca_fs_base_get_parent_dir ( char *filename, char **dirnamep)
 	 * but this code doesn't care if the target is really there
 	 * or not.
 	 */
-	int namelen;
-	char linkbuf[PATH_MAX+1];
-
-	namelen = readlink(filename, linkbuf, PATH_MAX);
-	if (namelen == -1) {
-	    /* something strange has happened between the time that
-	     * we determined that this was a link and the time that
-	     * we attempted to read it; punt and use the old name.
-	     */
-	    dir = strdup(filename);
-	}
-	else {
-	    /* successfully read the link */
-	    linkbuf[namelen] = '\0'; /* readlink doesn't null terminate */
-	    dir = strdup(linkbuf);
-	}
+        mca_fs_base_get_real_filename(filename, &dir);
     }
 
     slash = strrchr(dir, '/');

ompi/mca/fs/lustre/fs_lustre_file_open.c

@@ -70,7 +70,7 @@ mca_fs_lustre_file_open (struct ompi_communicator_t *comm,
     int fs_lustre_stripe_size = -1;
     int fs_lustre_stripe_width = -1;
     opal_cstring_t *stripe_str;
-
+    char *rfilename = (char *)filename;
     struct lov_user_md *lump=NULL;
 
     perm = mca_fs_base_get_file_perm(fh);
@@ -112,13 +112,25 @@ mca_fs_lustre_file_open (struct ompi_communicator_t *comm,
         fs_lustre_stripe_width = mca_fs_lustre_stripe_width;
     }
 
+    /* Check for soft links and replace filename by the actual
+       file used in case it is a soft link */
+    if (mca_fs_base_is_link(filename)) {
+        mca_fs_base_get_real_filename(filename, &rfilename);
+        /* make sure the real file is also on a Lustre file system */
+        if (LUSTRE != mca_fs_base_get_fstype(rfilename)) {
+            opal_output(1, "cannot use a soft-link between a LUSTRE and non-LUSTRE file system\n");
+            return OPAL_ERROR;
+        }
+    }
 
     /* Reset errno */
     errno = 0;
     if (OMPIO_ROOT == fh->f_rank) {
         if ( (fs_lustre_stripe_size>0 || fs_lustre_stripe_width>0) &&
              ( amode&O_CREAT)                                      && 
              ( (amode&O_RDWR)|| amode&O_WRONLY) ) {
+            /* this cannot be a soft-link since we are creating the file.
+               Not using rfilename here */
             llapi_file_create(filename,
                               fs_lustre_stripe_size,
                               -1, /* MSC need to change that */
@@ -136,7 +148,7 @@ mca_fs_lustre_file_open (struct ompi_communicator_t *comm,
         }
     }
 
-   comm->c_coll->coll_bcast ( &ret, 1, MPI_INT, 0, comm, comm->c_coll->coll_bcast_module);
+    comm->c_coll->coll_bcast ( &ret, 1, MPI_INT, 0, comm, comm->c_coll->coll_bcast_module);
     if ( OMPI_SUCCESS != ret ) {
         fh->fd = -1;
         return ret;
@@ -154,7 +166,7 @@ mca_fs_lustre_file_open (struct ompi_communicator_t *comm,
         fprintf(stderr,"Cannot allocate memory for extracting stripe size\n");
         return OMPI_ERROR;
     }
-    rc = llapi_file_get_stripe(filename, lump);
+    rc = llapi_file_get_stripe(rfilename, lump);
     if (rc != 0) {
         opal_output(1, "get_stripe failed: %d (%s)\n", errno, strerror(errno));
         free(lump);