Hi, this is my codes.

package main

import "net/http"
import "fmt"
import "github.com/streamrail/concurrent-map"
import (
	"crypto/rand"
	"encoding/hex"
	"log"
	_ "net/http/pprof"
	"runtime"
	"strconv"
)

var myMap = cmap.New()

func main() {
	runtime.GOMAXPROCS(runtime.NumCPU())
	log.Println("here start test")
	http.HandleFunc("/", myHandler)
	http.HandleFunc("/delete", deHandler)
	log.Fatal(http.ListenAndServe(":9090", nil))
}

func myHandler(w http.ResponseWriter, r *http.Request) {

	str := "aaaaaaaaaaaaaaaaaaaaaaaa"
	for i := 0; i < 10000000; i++ {
		key := str + strconv.Itoa(i)
		myMap.Set(key, "8ff98326-2187-4de2-924e-af5098921aba")
	}
	// myMap.Set(randString(), "1000")

	fmt.Println(myMap.Count())
	w.Write([]byte(strconv.Itoa(myMap.Count())))
}

func deHandler(w http.ResponseWriter, r *http.Request) {
	fmt.Println("delete")
	fmt.Println(myMap.Count())
	str := "aaaaaaaaaaaaaaaaaaaaaaaa"

	for i := 0; i < 10000000; i++ {
		key := str + strconv.Itoa(i)
		myMap.Remove(key)
	}
	runtime.GC()
	fmt.Println("after delete")
	fmt.Println(myMap.Count())
	w.Write([]byte(strconv.Itoa(myMap.Count())))
}

func randString() string {
	b := make([]byte, 10)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

The server consumes about 1.4G memory. And when I request the "localhost:9090/delete", myMap.Count() is 0. But the memory is not reduced.

I use pprof, the results is:

 go tool pprof http://localhost:9090/debug/pprof/heap
Fetching profile from http://localhost:9090/debug/pprof/heap
Saved profile in /home/user/pprof/pprof.for_go.localhost:9090.alloc_objects.alloc_space.inuse_objects.inuse_space.018.pb.gz
Entering interactive mode (type "help" for commands)
(pprof) top
582.01MB of 582.01MB total (  100%)
Dropped 9 nodes (cum <= 2.91MB)
Showing top 10 nodes out of 11 (cum >= 582.01MB)
      flat  flat%   sum%        cum   cum%
     544MB 93.47% 93.47%      544MB 93.47%  runtime.hashGrow
   32.51MB  5.59% 99.05%   582.01MB   100%  runtime.mapassign
    5.50MB  0.95%   100%     5.50MB  0.95%  runtime.evacuate
         0     0%   100%   582.01MB   100%  github.com/streamrail/concurrent-map.(*ConcurrentMap).Set
         0     0%   100%   582.01MB   100%  main.myHandler
         0     0%   100%   582.01MB   100%  net/http.(*ServeMux).ServeHTTP
         0     0%   100%   582.01MB   100%  net/http.(*conn).serve
         0     0%   100%   582.01MB   100%  net/http.HandlerFunc.ServeHTTP
         0     0%   100%   582.01MB   100%  net/http.serverHandler.ServeHTTP
         0     0%   100%   582.01MB   100%  runtime.goexit

And I read some posts, maybe map is not good at storing huge key/value ?

Hello!

I found myself beginning to port some personal projects to 1.18 and found that this library didn't yet have any support for generics. So, I figured I might as well take a swing at it.

I'm not exactly sure what the right path is for making this available is since 1.18 is still a beta. Regardless, hopefully this is useful to others :)

closes #17

allows for users to get a concurrent map with a configurable number of shards

made it an optional "list" so that the .New() call signature doesn't change.

I've also set SHARD_COUNT to be a constant. This could break people, but I think it's better for users not to be modifying globals to change the libraries behavior.

Hello,

Thanks for this package that has been useful for years. Golang 1.9 now provides a concurrent map in standard library. Consider mentioning it on the README for those using latest version of Golang.

Though, the package is still very useful for older versions :)

As mentioned by @efeller. When Using the current implementation, after calling Iter/IterBuffered, if the returned channel were not drained, some readlock may not be properly released. Which cause subsequent call to change the map, namely Set/Remove/Pop operations to block forever, thus a deadlock occurs.

We introduce two test functions (TestUnDrainedIter and TestUnDrainedIterBuffered) which demonstrate the bug.

This main idea of the fix is to take a snapshot of the current concurrentMap when calling Iter/IterBuffered. We factor out a snapshot function here, which returns an array of buffered channels(one per each shard, the readlocks are released after the related channel is populated). They then fan in to a return channel. In the buffered case, the size of the returned channel is the sum of length of all the channels. We avoid using m.Count() here since m may change between the function calls, which makes it inaccurate.

