SuperSeriousBusiness/GoToSocial
1
1
Fork 0
mirror of https://github.com/superseriousbusiness/gotosocial synced 2025-06-05 21:59:39 +02:00
Code Issues Projects Releases Wiki Activity

upgrade go-store

Browse Source
This commit is contained in:
tsmethurst
2022-01-24 17:35:13 +01:00
parent 667e7f112c
commit f28cf793ee
44 changed files with 653 additions and 4972 deletions
Download Patch File Download Diff File Expand all files Collapse all files

343
vendor/codeberg.org/gruf/go-mutexes/map.go generated vendored
View File

@@ -1,105 +1,322 @@
package mutexes
import (
"runtime"
"sync"
"sync/atomic"
)
// locktype defines maskable mutexmap lock types.
type locktype uint8
const (
// possible lock types.
lockTypeRead = locktype(1) << 0
lockTypeWrite = locktype(1) << 1
lockTypeMap = locktype(1) << 2
// possible mutexmap states.
stateUnlockd = uint8(0)
stateRLocked = uint8(1)
stateLocked = uint8(2)
stateInUse = uint8(3)
)
// permitLockType returns if provided locktype is permitted to go ahead in current state.
func permitLockType(state uint8, lt locktype) bool {
switch state {
// Unlocked state
// (all allowed)
case stateUnlockd:
return true
// Keys locked, no state lock.
// (don't allow map locks)
case stateInUse:
return lt&lockTypeMap == 0
// Read locked
// (only allow read locks)
case stateRLocked:
return lt&lockTypeRead != 0
// Write locked
// (none allowed)
case stateLocked:
return false
// shouldn't reach here
default:
panic("unexpected state")
}
}
// MutexMap is a structure that allows having a map of self-evicting mutexes
// by key. You do not need to worry about managing the contents of the map,
// only requesting RLock/Lock for keys, and ensuring to call the returned
// unlock functions.
type MutexMap struct {
// NOTE:
// Individual keyed mutexes should ONLY ever
// be locked within the protection of the outer
// mapMu lock. If you lock these outside the
// protection of this, there is a chance for
// deadlocks
mus map[string]RWMutex
mapMu sync.Mutex
pool sync.Pool
queue []func()
evict []func()
count int32
maxmu int32
state uint8
}
// NewMap returns a new MutexMap instance based on supplied
// RWMutex allocator function, nil implies use default
func NewMap(newFn func() RWMutex) MutexMap {
if newFn == nil {
newFn = NewRW
// NewMap returns a new MutexMap instance with provided max no. open mutexes.
func NewMap(max int32) MutexMap {
if max < 1 {
// Default = 128 * GOMAXPROCS
procs := runtime.GOMAXPROCS(0)
max = int32(procs * 128)
}
return MutexMap{
mus: make(map[string]RWMutex),
mapMu: sync.Mutex{},
mus: make(map[string]RWMutex),
pool: sync.Pool{
New: func() interface{} {
return newFn()
return NewRW()
},
},
maxmu: max,
}
}
func (mm *MutexMap) evict(key string, mu RWMutex) {
// Acquire map lock
mm.mapMu.Lock()
// acquire will either acquire a mutex from pool or alloc.
func (mm *MutexMap) acquire() RWMutex {
return mm.pool.Get().(RWMutex)
}
// Toggle mutex lock to
// ensure it is unused
unlock := mu.Lock()
unlock()
// Delete mutex key
delete(mm.mus, key)
mm.mapMu.Unlock()
// Release to pool
// release will release provided mutex to pool.
func (mm *MutexMap) release(mu RWMutex) {
mm.pool.Put(mu)
}
// RLock acquires a mutex read lock for supplied key, returning an RUnlock function
func (mm *MutexMap) RLock(key string) func() {
return mm.getLock(key, func(mu RWMutex) func() {
return mu.RLock()
})
}
// spinLock will wait (using a mutex to sleep thread) until 'cond()' returns true,
// returning with map lock. Note that 'cond' is performed within a map lock.
func (mm *MutexMap) spinLock(cond func() bool) {
mu := mm.acquire()
defer mm.release(mu)
// Lock acquires a mutex lock for supplied key, returning an Unlock function
func (mm *MutexMap) Lock(key string) func() {
return mm.getLock(key, func(mu RWMutex) func() {
return mu.Lock()
})
}
for {
// Get map lock
mm.mapMu.Lock()
func (mm *MutexMap) getLock(key string, doLock func(RWMutex) func()) func() {
// Get map lock
mm.mapMu.Lock()
// Check if return
if cond() {
return
}
// Look for mutex
mu, ok := mm.mus[key]
if ok {
// Lock and return
// its unlocker func
unlock := doLock(mu)
// Queue ourselves
unlock := mu.Lock()
mm.queue = append(mm.queue, unlock)
mm.mapMu.Unlock()
return unlock
// Wait on notify
mu.Lock()()
}
}
// lockMutex will acquire a lock on the mutex at provided key, handling earlier allocated mutex if provided. Unlocks map on return.
func (mm *MutexMap) lockMutex(key string, lt locktype) func() {
var unlock func()
// Incr counter
mm.count++
// Check for existing mutex at key
mu, ok := mm.mus[key]
if !ok {
// Alloc from pool
mu = mm.acquire()
mm.mus[key] = mu
// Queue mutex for eviction
mm.evict = append(mm.evict, func() {
delete(mm.mus, key)
mm.pool.Put(mu)
})
}
// Note: even though the mutex data structure is
// small, benchmarking does actually show that pooled
// alloc of mutexes here is faster
// If no state, set in use.
// State will already have been
// set if this is from LockState{}
if mm.state == stateUnlockd {
mm.state = stateInUse
}
// Acquire mu + add
mu = mm.pool.Get().(RWMutex)
mm.mus[key] = mu
switch {
// Read lock
case lt&lockTypeRead != 0:
unlock = mu.RLock()
// Lock mutex + unlock map
unlockFn := doLock(mu)
// Write lock
case lt&lockTypeWrite != 0:
unlock = mu.Lock()
// shouldn't reach here
default:
panic("unexpected lock type")
}
// Unlock map + return
mm.mapMu.Unlock()
return func() {
mm.mapMu.Lock()
unlock()
go mm.onUnlock()
}
}
// onUnlock is performed as the final (async) stage of releasing an acquired key / map mutex.
func (mm *MutexMap) onUnlock() {
// Decr counter
mm.count--
if mm.count < 1 {
// Perform all queued evictions
for i := 0; i < len(mm.evict); i++ {
mm.evict[i]()
}
// Notify all waiting goroutines
for i := 0; i < len(mm.queue); i++ {
mm.queue[i]()
}
// Reset the map state
mm.evict = nil
mm.queue = nil
mm.state = stateUnlockd
}
// Finally, unlock
mm.mapMu.Unlock()
}
// RLockMap acquires a read lock over the entire map, returning a lock state for acquiring key read locks.
// Please note that the 'unlock()' function will block until all keys locked from this state are unlocked.
func (mm *MutexMap) RLockMap() *LockState {
return mm.getMapLock(lockTypeRead)
}
// LockMap acquires a write lock over the entire map, returning a lock state for acquiring key read/write locks.
// Please note that the 'unlock()' function will block until all keys locked from this state are unlocked.
func (mm *MutexMap) LockMap() *LockState {
return mm.getMapLock(lockTypeWrite)
}
// RLock acquires a mutex read lock for supplied key, returning an RUnlock function.
func (mm *MutexMap) RLock(key string) (runlock func()) {
return mm.getLock(key, lockTypeRead)
}
// Lock acquires a mutex write lock for supplied key, returning an Unlock function.
func (mm *MutexMap) Lock(key string) (unlock func()) {
return mm.getLock(key, lockTypeWrite)
}
// getLock will fetch lock of provided type, for given key, returning unlock function.
func (mm *MutexMap) getLock(key string, lt locktype) func() {
// Spin until achieve lock
mm.spinLock(func() bool {
return permitLockType(mm.state, lt) &&
mm.count < mm.maxmu // not overloaded
})
// Perform actual mutex lock
return mm.lockMutex(key, lt)
}
// getMapLock will acquire a map lock of provided type, returning a LockState session.
func (mm *MutexMap) getMapLock(lt locktype) *LockState {
// Spin until achieve lock
mm.spinLock(func() bool {
return permitLockType(mm.state, lt|lockTypeMap) &&
mm.count < mm.maxmu // not overloaded
})
// Incr counter
mm.count++
switch {
// Set read lock state
case lt&lockTypeRead != 0:
mm.state = stateRLocked
// Set write lock state
case lt&lockTypeWrite != 0:
mm.state = stateLocked
default:
panic("unexpected lock type")
}
// Unlock + return
mm.mapMu.Unlock()
return &LockState{
mmap: mm,
ltyp: lt,
}
}
// LockState represents a window to a locked MutexMap.
type LockState struct {
wait sync.WaitGroup
mmap *MutexMap
done uint32
ltyp locktype
}
// Lock: see MutexMap.Lock() definition. Will panic if map only read locked.
func (st *LockState) Lock(key string) (unlock func()) {
return st.getLock(key, lockTypeWrite)
}
// RLock: see MutexMap.RLock() definition.
func (st *LockState) RLock(key string) (runlock func()) {
return st.getLock(key, lockTypeRead)
}
// UnlockMap will close this state and release the currently locked map.
func (st *LockState) UnlockMap() {
// Set state to finished (or panic if already done)
if !atomic.CompareAndSwapUint32(&st.done, 0, 1) {
panic("called UnlockMap() on expired state")
}
// Wait until done
st.wait.Wait()
// Async reset map
st.mmap.mapMu.Lock()
go st.mmap.onUnlock()
}
// getLock: see MutexMap.getLock() definition.
func (st *LockState) getLock(key string, lt locktype) func() {
st.wait.Add(1) // track lock
// Check if closed, or if write lock is allowed
if atomic.LoadUint32(&st.done) == 1 {
panic("map lock closed")
} else if lt&lockTypeWrite != 0 &&
st.ltyp&lockTypeWrite == 0 {
panic("called .Lock() on rlocked map")
}
// Spin until achieve map lock
st.mmap.spinLock(func() bool {
return st.mmap.count < st.mmap.maxmu
}) // i.e. not overloaded
// Perform actual mutex lock
unlock := st.mmap.lockMutex(key, lt)
return func() {
// Unlock mutex
unlockFn()
// Release function
go mm.evict(key, mu)
unlock()
st.wait.Done()
}
}

67
vendor/codeberg.org/gruf/go-mutexes/mutex_timeout.go generated vendored
View File

@@ -3,8 +3,6 @@ package mutexes
import (
"sync"
"time"
"codeberg.org/gruf/go-nowish"
)
// TimeoutMutex defines a Mutex with timeouts on locks
@@ -73,14 +71,6 @@ func (mu *timeoutRWMutex) RLockFunc(fn func()) func() {
return mutexTimeout(mu.rd, mu.mu.RLock(), fn)
}
// timeoutPool provides nowish.Timeout objects for timeout mutexes
var timeoutPool = sync.Pool{
New: func() interface{} {
t := nowish.NewTimeout()
return &t
},
}
// mutexTimeout performs a timed unlock, calling supplied fn if timeout is reached
func mutexTimeout(d time.Duration, unlock func(), fn func()) func() {
if d < 1 {
@@ -88,18 +78,65 @@ func mutexTimeout(d time.Duration, unlock func(), fn func()) func() {
return unlock
}
// Acquire timeout obj
t := timeoutPool.Get().(*nowish.Timeout)
// Acquire timer from pool
t := timerPool.Get().(*timer)
// Start the timeout with hook
t.Start(d, fn)
// Start the timer
go t.Start(d, fn)
// Return func cancelling timeout,
// replacing Timeout in pool and
// finally unlocking mutex
return func() {
defer timerPool.Put(t)
t.Cancel()
timeoutPool.Put(t)
unlock()
}
}
// timerPool is the global &timer{} pool.
var timerPool = sync.Pool{
New: func() interface{} {
return newtimer()
},
}
// timer represents a reusable cancellable timer.
type timer struct {
t *time.Timer
c chan struct{}
}
// newtimer returns a new timer instance.
func newtimer() *timer {
t := time.NewTimer(time.Minute)
t.Stop()
return &timer{t: t, c: make(chan struct{})}
}
// Start will start the timer with duration 'd', performing 'fn' on timeout.
func (t *timer) Start(d time.Duration, fn func()) {
t.t.Reset(d)
select {
// Timed out
case <-t.t.C:
fn()
// Cancelled
case <-t.c:
}
}
// Cancel will attempt to cancel the running timer.
func (t *timer) Cancel() {
select {
// cancel successful
case t.c <- struct{}{}:
if !t.t.Stop() {
<-t.t.C
} // stop timer
// already stopped
default:
}
}

9
vendor/codeberg.org/gruf/go-nowish/LICENSE generated vendored
View File

@@ -1,9 +0,0 @@
MIT License
Copyright (c) 2021 gruf
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

3
vendor/codeberg.org/gruf/go-nowish/README.md generated vendored
View File

@@ -1,3 +0,0 @@
a simple Go library with useful time utiities:
- Clock: a high performance clock giving a good "ish" representation of "now" (hence the name!)
- Timeout: a reusable structure for enforcing timeouts with a cancel

132
vendor/codeberg.org/gruf/go-nowish/clock.go generated vendored
View File

@@ -1,132 +0,0 @@
package nowish
import (
"sync"
"sync/atomic"
"time"
"unsafe"
)
// Start returns a new Clock instance initialized and
// started with the provided precision, along with the
// stop function for it's underlying timer
func Start(precision time.Duration) (*Clock, func()) {
c := Clock{}
return &c, c.Start(precision)
}
type Clock struct {
// format stores the time formatting style string
format string
// valid indicates whether the current value stored in .Format is valid
valid uint32
// mutex protects writes to .Format, not because it would be unsafe, but
// because we want to minimize unnnecessary allocations
mutex sync.Mutex
// nowfmt is an unsafe pointer to the last-updated time format string
nowfmt unsafe.Pointer
// now is an unsafe pointer to the last-updated time.Time object
now unsafe.Pointer
}
// Start starts the clock with the provided precision, the returned
// function is the stop function for the underlying timer. For >= 2ms,
// actual precision is usually within AT LEAST 10% of requested precision,
// less than this and the actual precision very quickly deteriorates.
func (c *Clock) Start(precision time.Duration) func() {
// Create ticker from duration
tick := time.NewTicker(precision / 10)
// Set initial time
t := time.Now()
atomic.StorePointer(&c.now, unsafe.Pointer(&t))
// Set initial format
s := ""
atomic.StorePointer(&c.nowfmt, unsafe.Pointer(&s))
// If formatting string unset, set default
c.mutex.Lock()
if c.format == "" {
c.format = time.RFC822
}
c.mutex.Unlock()
// Start main routine
go c.run(tick)
// Return stop fn
return tick.Stop
}
// run is the internal clock ticking loop.
func (c *Clock) run(tick *time.Ticker) {
for {
// Wait on tick
_, ok := <-tick.C
// Channel closed
if !ok {
break
}
// Update time
t := time.Now()
atomic.StorePointer(&c.now, unsafe.Pointer(&t))
// Invalidate format string
atomic.StoreUint32(&c.valid, 0)
}
}
// Now returns a good (ish) estimate of the current 'now' time.
func (c *Clock) Now() time.Time {
return *(*time.Time)(atomic.LoadPointer(&c.now))
}
// NowFormat returns the formatted "now" time, cached until next tick and "now" updates.
func (c *Clock) NowFormat() string {
// If format still valid, return this
if atomic.LoadUint32(&c.valid) == 1 {
return *(*string)(atomic.LoadPointer(&c.nowfmt))
}
// Get mutex lock
c.mutex.Lock()
// Double check still invalid
if atomic.LoadUint32(&c.valid) == 1 {
c.mutex.Unlock()
return *(*string)(atomic.LoadPointer(&c.nowfmt))
}
// Calculate time format
nowfmt := c.Now().Format(c.format)
// Update the stored value and set valid!
atomic.StorePointer(&c.nowfmt, unsafe.Pointer(&nowfmt))
atomic.StoreUint32(&c.valid, 1)
// Unlock and return
c.mutex.Unlock()
return nowfmt
}
// SetFormat sets the time format string used by .NowFormat().
func (c *Clock) SetFormat(format string) {
// Get mutex lock
c.mutex.Lock()
// Update time format
c.format = format
// Invalidate current format string
atomic.StoreUint32(&c.valid, 0)
// Unlock
c.mutex.Unlock()
}

233
vendor/codeberg.org/gruf/go-nowish/timeout.go generated vendored
View File

@@ -1,233 +0,0 @@
package nowish
import (
"sync"
"sync/atomic"
"time"
)
// Timeout provides a reusable structure for enforcing timeouts with a cancel.
type Timeout struct {
timer *time.Timer // timer is the underlying timeout-timer
cncl syncer // cncl is the cancel synchronization channel
next int64 // next is the next timeout duration to run on
state uint32 // state stores the current timeout state
mu sync.Mutex // mu protects state, and helps synchronize return of .Start()
}
// NewTimeout returns a new Timeout instance.
func NewTimeout() Timeout {
timer := time.NewTimer(time.Minute)
timer.Stop() // don't keep it running
return Timeout{
timer: timer,
cncl: make(syncer),
}
}
// startTimeout is the main timeout routine, handling starting the
// timeout runner at first and upon any time extensions, and handling
// any received cancels by stopping the running timer.
func (t *Timeout) startTimeout(hook func()) {
var cancelled bool
// Receive first timeout duration
d := atomic.SwapInt64(&t.next, 0)
// Indicate finished starting, this
// was left locked by t.start().
t.mu.Unlock()
for {
// Run supplied timeout
cancelled = t.runTimeout(d)
if cancelled {
break
}
// Check for extension or set timed out
d = atomic.SwapInt64(&t.next, 0)
if d < 1 {
if t.timedOut() {
// timeout reached
hook()
break
} else {
// already cancelled
t.cncl.wait()
cancelled = true
break
}
}
if !t.extend() {
// already cancelled
t.cncl.wait()
cancelled = true
break
}
}
if cancelled {
// Release the .Cancel()
defer t.cncl.notify()
}
// Mark as done
t.reset()
}
// runTimeout will until supplied timeout or cancel called.
func (t *Timeout) runTimeout(d int64) (cancelled bool) {
// Start the timer for 'd'
t.timer.Reset(time.Duration(d))
select {
// Timeout reached
case <-t.timer.C:
if !t.timingOut() {
// a sneaky cancel!
t.cncl.wait()
cancelled = true
}
// Cancel called
case <-t.cncl.wait():
cancelled = true
if !t.timer.Stop() {
<-t.timer.C
}
}
return cancelled
}
// Start starts the timer with supplied timeout. If timeout is reached before
// cancel then supplied timeout hook will be called. Panic will be called if
// Timeout is already running when calling this function.
func (t *Timeout) Start(d time.Duration, hook func()) {
if !t.start() {
t.mu.Unlock() // need to unlock
panic("timeout already started")
}
// Start the timeout
atomic.StoreInt64(&t.next, int64(d))
go t.startTimeout(hook)
// Wait until start
t.mu.Lock()
t.mu.Unlock()
}
// Extend will attempt to extend the timeout runner's time, returns false if not running.
func (t *Timeout) Extend(d time.Duration) bool {
var ok bool
if ok = t.running(); ok {
atomic.AddInt64(&t.next, int64(d))
}
return ok
}
// Cancel cancels the currently running timer. If a cancel is achieved, then
// this function will return after the timeout goroutine is finished.
func (t *Timeout) Cancel() {
if !t.cancel() {
return
}
t.cncl.notify()
<-t.cncl.wait()
}
// possible timeout states.
const (
stopped = 0
started = 1
timingOut = 2
cancelled = 3
timedOut = 4
)
// cas will perform a compare and swap where the compare is a provided function.
func (t *Timeout) cas(check func(uint32) bool, swap uint32) bool {
var cas bool
t.mu.Lock()
if cas = check(t.state); cas {
t.state = swap
}
t.mu.Unlock()
return cas
}
// start attempts to mark the timeout state as 'started', note DOES NOT unlock Timeout.mu.
func (t *Timeout) start() bool {
var ok bool
t.mu.Lock()
if ok = (t.state == stopped); ok {
t.state = started
}
// don't unlock
return ok
}
// timingOut attempts to mark the timeout state as 'timing out'.
func (t *Timeout) timingOut() bool {
return t.cas(func(u uint32) bool {
return (u == started)
}, timingOut)
}
// timedOut attempts mark the 'timing out' state as 'timed out'.
func (t *Timeout) timedOut() bool {
return t.cas(func(u uint32) bool {
return (u == timingOut)
}, timedOut)
}
// extend attempts to extend a 'timing out' state by moving it back to 'started'.
func (t *Timeout) extend() bool {
return t.cas(func(u uint32) bool {
return (u == started) ||
(u == timingOut)
}, started)
}
// running returns whether the state is anything other than 'stopped'.
func (t *Timeout) running() bool {
t.mu.Lock()
running := (t.state != stopped)
t.mu.Unlock()
return running
}
// cancel attempts to mark the timeout state as 'cancelled'.
func (t *Timeout) cancel() bool {
return t.cas(func(u uint32) bool {
return (u == started) ||
(u == timingOut)
}, cancelled)
}
// reset marks the timeout state as 'stopped'.
func (t *Timeout) reset() {
t.mu.Lock()
t.state = stopped
t.mu.Unlock()
}
// syncer provides helpful receiver methods for a synchronization channel.
type syncer (chan struct{})
// notify blocks on sending an empty value down channel.
func (s syncer) notify() {
s <- struct{}{}
}
// wait returns the underlying channel for blocking until '.notify()'.
func (s syncer) wait() <-chan struct{} {
return s
}

10
vendor/codeberg.org/gruf/go-nowish/util.go generated vendored
View File

@@ -1,10 +0,0 @@
package nowish
//nolint
type noCopy struct{}
//nolint
func (*noCopy) Lock() {}
//nolint
func (*noCopy) Unlock() {}

10
vendor/codeberg.org/gruf/go-store/kv/iterator.go generated vendored
View File

@@ -2,6 +2,7 @@ package kv
import (
"codeberg.org/gruf/go-errors"
"codeberg.org/gruf/go-mutexes"
"codeberg.org/gruf/go-store/storage"
)
@@ -17,10 +18,10 @@ var ErrIteratorClosed = errors.New("store/kv: iterator closed")
// have multiple iterators running concurrently
type KVIterator struct {
store *KVStore // store is the linked KVStore
state *mutexes.LockState
entries []storage.StorageEntry
index int
key string
onClose func()
}
// Next attempts to set the next key-value pair, the
@@ -43,13 +44,10 @@ func (i *KVIterator) Key() string {
// Release releases the KVIterator and KVStore's read lock
func (i *KVIterator) Release() {
// Reset key, path, entries
i.state.UnlockMap()
i.store = nil
i.key = ""
i.entries = nil
// Perform requested callback
i.onClose()
}
// Value returns the next value from the KVStore
@@ -60,5 +58,5 @@ func (i *KVIterator) Value() ([]byte, error) {
}
// Attempt to fetch from store
return i.store.get(i.store.mutexMap.RLock, i.key)
return i.store.get(i.state.RLock, i.key)
}

82
vendor/codeberg.org/gruf/go-store/kv/state.go generated vendored
View File

@@ -2,9 +2,9 @@ package kv
import (
"io"
"sync"
"codeberg.org/gruf/go-errors"
"codeberg.org/gruf/go-mutexes"
)
var ErrStateClosed = errors.New("store/kv: state closed")
@@ -16,61 +16,42 @@ var ErrStateClosed = errors.New("store/kv: state closed")
// then the state has zero guarantees
type StateRO struct {
store *KVStore
mutex sync.RWMutex
state *mutexes.LockState
}
func (st *StateRO) Get(key string) ([]byte, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return nil, ErrStateClosed
}
// Pass request to store
return st.store.get(st.store.mutexMap.RLock, key)
return st.store.get(st.state.RLock, key)
}
func (st *StateRO) GetStream(key string) (io.ReadCloser, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return nil, ErrStateClosed
}
// Pass request to store
return st.store.getStream(st.store.mutexMap.RLock, key)
return st.store.getStream(st.state.RLock, key)
}
func (st *StateRO) Has(key string) (bool, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return false, ErrStateClosed
}
// Pass request to store
return st.store.has(st.store.mutexMap.RLock, key)
return st.store.has(st.state.RLock, key)
}
func (st *StateRO) Release() {
// Get state write lock
st.mutex.Lock()
defer st.mutex.Unlock()
// Release the store
if st.store != nil {
st.store.mutex.RUnlock()
st.store = nil
}
st.state.UnlockMap()
st.store = nil
}
// StateRW provides a read-write window to the store. While this
@@ -80,101 +61,70 @@ func (st *StateRO) Release() {
// then the state has zero guarantees
type StateRW struct {
store *KVStore
mutex sync.RWMutex
state *mutexes.LockState
}
func (st *StateRW) Get(key string) ([]byte, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return nil, ErrStateClosed
}
// Pass request to store
return st.store.get(st.store.mutexMap.RLock, key)
return st.store.get(st.state.RLock, key)
}
func (st *StateRW) GetStream(key string) (io.ReadCloser, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return nil, ErrStateClosed
}
// Pass request to store
return st.store.getStream(st.store.mutexMap.RLock, key)
return st.store.getStream(st.state.RLock, key)
}
func (st *StateRW) Put(key string, value []byte) error {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return ErrStateClosed
}
// Pass request to store
return st.store.put(st.store.mutexMap.Lock, key, value)
return st.store.put(st.state.Lock, key, value)
}
func (st *StateRW) PutStream(key string, r io.Reader) error {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return ErrStateClosed
}
// Pass request to store
return st.store.putStream(st.store.mutexMap.Lock, key, r)
return st.store.putStream(st.state.Lock, key, r)
}
func (st *StateRW) Has(key string) (bool, error) {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return false, ErrStateClosed
}
// Pass request to store
return st.store.has(st.store.mutexMap.RLock, key)
return st.store.has(st.state.RLock, key)
}
func (st *StateRW) Delete(key string) error {
// Get state read lock
st.mutex.RLock()
defer st.mutex.RUnlock()
// Check not closed
if st.store == nil {
return ErrStateClosed
}
// Pass request to store
return st.store.delete(st.store.mutexMap.Lock, key)
return st.store.delete(st.state.Lock, key)
}
func (st *StateRW) Release() {
// Get state write lock
st.mutex.Lock()
defer st.mutex.Unlock()
// Release the store
if st.store != nil {
st.store.mutex.Unlock()
st.store = nil
}
st.state.UnlockMap()
st.store = nil
}

