cache resolve cache to lru to avoid leaking memory

License: MIT
Signed-off-by: Jeromy <[email protected]>

core/commands/ipns.go

@@ -76,11 +76,11 @@ Resolve the value of another name:
 
 		if local {
 			offroute := offline.NewOfflineRouter(n.Repo.Datastore(), n.PrivateKey)
-			resolver = namesys.NewRoutingResolver(offroute, 0)
+			resolver = namesys.NewRoutingResolver(offroute, 0, 0)
 		}
 
 		if nocache {
-			resolver = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), 0)
+			resolver = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), 0, 0)
 		}
 
 		var name string

core/core.go

@@ -226,13 +226,13 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
 	bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Routing)
 	n.Exchange = bitswap.New(ctx, n.Identity, bitswapNetwork, n.Blockstore, alwaysSendToPeer)
 
-	cachelife, err := n.getCacheLifetime()
+	cachelife, size, err := n.getCacheParams()
 	if err != nil {
 		return err
 	}
 
 	// setup name system
-	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), cachelife)
+	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), cachelife, size)
 
 	// setup ipns republishing
 	err = n.setupIpnsRepublisher()
@@ -243,21 +243,33 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
 	return nil
 }
 
-func (n *IpfsNode) getCacheLifetime() (time.Duration, error) {
+// getCacheParams returns cache life and cache size
+func (n *IpfsNode) getCacheParams() (time.Duration, int, error) {
 	cfg, err := n.Repo.Config()
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	ct := cfg.Ipns.ResolveCacheTime
+	var d time.Duration
 	if ct == "" {
-		return namesys.DefaultResolverCacheLife, nil
+		d = namesys.DefaultResolverCacheLife
+	} else {
+		parsed, err := time.ParseDuration(ct)
+		if err != nil {
+			return 0, 0, fmt.Errorf("error parsing cache life from Ipns.ResolveCacheTime: %s", err)
+		}
+		d = parsed
 	}
-	d, err := time.ParseDuration(ct)
-	if err != nil {
-		return 0, fmt.Errorf("error parsing cache life from Ipns.ResolveCacheTime: %s", err)
+
+	cs := cfg.Ipns.ResolveCacheSize
+	if cs == 0 {
+		cs = 128
+	}
+	if cs < 0 {
+		return 0, 0, fmt.Errorf("cannot specify negative resolve cache size")
 	}
-	return d, nil
+	return d, cs, nil
 }
 
 func (n *IpfsNode) setupIpnsRepublisher() error {
@@ -478,12 +490,12 @@ func (n *IpfsNode) SetupOfflineRouting() error {
 
 	n.Routing = offroute.NewOfflineRouter(n.Repo.Datastore(), n.PrivateKey)
 
-	cachelife, err := n.getCacheLifetime()
+	cachelife, size, err := n.getCacheParams()
 	if err != nil {
 		return err
 	}
 
-	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), cachelife)
+	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), cachelife, size)
 
 	return nil
 }

namesys/namesys.go

@@ -26,12 +26,12 @@ type mpns struct {
 }
 
 // NewNameSystem will construct the IPFS naming system based on Routing
