Fixes build given the new reply tracking changes

.circleci/config.yml

@@ -1,38 +1,6 @@
 version: 2.1
 
-executors:
-  edge:
-    docker:
-      - image: circleci/golang:1.13
-      - image: redislabs/redisearch:edge
-  latest:
-    docker:
-      - image: circleci/golang:1.13
-      - image: redislabs/redisearch:latest
-
 jobs:
-  ci-benchmark:
-    parameters:
-      redisearch_version:
-        type: executor
-      use_case:
-        type: string
-    executor: << parameters.redisearch_version >>
-    steps:
-      - checkout
-      - run: make
-      - run: sudo apt install python3.6 -y
-      - run: sudo apt install python3-pip -y
-      - run: python3 -m pip install wheel redisbench-admin==0.1.13
-      - run:
-          name: << parameters.use_case >> use case
-          command: |
-            redisbench-admin run \
-             --repetitions 3 \
-             --output-file-prefix circleci \
-             --upload-results-s3 \
-             --benchmark-config-file https://s3.amazonaws.com/benchmarks.redislabs/redisearch/datasets/<< parameters.use_case >>/<< parameters.use_case >>.redisearch.cfg.json
-
   build-edge: # test with redisearch:edge
     docker:
       - image: circleci/golang:1.13
@@ -54,63 +22,7 @@ jobs:
 workflows:
   commit:
     jobs:
+      - build-latest
       - build-edge:
           requires:
             - build-latest
-      - build-latest
-      - ci-benchmark:
-          name: edge-ecommerce-inventory
-          redisearch_version: edge
-          use_case: "ecommerce-inventory"
-          requires:
-            - build-edge
-      - ci-benchmark:
-          name: edge-nyc_taxis-ft.add
-          redisearch_version: edge
-          use_case: "nyc_taxis-ft.add"
-          requires:
-            - edge-ecommerce-inventory
-      - ci-benchmark:
-          name: edge-nyc_taxis-hashes
-          redisearch_version: edge
-          use_case: "nyc_taxis-hashes"
-          requires:
-            - edge-nyc_taxis-ft.add
-      - ci-benchmark:
-          name: edge-enwiki_abstract-hashes
-          redisearch_version: edge
-          use_case: "enwiki_abstract-hashes"
-          requires:
-            - edge-nyc_taxis-hashes
-
-  ci_benchmarks:
-    triggers:
-      - schedule:
-          cron: "0 0,6,12,18 * * *"
-          filters:
-            branches:
-              only:
-                - master
-    jobs:
-      - ci-benchmark:
-          name: edge-ecommerce-inventory
-          redisearch_version: edge
-          use_case: "ecommerce-inventory"
-      - ci-benchmark:
-          name: edge-nyc_taxis-ft.add
-          redisearch_version: edge
-          use_case: "nyc_taxis-ft.add"
-          requires:
-            - edge-ecommerce-inventory
-      - ci-benchmark:
-          name: edge-nyc_taxis-hashes
-          redisearch_version: edge
-          use_case: "nyc_taxis-hashes"
-          requires:
-            - edge-nyc_taxis-ft.add
-      - ci-benchmark:
-          name: edge-enwiki_abstract-hashes
-          redisearch_version: edge
-          use_case: "enwiki_abstract-hashes"
-          requires:
-            - edge-nyc_taxis-hashes

cmd/ftsb_redisearch/cmd_processor.go

@@ -66,6 +66,7 @@ func (p *processor) Init(workerNumber int, _ bool, totalWorkers int) {
 func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter bool) {
 	cmdSlots := make([][]radix.CmdAction, 0, 0)
 	timesSlots := make([][]time.Time, 0, 0)
+	replies := make([]interface{}, 0, 0)
 	clusterSlots := make([][2]uint16, 0, 0)
 	clusterAddr := make([]string, 0, 0)
 
@@ -98,10 +99,10 @@ func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter
 			time.Sleep(r.Delay())
 		}
 		if !clusterMode {
-			cmdSlots[slotP], timesSlots[slotP] = sendFlatCmd(p, p.vanillaClient, cmdType, cmdQueryId, cmd, docFields, bytelen, 1, cmdSlots[slotP], timesSlots[slotP])
+			cmdSlots[slotP], timesSlots[slotP] = sendFlatCmd(p, p.vanillaClient, cmdType, cmdQueryId, cmd, docFields, bytelen, cmdSlots[slotP], replies, timesSlots[slotP])
 		} else {
 			client, _ := p.vanillaCluster.Client(clusterAddr[slotP])
-			cmdSlots[slotP], timesSlots[slotP] = sendFlatCmd(p, client, cmdType, cmdQueryId, cmd, docFields, bytelen, 1, cmdSlots[slotP], timesSlots[slotP])
+			cmdSlots[slotP], timesSlots[slotP] = sendFlatCmd(p, client, cmdType, cmdQueryId, cmd, docFields, bytelen, cmdSlots[slotP], replies, timesSlots[slotP])
 		}
 	}
 	p.wg.Done()
@@ -122,7 +123,7 @@ func getRxLen(v interface{}) (res uint64) {
 	return
 }
 
-func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd string, docfields []string, txBytesCount, insertCount uint64, cmds []radix.CmdAction, replies []interface{}, times []time.Time) ([]radix.CmdAction, []time.Time) {
+func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd string, docfields []string, txBytesCount uint64, cmds []radix.CmdAction, replies []interface{}, times []time.Time) ([]radix.CmdAction, []time.Time) {
 	var err error = nil
 	var rcv interface{}
 	rxBytesCount := uint64(0)
@@ -131,11 +132,11 @@ func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd str
 	replies = append(replies, rcv)
 	start := time.Now()
 	times = append(times, start)
-	cmds, times = sendIfRequired(p, client, cmdType, cmdQueryId, cmds, err, times, rxBytesCount, rcv, txBytesCount)
+	cmds, times = sendIfRequired(p, client, cmdType, cmdQueryId, cmds, err, times, rxBytesCount, replies, txBytesCount)
 	return cmds, times
 }
 
-func sendIfRequired(p *processor, client radix.Client, cmdType string, cmdQueryId string, cmds []radix.CmdAction, err error, times []time.Time, rxBytesCount uint64, rcv interface{}, txBytesCount uint64) ([]radix.CmdAction, []time.Time) {
+func sendIfRequired(p *processor, client radix.Client, cmdType string, cmdQueryId string, cmds []radix.CmdAction, err error, times []time.Time, rxBytesCount uint64, replies []interface{}, txBytesCount uint64) ([]radix.CmdAction, []time.Time) {
 	cmdLen := len(cmds)
 	if cmdLen >= pipeline {
 		if cmdLen == 1 {