/**

  * 

  */

 package org.jscsi.target.storage;

 import static com.google.common.base.Preconditions.checkState;

 import java.io.File;

 import java.io.FileNotFoundException;

 import java.io.IOException;

 import java.security.InvalidKeyException;

 import java.security.Key;

 import java.security.NoSuchAlgorithmException;

 import java.util.Map;

 import java.util.Properties;

 import java.util.concurrent.Callable;

 import java.util.concurrent.CompletionService;

 import java.util.concurrent.ConcurrentHashMap;

 import java.util.concurrent.ExecutionException;

 import java.util.concurrent.ExecutorCompletionService;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.Future;

 import javax.crypto.Cipher;

 import javax.crypto.NoSuchPaddingException;

 import javax.crypto.spec.SecretKeySpec;

 import org.jclouds.ContextBuilder;

 import org.jclouds.blobstore.BlobStore;

 import org.jclouds.blobstore.BlobStoreContext;

 import org.jclouds.blobstore.domain.Blob;

 import org.jclouds.domain.Location;

 import org.jclouds.filesystem.reference.FilesystemConstants;

 import com.google.common.annotations.Beta;

 import com.google.common.cache.Cache;

 import com.google.common.cache.CacheBuilder;

 import com.google.common.io.ByteArrayDataOutput;

 import com.google.common.io.ByteStreams;

 /**

  * JClouds-Binding to store blocks as buckets in clouds-backends. This class utilizes caching as well as multithreaded

  * writing to improve performance.

  * 

  * @author Sebastian Graf, University of Konstanz

  * 

  */

 @Beta

 public class JCloudsStorageModule implements IStorageModule {

     // // START DEBUG CODE

     // private final static File writeFile = new

     // File("/Users/sebi/Desktop/writeaccess.txt");

     // private final static File readFile = new

     // File("/Users/sebi/Desktop/readaccess.txt");

     // private final static File uploadFile = new

     // File("/Users/sebi/Desktop/uploadaccess.txt");

     // private final static File downloadFile = new

     // File("/Users/sebi/Desktop/downloadaccess.txt");

     // static final FileWriter writer;

     // static final FileWriter reader;

     // static final FileWriter upload;

     // static final FileWriter download;

     // static {

     // try {

     // writer = new FileWriter(writeFile);

     // reader = new FileWriter(readFile);

     // upload = new FileWriter(uploadFile);

     // download = new FileWriter(downloadFile);

     // } catch (IOException e) {

     // throw new RuntimeException(e);

     // }

     // }

     /** Number of Blocks in one Cluster. */

     public static final int BLOCK_IN_CLUSTER = 512;

     private static final int BUCKETS_TO_PREFETCH = 3;

     private static final boolean ENCRYPT = false;

     private static final String ALGO = "AES";

     private static byte[] keyValue = new byte[] { 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k', 'k' };

     private static final Key KEY = new SecretKeySpec(keyValue, "AES");

     /** Number of Bytes in Bucket. */

     public final static int SIZE_PER_BUCKET = BLOCK_IN_CLUSTER * VIRTUAL_BLOCK_SIZE;

     public final static String CONTAINERNAME = "bench53473ResourcegraveISCSI9284";

     private final long mNumberOfCluster;

     private final String mContainerName;

     private final BlobStore mStore;

     private final BlobStoreContext mContext;

     private final Cache<Integer , byte[]> mByteCache;

     private int lastIndexWritten;

     private byte[] lastBlobWritten;

     private final CompletionService<Integer> mWriterService;

     private final CompletionService<Map.Entry<Integer , byte[]>> mReaderService;

     private final ConcurrentHashMap<Integer , Future<Integer>> mRunningWriteTasks;

     private final ConcurrentHashMap<Integer , Future<Map.Entry<Integer , byte[]>>> mRunningReadTasks;

     /**

      * Creates a new {@link JCloudsStorageModule} backed by the specified file. If no such file exists, a

      * {@link FileNotFoundException} will be thrown.

      * 

      * @param pSizeInBlocks blocksize for this module

      * @param pFile local storage, not used over here

      * 

      */

     public JCloudsStorageModule (final long pSizeInBlocks, final File pFile) {

         // number * 512 = size in bytes

         // 4gig, bench for iozone and bonnie++

         // mNumberOfCluster = 8388608 / BLOCK_IN_CLUSTER;

         // 512m, bench for fio

         mNumberOfCluster = 1048576 / BLOCK_IN_CLUSTER;

         mContainerName = CONTAINERNAME;

         String[] credentials = getCredentials();

         if (credentials.length == 0) {

             Properties properties = new Properties();

             properties.setProperty(FilesystemConstants.PROPERTY_BASEDIR, pFile.getAbsolutePath());

             mContext = ContextBuilder.newBuilder("filesystem").overrides(properties).credentials("testUser", "testPass").buildView(BlobStoreContext.class);

         } else {

             mContext = ContextBuilder.newBuilder("aws-s3").credentials(getCredentials()[0], getCredentials()[1]).buildView(BlobStoreContext.class);

         }

         // Create Container

         mStore = mContext.getBlobStore();

         if (!mStore.containerExists(mContainerName)) {

             Location locToSet = null;

             for (Location loc : mStore.listAssignableLocations()) {

                 if (loc.getId().equals("eu-west-1")) {

                     locToSet = loc;

                     break;

                 }

             }

             // System.out.println(locToSet);

             mStore.createContainerInLocation(locToSet, mContainerName);

         }

         final ExecutorService writerService = Executors.newFixedThreadPool(20);

         final ExecutorService readerService = Executors.newFixedThreadPool(20);

         mRunningWriteTasks = new ConcurrentHashMap<Integer , Future<Integer>>();

         mRunningReadTasks = new ConcurrentHashMap<Integer , Future<Map.Entry<Integer , byte[]>>>();

         mReaderService = new ExecutorCompletionService<Map.Entry<Integer , byte[]>>(readerService);

         final Thread readHashmapCleaner = new Thread(new ReadFutureCleaner());

         readHashmapCleaner.setDaemon(true);

         readHashmapCleaner.start();

         mWriterService = new ExecutorCompletionService<Integer>(writerService);

         final Thread writeHashmapCleaner = new Thread(new WriteFutureCleaner());

         writeHashmapCleaner.setDaemon(true);

         writeHashmapCleaner.start();

         mByteCache = CacheBuilder.newBuilder().maximumSize(100).build();

         lastIndexWritten = -1;

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public int checkBounds (long logicalBlockAddress, int transferLengthInBlocks) {

         if (logicalBlockAddress < 0 || logicalBlockAddress >= getSizeInBlocks()) {

             return 1;

         } else

         // if the logical block address is in bounds but the transferlength

         // either exceeds

         // the device size or is faulty return 2

         if (transferLengthInBlocks < 0 || logicalBlockAddress + transferLengthInBlocks > getSizeInBlocks()) {

             return 2;

         } else {

             return 0;

         }

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public long getSizeInBlocks () {

         return mNumberOfCluster * BLOCK_IN_CLUSTER;

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public synchronized void read (byte[] bytes, long storageIndex) throws IOException {

         final int bucketIndex = (int) (storageIndex / SIZE_PER_BUCKET);

         final int bucketOffset = (int) (storageIndex % SIZE_PER_BUCKET);

         try {

             storeBucket(-1, null);

             // // DEBUG CODE

             // reader.write(bucketIndex + "," + storageIndex + "," +

             // bucketOffset + "," + bytes.length +

             // "\n");

             // reader.flush();

             byte[] data = mByteCache.getIfPresent(bucketIndex);

             if (data == null) {

                 data = getAndprefetchBuckets(bucketIndex);

             }

             final ByteArrayDataOutput output = ByteStreams.newDataOutput(bytes.length);

             int length = -1;

             if (bucketOffset + bytes.length > SIZE_PER_BUCKET) {

                 length = SIZE_PER_BUCKET - bucketOffset;

             } else {

                 length = bytes.length;

             }

             output.write(data, bucketOffset, length);

             if (bucketOffset + bytes.length > SIZE_PER_BUCKET) {

                 data = mByteCache.getIfPresent(bucketIndex + 1);

                 if (data == null) {

                     data = getAndprefetchBuckets(bucketIndex + 1);

                 }

                 output.write(data, 0, bytes.length - (SIZE_PER_BUCKET - bucketOffset));

             }

             System.arraycopy(output.toByteArray(), 0, bytes, 0, bytes.length);

         } catch (ExecutionException | InterruptedException exc) {

             throw new IOException(exc);

         }

     }

     private final byte[] getAndprefetchBuckets (final int pBucketStartId) throws InterruptedException , ExecutionException {

         byte[] returnval = null;

         Future<Map.Entry<Integer , byte[]>> startTask = null;

         for (int i = pBucketStartId; i < pBucketStartId + BUCKETS_TO_PREFETCH; i++) {

             Future<Map.Entry<Integer , byte[]>> currentTask = mRunningReadTasks.remove(i);

             if (currentTask == null) {

                 currentTask = mReaderService.submit(new ReadTask(i));

                 mRunningReadTasks.put(i, currentTask);

             }

             if (i == pBucketStartId) {

                 startTask = currentTask;

             }

         }

         returnval = startTask.get().getValue();

         return returnval;

     }

     private final void storeBucket (int pBucketId, byte[] pData) throws InterruptedException , ExecutionException {

         if (lastIndexWritten != pBucketId && lastBlobWritten != null) {

             Future<Integer> writeTask = mRunningWriteTasks.remove(lastIndexWritten);

             if (writeTask != null) {

                 writeTask.cancel(false);

             }

             mRunningWriteTasks.put(lastIndexWritten, mWriterService.submit(new WriteTask(lastBlobWritten, lastIndexWritten)));

         }

         lastIndexWritten = pBucketId;

         lastBlobWritten = pData;

     }

     /**

      * {@inheritDoc}

      * 

      * @throws Exception

      */

     @Override

     public synchronized void write (byte[] bytes, long storageIndex) throws IOException {

         final int bucketIndex = (int) (storageIndex / SIZE_PER_BUCKET);

         final int bucketOffset = (int) (storageIndex % SIZE_PER_BUCKET);

         try {

             // // DEBUG CODE

             // writer.write(bucketIndex + "," + storageIndex + "," +

             // bucketOffset + "," + bytes.length +

             // "\n");

             // writer.flush();

             byte[] data = mByteCache.getIfPresent(bucketIndex);

             if (data == null) {

                 data = getAndprefetchBuckets(bucketIndex);

             }

             System.arraycopy(bytes, 0, data, bucketOffset, bytes.length + bucketOffset > SIZE_PER_BUCKET ? SIZE_PER_BUCKET - bucketOffset

                     : bytes.length);

             storeBucket(bucketIndex, data);

             mByteCache.put(bucketIndex, data);

             if (bucketOffset + bytes.length > SIZE_PER_BUCKET) {

                 data = mByteCache.getIfPresent(bucketIndex + 1);

                 if (data == null) {

                     data = getAndprefetchBuckets(bucketIndex + 1);

                 }

                 System.arraycopy(bytes, SIZE_PER_BUCKET - bucketOffset, data, 0, bytes.length - (SIZE_PER_BUCKET - bucketOffset));

                 storeBucket(bucketIndex + 1, data);

                 mByteCache.put(bucketIndex + 1, data);

             }

         } catch (ExecutionException | InterruptedException exc) {

             throw new IOException(exc);

         }

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public void close () throws IOException {

         mContext.close();

     }

     /**

      * Getting credentials for aws from homedir/.credentials

      * 

      * @return a two-dimensional String[] with login and password

      */

     private static String[] getCredentials () {

         return new String[0];

         // File userStore = new File(System.getProperty("user.home"),

         // new StringBuilder(".credentials").append(File.separator)

         // .append("aws.properties").toString());

         // if (!userStore.exists()) {

         // return new String[0];

         // } else {

         // Properties props = new Properties();

         // try {

         // props.load(new FileReader(userStore));

         // return new String[] { props.getProperty("access"),

         // props.getProperty("secret") };

         //

         // } catch (IOException exc) {

         // throw new RuntimeException(exc);

         // }

         // }

     }

     /**

      * Single task to write data to the cloud.

      * 

      * @author Sebastian Graf, University of Konstanz

      * 

      */

     class ReadTask implements Callable<Map.Entry<Integer , byte[]>> {

         final Cipher mCipher;

         /**

          * Bucket ID to be read.

          */

         final int mBucketId;

         ReadTask (final int pBucketId) {

             if (ENCRYPT) {

                 try {

                     mCipher = Cipher.getInstance(ALGO);

                     mCipher.init(Cipher.DECRYPT_MODE, KEY);

                 } catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidKeyException e) {

                     throw new RuntimeException(e);

                 }

             } else {

                 mCipher = null;

             }

             this.mBucketId = pBucketId;

         }

         @Override

         public Map.Entry<Integer , byte[]> call () throws Exception {

             byte[] data = mByteCache.getIfPresent(mBucketId);

             if (data == null) {

                 // long time = System.currentTimeMillis();

                 Blob blob = mStore.getBlob(mContainerName, Integer.toString(mBucketId));

                 if (blob == null) {

                     data = new byte[SIZE_PER_BUCKET];

                     // // DEBUG CODE

                     // download.write(Integer.toString(mBucketId) + ", empty, "

                     // + (System.currentTimeMillis() - time) + "\n");

                     // download.flush();

                 } else {

                     data = ByteStreams.toByteArray(blob.getPayload().getInput());

                     // download.write(Integer.toString(mBucketId) + "," +

                     // data.length + " , "

                     // + (System.currentTimeMillis() - time) + "\n");

                     // download.flush();

                     while (data.length < SIZE_PER_BUCKET) {

                         blob = mStore.getBlob(mContainerName, Integer.toString(mBucketId));

                         data = ByteStreams.toByteArray(blob.getPayload().getInput());

                         // // DEBUG CODE

                         // download.write(Integer.toString(mBucketId) + "," +

                         // data.length + " , "

                         // + (System.currentTimeMillis() - time) + "\n");

                         // download.flush();

                     }

                     if (ENCRYPT) {

                         data = mCipher.doFinal(data);

                     }

                 }

             }

             if (data.length < SIZE_PER_BUCKET) {

                 System.out.println(data.length);

                 // throw new IllegalStateEception("Bucket " + mBucketId

                 // +" invalid");

             }

             mByteCache.put(mBucketId, data);

             final byte[] finalizedData = data;

             return new Map.Entry<Integer , byte[]>() {

                 @Override

                 public byte[] setValue (byte[] value) {

                     throw new UnsupportedOperationException();

                 }

                 @Override

                 public byte[] getValue () {

                     return finalizedData;

                 }

                 @Override

                 public Integer getKey () {

                     return mBucketId;

                 }

             };

         }

     }

     /**

      * Single task to write data to the cloud.

      * 

      * @author Sebastian Graf, University of Konstanz

      * 

      */

     class WriteTask implements Callable<Integer> {

         /**

          * The bytes to buffer.

          */

         final byte[] mData;

         final int mBucketIndex;

         final Cipher mCipher;

         WriteTask (byte[] pData, int pBucketIndex) {

             checkState(pData.length == SIZE_PER_BUCKET);

             if (ENCRYPT) {

                 try {

                     mCipher = Cipher.getInstance(ALGO);

                     mCipher.init(Cipher.ENCRYPT_MODE, KEY);

                 } catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidKeyException e) {

                     throw new RuntimeException(e);

                 }

             } else {

                 mCipher = null;

             }

             this.mData = pData;

             this.mBucketIndex = pBucketIndex;

         }

         @Override

         public Integer call () throws Exception {

             boolean finished = false;

             while (!finished) {

                 try {

                     // long time = System.currentTimeMillis();

                     byte[] data = mData;

                     if (ENCRYPT) {

                         data = mCipher.doFinal(mData);

                     }

                     Blob blob = mStore.blobBuilder(Integer.toString(mBucketIndex)).build();

                     blob.setPayload(data);

                     mStore.putBlob(mContainerName, blob);

                     // // DEBUG CODE

                     // upload.write(Integer.toString(mBucketIndex) + ", " +

                     // (System.currentTimeMillis() -

                     // time)

                     // + "\n");

                     // upload.flush();

                 } catch (Exception exc) {

                 }

                 finished = true;

             }

             return mBucketIndex;

         }

     }

     class ReadFutureCleaner extends Thread {

         public void run () {

             while (true) {

                 try {

                     Future<Map.Entry<Integer , byte[]>> element = mReaderService.take();

                     if (!element.isCancelled()) {

                         mRunningReadTasks.remove(element.get().getKey());

                     }

                 } catch (Exception exc) {

                     throw new RuntimeException(exc);

                 }

             }

         }

     }

     class WriteFutureCleaner extends Thread {

         public void run () {

             while (true) {

                 try {

                     Future<Integer> element = mWriterService.take();

                     if (!element.isCancelled()) {

                         mRunningWriteTasks.remove(element.get());

                     }

                 } catch (Exception exc) {

                     throw new RuntimeException(exc);

                 }

             }

         }

     }

 }