如何实现 Go 并发 map 或 slice 以更快地管理使用中的资源？

Question

您有一个结构，该结构表示一次只有一个用户可以访问的资源。可能看起来像这样：

type Resource struct{
    InUse bool//or int32/int64 is you want to use atomics
    Resource string //parameters that map to the resource, think `/dev/chardeviceXXX`
}

这些资源的数量是有限的，用户会随机同时请求访问它们，因此您将它们打包在管理器中

type ResourceManager struct{
    Resources []*Resource //or a map 
}

我正在尝试找出让经理创建一个函数func (m *ResourceManager)GetUnusedResouce()(*Resouce,error) 的最佳、安全方法，该函数将：

• 遍历所有资源，直到找到未使用的资源
• 将其标记为 InUse 并将 *Resouce 返回到调用上下文/goroutine
• 我会锁定以避免任何系统级锁定 (flock) 并在 Go 中完成这一切
• 还需要一个函数来标记资源不再使用

现在我在管理器中使用互斥锁来锁定访问，因为我遍历整个切片。它是安全的，但我希望能够通过同时搜索已使用的资源并处理两个尝试将同一资源标记为 InUse 的 goroutine 来加快速度。

更新

我特别想知道是否将资源InUse 字段设置为int64，然后使用atomic.CompareAndSwapInt64 将允许资源管理器在发现未使用的资源时正确锁定：

func (m *ResourceManager)GetUnusedResouce()(*Resouce,error){
    for i := range Resources{
        if atomic.CompareAndSwapInt64(&Resouces[i].InUse,1){
            return Resouces[i],nil
        }
    }
    return nil, errors.New("all resouces in use")
}

任何可以更好地测试这一点的单元测试也将不胜感激。

Answer 1

问题中的GetUnusedResouce 函数可能会对所有资源执行比较和交换操作。根据资源数量和应用程序访问模式，执行受互斥体保护的少量操作会更快。

使用单链表实现快速get和put操作。

type Resource struct {
    next     *Resource
    Resource string
}

type ResourceManager struct {
    free *Resource
    mu   sync.Mutex
}

// Get gets a free resource from the manager or returns
// nil when the manager is empty.
func (m *ResourceManager) Get() *Resource {
    m.mu.Lock()
    defer m.mu.Unlock()
    result := m.free
    if m.free != nil {
        m.free = m.free.next
    }
    return result
}

// Put returns a resource to the pool.
func (m *ResourceManager) Put(r *Resource) {
    m.mu.Lock()
    defer m.mu.Unlock()
    r.next = m.free
    m.free = r
}

这是一个在测试中使用的示例：

func TestResourceManager(t *testing.T) {

    // Add free resources to a manager.
    var m ResourceManager
    m.Put(&Resource{Resource: "/dev/a"})
    m.Put(&Resource{Resource: "/dev/b"})

    // Test that we can get all resources from the pool.

    ra := m.Get()
    rb := m.Get()
    if ra.Resource > rb.Resource {
        // Sort ra, rb to make test independent of order.
        ra, rb = rb, ra
    }
    if ra == nil || ra.Resource != "/dev/a" {
        t.Errorf("ra is %v, want /dev/a", ra)
    }
    if rb == nil || rb.Resource != "/dev/b" {
        t.Errorf("rb is %v, want /dev/b", rb)
    }

    // Check for empty pool.

    r := m.Get()
    if r != nil {
        t.Errorf("r is %v, want nil", r)
    }

    // Return one resource and try again.

    m.Put(ra)
    ra = m.Get()
    if ra == nil || ra.Resource != "/dev/a" {
        t.Errorf("ra is %v, want /dev/a", ra)
    }
    r = m.Get()
    if r != nil {
        t.Errorf("r is %v, want nil", r)
    }

}

Run the test on the playground.

如果资源数量存在已知的合理限制，请使用频道。这种方法利用了运行时高度优化的通道实现。

type Resource struct {
    Resource string
}

type ResourceManager struct {
    free chan *Resource
}

// Get gets a free resource from the manager or returns
// nil when the manager is empty.
func (m *ResourceManager) Get() *Resource {
    select {
    case r := <-m.free:
        return r
    default:
        return nil
    }
}

// Put returns a resource to the pool.
func (m *ResourceManager) Put(r *Resource) {
    m.free <- r
}

// NewResourceManager returns a manager that can hold up to
// n free resources.
func NewResourceManager(n int) *ResourceManager {
    return &ResourceManager{free: make(chan *Resource, n)}
}

使用上述TestResourceManager 函数测试此实现，但将var m ResourceManager 替换为m := NewResourceManager(4)。

Run the test on the Go playground.

Answer 2

给定资源是否正在使用不是Resource 本身的属性，而是ResourceManager 的属性。

事实上，没有必要跟踪正在使用的资源（除非您出于问题中未提及的某些原因需要）。正在使用的资源在释放时可以简单地放回池中。

这是一个使用通道的可能实现。不需要一个互斥体，也不需要任何原子 CAS。

package main

import (
    fmt "fmt"
    "time"
)

type Resource struct {
    Data string
}

type ResourceManager struct {
    resources []*Resource
    closeCh   chan struct{}
    acquireCh chan *Resource
    releaseCh chan *Resource
}

func NewResourceManager() *ResourceManager {
    r := &ResourceManager{
        closeCh:   make(chan struct{}),
        acquireCh: make(chan *Resource),
        releaseCh: make(chan *Resource),
    }
    go r.run()
    return r
}

func (r *ResourceManager) run() {
    defer close(r.acquireCh)
    for {
        if len(r.resources) > 0 {
            select {
            case r.acquireCh <- r.resources[len(r.resources)-1]:
                r.resources = r.resources[:len(r.resources)-1]
            case res := <-r.releaseCh:
                r.resources = append(r.resources, res)
            case <-r.closeCh:
                return
            }
        } else {
            select {
            case res := <-r.releaseCh:
                r.resources = append(r.resources, res)
            case <-r.closeCh:
                return
            }
        }
    }
}

func (r *ResourceManager) AcquireResource() *Resource {
    return <-r.acquireCh
}

func (r *ResourceManager) ReleaseResource(res *Resource) {
    r.releaseCh <- res
}

func (r *ResourceManager) Close() {
    close(r.closeCh)
}

// small demo below ...

func test(id int, r *ResourceManager) {
    for {
        res := r.AcquireResource()
        fmt.Printf("test %d: %s\n", id, res.Data)
        time.Sleep(time.Millisecond)
        r.ReleaseResource(res)
    }
}

func main() {
    r := NewResourceManager()
    r.ReleaseResource(&Resource{"Resource A"}) // initial setup
    r.ReleaseResource(&Resource{"Resource B"}) // initial setup
    go test(1, r)
    go test(2, r)
    go test(3, r) // 3 consumers, but only 2 resources ...
    time.Sleep(time.Second)
    r.Close()
}