Merge pull request #259 from filipecosta90/core.and.redis.improvements

Redis and DynamoDB further improvements/tuning parameters

Author: dbsid

README.md

@@ -330,6 +330,9 @@ Common configurations:
 |dynamodb.ensure.clean.table|true|On load mode ensure that the table is clean at the begining. In case of true and if the table previously exists it will be deleted and recreated|
 |dynamodb.endpoint|""|Used endpoint for connection. If empty will use the default loaded configs|
 |dynamodb.region|""|Used region for connection ( should match endpoint ). If empty will use the default loaded configs|
+|dynamodb.consistent.reads|false|Reads on DynamoDB provide an eventually consistent read by default. If your benchmark/use-case requires a strongly consistent read, set this option to true|
+|dynamodb.delete.after.run.stage|false|Detele the database table after the run stage|
+
 
 
 ## TODO

db/dynamodb/db.go

@@ -25,10 +25,23 @@ type dynamodbWrapper struct {
 	primarykeyPtr      *string
 	readCapacityUnits  int64
 	writeCapacityUnits int64
+	consistentRead     bool
+	deleteAfterRun     bool
+	command            string
 }
 
 func (r *dynamodbWrapper) Close() error {
-	return nil
+	var err error = nil
+	if strings.Compare("run", r.command) == 0 {
+		log.Printf("Ensuring that the table is deleted after the run stage...\n")
+		if r.deleteAfterRun {
+			err = r.deleteTable()
+			if err != nil {
+				log.Printf("Couldn't delete table after run. Here's why: %v\n", err)
+			}
+		}
+	}
+	return err
 }
 
 func (r *dynamodbWrapper) InitThread(ctx context.Context, _ int, _ int) context.Context {
@@ -42,8 +55,9 @@ func (r *dynamodbWrapper) Read(ctx context.Context, table string, key string, fi
 	data = make(map[string][]byte, len(fields))
 
 	response, err := r.client.GetItem(context.TODO(), &dynamodb.GetItemInput{
-		Key:       r.GetKey(key),
-		TableName: r.tablename,
+		Key:            r.GetKey(key),
+		TableName:      r.tablename,
+		ConsistentRead: aws.Bool(r.consistentRead),
 	})
 	if err != nil {
 		log.Printf("Couldn't get info about %v. Here's why: %v\n", key, err)
@@ -180,9 +194,11 @@ func (r dynamoDbCreator) Create(p *properties.Properties) (ycsb.DB, error) {
 	rds.primarykeyPtr = aws.String(rds.primarykey)
 	rds.readCapacityUnits = p.GetInt64(readCapacityUnitsFieldName, readCapacityUnitsFieldNameDefault)
 	rds.writeCapacityUnits = p.GetInt64(writeCapacityUnitsFieldName, writeCapacityUnitsFieldNameDefault)
+	rds.consistentRead = p.GetBool(consistentReadFieldName, consistentReadFieldNameDefault)
+	rds.deleteAfterRun = p.GetBool(deleteTableAfterRunFieldName, deleteTableAfterRunFieldNameDefault)
 	endpoint := p.GetString(endpointField, endpointFieldDefault)
 	region := p.GetString(regionField, regionFieldDefault)
-	command, _ := p.Get(prop.Command)
+	rds.command, _ = p.Get(prop.Command)
 	var err error = nil
 	var cfg aws.Config
 	if strings.Contains(endpoint, "localhost") && strings.Compare(region, "localhost") != 0 {
@@ -213,7 +229,7 @@ func (r dynamoDbCreator) Create(p *properties.Properties) (ycsb.DB, error) {
 	rds.client = dynamodb.NewFromConfig(cfg)
 	exists, err := rds.tableExists()
 
-	if strings.Compare("load", command) == 0 {
+	if strings.Compare("load", rds.command) == 0 {
 		if !exists {
 			_, err = rds.createTable()
 		} else {
@@ -265,6 +281,13 @@ const (
 	endpointFieldDefault               = ""
 	regionField                        = "dynamodb.region"
 	regionFieldDefault                 = ""
+	// GetItem provides an eventually consistent read by default.
+	// If your application requires a strongly consistent read, set ConsistentRead to true.
+	// Although a strongly consistent read might take more time than an eventually consistent read, it always returns the last updated value.
+	consistentReadFieldName             = "dynamodb.consistent.reads"
+	consistentReadFieldNameDefault      = false
+	deleteTableAfterRunFieldName        = "dynamodb.delete.after.run.stage"
+	deleteTableAfterRunFieldNameDefault = false
 )
 
 func init() {

db/redis/db.go

@@ -264,6 +264,7 @@ const (
 	redisReadTimeout           = "redis.read_timeout"
 	redisWriteTimeout          = "redis.write_timeout"
 	redisPoolSize              = "redis.pool_size"
+	redisPoolSizeDefault       = 0
 	redisMinIdleConns          = "redis.min_idle_conns"
 	redisMaxConnAge            = "redis.max_conn_age"
 	redisPoolTimeout           = "redis.pool_timeout"
@@ -292,23 +293,28 @@ func parseTLS(p *properties.Properties) *tls.Config {
 
 func getOptionsSingle(p *properties.Properties) *goredis.Options {
 	opts := &goredis.Options{}
-	opts.Network = p.GetString(redisNetwork, redisNetworkDefault)
+
 	opts.Addr = p.GetString(redisAddr, redisAddrDefault)
-	opts.Password, _ = p.Get(redisPassword)
 	opts.DB = p.GetInt(redisDB, 0)
+	opts.Network = p.GetString(redisNetwork, redisNetworkDefault)
+	opts.Password, _ = p.Get(redisPassword)
 	opts.MaxRetries = p.GetInt(redisMaxRetries, 0)
 	opts.MinRetryBackoff = p.GetDuration(redisMinRetryBackoff, time.Millisecond*8)
 	opts.MaxRetryBackoff = p.GetDuration(redisMaxRetryBackoff, time.Millisecond*512)
 	opts.DialTimeout = p.GetDuration(redisDialTimeout, time.Second*5)
 	opts.ReadTimeout = p.GetDuration(redisReadTimeout, time.Second*3)
 	opts.WriteTimeout = p.GetDuration(redisWriteTimeout, opts.ReadTimeout)
-	opts.PoolSize = p.GetInt(redisPoolSize, 10)
+	opts.PoolSize = p.GetInt(redisPoolSize, redisPoolSizeDefault)
+	threadCount := p.MustGetInt("threadcount")
+	if opts.PoolSize == 0 {
+		opts.PoolSize = threadCount
+		fmt.Println(fmt.Sprintf("Setting %s=%d (from <threadcount>) given you haven't specified a value.", redisPoolSize, opts.PoolSize))
+	}
 	opts.MinIdleConns = p.GetInt(redisMinIdleConns, 0)
 	opts.MaxConnAge = p.GetDuration(redisMaxConnAge, 0)
 	opts.PoolTimeout = p.GetDuration(redisPoolTimeout, time.Second+opts.ReadTimeout)
 	opts.IdleTimeout = p.GetDuration(redisIdleTimeout, time.Minute*5)
 	opts.IdleCheckFrequency = p.GetDuration(redisIdleCheckFreq, time.Minute)
-
 	opts.TLSConfig = parseTLS(p)
 
 	return opts
@@ -330,7 +336,12 @@ func getOptionsCluster(p *properties.Properties) *goredis.ClusterOptions {
 	opts.DialTimeout = p.GetDuration(redisDialTimeout, time.Second*5)
 	opts.ReadTimeout = p.GetDuration(redisReadTimeout, time.Second*3)
 	opts.WriteTimeout = p.GetDuration(redisWriteTimeout, opts.ReadTimeout)
-	opts.PoolSize = p.GetInt(redisPoolSize, 10)
+	opts.PoolSize = p.GetInt(redisPoolSize, redisPoolSizeDefault)
+	threadCount := p.MustGetInt("threadcount")
+	if opts.PoolSize == 0 {
+		opts.PoolSize = threadCount
+		fmt.Println(fmt.Sprintf("Setting %s=%d (from <threadcount>) given you haven't specified a value.", redisPoolSize, opts.PoolSize))
+	}
 	opts.MinIdleConns = p.GetInt(redisMinIdleConns, 0)
 	opts.MaxConnAge = p.GetDuration(redisMaxConnAge, 0)
 	opts.PoolTimeout = p.GetDuration(redisPoolTimeout, time.Second+opts.ReadTimeout)

pkg/workload/core.go

@@ -408,10 +408,7 @@ func (c *core) nextKeyNum(state *coreState) int64 {
 			keyNum = c.transactionInsertKeySequence.Last() - c.keyChooser.Next(r)
 		}
 	} else {
-		keyNum = math.MaxInt64
-		for keyNum > c.transactionInsertKeySequence.Last() {
-			keyNum = c.keyChooser.Next(r)
-		}
+		keyNum = c.keyChooser.Next(r)
 	}
 	return keyNum
 }
@@ -650,31 +647,35 @@ func (coreCreator) Create(p *properties.Properties) (ycsb.Workload, error) {
 
 	c.keySequence = generator.NewCounter(insertStart)
 	c.operationChooser = createOperationGenerator(p)
+	var keyrangeLowerBound int64 = insertStart
+	var keyrangeUpperBound int64 = insertStart + insertCount - 1
 
 	c.transactionInsertKeySequence = generator.NewAcknowledgedCounter(c.recordCount)
 	switch requestDistrib {
 	case "uniform":
-		c.keyChooser = generator.NewUniform(insertStart, insertStart+insertCount-1)
+		c.keyChooser = generator.NewUniform(keyrangeLowerBound, keyrangeUpperBound)
 	case "sequential":
-		c.keyChooser = generator.NewSequential(insertStart, insertStart+insertCount-1)
+		c.keyChooser = generator.NewSequential(keyrangeLowerBound, keyrangeUpperBound)
 	case "zipfian":
 		insertProportion := p.GetFloat64(prop.InsertProportion, prop.InsertProportionDefault)
 		opCount := p.GetInt64(prop.OperationCount, 0)
 		expectedNewKeys := int64(float64(opCount) * insertProportion * 2.0)
-		c.keyChooser = generator.NewScrambledZipfian(insertStart, insertStart+insertCount+expectedNewKeys, generator.ZipfianConstant)
+		keyrangeUpperBound = insertStart + insertCount + expectedNewKeys
+		c.keyChooser = generator.NewScrambledZipfian(keyrangeLowerBound, keyrangeUpperBound, generator.ZipfianConstant)
 	case "latest":
 		c.keyChooser = generator.NewSkewedLatest(c.transactionInsertKeySequence)
 	case "hotspot":
 		hotsetFraction := p.GetFloat64(prop.HotspotDataFraction, prop.HotspotDataFractionDefault)
 		hotopnFraction := p.GetFloat64(prop.HotspotOpnFraction, prop.HotspotOpnFractionDefault)
-		c.keyChooser = generator.NewHotspot(insertStart, insertStart+insertCount-1, hotsetFraction, hotopnFraction)
+		c.keyChooser = generator.NewHotspot(keyrangeLowerBound, keyrangeUpperBound, hotsetFraction, hotopnFraction)
 	case "exponential":
 		percentile := p.GetFloat64(prop.ExponentialPercentile, prop.ExponentialPercentileDefault)
 		frac := p.GetFloat64(prop.ExponentialFrac, prop.ExponentialFracDefault)
 		c.keyChooser = generator.NewExponential(percentile, float64(c.recordCount)*frac)
 	default:
 		util.Fatalf("unknown request distribution %s", requestDistrib)
 	}
+	fmt.Println(fmt.Sprintf("Using request distribution '%s' a keyrange of [%d %d]", requestDistrib, keyrangeLowerBound, keyrangeUpperBound))
 
 	c.fieldChooser = generator.NewUniform(0, c.fieldCount-1)
 	switch scanLengthDistrib {