- Implementation of the performance pipelines for benchmarking

Author: giventocode

blobporter.go

@@ -12,6 +12,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/Azure/blobporter/sources"
+
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/transfer"
 	"github.com/Azure/blobporter/util"
@@ -37,6 +39,7 @@ var storageClientHTTPTimeout int
 
 var sourceParameters map[string]string
 var sourceAuthorization string
+var perfSourceDefinitions []sources.SourceDefinition
 
 var quietMode bool
 var calculateMD5 bool
@@ -54,7 +57,7 @@ const (
 	s3AccessKeyEnvVar = "S3_ACCESS_KEY"
 	s3SecretKeyEnvVar = "S3_SECRET_KEY"
 
-	programVersion = "0.5.23" // version number to show in help
+	programVersion = "0.5.24" // version number to show in help
 )
 
 const numOfWorkersFactor = 8

pipelinefactory.go

@@ -125,6 +125,24 @@ func getPipelines() ([]pipeline.SourcePipeline, pipeline.TargetPipeline, error)
 			pvSourceInfoForS3IsReq,
 			pvContainerIsReq,
 			pvBlockSizeCheckForBlockBlobs)
+	case transfer.PerfToBlock:
+		return getTransferPipelines(getPerfSourceToBlockPipelines,
+			pvBlobAuthInfoIsReq,
+			pvContainerIsReq,
+			pvSourceURIISReq,
+			pvBlockSizeCheckForBlockBlobs,
+			pvPerfSourceIsReq)
+	case transfer.PerfToPage:
+		return getTransferPipelines(getPerfSourceToPagePipelines,
+			pvBlobAuthInfoIsReq,
+			pvContainerIsReq,
+			pvSourceURIISReq,
+			pvBlockSizeCheckForPageBlobs,
+			pvPerfSourceIsReq)
+	case transfer.BlobToPerf:
+		return getTransferPipelines(getBlobToPerfPipelines,
+			pvSourceURIISReq,
+			pvSourceInfoForBlobIsReq)
 
 	}
 
@@ -245,6 +263,53 @@ func getBlobToPagePipelines() (source []pipeline.SourcePipeline, target pipeline
 	target = targets.NewAzurePage(storageAccountName, storageAccountKey, containerName)
 	return
 }
