Author: suresh
Date: Fri Oct 30 18:33:00 2009
New Revision: 831416
URL: http://svn.apache.org/viewvc?rev=831416&view=rev
Log:
HADOOP-6313. Implement Syncable interface in FSDataOutputStream to expose flush
APIs to application users. Contributed by Hairong Kuang.
Modified:
hadoop/common/trunk/CHANGES.txt
hadoop/common/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
hadoop/common/trunk/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
hadoop/common/trunk/src/java/org/apache/hadoop/fs/Syncable.java
hadoop/common/trunk/src/test/core/org/apache/hadoop/fs/TestLocalFileSystem.java
Modified: hadoop/common/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/CHANGES.txt?rev=831416&r1=831415&r2=831416&view=diff
==============================================================================
--- hadoop/common/trunk/CHANGES.txt (original)
+++ hadoop/common/trunk/CHANGES.txt Fri Oct 30 18:33:00 2009
@@ -246,6 +246,9 @@
HADOOP-6240. Add new FileContext rename operation that posix compliant
that allows overwriting existing destination. (suresh)
+ HADOOP-6313. Implement Syncable interface in FSDataOutputStream to expose
+ flush APIs to application users. (Hairong Kuang via suresh)
+
IMPROVEMENTS
HADOOP-4565. Added CombineFileInputFormat to use data locality information
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java?rev=831416&r1=831415&r2=831416&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
Fri Oct 30 18:33:00 2009
@@ -91,10 +91,29 @@
return wrappedStream;
}
- /** {...@inheritdoc} */
+ @Override // Syncable
+ @Deprecated
public void sync() throws IOException {
if (wrappedStream instanceof Syncable) {
((Syncable)wrappedStream).sync();
}
}
+
+ @Override // Syncable
+ public void hflush() throws IOException {
+ if (wrappedStream instanceof Syncable) {
+ ((Syncable)wrappedStream).hflush();
+ } else {
+ wrappedStream.flush();
+ }
+ }
+
+ @Override // Syncable
+ public void hsync() throws IOException {
+ if (wrappedStream instanceof Syncable) {
+ ((Syncable)wrappedStream).hsync();
+ } else {
+ wrappedStream.flush();
+ }
+ }
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java?rev=831416&r1=831415&r2=831416&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/fs/RawLocalFileSystem.java
Fri Oct 30 18:33:00 2009
@@ -180,7 +180,7 @@
/*********************************************************
* For create()'s FSOutputStream.
*********************************************************/
- class LocalFSFileOutputStream extends OutputStream implements Syncable {
+ class LocalFSFileOutputStream extends OutputStream {
private FileOutputStream fos;
private LocalFSFileOutputStream(Path f, boolean append) throws IOException
{
@@ -207,13 +207,8 @@
throw new FSError(e); // assume native fs error
}
}
-
- /** {...@inheritdoc} */
- public void sync() throws IOException {
- fos.getFD().sync();
- }
}
-
+
/** {...@inheritdoc} */
public FSDataOutputStream append(Path f, int bufferSize,
Progressable progress) throws IOException {
Modified: hadoop/common/trunk/src/java/org/apache/hadoop/fs/Syncable.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/fs/Syncable.java?rev=831416&r1=831415&r2=831416&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/fs/Syncable.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/fs/Syncable.java Fri Oct 30
18:33:00 2009
@@ -20,11 +20,23 @@
import java.io.IOException;
-/** This interface declare the sync() operation. */
+/** This interface for flush/sync operation. */
public interface Syncable {
/**
- * Synchronize all buffer with the underlying devices.
- * @throws IOException
+ * @deprecated As of HADOOP 0.21.0, replaced by hflush
+ * @see #hflush()
*/
- public void sync() throws IOException;
+ @Deprecated public void sync() throws IOException;
+
+ /** Flush out the data in client's user buffer. After the return of
+ * this call, new readers will see the data.
+ * @throws IOException if any error occurs
+ */
+ public void hflush() throws IOException;
+
+ /** Similar to posix fsync, flush out the data in client's user buffer
+ * all the way to the disk device (but the disk may have it in its cache).
+ * @throws IOException if error occurs
+ */
+ public void hsync() throws IOException;
}
Modified:
hadoop/common/trunk/src/test/core/org/apache/hadoop/fs/TestLocalFileSystem.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/test/core/org/apache/hadoop/fs/TestLocalFileSystem.java?rev=831416&r1=831415&r2=831416&view=diff
==============================================================================
---
hadoop/common/trunk/src/test/core/org/apache/hadoop/fs/TestLocalFileSystem.java
(original)
+++
hadoop/common/trunk/src/test/core/org/apache/hadoop/fs/TestLocalFileSystem.java
Fri Oct 30 18:33:00 2009
@@ -111,6 +111,43 @@
}
}
+ /**
+ * test Syncable interface on raw local file system
+ * @throws IOException
+ */
+ public void testSyncable() throws IOException {
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.getLocal(conf).getRawFileSystem();
+ Path file = new Path(TEST_ROOT_DIR, "syncable");
+ FSDataOutputStream out = fs.create(file);;
+ final int bytesWritten = 1;
+ byte[] expectedBuf = new byte[] {'0', '1', '2', '3'};
+ try {
+ out.write(expectedBuf, 0, 1);
+ out.hflush();
+ verifyFile(fs, file, bytesWritten, expectedBuf);
+ out.write(expectedBuf, bytesWritten, expectedBuf.length-bytesWritten);
+ out.hsync();
+ verifyFile(fs, file, expectedBuf.length, expectedBuf);
+ } finally {
+ out.close();
+ }
+ }
+
+ private void verifyFile(FileSystem fs, Path file, int bytesToVerify,
+ byte[] expectedBytes) throws IOException {
+ FSDataInputStream in = fs.open(file);
+ try {
+ byte[] readBuf = new byte[bytesToVerify];
+ in.readFully(readBuf, 0, bytesToVerify);
+ for (int i=0; i<bytesToVerify; i++) {
+ assertEquals(expectedBytes[i], readBuf[i]);
+ }
+ } finally {
+ in.close();
+ }
+ }
+
public void testCopy() throws IOException {
Configuration conf = new Configuration();
LocalFileSystem fs = FileSystem.getLocal(conf);
