Multiple concurrent dynamic locks and timeouts if failure to acquire locks

I have a use case where I need to lock on arguments of a function.

The function itself can be accessed concurrently

Function signature is something like

func (m objectType) operate(key string) (bool) {
    // get lock on "key" (return false if unable to get lock in X ms - eg: 100 ms)
	// operate
    // release lock on "key"
    return true;
}

The data space which can be locked is in the range of millions (~10 million)

Concurrent access to operate() is in the range of thousands (1 - 5k)

Expected contention is low though possible in case of hotspots in key (hence the lock)

What is the right way to implement this ? Few options I explored using a concurrent hash map

1. sync.Map - this is suited for cases with append only entries and high read ratio compared to writes. Hence not applicable here
2. sharded hashmap where each shard is locked by RWMutex - https://github.com/orcaman/concurrent-map - While this would work, concurrency is limited by no of shards rather than actual contention between keys. Also doesn't enable the timeout scenarios when lot of contention happens for a subset of keys

Though timeout is a P1 requirement, the P0 requirement would be to increase throughput here by granular locking if possible.

Is there a good way to achieve this ?

I would do it by using a map of buffered channels:

• to acquire a "mutex", try to fill a buffered channel with a value
• work
• when done, empty the buffered channel so that another goroutine can use it

Example:

package main

import (
	"fmt"
	"sync"
	"time"
)

type MutexMap struct {
	mut     sync.RWMutex        // handle concurrent access of chanMap
	chanMap map[int](chan bool) // dynamic mutexes map
}

func NewMutextMap() *MutexMap {
	var mut sync.RWMutex
	return &MutexMap{
		mut:     mut,
		chanMap: make(map[int](chan bool)),
	}
}

// Acquire a lock, with timeout
func (mm *MutexMap) Lock(id int, timeout time.Duration) error {
	// get global lock to read from map and get a channel
	mm.mut.Lock()
	if _, ok := mm.chanMap[id]; !ok {
		mm.chanMap[id] = make(chan bool, 1)
	}
	ch := mm.chanMap[id]
	mm.mut.Unlock()

	// try to write to buffered channel, with timeout
	select {
	case ch <- true:
		return nil
	case <-time.After(timeout):
		return fmt.Errorf("working on %v just timed out", id)
	}
}

// release lock
func (mm *MutexMap) Release(id int) {
	mm.mut.Lock()
	ch := mm.chanMap[id]
	mm.mut.Unlock()
	<-ch
}

func work(id int, mm *MutexMap) {
	// acquire lock with timeout
	if err := mm.Lock(id, 100*time.Millisecond); err != nil {
		fmt.Printf("ERROR: %s\n", err)
		return
	}
	fmt.Printf("working on task %v\n", id)
	// do some work...
	time.Sleep(time.Second)
	fmt.Printf("done working on %v\n", id)

	// release lock
	mm.Release(id)
}

func main() {
	mm := NewMutextMap()
	var wg sync.WaitGroup
	for i := 0; i < 50; i++ {
		wg.Add(1)
		id := i % 10
		go func(id int, mm *MutexMap, wg *sync.WaitGroup) {
			work(id, mm)
			defer wg.Done()
		}(id, mm, &wg)
	}
	wg.Wait()
}

EDIT: different version, where we also handle the concurrent access to the chanMap itself