HADOOP-4616. Fuse-dfs can handle bad values from FileSystem.read call.

(Pete Wyckoff via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@719785 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -1200,6 +1200,9 @@ Release 0.18.3 - Unreleased
     HADOOP-4647. NamenodeFsck should close the DFSClient it has created.
     (szetszwo)
 
+    HADOOP-4616. Fuse-dfs can handle bad values from FileSystem.read call.
+    (Pete Wyckoff via dhruba)
+
 Release 0.18.2 - 2008-11-03
 
   BUG FIXES

src/c++/libhdfs/hdfs.c

@@ -657,6 +657,9 @@ tSize hdfsRead(hdfsFS fs, hdfsFile f, void* buffer, tSize length)
             (*env)->GetByteArrayRegion(env, jbRarray, 0, noReadBytes, buffer);
         }  else {
             //This is a valid case: there aren't any bytes left to read!
+          if (noReadBytes == 0 || noReadBytes < -1) {
+            fprintf(stderr, "WARN: FSDataInputStream.read returned invalid return code - libhdfs returning EOF, i.e., 0: %d\n", noReadBytes);
+          }
             noReadBytes = 0;
         }
         errno = 0;
@@ -718,6 +721,9 @@ tSize hdfsPread(hdfsFS fs, hdfsFile f, tOffset position,
             (*env)->GetByteArrayRegion(env, jbRarray, 0, noReadBytes, buffer);
         }  else {
             //This is a valid case: there aren't any bytes left to read!
+          if (noReadBytes == 0 || noReadBytes < -1) {
+            fprintf(stderr, "WARN: FSDataInputStream.read returned invalid return code - libhdfs returning EOF, i.e., 0: %d\n", noReadBytes);
+          }
             noReadBytes = 0;
         }
         errno = 0;

src/contrib/fuse-dfs/src/fuse_dfs.c

