Implement XL layer - preliminary work.

xl-v1-createfile.go

@@ -0,0 +1,206 @@
+/*
+ * Minio Cloud Storage, (C) 2016 Minio, Inc.
+ *
+ * Licensed 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.
+ */
+
+package main
+
+import (
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"hash"
+	"io"
+	slashpath "path"
+	"strconv"
+	"time"
+
+	fastSha512 "github.com/minio/minio/pkg/crypto/sha512"
+)
+
+// Erasure block size.
+const erasureBlockSize = 4 * 1024 * 1024 // 4MiB.
+
+// Close and remove writers if they are safeFile.
+func closeAndRemoveWriters(writers ...io.WriteCloser) {
+	for _, writer := range writers {
+		safeCloseAndRemove(writer)
+	}
+}
+
+// WriteErasure reads predefined blocks, encodes them and writes to
+// configured storage disks.
+func (xl XL) writeErasure(volume, path string, reader *io.PipeReader) {
+	var writers = make([]io.WriteCloser, len(xl.storageDisks))
+	var sha512Writers = make([]hash.Hash, len(xl.storageDisks))
+	var metadataWriters = make([]io.WriteCloser, len(xl.storageDisks))
+
+	// Initialize storage disks, get all the writers and corresponding
+	// metadata writers.
+	for index, disk := range xl.storageDisks {
+		var err error
+		erasurePart := slashpath.Join(path, fmt.Sprintf("part.%d", index))
+		writers[index], err = disk.CreateFile(volume, erasurePart)
+		if err != nil {
+			// Remove previous temp writers for any failure.
+			closeAndRemoveWriters(writers...)
+			closeAndRemoveWriters(metadataWriters...)
+			deletePathAll(volume, path, xl.storageDisks...)
+			reader.CloseWithError(err)
+			return
+		}
+		metadataFilePath := slashpath.Join(path, metadataFile)
+		metadataWriters[index], err = disk.CreateFile(volume, metadataFilePath)
+		if err != nil {
+			// Remove previous temp writers for any failure.
+			closeAndRemoveWriters(writers...)
+			closeAndRemoveWriters(metadataWriters...)
+			deletePathAll(volume, path, xl.storageDisks...)
+			reader.CloseWithError(err)
+			return
+		}
+		sha512Writers[index] = fastSha512.New()
+	}
+
+	// Allocate 4MiB block size buffer for reading.
+	buffer := make([]byte, erasureBlockSize)
+	var totalSize int64 // Saves total incoming stream size.
+	for {
+		// Read up to allocated block size.
+		n, err := io.ReadFull(reader, buffer)
+		if err != nil {
+			// Any unexpected errors, close the pipe reader with error.
+			if err != io.ErrUnexpectedEOF && err != io.EOF {
+				// Remove all temp writers.
+				closeAndRemoveWriters(writers...)
+				closeAndRemoveWriters(metadataWriters...)
+				deletePathAll(volume, path, xl.storageDisks...)
+				reader.CloseWithError(err)
+				return
+			}
+		}
+		// At EOF break out.
+		if err == io.EOF {
+			break
+		}
+		if n > 0 {
+			// Split the input buffer into data and parity blocks.
+			var blocks [][]byte
+			blocks, err = xl.ReedSolomon.Split(buffer[0:n])
+			if err != nil {
+				// Remove all temp writers.
+				closeAndRemoveWriters(writers...)
+				closeAndRemoveWriters(metadataWriters...)
+				deletePathAll(volume, path, xl.storageDisks...)
+				reader.CloseWithError(err)
+				return
+			}
+			// Encode parity blocks using data blocks.
+			err = xl.ReedSolomon.Encode(blocks)
+			if err != nil {
+				// Remove all temp writers upon error.
+				closeAndRemoveWriters(writers...)
+				closeAndRemoveWriters(metadataWriters...)
+				deletePathAll(volume, path, xl.storageDisks...)
+				reader.CloseWithError(err)
+				return
+			}
+			// Loop through and write encoded data to all the disks.
+			for index, encodedData := range blocks {
+				_, err = writers[index].Write(encodedData)
+				if err != nil {
+					// Remove all temp writers upon error.
+					closeAndRemoveWriters(writers...)
+					closeAndRemoveWriters(metadataWriters...)
+					deletePathAll(volume, path, xl.storageDisks...)
+					reader.CloseWithError(err)
+					return
+				}
+				sha512Writers[index].Write(encodedData)
+			}
+			// Update total written.
+			totalSize += int64(n)
+		}
+	}
+
+	// Save additional erasureMetadata.
+	modTime := time.Now().UTC()
+
+	// Initialize metadata map, save all erasure related metadata.
+	metadata := make(map[string]string)
+	metadata["version"] = minioVersion
+	metadata["format.major"] = "1"
+	metadata["format.minor"] = "0"
+	metadata["format.patch"] = "0"
+	metadata["file.size"] = strconv.FormatInt(totalSize, 10)
+	metadata["file.modTime"] = modTime.Format(timeFormatAMZ)
+	metadata["file.xl.blockSize"] = strconv.Itoa(erasureBlockSize)
+	metadata["file.xl.dataBlocks"] = strconv.Itoa(xl.DataBlocks)
+	metadata["file.xl.parityBlocks"] = strconv.Itoa(xl.ParityBlocks)
+
+	// Write all the metadata.
+	for index, metadataWriter := range metadataWriters {
+		// Save sha512 checksum of each encoded blocks.
+		metadata["file.xl.block512Sum"] = hex.EncodeToString(sha512Writers[index].Sum(nil))
+
+		// Marshal metadata into json strings.
+		metadataBytes, err := json.Marshal(metadata)
+		if err != nil {
+			closeAndRemoveWriters(writers...)
+			closeAndRemoveWriters(metadataWriters...)
+			deletePathAll(volume, path, xl.storageDisks...)
+			reader.CloseWithError(err)
+			return
+		}
+
+		_, err = metadataWriter.Write(metadataBytes)
+		if err != nil {
+			closeAndRemoveWriters(writers...)
+			closeAndRemoveWriters(metadataWriters...)
+			deletePathAll(volume, path, xl.storageDisks...)
+			reader.CloseWithError(err)
+			return
+		}
+	}
+
+	// Close all writers and metadata writers in routines.
+	for index := range xl.storageDisks {
+		// Safely wrote, now rename to its actual location.
+		writers[index].Close()
+		metadataWriters[index].Close()
+	}
+
+	// Close the pipe reader and return.
+	reader.Close()
+	return
+}
+
+// CreateFile - create a file.
+func (xl XL) CreateFile(volume, path string) (writeCloser io.WriteCloser, err error) {
+	if !isValidVolname(volume) {
+		return nil, errInvalidArgument
+	}
+	if !isValidPath(path) {
+		return nil, errInvalidArgument
+	}
+
+	// Initialize pipe for data pipe line.
+	pipeReader, pipeWriter := io.Pipe()
+
+	// Start erasure encoding in routine, reading data block by block from pipeReader.
+	go xl.writeErasure(volume, path, pipeReader)
+
+	// Return the piped writer, caller should start writing to this.
+	return pipeWriter, nil
+}