Merge pull request #26 from gluster/buffering_writes

Configurable buffered writes in memory.

Author: jayunit100

conf/core-site.xml

@@ -35,4 +35,8 @@
     <value>Off</value>
   </property>
 
+  <property>
+  <name>fs.glusterfs.write.buffer.size</name>
+  <value>1024</value>
+  </property>
 </configuration>

src/main/java/org/apache/hadoop/fs/glusterfs/GlusterFUSEOutputStream.java

@@ -21,22 +21,29 @@
 
 import java.io.*;
 
-import org.apache.hadoop.fs.FSOutputSummer;
-import org.apache.hadoop.fs.FileSystem;
-
 public class GlusterFUSEOutputStream extends OutputStream{
     File f;
     long pos;
     boolean closed;
     OutputStream fuseOutputStream;
-
+    org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GlusterFUSEOutputStream.class);
+  
     public GlusterFUSEOutputStream(String file, boolean append) throws IOException{
+        this(file,append,0);
+    }
+
+    /**
+     * @param bufferSize : Size of buffer in bytes (if 0, then no buffer will be used).
+     */
+    public GlusterFUSEOutputStream(String file, boolean append, int bufferSize) throws IOException{
         this.f=new File(file); /* not needed ? */
         this.pos=0;
-        this.fuseOutputStream=new FileOutputStream(file, append);
+        fuseOutputStream=new FileOutputStream(file, append) ;
+        if(bufferSize > 0)
+        	fuseOutputStream = new BufferedOutputStream(fuseOutputStream, bufferSize);
         this.closed=false;
     }
-
+    
     public long getPos() throws IOException{
         return pos;
     }

src/main/java/org/apache/hadoop/fs/glusterfs/GlusterFileSystem.java

@@ -23,26 +23,26 @@
  */
 package org.apache.hadoop.fs.glusterfs;
 
-import java.io.*;
-import java.net.*;
-
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.URI;
 import java.util.StringTokenizer;
 import java.util.TreeMap;
-import java.util.regex.*;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,6 +65,7 @@ public class GlusterFileSystem extends FileSystem{
     private Path workingDir=null;
     private String glusterMount=null;
     private boolean mounted=false;
+    private int writeBufferSize = 0;
 
 
     /* for quick IO */
@@ -111,6 +112,7 @@ public void initialize(URI uri,Configuration conf) throws IOException{
         String remoteGFSServer=null;
         String needQuickRead=null;
         boolean autoMount=true;
+        
 
         if(this.mounted)
             return;
@@ -123,7 +125,8 @@ public void initialize(URI uri,Configuration conf) throws IOException{
             remoteGFSServer=conf.get("fs.glusterfs.server", null);
             needQuickRead=conf.get("quick.slave.io", null);
             autoMount=conf.getBoolean("fs.glusterfs.automount", true);
-
+            writeBufferSize = conf.getInt("fs.glusterfs.write.buffer.size", 0);
+            LOG.info("Gluster Output Buffering size configured to " + writeBufferSize + " bytes.");
             /*
              * bail out if we do not have enough information to do a FUSE mount
              */
@@ -421,7 +424,7 @@ public FSDataOutputStream create(Path path,FsPermission permission,boolean overw
         parent=path.getParent();
         mkdirs(parent);
 
-        glusterFileStream=new FSDataOutputStream(new GlusterFUSEOutputStream(f.getPath(), false));
+        glusterFileStream=new FSDataOutputStream(new GlusterFUSEOutputStream(f.getPath(), false, writeBufferSize));
 
         return glusterFileStream;
     }

src/test/java/org/gluster/test/TestGlusterFuseOutputStream.java

@@ -0,0 +1,36 @@
+package org.gluster.test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.TreeMap;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.fs.glusterfs.GlusterFSBrickClass;
+import org.apache.hadoop.fs.glusterfs.GlusterFUSEInputStream;
+import org.apache.hadoop.fs.glusterfs.GlusterFUSEOutputStream;
+import org.junit.Test;
+
+public class TestGlusterFuseOutputStream {
+
+	@Test
+    public void testOutputStream() throws IOException{
+      File out = new File("/tmp/testGlusterFuseOutputStream"+System.currentTimeMillis());
+      out.createNewFile();
+      //Create a 3 byte FuseOutputStream
+      final GlusterFUSEOutputStream stream = new GlusterFUSEOutputStream(out.getAbsolutePath(), true, 3);
+      
+      stream.write("ab".getBytes());
+      System.out.println(out.length());
+
+      Assert.assertEquals(0, out.length());
+      
+      stream.flush();
+      stream.close();
+      
+      System.out.println(out.length());
+      //Confirm that the buffer held 100 bytes.
+      Assert.assertEquals(2, out.length());
+      out.delete();
+    }
+}