- handle manager for source and targets

- refactoring

Author: giventocode

blobporter.go

@@ -15,7 +15,7 @@ import (
 	"github.com/Azure/blobporter/util"
 )
 
-const programVersion = "0.6.05"
+const programVersion = "0.6.06"
 
 var argsUtil paramParserValidator
 
@@ -111,7 +111,7 @@ func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo, numOfBatches int,
 			totalSize = totalSize + source.Size
 		}
 
-		if len(sourcesInfo) < 20 {
+		if len(sourcesInfo) < 0 {
 			fmt.Printf(summary)
 			return
 		}

targets/handleman.go renamed to internal/handleman.go

@@ -1,4 +1,4 @@
-package targets
+package internal
 
 import (
 	"fmt"
@@ -9,20 +9,32 @@ import (
 
 //There're two components here: poolHandle and the handle factory.
 //A pool is an asynchronous request/respone worker that runs on a single go-routine and keeps file handles for each file.
-//The number of file handles is constraint by the max number of handlers in cache (numberOfHandlersInCache) and the max number of handles per file (numOfHandlesPerFile).
+//The number of file handles is constraint by the max number of handlers in cache (maxFileHandlesInCache) and the max number of handles per file (numOfHandlesPerFile).
 //When the max number handles is reached file handles will be closed until space is available. The handle factory opens the file handles and initializes the
 //target file in case the folder structure and file need to be created. Since the factory tracks if a file has been initailized
 //, i.e. created or truncated at the begining of the transfer, only one instance of the factory is created.
 
-const maxFileHandlesInCache int32 = 600
+const maxFileHandlesInCache int = 600
 
-type fileHandlePool struct {
+//HandleMode TODO
+type HandleMode int
+
+const (
+	//Read read only file handles
+	Read HandleMode = iota
+	//Write write and append file handles
+	Write
+)
+
+//FileHandlePool TODO
+type FileHandlePool struct {
 	maxCacheSize      int
 	maxHandlesPerFile int
 	factory           *handleFactory
 	fileHandles       map[string][]*os.File
 	overwrite         bool
 	channels          poolChannels
+	mode              HandleMode
 }
 type poolChannels struct {
 	handleReq chan poolRequest
@@ -47,13 +59,15 @@ type poolResponse struct {
 	err    error
 }
 
-func newfileHandlePool(maxCacheSize int, maxHandlesPerFile int, overwrite bool) *fileHandlePool {
-	pool := fileHandlePool{
-		maxCacheSize:      maxCacheSize,
+//NewFileHandlePool TODO
+func NewFileHandlePool(maxHandlesPerFile int, mode HandleMode, overwrite bool) *FileHandlePool {
+	pool := FileHandlePool{
+		maxCacheSize:      maxFileHandlesInCache,
 		maxHandlesPerFile: maxHandlesPerFile,
-		factory:           newhandleFactory(overwrite),
+		factory:           newhandleFactory(mode, overwrite),
 		fileHandles:       make(map[string][]*os.File),
 		overwrite:         overwrite,
+		mode:              mode,
 		channels: poolChannels{
 			handleReq: make(chan poolRequest, 100),
 			closeReq:  make(chan poolCloseRequest, 100),
@@ -66,24 +80,28 @@ func newfileHandlePool(maxCacheSize int, maxHandlesPerFile int, overwrite bool)
 	return &pool
 }
 
-func (f *fileHandlePool) getHandle(path string) (*os.File, error) {
+//GetHandle TODO
+func (f *FileHandlePool) GetHandle(path string) (*os.File, error) {
 	respChan := make(chan poolResponse, 1)
 	req := poolRequest{path: path, response: respChan}
 	f.channels.handleReq <- req
 	resp := <-respChan
 	return resp.handle, resp.err
 }
 
-func (f *fileHandlePool) returnHandle(path string, handle *os.File) error {
+//ReturnHandle TODO
+func (f *FileHandlePool) ReturnHandle(path string, handle *os.File) error {
 	select {
 	case f.channels.returnReq <- poolReturnRequest{handle: handle, path: path}:
 	default:
-		//close the handle if channel is fool
+		//close the handle if channel is full
 		return handle.Close()
 	}
 	return nil
 }
-func (f *fileHandlePool) closeHandles(path string) error {
+
+//CloseHandles TODO
+func (f *FileHandlePool) CloseHandles(path string) error {
 	respChan := make(chan error, 1)
 	req := poolCloseRequest{path: path, err: respChan}
 	f.channels.closeReq <- req
@@ -93,12 +111,12 @@ func (f *fileHandlePool) closeHandles(path string) error {
 
 const fileHandleCacheDebug = "BP_FHC_DBG"
 
-func (f *fileHandlePool) startPool() {
-	//start := time.Now()
-	oc := 0
-	cc := 0
-	cm := 0
-	ch := 0
+func (f *FileHandlePool) startPool() {
+
+	oc := 0 //num of open requests
+	cc := 0 //num of close requests
+	cm := 0 //num of cache misses
+	ch := 0 //num of cache hits
 	dbg := os.Getenv(fileHandleCacheDebug)
 	go func() {
 		for {
@@ -204,6 +222,7 @@ func (f *fileHandlePool) startPool() {
 type handleFactory struct {
 	init       map[string]bool
 	factoryReq chan factoryRequest
+	mode       HandleMode
 }
 
 type factoryRequest struct {
@@ -216,11 +235,12 @@ type factoryResponse struct {
 	err    error
 }
 
-func newhandleFactory(overwrite bool) *handleFactory {
+func newhandleFactory(mode HandleMode, overwrite bool) *handleFactory {
 	reqChan := make(chan factoryRequest, 100)
 	fact := handleFactory{
 		init:       make(map[string]bool),
 		factoryReq: reqChan,
+		mode:       mode,
 	}
 
 	fact.startFactory(overwrite)
@@ -251,12 +271,17 @@ func (h *handleFactory) startFactory(overwrite bool) {
 			_, exists := h.init[req.path]
 			var fh *os.File
 			var err error
-			if !exists {
-				//fmt.Printf("init->%v\n", req.path)
-				fh, err = h.initFile(req.path, overwrite)
-			} else {
-				//fmt.Printf("open->%v\n", req.path)
-				fh, err = os.OpenFile(req.path, os.O_WRONLY, os.ModeAppend)
+			switch h.mode {
+			case Read:
+				fh, err = os.OpenFile(req.path, os.O_RDONLY, os.ModeAppend)
+			case Write:
+				if !exists {
+					fh, err = h.initFile(req.path, overwrite)
+				} else {
+					fh, err = os.OpenFile(req.path, os.O_WRONLY, os.ModeAppend)
+				}
+			default:
+				panic(fmt.Sprintf("Invalid handle mode:%v", h.mode))
 			}
 
 			select {

pipeline/pipeline.go

@@ -151,9 +151,7 @@ func createPartsInPartition(partitionSize int64, partitionOffSet int64, ordinalS
 
 //ConstructPartsPartition creates a slice of PartsPartition with a len of numberOfPartitions.
 func ConstructPartsPartition(numberOfPartitions int, size int64, blockSize int64, sourceURI string, targetAlias string, bufferQ chan []byte) []PartsPartition {
-	//bsib := uint64(blockSize)
 	numOfBlocks := int((size + blockSize - 1) / blockSize)
-
 	Partitions := make([]PartsPartition, numberOfPartitions)
 	//the size of the partition needs to be a multiple (blockSize * int) to make sure all but the last part/block
 	//are the same size
@@ -165,17 +163,27 @@ func ConstructPartsPartition(numberOfPartitions int, size int64, blockSize int64
 	var partOrdinal int
 	for p := 0; p < numberOfPartitions; p++ {
 		poffSet := int64(int64(p) * partitionSize)
+
 		if p == numberOfPartitions-1 {
 			partitionSize = int64(bytesLeft)
 		}
-		partition := PartsPartition{TotalNumOfParts: int64(numOfBlocks), TotalSize: size, Offset: poffSet, PartitionSize: partitionSize}
-		parts, partOrdinal, numOfPartsInPartition = createPartsInPartition(partitionSize, poffSet, partOrdinal, numOfBlocks, blockSize, sourceURI, targetAlias, bufferQ)
+
+		partition := PartsPartition{TotalNumOfParts: int64(numOfBlocks),
+			TotalSize:     size,
+			Offset:        poffSet,
+			PartitionSize: partitionSize}
+
+		parts, partOrdinal, numOfPartsInPartition = createPartsInPartition(partitionSize,
+			poffSet,
+			partOrdinal,
+			numOfBlocks,
+			blockSize,
+			sourceURI, targetAlias, bufferQ)
 
 		partition.Parts = parts
 		partition.NumOfParts = numOfPartsInPartition
 		Partitions[p] = partition
-
-		bytesLeft = bytesLeft - int64(partitionSize)
+	  	bytesLeft = bytesLeft - int64(partitionSize)
 	}
 
 	return Partitions
@@ -292,7 +300,7 @@ func (p *Part) MD5() string {
 }
 
 //MD5Bytes TODO
-func(p *Part) MD5Bytes()[]byte{
+func (p *Part) MD5Bytes() []byte {
 	p.MD5()
 	return p.md5Value
 }

sources/multifile.go

@@ -11,6 +11,7 @@ import (
 
 	"io"
 
+	"github.com/Azure/blobporter/internal"
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/util"
 )
@@ -21,12 +22,13 @@ import (
 
 // MultiFilePipeline  Contructs blocks queue and implements data readers
 type MultiFilePipeline struct {
-	FilesInfo           map[string]FileInfo
-	TotalNumberOfBlocks int
-	TotalSize           uint64
-	BlockSize           uint64
-	NumOfPartitions     int
+	filesInfo           map[string]FileInfo
+	totalNumberOfBlocks int
+	totalSize           uint64
+	blockSize           uint64
+	numOfPartitions     int
 	includeMD5          bool
+	handlePool          *internal.FileHandlePool
 }
 
 //FileInfo Contains the metadata associated with a file to be transferred
@@ -100,6 +102,8 @@ func NewMultiFile(params *MultiFileParams) []pipeline.SourcePipeline {
 	return pipelines
 }
 
+const maxNumOfHandlesPerFile int = 4
+
 func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint64, numOfPartitions int, md5 bool, keepDirStructure bool) pipeline.SourcePipeline {
 	totalNumberOfBlocks := 0
 	var totalSize uint64
@@ -142,19 +146,22 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 		fileInfos[files[f]] = fileInfo
 	}
 
-	return &MultiFilePipeline{FilesInfo: fileInfos,
-		TotalNumberOfBlocks: totalNumberOfBlocks,
-		BlockSize:           blockSize,
-		TotalSize:           totalSize,
-		NumOfPartitions:     numOfPartitions,
-		includeMD5:          md5}
+	handlePool := internal.NewFileHandlePool(maxNumOfHandlesPerFile, internal.Read, false)
+
+	return &MultiFilePipeline{filesInfo: fileInfos,
+		totalNumberOfBlocks: totalNumberOfBlocks,
+		blockSize:           blockSize,
+		totalSize:           totalSize,
+		numOfPartitions:     numOfPartitions,
+		includeMD5:          md5,
+		handlePool:          handlePool,
+	}
 }
 
 //ExecuteReader implements ExecuteReader from the pipeline.SourcePipeline Interface.
 //For each file the reader will maintain a open handle from which data will be read.
 // This implementation uses partitions (group of parts that can be read sequentially).
 func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
-	fileHandles := make(map[string]*os.File, len(f.FilesInfo))
 	var err error
 	var partition pipeline.PartsPartition
 
@@ -167,25 +174,30 @@ func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartiti
 		partition, ok = <-partitionsQ
 
 		if !ok {
-			for _, fh := range fileHandles {
-				fh.Close()
+			for _, finfo := range f.filesInfo {
+				err = f.handlePool.CloseHandles(finfo.SourceURI)
+				if err != nil {
+					log.Fatal(fmt.Errorf("error closing handle for file:%v. Error:%v", finfo.SourceURI, err))
+				}
 			}
 			return // no more blocks of file data to be read
 		}
 
+		//check if the partition is empty, as this may happen with small files
+		if len(partition.Parts) == 0 {
+			continue
+		}
+
 		var part pipeline.Part
 		for pip := 0; pip < len(partition.Parts); pip++ {
 			part = partition.Parts[pip]
 
-			fileURI = f.FilesInfo[part.SourceURI].SourceURI
-			fileHandle = fileHandles[fileURI]
+			fileURI = f.filesInfo[part.SourceURI].SourceURI
 
 			if fileHandle == nil {
-				if fileHandle, err = os.Open(fileURI); err != nil {
-					fmt.Printf("Error while opening the file %v \n", err)
-					log.Fatal(err)
+				if fileHandle, err = f.handlePool.GetHandle(fileURI); err != nil {
+					log.Fatal(fmt.Errorf(" error while opening the file.\nError:%v ", err))
 				}
-				fileHandles[fileURI] = fileHandle
 			}
 
 			if pip == 0 {
@@ -195,8 +207,7 @@ func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartiti
 			part.GetBuffer()
 
 			if _, err = fileHandle.Read(part.Data); err != nil && err != io.EOF {
-				fmt.Printf("Error while reading the file %v \n", err)
-				log.Fatal(err)
+				log.Fatal(fmt.Errorf(" error while reading the file.\nError:%v ", err))
 			}
 
 			util.PrintfIfDebug("ExecuteReader -> blockid:%v toread:%v name:%v read:%v ", part.BlockID, part.BytesToRead, part.TargetAlias, bytesRead)
@@ -207,16 +218,23 @@ func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartiti
 
 			readPartsQ <- part
 		}
+
+		//return handle
+		if err = f.handlePool.ReturnHandle(fileURI, fileHandle); err != nil {
+			log.Fatal(fmt.Errorf(" error returning the handle to the pool.\nPath: %v error:%v ", fileURI, err))
+		}
+
+		fileHandle = nil
 	}
 }
 
 //GetSourcesInfo implements GetSourcesInfo from the pipeline.SourcePipeline Interface.
 //Returns an an array of SourceInfo with the name, alias and size of the files to be transferred.
 func (f *MultiFilePipeline) GetSourcesInfo() []pipeline.SourceInfo {
 
-	sources := make([]pipeline.SourceInfo, len(f.FilesInfo))
+	sources := make([]pipeline.SourceInfo, len(f.filesInfo))
 	var i = 0
-	for _, file := range f.FilesInfo {
+	for _, file := range f.filesInfo {
 		sources[i] = pipeline.SourceInfo{SourceName: file.SourceURI, TargetAlias: file.TargetAlias, Size: uint64((*file.FileStats).Size())}
 		i++
 	}
@@ -255,18 +273,18 @@ func createPartsFromSource(size uint64, sourceNumOfBlocks int, blockSize uint64,
 // this implementation uses partitions to group parts into a set that can be read sequentially.
 // This is to avoid Window's memory pressure when calling SetFilePointer numerous times on the same handle
 func (f *MultiFilePipeline) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
-	numOfBlocks = f.TotalNumberOfBlocks
-	size = f.TotalSize
-	allPartitions := make([][]pipeline.PartsPartition, len(f.FilesInfo))
+	numOfBlocks = f.totalNumberOfBlocks
+	size = f.totalSize
+	allPartitions := make([][]pipeline.PartsPartition, len(f.filesInfo))
 	//size of the queue is equal to the number of partitions times the number of files to transfer.
 	//a lower value will block as this method is called before readers start
-	partitionsQ = make(chan pipeline.PartsPartition, f.NumOfPartitions*len(f.FilesInfo))
+	partitionsQ = make(chan pipeline.PartsPartition, f.numOfPartitions*len(f.filesInfo))
 	partsQ = nil
 	bufferQ := pipeline.NewBytesBufferChan(uint64(blockSize))
 	pindex := 0
 	maxpartitionNumber := 0
-	for _, source := range f.FilesInfo {
-		partitions := pipeline.ConstructPartsPartition(f.NumOfPartitions, (*source.FileStats).Size(), int64(blockSize), source.SourceURI, source.TargetAlias, bufferQ)
+	for _, source := range f.filesInfo {
+		partitions := pipeline.ConstructPartsPartition(f.numOfPartitions, (*source.FileStats).Size(), int64(blockSize), source.SourceURI, source.TargetAlias, bufferQ)
 		allPartitions[pindex] = partitions
 		if len(partitions) > maxpartitionNumber {
 			maxpartitionNumber = len(partitions)