This fix passes those two test functions mentioned before.

Got index out of range panic, https://github.com/streamrail/concurrent-map/blob/master/concurrent_map_template.txt#L34

No idea how that happened, afaik it should not. Marvell PJ4Bv7 Processor rev 2 (v7l)

If you declare a concurrentMap item without creating it with New() then calling map.Items blocks forever.

        var work cmap.ConcurrentMap
	for k, _ := range work.Items() {
		fmt.Println(k)
	}
        //Never reached

I think this should be treated as a bug, it should panic just like how other methods panic when the map hasn't been created.

The current implementation of iterCB is quite simple:

// Callback based iterator, cheapest way to read
// all elements in a map.
func (m ConcurrentMap) IterCb(fn IterCb) {
	for idx := range m {
		shard := (m)[idx]
		shard.RLock()
		for key, value := range shard.items {
			fn(key, value)
		}
		shard.RUnlock()
	}
}

We could create SHARD_COUNT go routines , one for each shard, and do some parrallelism here. If one callback function does lock the shard, that will keep the other ones unlocked.

Any specific reason why this is implemented this way ?

Context: concurrent-map uses FNV-1 (see: http://www.isthe.com/chongo/tech/comp/fnv/) to consistently map keys to one of its internal shards.

Surprisingly, it does not make use of hash/fnv, but instead relies on a custom implementation (see: https://github.com/orcaman/concurrent-map/blob/master/concurrent_map.go#L285).

I was about to make a PR to correct this, but I noticed that in the tests we actually check the behavior of the custom implementation against hash/fnv (see: https://github.com/orcaman/concurrent-map/blob/master/concurrent_map_test.go#L389).

What is the reason for that? I see that the native implementation is stateful but this can be work-around easily:

// Returns shard under given key
func (m ConcurrentMap) GetShard(key string) *ConcurrentMapShared {
	return m[uint(fnv32(key))%uint(SHARD_COUNT)]
}

would become:

// Returns shard under given key
func (m ConcurrentMap) GetShard(key string) *ConcurrentMapShared {
        hash := fnv.New32()
        hash.Write([]byte(key))
        hashKey := hash.Sum32()
        return m[hashKey%uint32(SHARD_COUNT)]
}

which is slightly more verbose, but makes us rely on a maintained implementation of FNV-1. Is there a reason I am not seeing for this?

hi,i use buffer channels as map value for thread safe,when test use 10 goroutines the value got from channel was not same with the one send in,any suggestion?

testmap := cmap.New()
    fmt.Println("SyncMapNew:    ", TestInParallel(&testmap, 10))

func TestInParallel(g *cmap.ConcurrentMap, n int) time.Duration {
    start := time.Now()
    var wait sync.WaitGroup

    for i := 0; i < n; i++ {
        wait.Add(1)
        go func() {
            TheTest(g, rand.New(rand.NewSource(int64(i*500))))
            wait.Done()
        }()
    }
    wait.Wait()
    return time.Now().Sub(start)
}
func TheTest(g *cmap.ConcurrentMap, rnd *rand.Rand) time.Duration {
    start := time.Now()
    var key string
    var value time.Time
    //var got time.Time
    for i := 0; i < 10000; i++ {
        key = strconv.Itoa(int(rnd.Int31n(50000)))
        if g.Has(key) == false {
            g.Set(key, make(chan time.Time, 100))
        }
        tchan, _ := g.Get(key)
        castchan := tchan.(chan time.Time)
        //castchan := make(chan time.Time, 100)
        value = time.Now()
        castchan <- value
        got := <-castchan
        g.Set(key, castchan)
        if value != got {
            panic(fmt.Sprintf("ERROR: expected %v, got %v", value, got))
        }
    }
    return time.Now().Sub(start)
}

Although I updated the comment and changed the method name as well, at the very least the comment should state what happens when the key already exists in the map (even if it becomes obvious when reading the source).

// Keys returns all keys as []string
func (m ConcurrentMap[V]) Keys() []string {
	count := m.Count()
	ch := make(chan string, count)
	go func() {
		// Foreach shard.
		wg := sync.WaitGroup{}
		wg.Add(SHARD_COUNT)
		for _, shard := range m {
			go func(shard *ConcurrentMapShared[V]) {
				// Foreach key, value pair.
				shard.RLock()
				for key := range shard.items {
					ch <- key
				}
				shard.RUnlock()
				wg.Done()
			}(shard)
		}
		wg.Wait()
		close(ch)
	}()

	// Generate keys
	keys := make([]string, 0, count)
	for k := range ch {
		keys = append(keys, k)
	}
	return keys
}

func (m ConcurrentMap[V]) Keys2() []string {
	keys := make([]string, 0, m.Count())

	wg := sync.WaitGroup{}
	wg.Add(SHARD_COUNT)
	for _, shard := range m {
		go func(shard *ConcurrentMapShared[V]) {
			shard.RLock()
			for key := range shard.items {
				keys = append(keys, key)
			}
			shard.RUnlock()
			wg.Done()
		}(shard)
	}
	wg.Wait()
	return keys
}

func (m ConcurrentMap[V]) Keys3() []string {
	keys := make([]string, 0, m.Count())
	for _, shard := range m {
		shard.RLock()
		for key := range shard.items {
			keys = append(keys, key)
		}
		shard.RUnlock()
	}
	return keys
}

func BenchmarkKeys(b *testing.B) {
	m := New[Animal]()

	// Insert 10000 elements.
	for i := 0; i < 10000; i++ {
		m.Set(strconv.Itoa(i), Animal{strconv.Itoa(i)})
	}

	b.ReportAllocs()
	b.ResetTimer()
	b.Run("Keys", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			m.Keys()
		}
	})
	b.Run("Keys2", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			m.Keys2()
		}
	})
	b.Run("Keys3", func(b *testing.B) {
		for i := 0; i < b.N; i++ {
			m.Keys3()
		}
	})
}

