Fix issues raised in review

- Check whether batching is supported when starting a batch.
- Better keying of batched writes.

Signed-off-by: Mark S. Lewis <[email protected]>

Author: bestbeforetoday

shim/batch.go

@@ -7,13 +7,19 @@ import (
 	"github.com/hyperledger/fabric-protos-go-apiv2/peer"
 )
 
+type batchKey struct {
+	Collection string
+	Key        string
+	Type       peer.WriteRecord_Type
+}
+
 type writeBatch struct {
-	writes map[string]*peer.WriteRecord
+	writes map[batchKey]*peer.WriteRecord
 }
 
 func newWriteBatch() *writeBatch {
 	return &writeBatch{
-		writes: make(map[string]*peer.WriteRecord),
+		writes: make(map[batchKey]*peer.WriteRecord),
 	}
 }
 
@@ -31,39 +37,44 @@ func (b *writeBatch) Writes() []*peer.WriteRecord {
 }
 
 func (b *writeBatch) PutState(collection string, key string, value []byte) {
-	b.writes[batchLedgerKey(collection, key)] = &peer.WriteRecord{
+	b.write(&peer.WriteRecord{
 		Key:        key,
 		Value:      value,
 		Collection: collection,
 		Type:       peer.WriteRecord_PUT_STATE,
-	}
+	})
 }
 
 func (b *writeBatch) PutStateMetadataEntry(collection string, key string, metakey string, metadata []byte) {
-	b.writes[batchLedgerKey(collection, key)] = &peer.WriteRecord{
+	b.write(&peer.WriteRecord{
 		Key:        key,
 		Collection: collection,
 		Metadata:   &peer.StateMetadata{Metakey: metakey, Value: metadata},
 		Type:       peer.WriteRecord_PUT_STATE_METADATA,
-	}
+	})
 }
 
 func (b *writeBatch) DelState(collection string, key string) {
-	b.writes[batchLedgerKey(collection, key)] = &peer.WriteRecord{
+	b.write(&peer.WriteRecord{
 		Key:        key,
 		Collection: collection,
 		Type:       peer.WriteRecord_DEL_STATE,
-	}
+	})
 }
 
 func (b *writeBatch) PurgeState(collection string, key string) {
-	b.writes[batchLedgerKey(collection, key)] = &peer.WriteRecord{
+	b.write(&peer.WriteRecord{
 		Key:        key,
 		Collection: collection,
 		Type:       peer.WriteRecord_PURGE_PRIVATE_DATA,
-	}
+	})
 }
 
-func batchLedgerKey(collection string, key string) string {
-	return prefixStateDataWriteBatch + collection + key
+func (b *writeBatch) write(record *peer.WriteRecord) {
+	key := batchKey{
+		Collection: record.Collection,
+		Key:        record.Key,
+		Type:       record.Type,
+	}
+	b.writes[key] = record
 }

shim/handler.go

@@ -19,9 +19,7 @@ const (
 	established state = "established" // connection established
 	ready       state = "ready"       // ready for requests
 
-	defaultMaxSizeWriteBatch  = 100
-	prefixMetaDataWriteBatch  = "m"
-	prefixStateDataWriteBatch = "s"
+	defaultMaxSizeWriteBatch = 100
 )
 
 // PeerChaincodeStream is the common stream interface for Peer - chaincode communication.
@@ -426,7 +424,11 @@ func (h *Handler) handlePurgeState(collection string, key string, channelID stri
 }
 
 // handleWriteBatch communicates with the peer to write batch to state all changes information into the ledger.
-func (h *Handler) handleWriteBatch(batch *peer.WriteBatchState, channelID string, txid string) error {
+func (h *Handler) handleWriteBatch(writes []*peer.WriteRecord, channelID string, txid string) error {
+	batch := &peer.WriteBatchState{
+		Rec: writes,
+	}
+
 	// Construct payload for PUT_STATE_BATCH
 	payloadBytes := marshalOrPanic(batch)
 
@@ -453,21 +455,14 @@ func (h *Handler) handleWriteBatch(batch *peer.WriteBatchState, channelID string
 }
 
 func (h *Handler) sendBatch(channelID string, txid string, writes []*peer.WriteRecord) error {
-	batch := &peer.WriteBatchState{}
-	for _, kv := range writes {
-		batch.Rec = append(batch.Rec, kv)
-		if len(batch.Rec) >= int(h.maxSizeWriteBatch) {
-			err := h.handleWriteBatch(batch, channelID, txid)
-			if err != nil {
-				return fmt.Errorf("failed send batch: %s", err)
-			}
-			batch.Rec = batch.Rec[:0]
+	for ; len(writes) > int(h.maxSizeWriteBatch); writes = writes[h.maxSizeWriteBatch:] {
+		if err := h.handleWriteBatch(writes[:h.maxSizeWriteBatch], channelID, txid); err != nil {
+			return fmt.Errorf("failed send batch: %s", err)
 		}
 	}
 
-	if len(batch.Rec) != 0 {
-		err := h.handleWriteBatch(batch, channelID, txid)
-		if err != nil {
+	if len(writes) > 0 {
+		if err := h.handleWriteBatch(writes, channelID, txid); err != nil {
 			return fmt.Errorf("failed send batch: %s", err)
 		}
 	}

shim/stub.go

@@ -623,7 +623,7 @@ func (s *ChaincodeStub) GetQueryResultWithPagination(query string, pageSize int3
 
 // StartWriteBatch documentation can be found in interfaces.go
 func (s *ChaincodeStub) StartWriteBatch() {
-	if s.writeBatch == nil {
+	if s.handler.usePeerWriteBatch && s.writeBatch == nil {
 		s.writeBatch = newWriteBatch()
 	}
 }