- removes dir from the sources

- fix default not working for transfer type
- new handle pool without locks

Author: giventocode

args.go

@@ -129,6 +129,7 @@ func newParamParserValidator() paramParserValidator {
 	}
 
 	args := &arguments{
+		keepDirStructure: 		true,
 		numberOfReaders:        defaultNumberOfReaders,
 		numberOfWorkers:        defaultNumberOfWorkers,
 		blockSizeStr:           defaultBlockSizeStr,
@@ -163,7 +164,8 @@ func (p *paramParserValidator) parseAndValidate() error {
 		p.pvgHTTPTimeOut,
 		p.pvgDupCheck,
 		p.pvgParseBlockSize,
-		p.pvgQuietMode)
+		p.pvgQuietMode,
+		p.pvgKeepDirectoryStructure)
 
 	if err != nil {
 		return err
@@ -246,6 +248,10 @@ func (p *paramParserValidator) getSourceRules() ([]parseAndValidationRule, error
 //**************************
 
 //Global rules....
+func (p *paramParserValidator) pvgKeepDirectoryStructure() error {
+	p.params.keepDirStructure = p.args.keepDirStructure
+	return nil
+}
 func (p *paramParserValidator) pvgQuietMode() error {
 	p.params.quietMode = p.args.quietMode
 	return nil

sources/multifile.go

@@ -104,7 +104,7 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 	totalNumberOfBlocks := 0
 	var totalSize uint64
 	var err error
-	fileInfos := make(map[string]FileInfo, len(files))
+	fileInfos := make(map[string]FileInfo)
 	useTargetAlias := len(targetAliases) == len(files)
 	for f := 0; f < len(files); f++ {
 		var fileStat os.FileInfo
@@ -114,6 +114,11 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 			log.Fatalf("Error: %v", err)
 		}
 
+		//directories are not allowed... so skipping them
+		if fileStat.IsDir() {
+			continue
+		}
+
 		if fileStat.Size() == 0 {
 			log.Fatalf("Empty files are not allowed. The file %v is empty", files[f])
 		}

targets/azureblock.go

@@ -88,13 +88,6 @@ func (t *AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize
 		return fmt.Errorf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
 	}
 
-	//clean uncommitted blocks for large files only (>1GB)
-	/*
-		if source.Size > util.GB {
-			return t.azutil.CleanUncommittedBlocks(source.TargetAlias)
-		}
-	*/
-
 	return nil
 }
 
@@ -126,7 +119,7 @@ func (t *AzureBlock) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo
 func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error) {
 	startTime = time.Now()
 	defer func() { duration = time.Now().Sub(startTime) }()
-	defer util.PrintfIfDebug("WritePart -> blockid:%v read:%v name:%v err:%v", part.BlockID, len(part.Data), part.TargetAlias, err)
+	util.PrintfIfDebug("WritePart -> blockid:%v read:%v name:%v err:%v", part.BlockID, len(part.Data), part.TargetAlias, err)
 	//computation of the MD5 happens is done by the readers.
 	var md5 []byte
 	if part.IsMD5Computed() {
@@ -140,6 +133,8 @@ func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, sta
 		return
 	}
 	reader := bytes.NewReader(part.Data)
+	util.PrintfIfDebug("WritePart -> blockid:%v reader:%v name:%v err:%v", part.BlockID, reader.Len(), part.TargetAlias, err)
+
 	err = t.azutil.PutBlock(t.container, part.TargetAlias, part.BlockID, reader)
 	return
 }

targets/azurepage.go

@@ -27,7 +27,12 @@ func NewAzurePagePipeline(params AzureTargetParams) pipeline.TargetPipeline {
 		log.Fatal(err)
 	}
 
-	_, err = az.CreateContainerIfNotExists()
+	var notfound bool
+	notfound, err = az.CreateContainerIfNotExists()
+
+	if notfound {
+		fmt.Printf("Info! Container was not found, creating it...\n")
+	}
 
 	if err != nil {
 		log.Fatal(err)

targets/handleman.go

@@ -0,0 +1,248 @@
+package targets
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+)
+
+type fileHandleFactory struct {
+	cacheSize      int
+	maxCacheSize   int
+	handleProvider handleProvider
+	handleReq      chan factoryRequest
+	closeReq       chan factoryCloseRequest
+	returnReq      chan factoryReturnRequest
+	fileHandles    map[string]*os.File
+	overwrite      bool
+}
+type factoryCloseRequest struct {
+	path string
+	err  chan error
+}
+type factoryReturnRequest struct {
+	path   string
+	handle *os.File
+}
+type factoryRequest struct {
+	path     string
+	response chan factoryResponse
+}
+
+type factoryResponse struct {
+	handle *os.File
+	err    error
+}
+
+func (f *fileHandleFactory) startFactory() {
+
+	go func() {
+		for {
+			select {
+			case req, ok := <-f.handleReq:
+				var err error
+
+				if !ok {
+					break
+				}
+
+				fh, exists := f.fileHandles[req.path]
+
+				if !exists {
+					responseChan := make(chan handleProviderResponse, 1)
+					f.handleProvider.handleReq <- handleProviderRequest{path: req.path,
+						response: responseChan}
+					resp := <-responseChan
+					fh = resp.handle
+					err = resp.err
+				} else {
+					delete(f.fileHandles, req.path)
+				}
+
+				resp := factoryResponse{handle: fh, err: err}
+
+				select {
+				case req.response <- resp:
+				default:
+				}
+
+			case ret := <-f.returnReq:
+				var err error
+				if f.cacheSize < f.maxCacheSize {
+					f.fileHandles[ret.path] = ret.handle
+					f.cacheSize++
+				} else {
+					err = ret.handle.Close()
+
+					if err != nil {
+						log.Fatalf("The handle failed close. Err:%v", err)
+					}
+				}
+
+			case close := <-f.closeReq:
+				var err error
+				if fh, ok := f.fileHandles[close.path]; ok {
+
+					err = fh.Close()
+					delete(f.fileHandles, close.path)
+					f.cacheSize--
+				}
+
+				select {
+				case close.err <- err:
+				default:
+				}
+
+			}
+		}
+	}()
+}
+
+type poolHandlerManager struct {
+	factories []fileHandleFactory
+	handleReq chan factoryRequest
+	returnReq chan factoryReturnRequest
+}
+
+func newpoolHandlerManager(numOfHandlesPerFile int, numberOfHandlersInCache int, overwrite bool) *poolHandlerManager {
+	factories := make([]fileHandleFactory, numOfHandlesPerFile)
+	handleReq := make(chan factoryRequest, 100)
+	returnReq := make(chan factoryReturnRequest, 100)
+	handleProvider := newhandleProvider()
+	handleProvider.startProvider(overwrite)
+
+	for i := 0; i < numOfHandlesPerFile; i++ {
+		//close request is unique for factory as handles need to close in all of them
+		closeReq := make(chan factoryCloseRequest, 100)
+
+		factory := fileHandleFactory{
+			maxCacheSize:   numberOfHandlersInCache / numOfHandlesPerFile,
+			overwrite:      overwrite,
+			handleReq:      handleReq,
+			returnReq:      returnReq,
+			fileHandles:    make(map[string]*os.File),
+			handleProvider: handleProvider,
+			closeReq:       closeReq}
+
+		factory.startFactory()
+
+		factories[i] = factory
+	}
+
+	return &poolHandlerManager{factories: factories, handleReq: handleReq, returnReq: returnReq}
+}
+func (p *poolHandlerManager) getHandle(path string) (*os.File, error) {
+	respChan := make(chan factoryResponse, 1)
+	req := factoryRequest{path: path, response: respChan}
+
+	p.handleReq <- req
+
+	response := <-respChan
+
+	return response.handle, response.err
+}
+
+func (p *poolHandlerManager) returnHandle(path string, fileHandle *os.File) error {
+	retrequest := factoryReturnRequest{path: path, handle: fileHandle}
+	p.returnReq <- retrequest
+	return nil
+}
+
+func (p *poolHandlerManager) closeCacheHandles(path string) error {
+
+	for _, factory := range p.factories {
+		errch := make(chan error, 1)
+		closerequest := factoryCloseRequest{path: path, err: errch}
+		factory.closeReq <- closerequest
+		err := <-errch
+
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type handleProvider struct {
+	init      map[string]bool
+	handleReq chan handleProviderRequest
+}
+
+type handleProviderRequest struct {
+	path     string
+	response chan handleProviderResponse
+}
+
+type handleProviderResponse struct {
+	handle *os.File
+	err    error
+}
+
+func newhandleProvider() handleProvider {
+	reqChan := make(chan handleProviderRequest, 100)
+	return handleProvider{
+		init:      make(map[string]bool),
+		handleReq: reqChan,
+	}
+}
+
+func (h *handleProvider) startProvider(overwrite bool) {
+	go func() {
+		for {
+			req, ok := <-h.handleReq
+
+			if !ok {
+				return
+			}
+
+			_, exists := h.init[req.path]
+			var fh *os.File
+			var err error
+			if !exists {
+				fh, err = h.initFile(req.path, overwrite)
+			} else {
+				fh, err = os.OpenFile(req.path, os.O_WRONLY, os.ModeAppend)
+			}
+
+			select {
+			case req.response <- handleProviderResponse{handle: fh, err: err}:
+			default:
+			}
+
+			h.init[req.path] = true
+
+		}
+	}()
+}
+
+func (h *handleProvider) initFile(filePath string, overwrite bool) (*os.File, error) {
+	var fh *os.File
+	var err error
+
+	path := filepath.Dir(filePath)
+
+	if path != "" {
+		err = os.MkdirAll(path, 0777)
+
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if _, err = os.Stat(filePath); os.IsExist(err) || !overwrite {
+		return nil, fmt.Errorf("The file already exists and file overwrite is disabled")
+	}
+
+	if fh, err = os.Create(filePath); os.IsExist(err) {
+		if err = os.Remove(filePath); err != nil {
+			return nil, err
+		}
+
+		if fh, err = os.Create(filePath); err != nil {
+			return nil, err
+		}
+	}
+
+	return fh, nil
+}