40
vendor/codeberg.org/gruf/go-store/kv/store.go generated vendored
View File

@@ -2,7 +2,6 @@ package kv
import (
"io"
"sync"
"codeberg.org/gruf/go-mutexes"
"codeberg.org/gruf/go-store/storage"
@@ -11,9 +10,8 @@ import (
// KVStore is a very simple, yet performant key-value store
type KVStore struct {
mutexMap mutexes.MutexMap // mutexMap is a map of keys to mutexes to protect file access
mutex sync.RWMutex // mutex is the total store mutex
storage storage.Storage // storage is the underlying storage
mutex mutexes.MutexMap // mutex is a map of keys to mutexes to protect file access
storage storage.Storage // storage is the underlying storage
}
func OpenFile(path string, cfg *storage.DiskConfig) (*KVStore, error) {
@@ -47,26 +45,19 @@ func OpenStorage(storage storage.Storage) (*KVStore, error) {
// Return new KVStore
return &KVStore{
mutexMap: mutexes.NewMap(mutexes.NewRW),
mutex: sync.RWMutex{},
storage: storage,
mutex: mutexes.NewMap(-1),
storage: storage,
}, nil
}
// RLock acquires a read-lock on supplied key, returning unlock function.
func (st *KVStore) RLock(key string) (runlock func()) {
st.mutex.RLock()
runlock = st.mutexMap.RLock(key)
st.mutex.RUnlock()
return runlock
return st.mutex.RLock(key)
}
// Lock acquires a write-lock on supplied key, returning unlock function.
func (st *KVStore) Lock(key string) (unlock func()) {
st.mutex.Lock()
unlock = st.mutexMap.Lock(key)
st.mutex.Unlock()
return unlock
return st.mutex.Lock(key)
}
// Get fetches the bytes for supplied key in the store
@@ -167,7 +158,7 @@ func (st *KVStore) Iterator(matchFn func(string) bool) (*KVIterator, error) {
}
// Get store read lock
st.mutex.RLock()
state := st.mutex.RLockMap()
// Setup the walk keys function
entries := []storage.StorageEntry{}
@@ -184,24 +175,24 @@ func (st *KVStore) Iterator(matchFn func(string) bool) (*KVIterator, error) {
// Walk keys in the storage
err := st.storage.WalkKeys(storage.WalkKeysOptions{WalkFn: walkFn})
if err != nil {
st.mutex.RUnlock()
state.UnlockMap()
return nil, err
}
// Return new iterator
return &KVIterator{
store: st,
state: state,
entries: entries,
index: -1,
key: "",
onClose: st.mutex.RUnlock,
}, nil
}
// Read provides a read-only window to the store, holding it in a read-locked state until release
func (st *KVStore) Read() *StateRO {
st.mutex.RLock()
return &StateRO{store: st}
state := st.mutex.RLockMap()
return &StateRO{store: st, state: state}
}
// ReadFn provides a read-only window to the store, holding it in a read-locked state until fn return.
@@ -216,8 +207,8 @@ func (st *KVStore) ReadFn(fn func(*StateRO)) {
// Update provides a read-write window to the store, holding it in a write-locked state until release
func (st *KVStore) Update() *StateRW {
st.mutex.Lock()
return &StateRW{store: st}
state := st.mutex.LockMap()
return &StateRW{store: st, state: state}
}
// UpdateFn provides a read-write window to the store, holding it in a write-locked state until fn return.
@@ -229,3 +220,8 @@ func (st *KVStore) UpdateFn(fn func(*StateRW)) {
// Pass to fn
fn(state)
}
// Close will close the underlying storage, the mutex map locking (e.g. RLock(), Lock() will still work).
func (st *KVStore) Close() error {
return st.storage.Close()
}

86
vendor/codeberg.org/gruf/go-store/storage/block.go generated vendored
View File

@@ -1,6 +1,7 @@
package storage
import (
"crypto/sha256"
"io"
"io/fs"
"os"
@@ -13,7 +14,6 @@ import (
"codeberg.org/gruf/go-hashenc"
"codeberg.org/gruf/go-pools"
"codeberg.org/gruf/go-store/util"
"github.com/zeebo/blake3"
)
var (
@@ -77,7 +77,7 @@ func getBlockConfig(cfg *BlockConfig) BlockConfig {
// BlockStorage is a Storage implementation that stores input data as chunks on
// a filesystem. Each value is chunked into blocks of configured size and these
// blocks are stored with name equal to their base64-encoded BLAKE3 hash-sum. A
// blocks are stored with name equal to their base64-encoded SHA256 hash-sum. A
// "node" file is finally created containing an array of hashes contained within
// this value
type BlockStorage struct {
@@ -87,7 +87,7 @@ type BlockStorage struct {
config BlockConfig // cfg is the supplied configuration for this store
hashPool sync.Pool // hashPool is this store's hashEncoder pool
bufpool pools.BufferPool // bufpool is this store's bytes.Buffer pool
lock *LockableFile // lock is the opened lockfile for this storage instance
lock *Lock // lock is the opened lockfile for this storage instance
// NOTE:
// BlockStorage does not need to lock each of the underlying block files
@@ -140,11 +140,9 @@ func OpenBlock(path string, cfg *BlockConfig) (*BlockStorage, error) {
}
// Open and acquire storage lock for path
lock, err := OpenLock(pb.Join(path, LockFile))
lock, err := OpenLock(pb.Join(path, lockFile))
if err != nil {
return nil, err
} else if err := lock.Lock(); err != nil {
return nil, err
}
// Figure out the largest size for bufpool slices
@@ -174,14 +172,23 @@ func OpenBlock(path string, cfg *BlockConfig) (*BlockStorage, error) {
// Clean implements storage.Clean()
func (st *BlockStorage) Clean() error {
nodes := map[string]*node{}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Acquire path builder
pb := util.GetPathBuilder()
defer util.PutPathBuilder(pb)
// Walk nodes dir for entries
nodes := map[string]*node{}
onceErr := errors.OnceError{}
// Walk nodes dir for entries
err := util.WalkDir(pb, st.nodePath, func(npath string, fsentry fs.DirEntry) {
// Only deal with regular files
if !fsentry.Type().IsRegular() {
@@ -303,6 +310,7 @@ func (st *BlockStorage) ReadBytes(key string) ([]byte, error) {
if err != nil {
return nil, err
}
defer rc.Close()
// Read all bytes and return
return io.ReadAll(rc)
@@ -316,9 +324,19 @@ func (st *BlockStorage) ReadStream(key string) (io.ReadCloser, error) {
return nil, err
}
// Track open
st.lock.Add()
// Check if open
if st.lock.Closed() {
st.lock.Done()
return nil, ErrClosed
}
// Attempt to open RO file
file, err := open(npath, defaultFileROFlags)
if err != nil {
st.lock.Done()
return nil, err
}
defer file.Close()
@@ -338,14 +356,16 @@ func (st *BlockStorage) ReadStream(key string) (io.ReadCloser, error) {
nil,
)
if err != nil {
st.lock.Done()
return nil, err
}
// Return new block reader
return util.NopReadCloser(&blockReader{
// Prepare block reader and return
rc := util.NopReadCloser(&blockReader{
storage: st,
node: &node,
}), nil
}) // we wrap the blockreader to decr lockfile waitgroup
return util.ReadCloserWithCallback(rc, st.lock.Done), nil
}
func (st *BlockStorage) readBlock(key string) ([]byte, error) {
@@ -383,6 +403,15 @@ func (st *BlockStorage) WriteStream(key string, r io.Reader) error {
return err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Check if this exists
ok, err := stat(key)
if err != nil {
@@ -567,6 +596,15 @@ func (st *BlockStorage) Stat(key string) (bool, error) {
return false, err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return false, ErrClosed
}
// Check for file on disk
return stat(kpath)
}
@@ -579,18 +617,35 @@ func (st *BlockStorage) Remove(key string) error {
return err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Attempt to remove file
return os.Remove(kpath)
}
// Close implements Storage.Close()
func (st *BlockStorage) Close() error {
defer st.lock.Close()
return st.lock.Unlock()
return st.lock.Close()
}
// WalkKeys implements Storage.WalkKeys()
func (st *BlockStorage) WalkKeys(opts WalkKeysOptions) error {
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Acquire path builder
pb := util.GetPathBuilder()
defer util.PutPathBuilder(pb)
@@ -800,7 +855,7 @@ var (
// encodedHashLen is the once-calculated encoded hash-sum length
encodedHashLen = base64Encoding.EncodedLen(
blake3.New().Size(),
sha256.New().Size(),
)
)
@@ -812,9 +867,8 @@ type hashEncoder struct {
// newHashEncoder returns a new hashEncoder instance
func newHashEncoder() *hashEncoder {
hash := blake3.New()
return &hashEncoder{
henc: hashenc.New(hash, base64Encoding),
henc: hashenc.New(sha256.New(), base64Encoding),
ebuf: make([]byte, encodedHashLen),
}
}

67
vendor/codeberg.org/gruf/go-store/storage/disk.go generated vendored
View File

@@ -71,7 +71,7 @@ type DiskStorage struct {
path string // path is the root path of this store
bufp pools.BufferPool // bufp is the buffer pool for this DiskStorage
config DiskConfig // cfg is the supplied configuration for this store
lock *LockableFile // lock is the opened lockfile for this storage instance
lock *Lock // lock is the opened lockfile for this storage instance
}
// OpenFile opens a DiskStorage instance for given folder path and configuration
@@ -118,11 +118,9 @@ func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) {
}
// Open and acquire storage lock for path
lock, err := OpenLock(pb.Join(path, LockFile))
lock, err := OpenLock(pb.Join(path, lockFile))
if err != nil {
return nil, err
} else if err := lock.Lock(); err != nil {
return nil, err
}
// Return new DiskStorage
@@ -136,6 +134,11 @@ func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) {
// Clean implements Storage.Clean()
func (st *DiskStorage) Clean() error {
st.lock.Add()
defer st.lock.Done()
if st.lock.Closed() {
return ErrClosed
}
return util.CleanDirs(st.path)
}
@@ -160,9 +163,18 @@ func (st *DiskStorage) ReadStream(key string) (io.ReadCloser, error) {
return nil, err
}
// Track open
st.lock.Add()
// Check if open
if st.lock.Closed() {
return nil, ErrClosed
}
// Attempt to open file (replace ENOENT with our own)
file, err := open(kpath, defaultFileROFlags)
if err != nil {
st.lock.Done()
return nil, errSwapNotFound(err)
}
@@ -170,12 +182,14 @@ func (st *DiskStorage) ReadStream(key string) (io.ReadCloser, error) {
cFile, err := st.config.Compression.Reader(file)
if err != nil {
file.Close() // close this here, ignore error
st.lock.Done()
return nil, err
}
// Wrap compressor to ensure file close
return util.ReadCloserWithCallback(cFile, func() {
file.Close()
st.lock.Done()
}), nil
}
@@ -192,6 +206,15 @@ func (st *DiskStorage) WriteStream(key string, r io.Reader) error {
return err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Ensure dirs leading up to file exist
err = os.MkdirAll(path.Dir(kpath), defaultDirPerms)
if err != nil {
@@ -242,6 +265,15 @@ func (st *DiskStorage) Stat(key string) (bool, error) {
return false, err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return false, ErrClosed
}
// Check for file on disk
return stat(kpath)
}
@@ -254,18 +286,35 @@ func (st *DiskStorage) Remove(key string) error {
return err
}
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Attempt to remove file
return os.Remove(kpath)
}
// Close implements Storage.Close()
func (st *DiskStorage) Close() error {
defer st.lock.Close()
return st.lock.Unlock()
return st.lock.Close()
}
// WalkKeys implements Storage.WalkKeys()
func (st *DiskStorage) WalkKeys(opts WalkKeysOptions) error {
// Track open
st.lock.Add()
defer st.lock.Done()
// Check if open
if st.lock.Closed() {
return ErrClosed
}
// Acquire path builder
pb := util.GetPathBuilder()
defer util.PutPathBuilder(pb)
@@ -286,13 +335,13 @@ func (st *DiskStorage) WalkKeys(opts WalkKeysOptions) error {
// filepath checks and returns a formatted filepath for given key
func (st *DiskStorage) filepath(key string) (string, error) {
// Calculate transformed key path
key = st.config.Transform.KeyToPath(key)
// Acquire path builder
pb := util.GetPathBuilder()
defer util.PutPathBuilder(pb)
// Calculate transformed key path
key = st.config.Transform.KeyToPath(key)
// Generated joined root path
pb.AppendString(st.path)
pb.AppendString(key)

14
vendor/codeberg.org/gruf/go-store/storage/errors.go generated vendored
View File

@@ -19,6 +19,9 @@ func (e errorString) Extend(s string, a ...interface{}) errorString {
}
var (
// ErrClosed is returned on operations on a closed storage
ErrClosed = errorString("store/storage: closed")
// ErrNotFound is the error returned when a key cannot be found in storage
ErrNotFound = errorString("store/storage: key not found")
@@ -39,6 +42,9 @@ var (
// errCorruptNodes is returned when nodes with missing blocks are found during a BlockStorage clean
errCorruptNodes = errorString("store/storage: corrupted nodes")
// ErrAlreadyLocked is returned on fail opening a storage lockfile
ErrAlreadyLocked = errorString("store/storage: storage lock already open")
)
// errSwapNoop performs no error swaps
@@ -61,3 +67,11 @@ func errSwapExist(err error) error {
}
return err
}
// errSwapUnavailable swaps syscall.EAGAIN for ErrAlreadyLocked
func errSwapUnavailable(err error) error {
if err == syscall.EAGAIN {
return ErrAlreadyLocked
}
return err
}

9
vendor/codeberg.org/gruf/go-store/storage/fs.go generated vendored
View File

@@ -8,11 +8,14 @@ import (
)
const (
defaultDirPerms = 0755
defaultFilePerms = 0644
// default file permission bits
defaultDirPerms = 0755
defaultFilePerms = 0644
// default file open flags
defaultFileROFlags = syscall.O_RDONLY
defaultFileRWFlags = syscall.O_CREAT | syscall.O_RDWR
defaultFileLockFlags = syscall.O_RDONLY | syscall.O_EXCL | syscall.O_CREAT
defaultFileLockFlags = syscall.O_RDONLY | syscall.O_CREAT
)
// NOTE:

83
vendor/codeberg.org/gruf/go-store/storage/lock.go generated vendored
View File

@@ -1,38 +1,81 @@
package storage
import (
"os"
"sync"
"sync/atomic"
"syscall"
"codeberg.org/gruf/go-store/util"
)
// LockFile is our standard lockfile name.
const LockFile = "store.lock"
// lockFile is our standard lockfile name.
var lockFile = "store.lock"
type LockableFile struct {
*os.File
// IsLockKey returns whether storage key is our lockfile.
func IsLockKey(key string) bool {
return key == lockFile
}
// Lock represents a filesystem lock to ensure only one storage instance open per path.
type Lock struct {
fd int
wg sync.WaitGroup
st uint32
}
// OpenLock opens a lockfile at path.
func OpenLock(path string) (*LockableFile, error) {
file, err := open(path, defaultFileLockFlags)
func OpenLock(path string) (*Lock, error) {
var fd int
// Open the file descriptor at path
err := util.RetryOnEINTR(func() (err error) {
fd, err = syscall.Open(path, defaultFileLockFlags, defaultFilePerms)
return
})
if err != nil {
return nil, err
}
return &LockableFile{file}, nil
}
func (f *LockableFile) Lock() error {
return f.flock(syscall.LOCK_EX | syscall.LOCK_NB)
}
func (f *LockableFile) Unlock() error {
return f.flock(syscall.LOCK_UN | syscall.LOCK_NB)
}
func (f *LockableFile) flock(how int) error {
return util.RetryOnEINTR(func() error {
return syscall.Flock(int(f.Fd()), how)
// Get a flock on the file descriptor
err = util.RetryOnEINTR(func() error {
return syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB)
})
if err != nil {
return nil, errSwapUnavailable(err)
}
return &Lock{fd: fd}, nil
}
// Add will add '1' to the underlying sync.WaitGroup.
func (f *Lock) Add() {
f.wg.Add(1)
}
// Done will decrememnt '1' from the underlying sync.WaitGroup.
func (f *Lock) Done() {
f.wg.Done()
}
// Close will attempt to close the lockfile and file descriptor.
func (f *Lock) Close() error {
var err error
if atomic.CompareAndSwapUint32(&f.st, 0, 1) {
// Wait until done
f.wg.Wait()
// Ensure gets closed
defer syscall.Close(f.fd)
// Call funlock on the file descriptor
err = util.RetryOnEINTR(func() error {
return syscall.Flock(f.fd, syscall.LOCK_UN|syscall.LOCK_NB)
})
}
return err
}
// Closed will return whether this lockfile has been closed (and unlocked).
func (f *Lock) Closed() bool {
return (atomic.LoadUint32(&f.st) == 1)
}

78
vendor/codeberg.org/gruf/go-store/storage/memory.go generated vendored
View File

@@ -14,6 +14,7 @@ type MemoryStorage struct {
ow bool // overwrites
fs map[string][]byte
mu sync.Mutex
st uint32
}
// OpenMemory opens a new MemoryStorage instance with internal map of 'size'.
@@ -27,13 +28,26 @@ func OpenMemory(size int, overwrites bool) *MemoryStorage {
// Clean implements Storage.Clean().
func (st *MemoryStorage) Clean() error {
st.mu.Lock()
defer st.mu.Unlock()
if st.st == 1 {
return ErrClosed
}
return nil
}
// ReadBytes implements Storage.ReadBytes().
func (st *MemoryStorage) ReadBytes(key string) ([]byte, error) {
// Safely check store
// Lock storage
st.mu.Lock()
// Check store open
if st.st == 1 {
st.mu.Unlock()
return nil, ErrClosed
}
// Check for key
b, ok := st.fs[key]
st.mu.Unlock()
@@ -48,8 +62,16 @@ func (st *MemoryStorage) ReadBytes(key string) ([]byte, error) {
// ReadStream implements Storage.ReadStream().
func (st *MemoryStorage) ReadStream(key string) (io.ReadCloser, error) {
// Safely check store
// Lock storage
st.mu.Lock()
// Check store open
if st.st == 1 {
st.mu.Unlock()
return nil, ErrClosed
}
// Check for key
b, ok := st.fs[key]
st.mu.Unlock()
@@ -66,19 +88,24 @@ func (st *MemoryStorage) ReadStream(key string) (io.ReadCloser, error) {
// WriteBytes implements Storage.WriteBytes().
func (st *MemoryStorage) WriteBytes(key string, b []byte) error {
// Safely check store
// Lock storage
st.mu.Lock()
defer st.mu.Unlock()
// Check store open
if st.st == 1 {
return ErrClosed
}
_, ok := st.fs[key]
// Check for already exist
if ok && !st.ow {
st.mu.Unlock()
return ErrAlreadyExists
}
// Write + unlock
st.fs[key] = bytes.Copy(b)
st.mu.Unlock()
return nil
}
@@ -96,43 +123,66 @@ func (st *MemoryStorage) WriteStream(key string, r io.Reader) error {
// Stat implements Storage.Stat().
func (st *MemoryStorage) Stat(key string) (bool, error) {
// Lock storage
st.mu.Lock()
defer st.mu.Unlock()
// Check store open
if st.st == 1 {
return false, ErrClosed
}
// Check for key
_, ok := st.fs[key]
st.mu.Unlock()
return ok, nil
}
// Remove implements Storage.Remove().
func (st *MemoryStorage) Remove(key string) error {
// Safely check store
// Lock storage
st.mu.Lock()
_, ok := st.fs[key]
defer st.mu.Unlock()
// Check in store
// Check store open
if st.st == 1 {
return ErrClosed
}
// Check for key
_, ok := st.fs[key]
if !ok {
st.mu.Unlock()
return ErrNotFound
}
// Delete + unlock
// Remove from store
delete(st.fs, key)
st.mu.Unlock()
return nil
}
// Close implements Storage.Close().
func (st *MemoryStorage) Close() error {
st.mu.Lock()
st.st = 1
st.mu.Unlock()
return nil
}
// WalkKeys implements Storage.WalkKeys().
func (st *MemoryStorage) WalkKeys(opts WalkKeysOptions) error {
// Safely walk storage keys
// Lock storage
st.mu.Lock()
defer st.mu.Unlock()
// Check store open
if st.st == 1 {
return ErrClosed
}
// Walk store keys
for key := range st.fs {
opts.WalkFn(entry(key))
}
st.mu.Unlock()
return nil
}