-func NewNameSystem(r routing.IpfsRouting, ds ds.Datastore, cachelife time.Duration) NameSystem {
+func NewNameSystem(r routing.IpfsRouting, ds ds.Datastore, cachelife time.Duration, cachesize int) NameSystem {
 	return &mpns{
 		resolvers: map[string]resolver{
 			"dns":      newDNSResolver(),
 			"proquint": new(ProquintResolver),
-			"dht":      NewRoutingResolver(r, cachelife),
+			"dht":      NewRoutingResolver(r, cachelife, cachesize),
 		},
 		publishers: map[string]Publisher{
 			"/ipns/": NewRoutingPublisher(r, ds),

namesys/republisher/repub_test.go

@@ -36,7 +36,7 @@ func TestRepublish(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		nd.Namesys = namesys.NewNameSystem(nd.Routing, nd.Repo.Datastore(), 0)
+		nd.Namesys = namesys.NewNameSystem(nd.Routing, nd.Repo.Datastore(), 0, 0)
 
 		nodes = append(nodes, nd)
 	}

namesys/resolve_test.go

@@ -19,7 +19,7 @@ func TestRoutingResolve(t *testing.T) {
 	d := mockrouting.NewServer().Client(testutil.RandIdentityOrFatal(t))
 	dstore := ds.NewMapDatastore()
 
-	resolver := NewRoutingResolver(d, 0)
+	resolver := NewRoutingResolver(d, 0, 0)
 	publisher := NewRoutingPublisher(d, dstore)
 
 	privk, pubk, err := testutil.RandTestKeyPair(512)
@@ -53,7 +53,7 @@ func TestPrexistingExpiredRecord(t *testing.T) {
 	dstore := ds.NewMapDatastore()
 	d := mockrouting.NewServer().ClientWithDatastore(context.Background(), testutil.RandIdentityOrFatal(t), dstore)
 
-	resolver := NewRoutingResolver(d, 0)
+	resolver := NewRoutingResolver(d, 0, 0)
 	publisher := NewRoutingPublisher(d, dstore)
 
 	privk, pubk, err := testutil.RandTestKeyPair(512)
@@ -90,7 +90,7 @@ func TestPrexistingRecord(t *testing.T) {
 	dstore := ds.NewMapDatastore()
 	d := mockrouting.NewServer().ClientWithDatastore(context.Background(), testutil.RandIdentityOrFatal(t), dstore)
 
-	resolver := NewRoutingResolver(d, 0)
+	resolver := NewRoutingResolver(d, 0, 0)
 	publisher := NewRoutingPublisher(d, dstore)
 
 	privk, pubk, err := testutil.RandTestKeyPair(512)

namesys/routing.go

@@ -2,10 +2,10 @@ package namesys
 
 import (
 	"fmt"
-	"sync"
 	"time"
 
 	proto "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/gogo/protobuf/proto"
+	lru "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/hashicorp/golang-lru"
 	mh "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multihash"
 	"github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
 	logging "github.com/ipfs/go-ipfs/vendor/QmTBXYb6y2ZcJmoXVKk3pf9rzSEjbCg7tQaJW7RSuH14nv/go-log"
@@ -23,23 +23,38 @@ var log = logging.Logger("namesys")
 type routingResolver struct {
 	routing routing.IpfsRouting
 
-	cache     map[string]cacheEntry
-	cachelock sync.Mutex
+	cache     *lru.Cache
 	cachelife time.Duration
 }
 
 func (r *routingResolver) cacheGet(name string) (path.Path, bool) {
-	r.cachelock.Lock()
-	entry, ok := r.cache[name]
-	r.cachelock.Unlock()
-	if ok && time.Now().Before(entry.eol) {
+	if r.cache == nil {
+		return "", false
+	}
+
+	ientry, ok := r.cache.Get(name)
+	if !ok {
+		return "", false
+	}
+
+	entry, ok := ientry.(cacheEntry)
+	if !ok {
+		// should never happen, purely for sanity
+		return "", false
+	}
+
+	if time.Now().Before(entry.eol) {
 		return entry.val, true
 	}
 
 	return "", false
 }
 
 func (r *routingResolver) cacheSet(name string, val path.Path, rec *pb.IpnsEntry) {
+	if r.cache == nil {
+		return
+	}
+
 	ttl := r.cachelife
 	if rec.Ttl != nil {
 		// if the record has a ttl set, and its less than ours, use it instead
@@ -55,12 +70,10 @@ func (r *routingResolver) cacheSet(name string, val path.Path, rec *pb.IpnsEntry
 		cacheTil = eol
 	}
 
-	r.cachelock.Lock()
-	r.cache[name] = cacheEntry{
+	r.cache.Add(name, cacheEntry{
 		val: val,
 		eol: cacheTil,
-	}
-	r.cachelock.Unlock()
+	})
 }
 
 type cacheEntry struct {
@@ -70,14 +83,19 @@ type cacheEntry struct {
 
 // NewRoutingResolver constructs a name resolver using the IPFS Routing system
 // to implement SFS-like naming on top.
-func NewRoutingResolver(route routing.IpfsRouting, cachelife time.Duration) *routingResolver {
+func NewRoutingResolver(route routing.IpfsRouting, cachelife time.Duration, cachesize int) *routingResolver {
 	if route == nil {
 		panic("attempt to create resolver with nil routing system")
 	}
 
+	if cachesize < 0 {
+		cachesize = 0
+	}
+	cache, _ := lru.New(cachesize)
+
 	return &routingResolver{
 		routing:   route,
-		cache:     make(map[string]cacheEntry),
+		cache:     cache,
 		cachelife: cachelife,
 	}
 }

repo/config/ipns.go

@@ -5,4 +5,5 @@ type Ipns struct {
 	RecordLifetime  string
 
 	ResolveCacheTime string
+	ResolveCacheSize int
 }

routing/dht/routing.go

@@ -84,10 +84,7 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key key.Key) ([]byte, error) {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
-	// retrieve a majority of the expected record count
-	majority := (KValue / 2) + 1
-
-	vals, err := dht.GetValues(ctx, key, majority)
+	vals, err := dht.GetValues(ctx, key, 16)
 	if err != nil {
 		return nil, err
 	}