+func getPerfSourceToPagePipelines() (source []pipeline.SourcePipeline, target pipeline.TargetPipeline, err error) {
+
+	defs := perfSourceDefinitions
+
+	params := sources.PerfSourceParams{
+		Definitions: defs,
+		SourceParams: sources.SourceParams{
+			CalculateMD5: calculateMD5}}
+
+	source = sources.NewPerfSourcePipeline(params, blockSize)
+	target = targets.NewAzurePage(storageAccountName, storageAccountKey, containerName)
+	return
+}
+
+func getPerfSourceToBlockPipelines() (source []pipeline.SourcePipeline, target pipeline.TargetPipeline, err error) {
+
+	defs := perfSourceDefinitions
+
+	params := sources.PerfSourceParams{
+		Definitions: defs,
+		SourceParams: sources.SourceParams{
+			CalculateMD5: calculateMD5}}
+
+	source = sources.NewPerfSourcePipeline(params, blockSize)
+	target = targets.NewAzureBlock(storageAccountName, storageAccountKey, containerName)
+	return
+}
+func getBlobToPerfPipelines() (source []pipeline.SourcePipeline, target pipeline.TargetPipeline, err error) {
+
+	blobNames = []string{sourceParameters["PREFIX"]}
+
+	params := &sources.AzureBlobParams{
+		Container:   sourceParameters["CONTAINER"],
+		BlobNames:   blobNames,
+		AccountName: sourceParameters["ACCOUNT_NAME"],
+		AccountKey:  sourceAuthorization,
+		SourceParams: sources.SourceParams{
+			CalculateMD5:      calculateMD5,
+			UseExactNameMatch: exactNameMatch,
+			FilesPerPipeline:  numberOfFilesInBatch,
+			//default to always true so blob names are kept
+			KeepDirStructure: true}}
+
+	source = sources.NewAzureBlob(params)
+	target = targets.NewPerfTargetPipeline()
+	return
+}
 func getBlobToBlockPipelines() (source []pipeline.SourcePipeline, target pipeline.TargetPipeline, err error) {
 
 	blobNames = []string{sourceParameters["PREFIX"]}
@@ -521,6 +586,17 @@ func pvContainerIsReq() error {
 	return nil
 }
 
+func pvPerfSourceIsReq() error {
+	var err error
+	perfSourceDefinitions, err = sources.ParseSourceDefinitions(sourceURIs[0])
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func pvBlobAuthInfoIsReq() error {
 
 	// wasn't specified, try the environment variable

sources/perfsource.go

@@ -0,0 +1,179 @@
+package sources
+
+import (
+	"fmt"
+	"math/rand"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Azure/blobporter/pipeline"
+	"github.com/Azure/blobporter/util"
+)
+
+//SourceDefinition TODO
+type SourceDefinition struct {
+	Names           []string
+	Size            uint64
+	NumberOfSources int
+}
+
+//ParseSourceDefinitions TODO
+func ParseSourceDefinitions(def string) ([]SourceDefinition, error) {
+	data := strings.Split(def, ";")
+	defs := make([]SourceDefinition, len(data))
+	for s := 0; s < len(data); s++ {
+		def, err := newSourceDefinition(data[s])
+		if err != nil {
+			return nil, err
+		}
+		defs[s] = *def
+	}
+	return defs, nil
+}
+
+func newSourceDefinition(def string) (*SourceDefinition, error) {
+	data := strings.Split(def, ":")
+
+	if len(data) != 2 {
+		return nil, fmt.Errorf("Invalid format: %s\nThe source definition must be [Size]:[NumOfSources]", data)
+	}
+
+	size, err := util.ByteCountFromSizeString(data[0])
+	if err != nil {
+		return nil, fmt.Errorf("Invalid format. The source definition must be [Size]:[NumOfSources]. The size is invalid. Error:%v", err)
+	}
+	var numOfSrcs int
+	numOfSrcs, err = strconv.Atoi(data[1])
+	if err != nil {
+		return nil, fmt.Errorf("Invalid format. The source definition must be [Size]:[NumOfSources]. Failed to parse the number of sources. Error:%v", err)
+	}
+	names := make([]string, numOfSrcs)
+	for n := 0; n < numOfSrcs; n++ {
+		names[n] = fmt.Sprintf("%s%v.dat", strings.Replace(def,":","_",1), time.Now().Nanosecond())
+	}
+
+	return &SourceDefinition{Size: size, NumberOfSources: numOfSrcs, Names: names}, nil
+}
+
+func (d *SourceDefinition) getSourceInfo() []pipeline.SourceInfo {
+	infos := make([]pipeline.SourceInfo, len(d.Names))
+	for i, n := range d.Names {
+		infos[i] = pipeline.SourceInfo{SourceName: n, TargetAlias: n, Size: d.Size}
+	}
+	return infos
+}
+
+//PerfSourcePipeline TODO
+type PerfSourcePipeline struct {
+	definitions []SourceDefinition
+	blockSize   uint64
+	dataBlock   []byte
+	includeMD5  bool
+}
+
+//PerfSourceParams TODO
+type PerfSourceParams struct {
+	SourceParams
+	Definitions []SourceDefinition
+}
+
+//NewPerfSourcePipeline TODO
+func NewPerfSourcePipeline(params PerfSourceParams, blockSize uint64) []pipeline.SourcePipeline {
+	ssps := make([]pipeline.SourcePipeline, 1)
+	ssp := PerfSourcePipeline{
+		definitions: params.Definitions,
+		blockSize:   blockSize,
+		includeMD5:  params.CalculateMD5}
+	ssp.setSharedDataBlock()
+	ssps[0] = &ssp
+	return ssps
+}
+
+//ConstructBlockInfoQueue TODO
+func (s *PerfSourcePipeline) ConstructBlockInfoQueue(blockSize uint64) (partitionQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
+	allParts := make([][]pipeline.Part, len(s.definitions))
+	//disable memory buffer for parts (bufferQ == nil)
+	var bufferQ chan []byte
+	largestNumOfParts := 0
+	for i, source := range s.definitions {
+		srcParts := make([]pipeline.Part, 0)
+		for _, name := range source.Names {
+			size = size + source.Size
+			parts, sourceNumOfBlocks := pipeline.ConstructPartsQueue(source.Size, blockSize, name, name, bufferQ)
+			srcParts = append(srcParts, parts...)
+			numOfBlocks = numOfBlocks + sourceNumOfBlocks
+		}
+		allParts[i] = srcParts
+		if largestNumOfParts < len(srcParts) {
+			largestNumOfParts = len(srcParts)
+		}
+	}
+
+	partsQ = make(chan pipeline.Part, numOfBlocks)
+
+	for i := 0; i < largestNumOfParts; i++ {
+		for _, ps := range allParts {
+			if i < len(ps) {
+				partsQ <- ps[i]
+			}
+		}
+	}
+
+	close(partsQ)
+
+	return
+}
+
+//ExecuteReader TODO
+func (s *PerfSourcePipeline) ExecuteReader(partitionQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
+	var blocksHandled = 0
+	defer wg.Done()
+	for {
+		p, ok := <-partsQ
+
+		if !ok {
+			return // no more blocks of file data to be read
+		}
+		if s.includeMD5 {
+			p.MD5()
+		}
+
+		p.Data = s.dataBlock
+
+		if len(s.dataBlock) != int(p.BytesToRead) {
+			small := s.dataBlock[:p.BytesToRead]
+			p.Data = small
+		}
+
+		readPartsQ <- p
+		blocksHandled++
+	}
+}
+
+//GetSourcesInfo TODO
+func (s *PerfSourcePipeline) GetSourcesInfo() []pipeline.SourceInfo {
+	srcInfo := make([]pipeline.SourceInfo, 0)
+	for _, def := range s.definitions {
+		srcInfo = append(srcInfo, def.getSourceInfo()...)
+	}
+	return srcInfo
+}
+
+var data = []byte("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ")
+var block []byte
+
+func (s *PerfSourcePipeline) setSharedDataBlock() {
+	s.dataBlock = s.getDataBlock(s.blockSize)
+}
+
+func (s *PerfSourcePipeline) getDataBlock(size uint64) []byte {
+	b := make([]byte, size)
+	ln := len(data)
+	for i := 0; i < len(b); i++ {
+		b[i] = data[rand.Intn(ln)]
+	}
+
+	return b
+}

targets/perftarget.go

@@ -0,0 +1,51 @@
+package targets
+
+import (
+	"time"
+
+	"github.com/Azure/blobporter/pipeline"
+)
+
+////////////////////////////////////////////////////////////
+///// Perf Target
+////////////////////////////////////////////////////////////
+
+//PerfTarget TODO
+type PerfTarget struct {
+}
+
+//NewPerfTargetPipeline creates a new Azure Block target
+func NewPerfTargetPipeline() pipeline.TargetPipeline {
+	return &PerfTarget{}
+}
+
+//PreProcessSourceInfo TODO.
+func (t *PerfTarget) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
+	return nil
+}
+
+//CommitList TODO
+func (t *PerfTarget) CommitList(listInfo *pipeline.TargetCommittedListInfo, NumberOfBlocks int, targetName string) (msg string, err error) {
+	msg = "Perf test committed"
+	err = nil
+	return
+}
+
+//ProcessWrittenPart TODO
+func (t *PerfTarget) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo *pipeline.TargetCommittedListInfo) (requeue bool, err error) {
+	requeue = false
+	err = nil
+	return
+}
+
+//WritePart TODO
+func (t *PerfTarget) WritePart(part *pipeline.Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error) {
+
+	s := time.Now()
+	numOfRetries = 0
+	err = nil
+	startTime = s
+	duration = time.Now().Sub(s)
+
+	return
+}

transfer/transfer.go

@@ -84,6 +84,9 @@ const (
 	BlobToPage             = "blob-pageblob"
 	S3ToBlock              = "s3-blockblob"
 	S3ToPage               = "s3-pageblob"
+	PerfToBlock            = "perf-block"
+	PerfToPage             = "perf-page"
+	BlobToPerf             = "blob-perf"
 	none                   = "none"
 )
 
@@ -121,6 +124,12 @@ func ParseTransferDefinition(str string) (Definition, error) {
 		return S3ToBlock, nil
 	case "s3-pageblob":
 		return S3ToPage, nil
+	case "perf-block":
+		return PerfToBlock, nil
+	case "perf-page":
+		return PerfToPage, nil
+	case "blob-perf":
+		return BlobToPerf, nil
 	default:
 		return none, fmt.Errorf("%v is not a valid transfer definition value.\n Valid values: file-blockblob, http-blockblob,file-pageblob, http-pageblob, pageblob-file, blockblob-file, http-file", str)
 	}