Repository: olric-data/olric

Issue:
Possible memory leak at DMap.Scan()
Hi!

I spotted that each Scan()  call leaves a couple of goroutines alive (an iterator is closed).  Minimum reproducing example: 

```golang
package main

import (
	"context"
	"fmt"
	"log"
	"runtime"
	"time"

	"github.com/buraksezer/olric"
	"github.com/buraksezer/olric/config"
)

func main() {
	// Sample for Olric v0.5.x

	// Deployment scenario: embedded-member
	// This creates a single-node Olric cluster. It's good enough for experimenting.

	// config.New returns a new config.Config with sane defaults. Available values for env:
	// local, lan, wan
	cfg := config.New("local")

	// Callback function. It's called when this node is ready to accept connections.
	ready := make(chan struct{}, 1)
	cfg.Started = func() {
		ready <- struct{}{}
		log.Println("[INFO] Olric is ready to accept connections")
	}
	
	// Create a new Olric instance.
	db, err := olric.New(cfg)
	if err != nil {
		log.Fatalf("Failed to create Olric instance: %v", err)
	}

	// Start the instance. It will form a single-node cluster.
	go func() {
		// Call Start at background. It's a blocker call.
		err = db.Start()
		if err != nil {
			log.Fatalf("olric.Start returned an error: %v", err)
		}
	}()

	<-ready

	// In embedded-member scenario, you can use the EmbeddedClient. It implements
	// the Client interface.
	client := db.NewEmbeddedClient()

	dm, err := client.NewDMap("bucket-of-arbitrary-items")
	if err != nil {
		log.Fatalf("olric.NewDMap returned an error: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

	// Create N key-value pairs:
	const N = 10
	for i := range N {
		key := fmt.Sprintf("key-%d", i)
		value := fmt.Sprintf("value-%d", i)
		if err = dm.Put(ctx, key, value); err != nil {
			log.Fatalf("DMap.Put returned an error: %v", err)
		}
	}

	// Iterate M times over N keys:
	const M = 10
	for range M {
		iter, err := dm.Scan(ctx)
		if err != nil {
			log.Fatalf("DMap.Scan returned an error: %v", err)
		}
		for iter.Next() {
			// Do nothing
		}
		iter.Close() // close the iterator explicitly
	}

	// Dispose all the allocated resources:
	if err := dm.Destroy(ctx); err != nil {
		log.Fatalf("Failed to destroy DMap: %v", err)
	}
	if err := client.Close(ctx); err != nil {
		log.Fatalf("Failed to close EmbeddedClient: %v", err)
	}
	if err := db.Shutdown(ctx); err != nil {
		log.Fatalf("Failed to shutdown Olric: %v", err)
	}

	cancel()                // cancel the context, passed to the all methods above, explicitly
	runtime.GC()            // run GC explicitly
	time.Sleep(time.Second) // wait a bit

	s := runtime.MemStats{}
	runtime.ReadMemStats(&s)
	const KB = 1 << 10
	fmt.Printf("Non-Freed objects: %d, Mem in use(KB): %d\n", s.Mallocs-s.Frees, s.HeapAlloc/KB)
	fmt.Printf("GoRoutines remained: %d\n", runtime.NumGoroutine())
}
```
Outputs: 
```
Non-Freed objects: 7597, Mem in use(KB): 966
GoRoutines remained: 22
```

Goroutines are got stuck at: 
```
runtime.gopark (proc.go:425) runtime
runtime.selectgo (select.go:335) runtime
olric.(*ClusterClient).fetchRoutingTablePeriodically (cluster_client.go:757) github.com/buraksezer/olric
olric.NewClusterClient.gowrap1 (cluster_client.go:836) github.com/buraksezer/olric
runtime.goexit (asm_arm64.s:1223) runtime
 - Async Stack Trace
olric.NewClusterClient (cluster_client.go:836) github.com/buraksezer/olric
```



Discussion:

Solution patch:
diff --git a/client.go b/client.go
index 11b889b..45f5c5a 100644
--- a/client.go
+++ b/client.go
@@ -237,6 +237,9 @@ type DMap interface {
 	// Redis client has retransmission logic in case of timeouts, pipeline
 	// can be retransmitted and commands can be executed more than once.
 	Pipeline(opts ...PipelineOption) (*DMapPipeline, error)
+
+	// Close stops background routines and frees allocated resources.
+	Close(ctx context.Context) error
 }
 
 // PipelineOption is a function for defining options to control behavior of the Pipeline command.
diff --git a/cluster_client.go b/cluster_client.go
index 288b77f..b7412a6 100644
--- a/cluster_client.go
+++ b/cluster_client.go
@@ -395,6 +395,11 @@ func (dm *ClusterDMap) LockWithTimeout(ctx context.Context, key string, timeout,
 	}, nil
 }
 
