client, wire: Remove context from Cache

According to the package documentation, a context should not be stored
in a struct type. Furthermore, the context had no major relevance.

Signed-off-by: Matthias Geihs <matthias@perun.network>

Author: matthiasgeihs

client/channelconn.go

@@ -46,7 +46,8 @@ func newChannelConn(id channel.ID, peers []wire.Address, idx channel.Index, sub
 	// relay to receive all update responses
 	relay := wire.NewRelay()
 	// we cache all responses for the lifetime of the relay
-	relay.Cache(context.Background(), func(*wire.Envelope) bool { return true })
+	cacheAll := func(*wire.Envelope) bool { return true }
+	relay.Cache(&cacheAll)
 	// Close the relay if anything goes wrong in the following.
 	// We could have a leaky subscription otherwise.
 	defer func() {

client/proposal.go

@@ -265,7 +265,8 @@ func (c *Client) acceptChannelProposal(
 	// enables caching of incoming version 0 signatures before sending any message
 	// that might trigger a fast peer to send those. We don't know the channel id
 	// yet so the cache predicate is coarser than the later subscription.
-	enableVer0Cache(ctx, c.conn)
+	pred := enableVer0Cache(c.conn)
+	defer c.conn.ReleaseCache(pred)
 
 	if err := c.conn.pubMsg(ctx, acc, p); err != nil {
 		c.logPeer(p).Errorf("error sending proposal acceptance: %v", err)
@@ -302,7 +303,8 @@ func (c *Client) proposeTwoPartyChannel(
 	// enables caching of incoming version 0 signatures before sending any message
 	// that might trigger a fast peer to send those. We don't know the channel id
 	// yet so the cache predicate is coarser than the later subscription.
-	enableVer0Cache(ctx, c.conn)
+	pred := enableVer0Cache(c.conn)
+	defer c.conn.ReleaseCache(pred)
 
 	proposalID := proposal.ProposalID()
 	isResponse := func(e *wire.Envelope) bool {
@@ -680,11 +682,13 @@ func (c *Client) fundSubchannel(ctx context.Context, prop *SubChannelProposal, s
 }
 
 // enableVer0Cache enables caching of incoming version 0 signatures.
-func enableVer0Cache(ctx context.Context, c wire.Cacher) {
-	c.Cache(ctx, func(m *wire.Envelope) bool {
+func enableVer0Cache(c wire.Cacher) *wire.Predicate {
+	p := func(m *wire.Envelope) bool {
 		return m.Msg.Type() == wire.ChannelUpdateAcc &&
 			m.Msg.(*msgChannelUpdateAcc).Version == 0
-	})
+	}
+	c.Cache(&p)
+	return &p
 }
 
 func (c *Client) enableVer1Cache() {

wire/cache.go

@@ -14,57 +14,53 @@
 
 package wire
 
-import "context"
-
 type (
-	// Cache is a message cache. The default value is a valid empty cache.
+	// Cache is a message cache.
 	Cache struct {
 		msgs  []*Envelope
-		preds []ctxPredicate
+		preds map[*Predicate]struct{}
 	}
 
 	// A Predicate defines a message filter.
 	Predicate = func(*Envelope) bool
 
-	ctxPredicate struct {
-		ctx context.Context
-		p   Predicate
-	}
-
 	// A Cacher has the Cache method to enable caching of messages.
 	Cacher interface {
 		// Cache should enable the caching of messages
-		Cache(context.Context, Predicate)
+		Cache(*Predicate)
 	}
 )
 
+// MakeCache creates a new cache.
+func MakeCache() Cache {
+	return Cache{
+		preds: make(map[*func(*Envelope) bool]struct{}),
+	}
+}
+
 // Cache is a message cache. The default value is a valid empty cache.
-func (c *Cache) Cache(ctx context.Context, p Predicate) {
-	c.preds = append(c.preds, ctxPredicate{ctx, p})
+func (c *Cache) Cache(p *Predicate) {
+	c.preds[p] = struct{}{}
+}
+
+// Release releases the cache predicate.
+func (c *Cache) Release(p *Predicate) {
+	delete(c.preds, p)
 }
 
 // Put puts the message into the cache if it matches any active predicate.
 // If it matches several predicates, it is still only added once to the cache.
 func (c *Cache) Put(e *Envelope) bool {
 	// we filter the predicates for non-active and lazily remove them
-	preds := c.preds[:0]
 	any := false
-	for _, p := range c.preds {
-		select {
-		case <-p.ctx.Done():
-			continue // skip done predicate
-		default:
-			preds = append(preds, p)
-		}
-
-		any = any || p.p(e)
+	for p := range c.preds {
+		any = any || (*p)(e)
 	}
 
 	if any {
 		c.msgs = append(c.msgs, e)
 	}
 
-	c.preds = preds
 	return any
 }
 

wire/cache_internal_test.go

@@ -15,7 +15,6 @@
 package wire
 
 import (
-	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -28,7 +27,7 @@ func TestCache(t *testing.T) {
 	assert, require := assert.New(t), require.New(t)
 	rng := test.Prng(t)
 
-	var c Cache
+	c := MakeCache()
 	require.Zero(c.Size())
 
 	ping0 := NewRandomEnvelope(rng, NewPingMsg())
@@ -42,8 +41,7 @@ func TestCache(t *testing.T) {
 	assert.Zero(c.Size())
 
 	isPing := func(e *Envelope) bool { return e.Msg.Type() == Ping }
-	ctx, cancel := context.WithCancel(context.Background())
-	c.Cache(ctx, isPing)
+	c.Cache(&isPing)
 	assert.True(c.Put(ping0), "Put into cache with predicate")
 	assert.Equal(1, c.Size())
 	assert.False(c.Put(pong), "Put into cache with non-matching prediacte")
@@ -54,7 +52,7 @@ func TestCache(t *testing.T) {
 	empty := c.Messages(func(*Envelope) bool { return false })
 	assert.Len(empty, 0)
 
-	cancel()
+	c.Release(&isPing)
 	assert.False(c.Put(ping2), "Put into cache with canceled predicate")
 	assert.Equal(2, c.Size())
 	assert.Len(c.preds, 0, "internal: Put should have removed canceled predicate")
@@ -67,7 +65,7 @@ func TestCache(t *testing.T) {
 	require.Len(msgs, 1)
 	assert.Same(msgs[0], ping0)
 
-	c.Cache(context.Background(), isPing)
+	c.Cache(&isPing)
 	c.Flush()
 	assert.Equal(0, c.Size())
 	assert.False(c.Put(ping0), "flushed cache should not hold any predicates")

wire/relay.go

@@ -15,7 +15,6 @@
 package wire
 
 import (
-	"context"
 	stdsync "sync"
 
 	"github.com/pkg/errors"
@@ -41,7 +40,10 @@ type subscription struct {
 
 // NewRelay returns a new Relay which logs unhandled messages.
 func NewRelay() *Relay {
-	return &Relay{defaultMsgHandler: logUnhandledMsg}
+	return &Relay{
+		defaultMsgHandler: logUnhandledMsg,
+		cache:             MakeCache(),
+	}
 }
 
 // Close closes the relay.
@@ -64,16 +66,24 @@ func (p *Relay) Close() error {
 }
 
 // Cache enables caching of messages that don't match any consumer. They are
-// only cached if they match the given predicate, within the given context.
-func (p *Relay) Cache(ctx context.Context, predicate Predicate) {
+// only cached if they match the given predicate.
+func (p *Relay) Cache(predicate *Predicate) {
 	p.mutex.Lock()
 	defer p.mutex.Unlock()
 
 	if p.IsClosed() {
 		return
 	}
 
-	p.cache.Cache(ctx, predicate)
+	p.cache.Cache(predicate)
+}
+
+// ReleaseCache disable caching for the given predicate.
+func (p *Relay) ReleaseCache(predicate *Predicate) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	p.cache.Release(predicate)
 }
 
 // Subscribe adds a Consumer to the subscriptions.

wire/relay_internal_test.go

@@ -126,7 +126,7 @@ func TestProducer_caching(t *testing.T) {
 	prod.SetDefaultMsgHandler(func(e *Envelope) { unhandlesMsg = append(unhandlesMsg, e) })
 
 	ctx := context.Background()
-	prod.Cache(ctx, isPing)
+	prod.Cache(&isPing)
 
 	rng := test.Prng(t)
 	ping0 := NewRandomEnvelope(rng, NewPingMsg())
@@ -141,7 +141,7 @@ func TestProducer_caching(t *testing.T) {
 	assert.Equal(1, prod.cache.Size())
 	assert.Len(unhandlesMsg, 1)
 
-	prod.Cache(ctx, isPong)
+	prod.Cache(&isPong)
 	prod.Put(pong2)
 	assert.Equal(2, prod.cache.Size())
 	assert.Len(unhandlesMsg, 1)
@@ -161,7 +161,8 @@ func TestProducer_caching(t *testing.T) {
 	assert.Contains(err.Error(), "cache")
 	assert.Zero(prod.cache.Size(), "producer.Close should flush the cache")
 
-	prod.Cache(ctx, func(*Envelope) bool { return true })
+	p := func(*Envelope) bool { return true }
+	prod.Cache(&p)
 	prod.cache.Put(ping0)
 	assert.Zero(prod.cache.Size(), "Cache on closed producer should not enable caching")
 }