Concurrently read map entries into a channel

Question

I have a scenario where I need to range over (as many as possible) map entries and send them into a channel. The operation at the other end of the channel can take a long amount of time and the map is accessed concurrently (and protected by an RWMutex). The map is also rather big and I want to avoid creating a temporary copy of it.

Assume I have a struct like this:

type Example struct {
    sync.RWMutex
    m map[string]struct{}
}

Now I came up with something like this:

func (e *Example) StreamAll() <-chan string {
    toReturn := make(chan string)
    go func() {
        e.RLock()
        defer e.RUnlock()
        for k := range e.m {
            e.RUnlock()
            toReturn <- k
            e.RLock()
        }
        close(toReturn)
    }()
    return toReturn
}

The language specification has this interesting bit about ranging over maps:

If map entries that have not yet been reached are removed during iteration, the corresponding iteration values will not be produced. If map entries are created during iteration, that entry may be produced during the iteration or may be skipped.

Now, what I'd like to know is this: Is there a guarantee that my method of ranging over the map works even if the map is changed between iterations? Including the case where the key I last read is deleted? I don't need all of the map entries, but most of them.

Here is a complete example:

package main

import (
    "fmt"
    "sync"
)

type Example struct {
    sync.RWMutex
    m map[string]struct{}
}

func NewExample() *Example {
    return &Example{
        m: make(map[string]struct{}),
    }
}

func (e *Example) Put(s string) {
    e.Lock()
    defer e.Unlock()
    e.m[s] = struct{}{}
}

func (e *Example) Delete(s string) {
    e.Lock()
    defer e.Unlock()
    delete(e.m, s)
}

func (e *Example) StreamAll() <-chan string {
    toReturn := make(chan string)
    go func() {
        e.RLock()
        defer e.RUnlock()
        for k := range e.m {
            e.RUnlock()
            toReturn <- k
            e.RLock()
        }
        close(toReturn)
    }()
    return toReturn
}

func main() {
    e := NewExample()
    e.Put("a")
    e.Put("b")

    values := e.StreamAll()
    // Assume other goroutines concurrently call Put and Delete on e
    for k := range values {
        fmt.Println(k)
    }
}

Concurrently read map entries into a channel

Answers (1)

Related Questions