/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */



import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.concurrent.Callable;

//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.TachyonURI;
import tachyon.Version;
import tachyon.client.OutStream;
import tachyon.client.TachyonByteBuffer;
import tachyon.client.TachyonFile;
import tachyon.client.TachyonFS;
import tachyon.client.WriteType;
import tachyon.conf.TachyonConf;

public class SingleFileWriter {
  //private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  private final TachyonURI mMasterLocation;
  private final TachyonURI mFilePath;
  private final WriteType mWriteType;
  private final TachyonFS mTachyonClient; 
  private final int mNumbers = 20;

  public SingleFileWriter(TachyonURI masterLocation, TachyonURI filePath, WriteType writeType) 
  throws Exception {
    mMasterLocation = masterLocation;
    mFilePath = filePath;
    mWriteType = writeType;
    mTachyonClient = TachyonFS.get(mMasterLocation, new TachyonConf());
  }

  
  private void createFile() throws IOException {
    int fileId = mTachyonClient.createFile(mFilePath);
    System.out.println("Creating file " + mFilePath + " id =" + fileId );
  }

  private void writeFile() throws IOException {
    System.out.println("Start writing " + mFilePath);
    ByteBuffer buf = ByteBuffer.allocate(mNumbers * 4);
    buf.order(ByteOrder.nativeOrder());
    for (int k = 0; k < mNumbers; k ++) {
      buf.putInt(k);
    }

    buf.flip();

    TachyonFile file = mTachyonClient.getFile(mFilePath);
    OutStream os = file.getOutStream(mWriteType);
    os.write(buf.array());
    os.close();
    System.out.println("Finish writing " + mFilePath);
  }

  private void deleteFile() throws IOException {
    mTachyonClient.delete(mFilePath, false);
    System.out.println("Deleting file " + mFilePath);
  }

  public static void main(String[] args) throws IllegalArgumentException {
    if (args.length != 3) {
      System.out.println("java -cp pdct-test-1.0.jar  " 
         + "SingleFileWriter <TachyonMasterAddress> <FilePath> <WriteType>");
      System.exit(-1);
    }

   SingleFileWriter sfw;
   try {
    sfw = new SingleFileWriter(new TachyonURI(args[0]), new TachyonURI(args[1]), 
	WriteType.valueOf(args[2]));
    sfw.createFile();
    sfw.writeFile();
    //sfw.deleteFile();
   } catch (Exception e) {
     System.err.println("Exception: " + e.getMessage());
   }
   return;
  }
}

@@ -1565,20 +1565,61 @@
 nt m
-Numbers = 20
+BufferSize= 64*1024;%0A  private final double mFileSize
 ;%0A%0A 
@@ -1710,16 +1710,30 @@
 riteType
+, double fsize
 ) %0A  thr
@@ -1916,16 +1916,35 @@
 ));%0A
+    mFileSize= fsize;%0A
   %7D%0A%0A
-  %0A
   pr
@@ -2220,24 +2220,48 @@
 mFilePath);%0A
+  // prepare input data%0A
     ByteBuff
@@ -2294,19 +2294,18 @@
 te(m
-Numbers * 4
+BufferSize
 );%0A 
@@ -2372,15 +2372,19 @@
  %3C m
-Numbers
+BufferSize 
 ; k 
@@ -2434,16 +2434,58 @@
 lip();%0A%0A
+  // we just open one file for the moment%0A
     Tach
@@ -2580,24 +2580,121 @@
 WriteType);%0A
+%0A    double numIters = mFileSize / (double)mBufferSize; %0A%0A    for (int i=0; i%3CnumIters; i++) %7B%0A  
     os.write
@@ -2704,24 +2704,49 @@
 f.array());%0A
+      buf.flip();%0A    %7D%0A%0A
     os.close
@@ -3196,16 +3196,27 @@
 iteType%3E
+ %3CFileSize%3E
 %22);%0A    
@@ -3241,16 +3241,430 @@
 %0A    %7D%0A%0A
+    double fileSize;%0A    if (args%5B3%5D.equals(%221g%22)) %7B%0A       fileSize = 1024.0*1024*1024;%0A    %7D else if (args%5B3%5D.equals(%2210g%22)) %7B%0A       fileSize = 10240.0*1024*1024;%0A    %7D else if (args%5B3%5D.equals(%22100g%22)) %7B%0A       fileSize = 102400.0*1024*1024;%0A    %7D else if (args%5B3%5D.equals(%22200g%22)) %7B%0A       fileSize = 204800.0*1024*1024;%0A    %7D else %7B%0A       throw new IllegalArgumentException(%22Invalid arg: %22 + args%5B0%5D);%0A    %7D%0A%0A
    Singl
@@ -3798,16 +3798,26 @@
 args%5B2%5D)
+, fileSize
 );%0A    s

update singleFileWriter