From 68cdd5a33946028b59b5fe37d0b69c9854454d88 Mon Sep 17 00:00:00 2001
From: Alex Baranau <baranau.a@gmail.com>
Date: Thu, 27 Jan 2011 22:10:01 +0200
Subject: [PATCH] FLUME-320: Race condition in new NIO based TailSource

---
 .../cloudera/flume/handlers/text/TailSource.java   |  196 ++++++++++----------
 .../flume/handlers/text/TestTailSourceCursor.java  |   47 ++---
 2 files changed, 120 insertions(+), 123 deletions(-)

diff --git a/src/java/com/cloudera/flume/handlers/text/TailSource.java b/src/java/com/cloudera/flume/handlers/text/TailSource.java
index f4c5612..267a288 100644
--- a/src/java/com/cloudera/flume/handlers/text/TailSource.java
+++ b/src/java/com/cloudera/flume/handlers/text/TailSource.java
@@ -290,14 +290,6 @@ public class TailSource extends EventSource.Base {
 
       // oh! f exists and is a file
       try {
-        if (in != null) {
-          if (lastFileMod == file.lastModified()
-              && lastChannelPos == file.length()) {
-            LOG.debug("Tail '" + file + "': recheck still the same");
-            return false;
-          }
-        }
-
         // let's open the file
         raf = new RandomAccessFile(file, "r");
         lastFileMod = file.lastModified();
@@ -317,7 +309,7 @@ public class TailSource extends EventSource.Base {
       }
     }
 
-    boolean extractLines(ByteBuffer buf, long fmod) throws IOException,
+    boolean extractLines(ByteBuffer buf) throws IOException,
         InterruptedException {
       boolean madeProgress = false;
       int start = buf.position();
@@ -335,10 +327,6 @@ public class TailSource extends EventSource.Base {
           buf.mark(); // new mark.
           start = buf.position();
 
-          // this may be racy.
-          lastChannelPos = in.position();
-          lastFileMod = fmod;
-
           Event e = new EventImpl(body);
           e.set(A_TAILSRCFILE, file.getName().getBytes());
           sync.put(e);
@@ -365,98 +353,76 @@ public class TailSource extends EventSource.Base {
           return checkForUpdates();
         }
 
-        // get stats from raf and from f.
-        long flen = file.length(); // length of filename
-        long chlen = in.size(); // length of file.
-        long fmod = file.lastModified(); // ideally this has raf's last
-        // modified.
-
-        lastChannelSize = chlen;
-
-        // cases:
-        if (chlen == flen && lastChannelPos == flen) {
-          if (lastFileMod == fmod) {
-            // // 3) raf len == file len, last read == file len, lastMod same ->
-            // no change
-            LOG.debug("tail " + file + " : no change");
-            return false;
-          } else {
-            // // 4) raf len == file len, last read == file len, lastMod diff ?!
-            // ->
-            // restart file.
-            LOG.debug("tail " + file
-                + " : same file len, but new last mod time" + " -> reset");
-            resetRAF();
-            return true;
-          }
-        }
-
-        // file has changed
-        LOG.debug("tail " + file + " : file changed");
-        LOG.debug("tail " + file + " : old size, mod time " + lastChannelPos
-            + "," + lastFileMod);
-        LOG.debug("tail " + file + " : new size, " + "mod time " + flen + ","
-            + fmod);
-
-        // // 1) truncated file? -> restart file
-        // file truncated?
-        if (lastChannelPos > flen) {
-          LOG.debug("tail " + file + " : file truncated!?");
-
-          // normally we would check the inode, but since we cannot, we restart
-          // the file.
-          resetRAF();
+        long chlen = in.size();
+        boolean madeProgress = readAllFromChannel();
+
+        if (madeProgress) {
+          lastChannelSize = lastChannelPos; // this is more safe than in.size()
+          // due to possible race conditions
+          // NOTE: this is racy (but very very small chance): if file was
+          // rotated right before execution of next line with the file of the
+          // same length and this new file is never modified the logic in this
+          // method will never discover the rotation.
+          lastFileMod = file.lastModified();
+          LOG.debug("tail " + file + " : new data found");
           return true;
         }
 
-        // I make this a rendezvous because this source is being pulled
-        // copy data from current file pointer to EOF to dest.
-        boolean madeProgress = false;
-
-        int rd;
-        while ((rd = in.read(buf)) > 0) {
-          // need char encoder to find line breaks in buf.
-          lastChannelPos += (rd < 0 ? 0 : rd); // rd == -1 if at end of
-          // stream.
-
-          int lastRd = 0;
-          int loops = 0;
-          boolean progress = false;
-          do {
-
-            if (lastRd == -1 && rd == -1) {
-              return madeProgress;
-            }
-
-            buf.flip();
-
-            // extract lines
-            progress = extractLines(buf, fmod);
-            if (progress) {
-              madeProgress = true;
-            }
-
-            lastRd = rd;
-            loops++;
-          } while (progress); // / potential race
+        // this may seem racy but race conds handled properly with
+        // extra checks below
+        long fmod = file.lastModified();
+        long flen = file.length(); // length of filename
 
-          // if the amount read catches up to the size of the file, we can fall
-          // out and let another fileChannel be read. If the last buffer isn't
-          // read, then it remain in the byte buffer.
+        // If nothing can be read from channel, then cases:
+        // 1) no change -> return
+        if (flen == lastChannelSize && fmod == lastFileMod) {
+          LOG.debug("tail " + file + " : no change");
+          return false;
+        }
 
+        // 2) file rotated
+        LOG.debug("tail " + file + " : file rotated?");
+        // a) rotated with file of same length
+        if (flen == lastChannelSize && fmod != lastFileMod) {
+          // this is not trivial situation: it can
+          // be "false positive", so we want to be sure rotation with same
+          // file length really happened by additional checks.
+          // Situation is ultimately rare so doing time consuming/heavy
+          // things is OK here
+          LOG.debug("tail " + file + " : file rotated with new one with " +
+                  "same length?");
+          raf.getFD().sync(); // Alex: not sure this helps at all...
+          Thread.sleep(1000); // sanity interval: more data may be written
         }
 
-        if (rd == -1 && flen != lastChannelSize) {
-          // we've rotated with a longer file.
-          LOG.debug("tail " + file
-              + " : no progress but raflen != filelen, resetting");
-          resetRAF();
+        // b) "false positive" for file rotation due to race condition
+        // during fetching file stats: actually more data was added into the
+        // file (and hence it is visible in channel)
+        if (in.size() != chlen) {
+          LOG.debug("tail " + file + " : there's extra data to be read from " +
+                  "file, aborting file rotation handling");
           return true;
+        }
 
+        // c) again "false positive" for file rotation: file was truncated
+        if (chlen < lastChannelSize) {
+          LOG.debug("tail " + file + " : file was truncated, " +
+                  "aborting file rotation handling");
+          lastChannelSize = chlen;
+          lastChannelPos = chlen;
+          lastFileMod = file.lastModified();
+          in.position(chlen); // setting cursor to the last position of
+          // truncated file
+          return false;
         }
 
-        // LOG.debug("tail " + file + ": read " + len + " bytes");
-        LOG.debug("tail " + file + ": read " + lastChannelPos + " bytes");
+        LOG.debug("tail " + file + " : file rotated!");
+        resetRAF(); // resetting raf to catch up new file
+
+        // if file is not empty report true to start reading from it without a
+        // delay
+        return flen > 0;
+
       } catch (IOException e) {
         LOG.debug(e.getMessage(), e);
         in = null;
@@ -476,7 +442,47 @@ public class TailSource extends EventSource.Base {
       }
       return true;
     }
-  };
+
+    private boolean readAllFromChannel() throws IOException, InterruptedException {
+      // I make this a rendezvous because this source is being pulled
+      // copy data from current file pointer to EOF to dest.
+      boolean madeProgress = false;
+
+      int rd;
+      while ((rd = in.read(buf)) > 0) {
+        madeProgress = true;
+
+        // need char encoder to find line breaks in buf.
+        lastChannelPos += (rd < 0 ? 0 : rd); // rd == -1 if at end of
+        // stream.
+
+        int lastRd = 0;
+        boolean progress = false;
+        do {
+
+          if (lastRd == -1 && rd == -1) {
+            return true;
+          }
+
+          buf.flip();
+
+          // extract lines
+          extractLines(buf);
+
+          lastRd = rd;
+        } while (progress); // / potential race
+
+        // if the amount read catches up to the size of the file, we can fall
+        // out and let another fileChannel be read. If the last buffer isn't
+        // read, then it remain in the byte buffer.
+
+      }
+
+      LOG.debug("tail " + file + ": last read position " + lastChannelPos + ", madeProgress: " + madeProgress);
+
+      return madeProgress;
+    }
+  }
 
   /**
    * This is the main driver thread that runs through the file cursor list
diff --git a/src/javatest/com/cloudera/flume/handlers/text/TestTailSourceCursor.java b/src/javatest/com/cloudera/flume/handlers/text/TestTailSourceCursor.java
index e4b236d..a981ee4 100644
--- a/src/javatest/com/cloudera/flume/handlers/text/TestTailSourceCursor.java
+++ b/src/javatest/com/cloudera/flume/handlers/text/TestTailSourceCursor.java
@@ -200,6 +200,7 @@ public class TestTailSourceCursor {
     assertEquals(5, q.size()); // should be 5 in queue.
 
     assertTrue(c.tailBody()); // notice raflen!= filelen, reset
+    assertTrue(c.tailBody()); // open new file
     assertTrue(c.tailBody()); // read
     assertFalse(c.tailBody()); // no more to read
     assertEquals(11, q.size()); // should be 5 in queue.
@@ -238,12 +239,12 @@ public class TestTailSourceCursor {
     assertTrue(c.tailBody()); // finish reading the first file
     assertEquals(5, q.size()); // should be 5 in queue.
 
-    assertTrue(c.tailBody()); // notice raflen!= filelen, reset
+    assertTrue(c.tailBody()); // notice file rotation, reset
     assertTrue(c.tailBody()); // attempt to open file again.
+    assertTrue(c.tailBody()); // read 4 lines from new file
     assertFalse(c.tailBody()); // no more to read
 
-    // This should be 10, but actually results in 5.
-    assertEquals(9, q.size()); // should be 5 in queue.
+    assertEquals(9, q.size()); // should be 5 + 4 in queue.
   }
 
   /**
@@ -311,18 +312,18 @@ public class TestTailSourceCursor {
     assertTrue(c.tailBody()); // finish reading the first file
     assertEquals(5, q.size()); // should be 5 in queue.
 
-    assertFalse(c.tailBody()); // attempt to file again. (not there)
-    assertFalse(c.tailBody()); // attempt to file again. (not there)
+    assertFalse(c.tailBody()); // no more to read
+    assertFalse(c.tailBody()); // no more to read
 
     // wait a second to force a new last modified time.
     f.renameTo(f2); // move the file (should be no problem).
     Clock.sleep(1000);
     appendData(f, 5, 5);
 
-    assertTrue(c.tailBody()); // notice new mod time and reset
+    assertTrue(c.tailBody()); // notice new mod time and reset, file has data to read
     assertTrue(c.tailBody()); // open the new file
     assertTrue(c.tailBody()); // read new file
-    assertFalse(c.tailBody()); // fails this time
+    assertFalse(c.tailBody()); // no more to read
     assertEquals(10, q.size());
   }
 
@@ -407,6 +408,7 @@ public class TestTailSourceCursor {
     assertTrue(null != c.in);
 
     assertTrue(c.tailBody()); // read but since of EOL, buffer (no progress)
+    assertEquals(0, q.size()); // no events since no EOL found
     assertTrue(0 != c.lastChannelSize);
     assertTrue(null != c.in);
 
@@ -459,32 +461,21 @@ public class TestTailSourceCursor {
     raf.setLength(10);
     raf.close();
 
-    // TODO (jon): When a file gets truncated, because we don't have an
-    // inode we don't know if this is a new file or a truncated version of the
-    // original file. This is not a fundamental flaw -- safe assumption here is
-    // that it is a new file and according to our semantics, we are just
-    // duplicating information. NOTE: This is another place where this tail
-    // differs from the semantics of gnu-tail.
-
-    assertTrue(c.tailBody()); // reset changed file
-    assertEquals(0, c.lastChannelPos);
-    assertEquals(null, c.in);
-
-    assertTrue(c.tailBody()); // re-opens length changed file
-    assertTrue(0 != c.lastChannelSize);
+    assertFalse(c.tailBody()); // detect file truncation, no data to read
+    assertEquals(5, q.size()); // should be 5 in queue.
+    assertEquals(10, c.lastChannelPos);
     assertTrue(null != c.in);
 
-    assertTrue(c.tailBody()); // finish reading the file
-    assertTrue(0 != c.lastChannelSize);
-    assertTrue(null != c.in);
+    assertFalse(c.tailBody()); // no data to read
+    assertEquals(5, q.size()); // should be 5 in queue.
 
-    assertFalse(c.tailBody()); // attempt to open file again.
-    assertEquals(6, q.size()); // should be 5 in queue.
+    appendData(f, 5, 5); // appending data after truncation
 
-    c.flush();
-    assertTrue(c.tailBody()); // attempt to open file again.
-    assertEquals(7, q.size()); // should be 5 in queue.
+    assertTrue(c.tailBody()); // reading appended data
+    assertEquals(10, q.size()); // should be 5 + 5 in queue.
 
+    assertFalse(c.tailBody()); // no data to read
+    assertEquals(10, q.size()); // should be 5 + 5 in queue.
   }
 
   /**
-- 
1.7.0.4

