[FAB-11047] Enlarge discovery client memoization cache

The discovery server side has a credential cache that reuses
results of access control checks for tuples of requests and signatures
to save CPU cycles.

The idea is that when clients send a request that was sent in the past, they can
skip signinig it and send the signature that was produced in the past in order
for the service on the peer to find the signature and request digest in its cache
and not do costly signature checks.

The discovery client implementation which is used by the go-SDK, caches
only the latest request sent to the service which isn't ideal.

This change set changes this, and introduces a small cache to the
client implementation.

Change-Id: Iade112d2a0cd1fac7dc7968105c652ad7f53d88c
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

discovery/client/client.go

@@ -7,7 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 package discovery
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -27,8 +26,6 @@ var (
 
 // Client interacts with the discovery server
 type Client struct {
-	lastRequest      []byte
-	lastSignature    []byte
 	createConnection Dialer
 	signRequest      Signer
 }
@@ -161,23 +158,11 @@ func (c *Client) Send(ctx context.Context, req *Request, auth *discovery.AuthInf
 	if err != nil {
 		return nil, errors.Wrap(err, "failed marshaling Request to bytes")
 	}
-	sig := c.lastSignature
-	// Only sign the Request if it is different than the previous Request sent.
-	// Otherwise, use the last signature from the previous send.
-	// This is not only to save CPU cycles in the Client-side,
-	// but also for the server side to be able to memoize the signature verification.
-	// We have the use the previous signature, because many signature schemes are not deterministic.
-	if !bytes.Equal(c.lastRequest, payload) {
-		sig, err = c.signRequest(payload)
-		if err != nil {
-			return nil, errors.Wrap(err, "failed signing Request")
-		}
-	}
 
-	// Remember this Request and the corresponding signature, in order to skip signing next time
-	// and reuse the signature
-	c.lastRequest = payload
-	c.lastSignature = sig
+	sig, err := c.signRequest(payload)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed signing Request")
+	}
 
 	conn, err := c.createConnection()
 	if err != nil {
@@ -550,10 +535,10 @@ type endorsementDescriptor struct {
 }
 
 // NewClient creates a new Client instance
-func NewClient(createConnection Dialer, s Signer) *Client {
+func NewClient(createConnection Dialer, s Signer, signerCacheSize int) *Client {
 	return &Client{
 		createConnection: createConnection,
-		signRequest:      s,
+		signRequest:      NewMemoizeSigner(s, signerCacheSize).Sign,
 	}
 }
 

discovery/client/client_test.go

@@ -41,6 +41,10 @@ import (
 	"google.golang.org/grpc/credentials"
 )
 
+const (
+	signerCacheSize = 1
+)
+
 var (
 	ctx = context.Background()
 
@@ -328,7 +332,7 @@ func TestClient(t *testing.T) {
 		ClientIdentity:    []byte{1, 2, 3},
 		ClientTlsCertHash: util.ComputeSHA256(clientTLSCert.Certificate[0]),
 	}
-	cl := NewClient(connect, signer)
+	cl := NewClient(connect, signer, signerCacheSize)
 
 	sup.On("PeersOfChannel").Return(channelPeersWithoutChaincodes).Times(2)
 	req := NewRequest()
@@ -450,7 +454,7 @@ func TestUnableToSign(t *testing.T) {
 	authInfo := &discovery.AuthInfo{
 		ClientIdentity: []byte{1, 2, 3},
 	}
-	cl := NewClient(failToConnect, signer)
+	cl := NewClient(failToConnect, signer, signerCacheSize)
 	req := NewRequest()
 	req = req.OfChannel("mychannel")
 	resp, err := cl.Send(ctx, req, authInfo)
@@ -468,7 +472,7 @@ func TestUnableToConnect(t *testing.T) {
 	auth := &discovery.AuthInfo{
 		ClientIdentity: []byte{1, 2, 3},
 	}
-	cl := NewClient(failToConnect, signer)
+	cl := NewClient(failToConnect, signer, signerCacheSize)
 	req := NewRequest()
 	req = req.OfChannel("mychannel")
 	resp, err := cl.Send(ctx, req, auth)
@@ -491,7 +495,7 @@ func TestBadResponses(t *testing.T) {
 	auth := &discovery.AuthInfo{
 		ClientIdentity: []byte{1, 2, 3},
 	}
-	cl := NewClient(connect, signer)
+	cl := NewClient(connect, signer, signerCacheSize)
 
 	// Scenario I: discovery service sends back an error
 	svc.On("Discover").Return(nil, errors.New("foo")).Once()

discovery/client/signer.go

@@ -0,0 +1,94 @@
+/*
+Copyright IBM Corp All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package discovery
+
+import (
+	"encoding/hex"
+	"sync"
+
+	"github.com/hyperledger/fabric/common/util"
+)
+
+// MemoizeSigner signs messages with the same signature
+// if the message was signed recently
+type MemoizeSigner struct {
+	maxMemorySize int
+	sync.RWMutex
+	memory map[string][]byte
+	sign   Signer
+}
+
+// NewMemoizeSigner creates a new MemoizeSigner that signs
+// message with the given sign function
+func NewMemoizeSigner(signFunc Signer, maxMemorySize int) *MemoizeSigner {
+	return &MemoizeSigner{
+		maxMemorySize: maxMemorySize,
+		memory:        make(map[string][]byte),
+		sign:          signFunc,
+	}
+}
+
+// Signer signs a message and returns the signature and nil,
+// or nil and error on failure
+func (ms *MemoizeSigner) Sign(msg []byte) ([]byte, error) {
+	sig, isInMemory := ms.lookup(msg)
+	if isInMemory {
+		return sig, nil
+	}
+	sig, err := ms.sign(msg)
+	if err != nil {
+		return nil, err
+	}
+	ms.memorize(msg, sig)
+	return sig, nil
+}
+
+// lookup looks up the given message in memory and returns
+// the signature, if the message is in memory
+func (ms *MemoizeSigner) lookup(msg []byte) ([]byte, bool) {
+	ms.RLock()
+	defer ms.RUnlock()
+	sig, exists := ms.memory[msgDigest(msg)]
+	return sig, exists
+}
+
+func (ms *MemoizeSigner) memorize(msg, signature []byte) {
+	ms.RLock()
+	shouldShrink := len(ms.memory) >= ms.maxMemorySize
+	ms.RUnlock()
+
+	if shouldShrink {
+		ms.shrinkMemory()
+	}
+
+	ms.Lock()
+	defer ms.Unlock()
+	ms.memory[msgDigest(msg)] = signature
+}
+
+// evict evicts random messages from memory
+// until its size is smaller than maxMemorySize
+func (ms *MemoizeSigner) shrinkMemory() {
+	ms.Lock()
+	defer ms.Unlock()
+	for len(ms.memory) > ms.maxMemorySize {
+		ms.evictFromMemory()
+	}
+}
+
+// evictFromMemory evicts a random message from memory
+func (ms *MemoizeSigner) evictFromMemory() {
+	for dig := range ms.memory {
+		delete(ms.memory, dig)
+		return
+	}
+}
+
+// msgDigest returns a digest of a given message
+func msgDigest(msg []byte) string {
+	return hex.EncodeToString(util.ComputeSHA256(msg))
+}

discovery/client/signer_test.go

@@ -0,0 +1,88 @@
+/*
+Copyright IBM Corp All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package discovery
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/hyperledger/fabric/bccsp/factory"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+func init() {
+	factory.InitFactories(nil)
+}
+
+func TestSameMessage(t *testing.T) {
+	var signedInvokedCount int
+	sign := func(msg []byte) ([]byte, error) {
+		signedInvokedCount++
+		return msg, nil
+	}
+
+	ms := NewMemoizeSigner(sign, 10)
+	for i := 0; i < 5; i++ {
+		sig, err := ms.Sign([]byte{1, 2, 3})
+		assert.NoError(t, err)
+		assert.Equal(t, []byte{1, 2, 3}, sig)
+		assert.Equal(t, 1, signedInvokedCount)
+	}
+}
+
+func TestDifferentMessages(t *testing.T) {
+	n := 5
+	var signedInvokedCount uint32
+	sign := func(msg []byte) ([]byte, error) {
+		atomic.AddUint32(&signedInvokedCount, 1)
+		return msg, nil
+	}
+
+	ms := NewMemoizeSigner(sign, n)
+	parallelSignRange := func(start, end int) {
+		var wg sync.WaitGroup
+		wg.Add(end - start)
+		for i := start; i < end; i++ {
+			i := i
+			go func() {
+				defer wg.Done()
+				sig, err := ms.Sign([]byte{byte(i)})
+				assert.NoError(t, err)
+				assert.Equal(t, []byte{byte(i)}, sig)
+			}()
+		}
+		wg.Wait()
+	}
+
+	// Query once
+	parallelSignRange(0, n)
+	assert.Equal(t, uint32(n), atomic.LoadUint32(&signedInvokedCount))
+
+	// Query twice
+	parallelSignRange(0, n)
+	assert.Equal(t, uint32(n), atomic.LoadUint32(&signedInvokedCount))
+
+	// Query thrice on a disjoint range
+	parallelSignRange(n+1, 2*n)
+	oldSignedInvokedCount := atomic.LoadUint32(&signedInvokedCount)
+
+	// Ensure that some of the early messages 0-n were purged from memory
+	parallelSignRange(0, n)
+	assert.True(t, oldSignedInvokedCount < atomic.LoadUint32(&signedInvokedCount))
+}
+
+func TestFailure(t *testing.T) {
+	sign := func(_ []byte) ([]byte, error) {
+		return nil, errors.New("something went wrong")
+	}
+
+	ms := NewMemoizeSigner(sign, 1)
+	_, err := ms.Sign([]byte{1, 2, 3})
+	assert.Equal(t, "something went wrong", err.Error())
+}

discovery/cmd/stub.go

@@ -61,7 +61,7 @@ func (stub *ClientStub) Send(server string, conf common.Config, req *discovery.R
 	timeout, cancel := context.WithTimeout(context.Background(), defaultTimeout)
 	defer cancel()
 
-	disc := discovery.NewClient(comm.NewDialer(server), signer.Sign)
+	disc := discovery.NewClient(comm.NewDialer(server), signer.Sign, 10)
 
 	resp, err := disc.Send(timeout, req, &AuthInfo{
 		ClientIdentity:    signer.Creator,

discovery/test/integration_test.go

@@ -494,7 +494,8 @@ func createClientAndService(t *testing.T, testdir string) (*client, *service) {
 	assert.NoError(t, err)
 
 	wrapperClient := &client{AuthInfo: authInfo, conn: conn}
-	c := disc.NewClient(wrapperClient.newConnection, signer.Sign)
+	signerCacheSize := 10
+	c := disc.NewClient(wrapperClient.newConnection, signer.Sign, signerCacheSize)
 	wrapperClient.Client = c
 	service := &service{Server: gRPCServer.Server(), lc: lc, sup: sup}
 	return wrapperClient, service