+// Close stops background routines and frees allocated resources.
+func (dm *ClusterDMap) Close(_ context.Context) error {
+	return nil
+}
+
 // Unlock releases the distributed lock associated with the current context by using the provided context for execution.
 func (c *ClusterLockContext) Unlock(ctx context.Context) error {
 	rc, err := c.dm.clusterClient.smartPick(c.dm.name, c.key)
@@ -750,7 +755,7 @@ func WithRoutingTableFetchInterval(interval time.Duration) ClusterClientOption {
 // fetchRoutingTable updates the cluster routing table by fetching the latest version from the cluster.
 // It initializes the partition count if it's the first invocation. Returns an error if fetching fails.
 func (cl *ClusterClient) fetchRoutingTable() error {
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(cl.ctx)
 	defer cancel()
 
 	routingTable, err := cl.RoutingTable(ctx)
diff --git a/embedded_client.go b/embedded_client.go
index cc7e75a..a463856 100644
--- a/embedded_client.go
+++ b/embedded_client.go
@@ -131,7 +131,7 @@ func (e *EmbeddedClient) RefreshMetadata(_ context.Context) error {
 // * Count
 // * Match
 func (dm *EmbeddedDMap) Scan(ctx context.Context, options ...ScanOption) (Iterator, error) {
-	cc, err := NewClusterClient([]string{dm.client.db.rt.This().String()})
+	cc, err := dm.setOrGetClusterClient()
 	if err != nil {
 		return nil, err
 	}
@@ -280,6 +280,18 @@ func (dm *EmbeddedDMap) Put(ctx context.Context, key string, value interface{},
 	return nil
 }
 
+// Close stops background routines and frees allocated resources.
+func (dm *EmbeddedDMap) Close(ctx context.Context) error {
+	dm.mtx.RLock()
+	clusterClient := dm.clusterClient
+	dm.mtx.RUnlock()
+
+	if clusterClient != nil {
+		return dm.clusterClient.Close(ctx)
+	}
+	return nil
+}
+
 func (e *EmbeddedClient) NewDMap(name string, options ...DMapOption) (DMap, error) {
 	dm, err := e.db.dmap.NewDMap(name)
 	if err != nil {



Test patch:
diff --git a/embedded_client_test.go b/embedded_client_test.go
index fd7d8f5..60a8cc1 100644
--- a/embedded_client_test.go
+++ b/embedded_client_test.go
@@ -17,6 +17,7 @@ package olric
 import (
 	"context"
 	"fmt"
+	"runtime"
 	"testing"
 	"time"
 
@@ -639,3 +640,62 @@ func TestEmbeddedClient_DMap_Put_PX_With_NX(t *testing.T) {
 	require.NoError(t, err)
 	assert.NotZero(t, gr.TTL())
 }
+
+func TestEmbeddedClient_Issue263(t *testing.T) {
+	initNumRoutines := runtime.NumGoroutine()
+
+	cluster := newTestOlricCluster(t)
+	db := cluster.addMember(t)
+
+	e := db.NewEmbeddedClient()
+	ctx, cancel := context.WithCancel(context.Background())
+	dm, err := e.NewDMap("mydmap")
+	require.NoError(t, err)
+
+	// Create N key-value pairs:
+	const N = 100
+	for i := range N {
+		key := fmt.Sprintf("key-%d", i)
+		value := fmt.Sprintf("value-%d", i)
+		err := dm.Put(ctx, key, value)
+		require.NoError(t, err)
+	}
+
+	// Iterate M times over N keys:
+	const M = 100
+	for range M {
+		iter, err := dm.Scan(ctx)
+		require.NoError(t, err)
+		for iter.Next() {
+			// Do nothing
+		}
+		iter.Close()
+	}
+
+	require.NoError(t, dm.Close(ctx))
+	require.NoError(t, e.Close(ctx))
+	require.NoError(t, db.Shutdown(ctx))
+
+	cancel()
+
+	assert.Equal(t, initNumRoutines, runtime.NumGoroutine())
+
+	runtime.GC()
+	time.Sleep(time.Second)
+
+	s := runtime.MemStats{}
+	runtime.ReadMemStats(&s)
+
+	const (
+		KB = 1 << 10
+		MB = KB << 10
+	)
+
+	buf := make([]byte, MB)
+	stackSize := runtime.Stack(buf, true)
+
+	t.Logf("Non-freed objects: %d\n", s.Mallocs-s.Frees)
+	t.Logf("Mem in use (KB): %d\n", s.HeapAlloc/KB)
+	t.Logf("Go-routines remained: %d\n", runtime.NumGoroutine())
+	t.Logf("Stack traces:\n%s\n", buf[:stackSize])
+}