@@ -840,17 +840,26 @@ static int dfs_read(const char *path, char *buf, size_t size, off_t offset,
   assert(fh->fs != NULL);
   assert(fh->hdfsFH != NULL);
 
+  // special case this as simplifies the rest of the logic to know the caller wanted > 0 bytes
+  if (size == 0)
+    return 0;
+
+  // If size is bigger than the read buffer, then just read right into the user supplied buffer
   if (size >= dfs->rdbuffer_size) {
     int num_read;
-    int total_read = 0;
+    size_t total_read = 0;
     while (size - total_read > 0 && (num_read = hdfsPread(fh->fs, fh->hdfsFH, offset + total_read, buf + total_read, size - total_read)) > 0) {
       total_read += num_read;
     }
+    // if there was an error before satisfying the current read, this logic declares it an error
+    // and does not try to return any of the bytes read. Don't think it matters, so the code
+    // is just being conservative.
+    if (total_read < size && num_read < 0) {
+      total_read = -EIO;
+    }
     return total_read;
   }
 
-  assert(fh->bufferSize >= 0);
-
   //
   // Critical section - protect from multiple reads in different threads accessing the read buffer
   // (no returns until end)
@@ -872,7 +881,7 @@ static int dfs_read(const char *path, char *buf, size_t size, off_t offset,
       offset + size > fh->buffersStartOffset + fh->bufferSize) 
     {
       // Read into the buffer from DFS
-      size_t num_read;
+      int num_read = 0;
       size_t total_read = 0;
 
       while (dfs->rdbuffer_size  - total_read > 0 && 
@@ -880,21 +889,31 @@ static int dfs_read(const char *path, char *buf, size_t size, off_t offset,
         total_read += num_read;
       }
 
-      if (num_read < 0) {
+      // if there was an error before satisfying the current read, this logic declares it an error
+      // and does not try to return any of the bytes read. Don't think it matters, so the code
+      // is just being conservative.
+      if (total_read < size && num_read < 0) {
         // invalidate the buffer 
         fh->bufferSize = 0; 
         syslog(LOG_ERR, "Read error - pread failed for %s with return code %d %s:%d", path, (int)num_read, __FILE__, __LINE__);
         ret = -EIO;
       } else {
+        // Either EOF, all read or read beyond size, but then there was an error
         fh->bufferSize = total_read;
         fh->buffersStartOffset = offset;
 
         if (dfs->rdbuffer_size - total_read > 0) {
+          // assert(num_read == 0); this should be true since if num_read < 0 handled above.
           isEOF = 1;
         }
       }
     }
-  if (ret == 0) {
+
+  //
+  // NOTE on EOF, fh->bufferSize == 0 and ret = 0 ,so the logic for copying data into the caller's buffer is bypassed, and
+  //  the code returns 0 as required
+  //
+  if (ret == 0 && fh->bufferSize > 0) {
 
     assert(offset >= fh->buffersStartOffset);
     assert(fh->buf);
@@ -911,8 +930,6 @@ static int dfs_read(const char *path, char *buf, size_t size, off_t offset,
     
     memcpy(buf, offsetPtr, amount);
 
-    // fuse requires the below and the code should guarantee this assertion
-    assert(amount == size || isEOF);
     ret = amount;
   }
 
@@ -921,6 +938,13 @@ static int dfs_read(const char *path, char *buf, size_t size, off_t offset,
   //
   pthread_mutex_unlock(&fh->mutex);
 
+  // fuse requires the below and the code should guarantee this assertion
+  // 3 cases on return:
+  //   1. entire read satisfied
+  //   2. partial read and isEOF - including 0 size read
+  //   3. error 
+  assert(ret == size || isEOF || ret < 0);
+
   return ret;
 }
 
@@ -1469,8 +1493,9 @@ int dfs_release (const char *path, struct fuse_file_info *fi) {
 
     if (fh->buf != NULL) {
       free(fh->buf);
-      pthread_mutex_destroy(&fh->mutex);
     }
+    // this is always created and initialized, so always destroy it. (see dfs_open)
+      pthread_mutex_destroy(&fh->mutex);
 
     free(fh);
 

src/contrib/fuse-dfs/src/test/TestFuseDFS.java

@@ -58,7 +58,7 @@ public class TestFuseDFS extends TestCase {
     System.err.println("LD_LIBRARY_PATH=" + lp);
     String cmd[] =  {  fuse_cmd, "dfs://" + dfs.getHost() + ":" + String.valueOf(dfs.getPort()), 
                        mountpoint, "-obig_writes", "-odebug", "-oentry_timeout=1",  "-oattribute_timeout=1", "-ousetrash", "rw", "-oinitchecks",
-                       "-ordbuffer=5000"};
+                       "-ordbuffer=32768"};
     final String [] envp = {
       "CLASSPATH="+  cp,
       "LD_LIBRARY_PATH=" + lp,
@@ -432,10 +432,15 @@ public class TestFuseDFS extends TestCase {
       FileStatus foo = fileSys.getFileStatus(myPath);
       assertTrue(foo.getLen() >= 9 * 1024);
 
+
       {
         // cat the file
         DataInputStream is = new DataInputStream(new FileInputStream(mpoint + "/test/hello.reads"));
         byte buf [] = new byte[4096];
+        // test reading 0 length
+        assertTrue(is.read(buf, 0, 0) == 0);
+
+        // test real reads
         assertTrue(is.read(buf, 0, 1024) == 1024);
         assertTrue(is.read(buf, 0, 4096) == 4096);
         assertTrue(is.read(buf, 0, 4096) == 4096);
@@ -458,6 +463,21 @@ public class TestFuseDFS extends TestCase {
           assertTrue(read >= 9 * 1024);
         }
       }
+
+      // check reading an empty file for EOF
+      {
+        // create the file
+        myPath = new Path("/test/hello.reads2");
+        s = fileSys.create(myPath);
+        s.close();
+
+        FSDataInputStream fs = fileSys.open(myPath);
+        assertEquals(-1,  fs.read());
+
+        FileInputStream f = new FileInputStream(mpoint + "/test/hello.reads2");
+        assertEquals(-1, f.read());
+      }
+
     } catch(Exception e) {
       e.printStackTrace();
     } finally {
@@ -469,7 +489,6 @@ public class TestFuseDFS extends TestCase {
    * Use filesys to create the hello world! file and then cat it and see its contents are correct.
    */
   public void testCat() throws IOException,InterruptedException  {
-    if(true) return;
     try {
       // First create a new directory with mkdirs
       Runtime r = Runtime.getRuntime();