BenchmarkKeys
BenchmarkKeys/Keys
BenchmarkKeys/Keys-8                1406            859256 ns/op          329637 B/op         69 allocs/op
BenchmarkKeys/Keys2
BenchmarkKeys/Keys2-8              12532             96416 ns/op          165675 B/op         67 allocs/op
BenchmarkKeys/Keys3
BenchmarkKeys/Keys3-8              10000            114136 ns/op          163840 B/op          1 allocs/op

I really don't know what the advantages of the current Keys method, why not just use the sync solution? It's not the fastest, but it's gc friendly.

hello,as the title ,can you add a clone func to deep copy a cmap to a new one? and for the values like string,add a function maybe named append to append new string to the old ones stored before,because if the string stored in cmap is large,i get it and cancat with another large string will cost a lot of memory

I have personally had a requirement to do multiple consecutive operations using the same key with the requirement that nothing can change from start to end. To enable this, I propose explicitly named versions of the Get, Set and Has functions which do not take out locks on the shard. Instead, it is up to the user to use the GetShard method and lock/unlock it correctly.

The three functions would look like

func (m ConcurrentMap) UnlockedSet(key string, value interface{})

func (m ConcurrentMap) UnlockedGet(key string) (interface{}, bool)

func (m ConcurrentMap) UnlockedHas(key string) bool

I'm not sure about the naming. Maybe UnsafeGet, UnsafeSet and UnsafeHas is better?

Here's an example of how I think this could be useful

shard := conMap.GetShard(key)
shard.RLock()
pointerToMyStruct, ok := conMap.UnlockedGet(key)
// Some other thread might be trying to write to this same key but, we still have a lock!
if ok {
    pointerToMyStruct.ReadSomeCoolInfo()
}
shard.RUnlock()
// Now the write on the other thread will go-ahead

In my case, I'm developing a cache where entries expire. I want to ensure that if the value has expired when the Get call is made, it's still expired when when ReadSomeCoolInfo call is made. If another thread were allowed to refresh the value between the two functions, it would cause me to return the wrong information.

The alternative to these functions would be to expose items on the shards to other packages and let the developer dig in and do it all manually.

When we are traversing frequently, using IterBuffered() will generate a lot of memory garbage, which will cause a lot of burden on the GC of the program. At this time, we need to traverse externally, so we have to provide the items method

func (m ConcurrentMap) Keys() []string {
	count := m.Count()

	time.Sleep(5*time.Second)

	ch := make(chan string, count)
	go func() {
		// Foreach shard.
		wg := sync.WaitGroup{}
		wg.Add(SHARD_COUNT)
		for _, shard := range m {
			go func(shard *ConcurrentMapShared) {
				// Foreach key, value pair.
				shard.RLock()
				for key := range shard.items {
					ch <- key
				}
				shard.RUnlock()
				wg.Done()
			}(shard)
		}
		wg.Wait()
		close(ch)
	}()

	// Generate keys
	keys := make([]string, 0, count)
	for k := range ch {
		keys = append(keys, k)
	}
	return keys
}

as the func shows, m.Count() is called before loop keys from shards. However, there was no direct connect between Cont() res and the length of keys. So why the func is designed to count first ?