Client should now respect the maxBatchSize setting from the server

cmd/casload/main.go

@@ -122,14 +122,19 @@ func main() {
 	}
 	defer cs.Close()
 
+	var finalMaxBatchBlobSize = *maxBatchBlobSize
+	if cs.maxBatchBlobSize > 0 && cs.maxBatchBlobSize < finalMaxBatchBlobSize {
+		finalMaxBatchBlobSize = cs.maxBatchBlobSize
+	}
+
 	actionContext := load.ActionContext{
 		InstanceName:     instanceName,
 		CasClient:        cs.casClient,
 		BytestreamClient: cs.bytestreamClient,
 		Ctx:              ctx,
 		KnownDigests:     make(map[string]bool),
 		WriteChunkSize:   *writeChunkSize,
-		MaxBatchBlobSize: *maxBatchBlobSize,
+		MaxBatchBlobSize: finalMaxBatchBlobSize,
 	}
 
 	for _, action := range actions {

pkg/load/generate.go

@@ -39,10 +39,12 @@ func (g *generateAction) String() string {
 }
 
 type generateResult struct {
-	startTime time.Time
-	endTime   time.Time
-	success   int
-	errors    int
+	startTime        time.Time
+	endTime          time.Time
+	success          int
+	errors           int
+	byteStreamWrites int
+	blobWrites       int
 }
 
 type generateWorkItem struct {
@@ -180,6 +182,12 @@ func (g *generateAction) RunAction(actionContext *ActionContext) error {
 		}
 		elapsedTimes[i] = r.elapsed
 
+		if int64(r.blobSize) < actionContext.MaxBatchBlobSize {
+			result.blobWrites += 1
+		} else {
+			result.byteStreamWrites += 1
+		}
+
 		if i%100 == 0 {
 			log.Debugf("progress: %d / %d", i, g.numRequests)
 		}
@@ -189,12 +197,14 @@ func (g *generateAction) RunAction(actionContext *ActionContext) error {
 
 	close(resultChan)
 
-	fmt.Printf("program: %s\n  startTime: %s\n  endTime: %s\n  success: %d\n  errors: %d\n",
+	fmt.Printf("program: %s\n  startTime: %s\n  endTime: %s\n  success: %d\n  errors: %d\n  byteStreamWrites: %d\n  blobWrites: %d\n",
 		g.String(),
 		result.startTime.String(),
 		result.endTime.String(),
 		result.success,
 		result.errors,
+		result.byteStreamWrites,
+		result.blobWrites,
 	)
 
 	stats.PrintTimingStats(elapsedTimes)

pkg/load/read.go

@@ -41,10 +41,12 @@ func (self *readAction) String() string {
 }
 
 type readResult struct {
-	startTime time.Time
-	endTime   time.Time
-	success   int
-	errors    int
+	startTime       time.Time
+	endTime         time.Time
+	success         int
+	errors          int
+	byteStreamReads int
+	blobReads       int
 }
 
 type readWorkItem struct {
@@ -208,6 +210,11 @@ func (self *readAction) RunAction(actionContext *ActionContext) error {
 			}).Error("request error")
 		}
 		elapsedTimes[i] = r.elapsed
+		if int64(r.digest.SizeBytes) < actionContext.MaxBatchBlobSize {
+			result.blobReads += 1
+		} else {
+			result.byteStreamReads += 1
+		}
 
 		if i%100 == 0 {
 			log.Debugf("progress: %d / %d", i, len(workItems))
@@ -218,12 +225,14 @@ func (self *readAction) RunAction(actionContext *ActionContext) error {
 
 	close(resultChan)
 
-	fmt.Printf("program: %s\n  startTime: %s\n  endTime: %s\n  success: %d\n  errors: %d\n",
+	fmt.Printf("program: %s\n  startTime: %s\n  endTime: %s\n  success: %d\n  errors: %d\n  byteStreamReads: %d\n  blobReads: %d\n",
 		self.String(),
 		result.startTime.String(),
 		result.endTime.String(),
 		result.success,
 		result.errors,
+		result.byteStreamReads,
+		result.blobReads,
 	)
 
 	stats.PrintTimingStats(elapsedTimes)