fix the bug

Author: tony4248

bin/blobfs.conf

@@ -2,7 +2,7 @@
 # ++++++++++++++++++ blobfs configurations  +++++++++++++++++++++++++++++++++++++++
 # ================================================================================
 # Azure storage account connection string
-Storage_Connection_String = DefaultEndpointsProtocol=https;AccountName=dev2107;AccountKey=wJNRBizfsyQ2JUBBBJjBFAsxjBmvVpkK+61ZqKV/n3HaI/jSEdS3oGoyBGdmvBhVVeUy5S2hlGSAuipeyXXSww==;EndpointSuffix=core.chinacloudapi.cn
+Storage_Connection_String = xxx
 
 # the name of the queue, which will be used for synchronizing the file cache across the nodes.
 queue_name = blobfs

bin/bloblib-1.0.0.jar

@@ -31,10 +31,6 @@ public class BlobBufferedIns extends InputStream {
 
 	boolean isBlobEOF = false;
 
-	/* get the parallel downloader instance */
-	ParallelDownloader parallelDownloader = ParallelDownloader.getInstance();
-	
-	
 	@SuppressWarnings("static-access")
 	public BlobBufferedIns(BlobReqParams reqParams) throws BfsException {
 		this.blob = BlobService.getBlobReference(reqParams);
@@ -44,15 +40,14 @@ public BlobBufferedIns(BlobReqParams reqParams) throws BfsException {
 		this.fullBlobPath = reqParams.getBlobFullPath();
 		this.dwLocalBufferSize = (int) Math.min(Constants.BLOB_BUFFERED_INS_DOWNLOAD_SIZE * 2, this.blobSize);
 	}
-	
 	public CloudBlob getBlob() {
 		return blob;
 	}
 
 	public long getBlobSize() {
 		return blobSize;
 	}
-
+	
 	@Override
 	public synchronized int read(byte[] outputBuffer, int offset, int len) throws IOException {
 		int numOfBytesReaded = 0;
@@ -220,9 +215,8 @@ public synchronized final byte[] downloadBlobChunk (int offset, long len) throws
 		len = Math.min(blobSize - offset, len) ;
         dwLocalBuffer = new byte[(int) len];
 		try {
-			dwLocalBuffer = parallelDownloader.downloadBlobWithParallelThreads(blob, offset, len);
-			bytesDownloaded = dwLocalBuffer.length;
-		} catch (Exception ex) {
+			bytesDownloaded = blob.downloadRangeToByteArray(offset, len, dwLocalBuffer, 0);
+		} catch (StorageException ex) {
 			String errMessage = "Unexpected exception occurred when downloading from the blob : " 
 						+ this.fullBlobPath + ". " + ex.getMessage(); 
 			BfsUtility.throwBlobfsException(ex, errMessage);

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

@@ -3,9 +3,12 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 
+import javax.xml.bind.DatatypeConverter;
+
 import org.pmw.tinylog.Logger;
 
 import com.microsoft.azure.storage.AccessCondition;
@@ -33,7 +36,7 @@ public class BlobBufferedOus extends OutputStream {
 	long totalDataUploaded = 0;
 	long localFileSize = 0;
 	/* the upload chunk size, the should be smaller than the buffer size */
-	int chunkSizeOfBB = Constants.BLOB_BUFFERED_OUTS_BLOCKBLOB_CHUNK_SIZE * 2;
+	int chunkSizeOfBB = Constants.BLOB_BUFFERED_OUTS_BLOCKBLOB_CHUNK_SIZE;
 	int chunkNumber = 0;
 	int chunkSizeOfAB = Constants.BLOB_BUFFERED_OUTS_APPENDBLOB_CHUNK_SIZE;
 	int chunkSizeOfPB = Constants.BLOB_BUFFERED_OUTS_PAGEBLOB_CHUNK_SIZE;
@@ -51,9 +54,6 @@ public class BlobBufferedOus extends OutputStream {
 
 	/* list of all block ids we will be uploading - need it for the commit at the end */
     List<BlockEntry> blockList;
-    
-    /* get the parallel uploader instance */
-    ParallelUploader parallelUploader = ParallelUploader.getInstance();
 
 	@SuppressWarnings("static-access")
 	public BlobBufferedOus(BlobReqParams reqParams) throws BfsException, StorageException {
@@ -99,7 +99,6 @@ 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){
@@ -123,12 +122,6 @@ 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 */
@@ -205,27 +198,34 @@ public synchronized final int uploadBlobChunk (byte[] rawData,  int offset, int
 		try {
 			/* 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
-			{
-				/* 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 */
+			
+			ByteArrayInputStream bInput = new ByteArrayInputStream(rawData, offset, length);
+			/* update the chunk counter */
+			chunkNumber ++;
+			if (BfsBlobType.BLOCKBLOB.equals(this.blobType)){
+				/* save chunk id in array (must be base64) */
+				String chunkId = DatatypeConverter.printBase64Binary(String.format("BlockId%07d", chunkNumber).getBytes(StandardCharsets.UTF_8));
+                //String chunkId = Base64.getEncoder().encodeToString(String.format("BlockId%07d", chunkNumber).getBytes(StandardCharsets.UTF_8));
+                BlockEntry chunk = new BlockEntry(chunkId);
+                blockList.add(chunk);               
+				((CloudBlockBlob) blob).uploadBlock(chunkId, bInput, (long)length, accCondtion, null, null);
+			} else if (BfsBlobType.APPENDBLOB.equals(this.blobType)){				
+				((CloudAppendBlob) blob).appendBlock(bInput, (long)length, accCondtion, null, null);
+			} else if (BfsBlobType.PAGEBLOB.equals(this.blobType)){
 					((CloudPageBlob) blob).upload(bInput, length, accCondtion, null, null);
-				}
-				bInput.close();
 			}
+			bInput.close();
 			dataUploadedThisChunk = length;
-	} catch (Exception ex) {
+			/* save the block ID list , secure the data */ 
+			if (BfsBlobType.BLOCKBLOB.equals(this.blobType)){
+				/* set leaseID in the meta data */
+				BlobReqParams cbParams = new BlobReqParams();
+				cbParams.setBlobInstance(blob);
+				cbParams.setLeaseID(leaseID);
+				BlobService.setBlobMetadata(cbParams,Constants.BLOB_META_DATA_COMMITED_BLOBKS_KEY, BfsUtility.blockIds(blockList));
+			}
+			
+		} catch (StorageException | IOException ex) {
 			String errMessage = "Unexpected exception occurred when uploading to the blob : " 
 						+ this.fullBlobPath + ", No. of chunk: " + chunkNumber + "." + ex.getMessage(); 
 			BfsUtility.throwBlobfsException(ex, errMessage);
@@ -234,21 +234,14 @@ 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){ result = true;}
+			if (centralBufOffset > chunkSizeOfBB){ return result = true;}
 		} else if (BfsBlobType.APPENDBLOB.equals(this.blobType)){
-			if (centralBufOffset > chunkSizeOfAB){ result = true;}
-		} else if(BfsBlobType.PAGEBLOB.equals(this.blobType)){
-			if (centralBufOffset > chunkSizeOfPB && BfsUtility.is512BytesAligned(centralBufOffset)){
-				result = true;
-			}
+			if (centralBufOffset > chunkSizeOfAB){ return result = true;}
 		}
-		/* page blob */
+		/* page blob */		 
 		return result;
 	}
 
@@ -277,16 +270,14 @@ public synchronized void verifyUploadConditions() throws BfsException {
 				String errMessage = "The size of the source file exceeds the size limit: " + blobSizeLimit + "."; 
 				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);
-				}
+			
+			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);
 
 		}