bloblib v0.0.4

Author: tony4248

README.md

@@ -11,8 +11,13 @@ BlobLib is the core libraries of the blobfs and blobfs-win, blobfs is a distribu
 Packaged the core operations of the azure blob storage as a separated library, this will make it easier for supporting different OS platform. This library can also be used alone with various projects.
 
 
-## Features:
-* New Features since v0.0.4 :
+## Features and Updates:
+### New features since v0.0.4 :
+
+* upload and down with multiple threads, more faster and stable.
+* Page blob is full supported.
+
+### New features since v0.0.3 :
 
 * Support man operations of the container,virtual directory and blob. such as create, delete, list, read, write etc.
 * New extension class of the InputStream and OutputStream. add a new cache layer and is optimized for the blob read and write.

src/main/java/com/wesley/bloblib/BlobBufferedOus.java

@@ -36,6 +36,7 @@ public class BlobBufferedOus extends OutputStream {
 	int chunkSizeOfBB = Constants.BLOB_BUFFERED_OUTS_BLOCKBLOB_CHUNK_SIZE * 2;
 	int chunkNumber = 0;
 	int chunkSizeOfAB = Constants.BLOB_BUFFERED_OUTS_APPENDBLOB_CHUNK_SIZE;
+	int chunkSizeOfPB = Constants.BLOB_BUFFERED_OUTS_PAGEBLOB_CHUNK_SIZE;
 	/* the available bytes in central buffer */
 	long blobOffset = 0;
 	/* the path of the blob */
@@ -98,6 +99,7 @@ public synchronized void write(final byte[] data, final int offset, final int le
 		/* write data to the buffer */
 		writeToBuffer(data, offset, length);
 		/* check the buffered data and the chunk size threshold */
+		/* make sure the data is 512 Bytes aligned for page blob */
 		if (isBufferedDataReadyToUpload()){
 			int numOfdataUploaded = 0;
 			if ((numOfdataUploaded = uploadBlobChunk(centralBuffer, 0, centralBufOffset)) > 0){
@@ -121,6 +123,12 @@ public synchronized final void flush() throws IOException {
 		try {
 			if (centralBufOffset > 0) {
 				int numOfdataUploaded = 0;
+				/* for page blob, we should make sure the length is 512 Bytes aligned */
+				if (BfsBlobType.PAGEBLOB.equals(this.blobType)) {
+					/* make sure the data is 512 Bytes aligned for page blob */
+					this.centralBufOffset = (int) BfsUtility.extendTo512BytesAlignedLength(centralBufOffset);
+					this.centralBuffer = BfsUtility.extendTo512BytesAlignedData(this.centralBuffer);
+				}
 				if ((numOfdataUploaded = uploadBlobChunk(centralBuffer, 0, centralBufOffset)) > 0) {
 					totalDataUploaded += numOfdataUploaded;
 					/* reset the chunk count */
@@ -198,17 +206,20 @@ public synchronized final int uploadBlobChunk (byte[] rawData,  int offset, int
 			/* renew the lease firstly, otherwise the lease may be expired, this will cause error */
 			blob.renewLease(accCondtion);
      		if (BfsBlobType.BLOCKBLOB.equals(this.blobType)){
+     			/* is the block blob */
      			int uploadRes[] = parallelUploader.uploadBlobWithParallelThreads(rawData, blob, offset, length, 
      					blockList, accCondtion, leaseID, chunkNumber);
 				/* update the chunk counter */
 				chunkNumber = uploadRes[1];
 			} 
      		else
 			{
-				ByteArrayInputStream bInput = new ByteArrayInputStream(rawData, offset, length);
+				/* is the append blob or page blob */
+     			ByteArrayInputStream bInput = new ByteArrayInputStream(rawData, offset, length);
 				if (BfsBlobType.APPENDBLOB.equals(this.blobType)){
 					((CloudAppendBlob) blob).appendBlock(bInput, (long)length, accCondtion, null, null);
 				}else if (BfsBlobType.PAGEBLOB.equals(this.blobType)){
+					/* for page blob, we already make sure the length is 512B aligned */
 					((CloudPageBlob) blob).upload(bInput, length, accCondtion, null, null);
 				}
 				bInput.close();
@@ -223,12 +234,19 @@ public synchronized final int uploadBlobChunk (byte[] rawData,  int offset, int
 		return dataUploadedThisChunk;
 	}
 
+	/** make sure the length is valid
+	 * @return
+	 */
 	public synchronized boolean isBufferedDataReadyToUpload() {
 		boolean result = false;
 		if (BfsBlobType.BLOCKBLOB.equals(this.blobType)){
-			if (centralBufOffset > chunkSizeOfBB){ return result = true;}
+			if (centralBufOffset > chunkSizeOfBB){ result = true;}
 		} else if (BfsBlobType.APPENDBLOB.equals(this.blobType)){
-			if (centralBufOffset > chunkSizeOfAB){ return result = true;}
+			if (centralBufOffset > chunkSizeOfAB){ result = true;}
+		} else if(BfsBlobType.PAGEBLOB.equals(this.blobType)){
+			if (centralBufOffset > chunkSizeOfPB && BfsUtility.is512BytesAligned(centralBufOffset)){
+				result = true;
+			}
 		}
 		/* page blob */
 		return result;
@@ -259,14 +277,16 @@ public synchronized void verifyUploadConditions() throws BfsException {
 				String errMessage = "The size of the source file exceeds the size limit: " + blobSizeLimit + "."; 
 				throw new BfsException(errMessage);
 			}
-			
-			if (numOfCommitedBlocks + chunkNumber > Constants.BLOB_BLOCK_NUMBER_LIMIT - 1){ //50000
-				String errMessage = "The block count of target blob: " + fullBlobPath + " exceeds the " + Constants.BLOB_BLOCK_NUMBER_LIMIT + " count limit.";
-				throw new BfsException(errMessage);
+			/* for block blob and append blob */
+			if (BfsBlobType.BLOCKBLOB.equals(this.blobType) || (BfsBlobType.APPENDBLOB.equals(this.blobType))){
+				if (numOfCommitedBlocks + chunkNumber > Constants.BLOB_BLOCK_NUMBER_LIMIT - 1){ //50000
+					String errMessage = "The block count of target blob: " + fullBlobPath + " exceeds the " + Constants.BLOB_BLOCK_NUMBER_LIMIT + " count limit.";
+					throw new BfsException(errMessage);
+				}
 			}
 		} catch (Exception ex) {
 			String errMessage = "Unexpected exception occurred when verifing the upload conditons for to the blob : " 
-					+ this.fullBlobPath + "." + ex.getMessage(); 
+					+ this.fullBlobPath + ". " + ex.getMessage(); 
 			BfsUtility.throwBlobfsException(ex, errMessage);
 
 		}

src/main/java/com/wesley/bloblib/BlobLib.java

@@ -7,6 +7,8 @@ public class BlobLib {
 	@SuppressWarnings("static-access")
 	public static void main(String[] args) {
 		try {
+//			long len = get512BytesAlignedLength(23435543L);
+//			System.out.println(len);
 			//BfsPath bfsPath = new BfsPath("/index.html");
 			//PathProperties pathProperties = bfsPath.getBfsPathProperties();
 //			MsgsToBeSentCache.getInstance().startAutoSendService();
@@ -36,9 +38,9 @@ public static void main(String[] args) {
 //            BlobBufferedIns bbIns = new BlobBufferedIns(insParams);
 //			BlobReqParams ousParams = new BlobReqParams();
 //            ousParams.setContainer("music2017");
-//            ousParams.setBlob("blobfs-2017-11-23.log");
+//            ousParams.setBlob("blobfs-2017-10-21.vhd");
 ////            /* get the blob type */
-//            ousParams.setBfsBlobType(BfsBlobType.BLOCKBLOB);
+//            ousParams.setBfsBlobType(BfsBlobType.PAGEBLOB);
 //            BlobService.createBlob(ousParams);
 //            BlobBufferedOus bbOus = new BlobBufferedOus(ousParams);	
 //			FileReader fileRead = new FileReader("C:\\Downloads\\Mydevs\\blobfs-win\\blobfs-win\\bin\\Debug\\blobfs-2017-11-19.log");

src/main/java/com/wesley/bloblib/Constants.java

@@ -20,6 +20,7 @@ public final class Constants {
 	public static final int 	BLOB_BUFFERED_OUTS_BUFFER_SIZE = 4 * 1024 * 1024; //default is 4MB:  4 * 1024 * 1024 = 4194304
 	public static final int 	BLOB_BUFFERED_OUTS_BLOCKBLOB_CHUNK_SIZE = 4 * 1024 * 1024; //default is 4MB:  4 * 1024 * 1024 = 4194304
 	public static final int 	BLOB_BUFFERED_OUTS_APPENDBLOB_CHUNK_SIZE = 4 * 1024 * 1024; //default is 4MB:  4 * 1024 * 1024 = 4194304
+	public static final int 	BLOB_BUFFERED_OUTS_PAGEBLOB_CHUNK_SIZE = 4 * 1024 * 1024; //default is 4MB:  4 * 1024 * 1024 = 4194304
 
 	public static final int 	BFS_FILES_CACHE_INIT_CAPACITY = 1024; //1024
 	public static final int 	BFS_FILES_CACHE_MAX_CAPACITY = 65535; //opened file

src/main/java/com/wesley/bloblib/ParallelUploader.java

@@ -121,6 +121,7 @@ private final List<UploadTask> splitJobIntoTheParallelTasks(int blobOffset, long
             } else {
               bytesToWrite = (int)bytesLeft;
             }
+            
             chunkNumber ++;
         	/* save chunk id in array (must be base64) */
             String chunkId = DatatypeConverter.printBase64Binary(String.format("BlockId%07d", chunkNumber).getBytes(StandardCharsets.UTF_8));

src/main/java/com/wesley/bloblib/utils/BfsUtility.java

@@ -15,6 +15,7 @@
 
 import org.pmw.tinylog.Logger;
 
+import com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper;
 import com.microsoft.azure.storage.blob.BlockEntry;
 import com.wesley.bloblib.Constants;
 import com.wesley.bloblib.BfsException;
@@ -109,6 +110,10 @@ public static int commandExecutor(final String command, final StringBuilder outp
         return process.exitValue();
     }
 
+    /**
+     * @param process
+     * @return
+     */
     public static boolean isRunning(Process process) {
         try {
             process.exitValue();
@@ -133,6 +138,10 @@ public static int getIdOnUnix(String cmd) throws IOException{
 		return uid;		
 	}
 
+	/**
+	 * @param path
+	 * @return
+	 */
 	public static String removeLastSlash (String path){
 		String tmpPath = "";
 		if (path.endsWith("/")) {
@@ -143,24 +152,41 @@ public static String removeLastSlash (String path){
 		return tmpPath;
 	}
 
+	/**
+	 * @param date
+	 * @param timeZone
+	 * @return
+	 */
 	public static String changeDateTimeZone(Date date , String timeZone){
 		DateFormat dateFormat = new SimpleDateFormat("EEE MMM dd HH:mm:ss zzz yyyy", Locale.US);
 		dateFormat.setTimeZone(TimeZone.getTimeZone(timeZone));
 		return dateFormat.format(date);
 	}
 
+	/**
+	 * @param fileOrDirPath
+	 * @return
+	 */
 	public static String getParentDirPath(String fileOrDirPath) {
 	    boolean endsWithSlash = fileOrDirPath.endsWith(File.separator);
 	    return fileOrDirPath.substring(0, fileOrDirPath.lastIndexOf(File.separatorChar, 
 	            endsWithSlash ? fileOrDirPath.length() - 2 : fileOrDirPath.length() - 1));
 	}
 
 
+	/**
+	 * @param date
+	 * @return
+	 */
 	public static String convertDateToString(Date date){
 		DateFormat dateFormat = new SimpleDateFormat("EEE MMM dd HH:mm:ss zzz yyyy", Locale.US);
 		return dateFormat.format(date);
 	}
 
+	/**
+	 * @param dateString
+	 * @return
+	 */
 	public static Date convertStringToDate(String dateString){
 		DateFormat dateFormat = new SimpleDateFormat("EEE MMM dd HH:mm:ss zzz yyyy", Locale.US);
 		Date date = null;
@@ -173,8 +199,63 @@ public static Date convertStringToDate(String dateString){
 		return date;
 	}
 
+	/**
+	 * @return
+	 */
 	public static boolean isWindows() {		
         return (OS.indexOf("win") >= 0);
     }
 
+	/**
+	 * @param length
+	 * @return
+	 */
+	public static boolean is512BytesAligned(long length){
+	    if (length % 512 == 0){return true;}
+	    return false;
+	}
+
+	/**
+	 * get the 512 bytes aligned length
+	 * @param length
+	 * @return
+	 */
+	public static long shrinkTo512BytesAlignedLength(long length){
+		long tmpLen = 0;
+	    if (length % 512 == 0){tmpLen = length; }
+	    else{
+	    	if (length < 512){tmpLen = 0; }
+	    	else{
+	    		tmpLen = (length - length % 512);
+	    	}
+	    }
+	    return tmpLen;
+	}
+	
+	public static long extendTo512BytesAlignedLength(long length){
+		long tmpLen = 0;
+	    if (length % 512 == 0){ tmpLen = length; }
+	    else{
+	    	tmpLen = length + (512 - length % 512);
+	    }
+	    return tmpLen;
+	}
+	
+	/**
+	 * make data 512 bytes aligned by padding the rest slots with zero;
+	 * @param origData
+	 * @return
+	 */
+	public static byte[]  extendTo512BytesAlignedData(byte[] origData){
+		long length = origData.length;
+	    if (length % 512 == 0){ return  origData; }
+	    else{
+	    	long tmpLen = length + (512 - length % 512);
+	    	byte[] tgtData = new byte [(int) tmpLen]; // all bytes are initialized with 0's
+	    	System.arraycopy(origData, 0, tgtData, 0, (int) length);
+	    	return tgtData;
+	    }
+	}
+	
+	
 }