From 157d64b4236d7edae44a325b0476452450e92e99 Mon Sep 17 00:00:00 2001 From: diamondburned Date: Sat, 19 Dec 2020 02:35:13 -0800 Subject: [PATCH] Moreatomic: Replaced invalid mutex with modified sync.Map impl --- go.mod | 1 - go.sum | 2 - internal/moreatomic/syncmap.go | 49 ++--- internal/moreatomic/syncmod/syncmod.go | 263 +++++++++++++++++++++++++ 4 files changed, 275 insertions(+), 40 deletions(-) create mode 100644 internal/moreatomic/syncmod/syncmod.go diff --git a/go.mod b/go.mod index 90721c5..6d8e5e6 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,6 @@ go 1.13 require ( github.com/gorilla/schema v1.2.0 github.com/gorilla/websocket v1.4.2 - github.com/kawasin73/umutex v0.2.1 github.com/pkg/errors v0.9.1 golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897 golang.org/x/sys v0.0.0-20201018230417-eeed37f84f13 // indirect diff --git a/go.sum b/go.sum index e55a9f1..07e39d5 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,6 @@ github.com/gorilla/schema v1.2.0 h1:YufUaxZYCKGFuAq3c96BOhjgd5nmXiOY9NGzF247Tsc= github.com/gorilla/schema v1.2.0/go.mod h1:kgLaKoK1FELgZqMAVxx/5cbj0kT+57qxUrAlIO2eleU= github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -github.com/kawasin73/umutex v0.2.1 h1:Onkzz3LKs1HThskVwdhhBocqdRQqwCZ03quDJzuPzPo= -github.com/kawasin73/umutex v0.2.1/go.mod h1:A02N2muKVFMvFlp5c+hBycgdH964YtieGs+7mYB16NU= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M= diff --git a/internal/moreatomic/syncmap.go b/internal/moreatomic/syncmap.go index 389020b..481892c 100644 --- a/internal/moreatomic/syncmap.go +++ b/internal/moreatomic/syncmap.go @@ -1,64 +1,39 @@ package moreatomic -import "github.com/kawasin73/umutex" +import ( + "sync/atomic" + + "github.com/diamondburned/arikawa/v2/internal/moreatomic/syncmod" +) // Map is a thread-safe map that is a wrapper around sync.Map with slight API // additions. type Map struct { - upmu umutex.UMutex - smap map[interface{}]interface{} + val atomic.Value ctor func() interface{} } -type sentinelType struct{} - -var sentinel = sentinelType{} - func NewMap(ctor func() interface{}) *Map { - return &Map{ - smap: map[interface{}]interface{}{}, - ctor: ctor, - } + sm := &Map{ctor: ctor} + sm.Reset() + return sm } // Reset swaps the internal map out with a fresh one, dropping the old map. This // method never errors. func (sm *Map) Reset() error { - sm.upmu.Lock() - sm.smap = map[interface{}]interface{}{} - sm.upmu.Unlock() + sm.val.Store(&syncmod.Map{New: sm.ctor}) return nil } // LoadOrStore loads an existing value or stores a new value created from the // given constructor then return that value. func (sm *Map) LoadOrStore(k interface{}) (lv interface{}, loaded bool) { - sm.upmu.RLock() - - lv, loaded = sm.smap[k] - if !loaded { - lv = sm.ctor() - - // Wait until upgrade succeeds. - for !sm.upmu.Upgrade() { - } - - sm.smap[k] = lv - - sm.upmu.Unlock() - return - } - - sm.upmu.RUnlock() - return + return sm.val.Load().(*syncmod.Map).LoadOrStore(k) } // Load loads an existing value; it returns ok set to false if there is no // value with that key. func (sm *Map) Load(k interface{}) (lv interface{}, ok bool) { - sm.upmu.RLock() - defer sm.upmu.RUnlock() - - lv, ok = sm.smap[k] - return + return sm.val.Load().(*syncmod.Map).Load(k) } diff --git a/internal/moreatomic/syncmod/syncmod.go b/internal/moreatomic/syncmod/syncmod.go new file mode 100644 index 0000000..5478514 --- /dev/null +++ b/internal/moreatomic/syncmod/syncmod.go @@ -0,0 +1,263 @@ +// Copyright 2016 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package syncmod contains a clone of package sync's map.go file with unused +// methods removed and some tweaks with LoadOrStore. +package syncmod + +import ( + "sync" + "sync/atomic" + "unsafe" +) + +// Map is like a Go map[interface{}]interface{} but is safe for concurrent use +// by multiple goroutines without additional locking or coordination. +// Loads, stores, and deletes run in amortized constant time. +// +// The Map type is specialized. Most code should use a plain Go map instead, +// with separate locking or coordination, for better type safety and to make it +// easier to maintain other invariants along with the map content. +// +// The Map type is optimized for two common use cases: (1) when the entry for a given +// key is only ever written once but read many times, as in caches that only grow, +// or (2) when multiple goroutines read, write, and overwrite entries for disjoint +// sets of keys. In these two cases, use of a Map may significantly reduce lock +// contention compared to a Go map paired with a separate Mutex or RWMutex. +// +// The zero Map is empty and ready for use. A Map must not be copied after first use. +type Map struct { + New func() interface{} + + mu sync.Mutex + + // read contains the portion of the map's contents that are safe for + // concurrent access (with or without mu held). + // + // The read field itself is always safe to load, but must only be stored with + // mu held. + // + // Entries stored in read may be updated concurrently without mu, but updating + // a previously-expunged entry requires that the entry be copied to the dirty + // map and unexpunged with mu held. + read atomic.Value // readOnly + + // dirty contains the portion of the map's contents that require mu to be + // held. To ensure that the dirty map can be promoted to the read map quickly, + // it also includes all of the non-expunged entries in the read map. + // + // Expunged entries are not stored in the dirty map. An expunged entry in the + // clean map must be unexpunged and added to the dirty map before a new value + // can be stored to it. + // + // If the dirty map is nil, the next write to the map will initialize it by + // making a shallow copy of the clean map, omitting stale entries. + dirty map[interface{}]*entry + + // misses counts the number of loads since the read map was last updated that + // needed to lock mu to determine whether the key was present. + // + // Once enough misses have occurred to cover the cost of copying the dirty + // map, the dirty map will be promoted to the read map (in the unamended + // state) and the next store to the map will make a new dirty copy. + misses int +} + +// readOnly is an immutable struct stored atomically in the Map.read field. +type readOnly struct { + m map[interface{}]*entry + amended bool // true if the dirty map contains some key not in m. +} + +// expunged is an arbitrary pointer that marks entries which have been deleted +// from the dirty map. +var expunged = unsafe.Pointer(new(interface{})) + +// An entry is a slot in the map corresponding to a particular key. +type entry struct { + // p points to the interface{} value stored for the entry. + // + // If p == nil, the entry has been deleted and m.dirty == nil. + // + // If p == expunged, the entry has been deleted, m.dirty != nil, and the entry + // is missing from m.dirty. + // + // Otherwise, the entry is valid and recorded in m.read.m[key] and, if m.dirty + // != nil, in m.dirty[key]. + // + // An entry can be deleted by atomic replacement with nil: when m.dirty is + // next created, it will atomically replace nil with expunged and leave + // m.dirty[key] unset. + // + // An entry's associated value can be updated by atomic replacement, provided + // p != expunged. If p == expunged, an entry's associated value can be updated + // only after first setting m.dirty[key] = e so that lookups using the dirty + // map find the entry. + p unsafe.Pointer // *interface{} +} + +func newEntry(i interface{}) *entry { + return &entry{p: unsafe.Pointer(&i)} +} + +// Load returns the value stored in the map for a key, or nil if no +// value is present. +// The ok result indicates whether value was found in the map. +func (m *Map) Load(key interface{}) (value interface{}, ok bool) { + read, _ := m.read.Load().(readOnly) + e, ok := read.m[key] + if !ok && read.amended { + m.mu.Lock() + // Avoid reporting a spurious miss if m.dirty got promoted while we were + // blocked on m.mu. (If further loads of the same key will not miss, it's + // not worth copying the dirty map for this key.) + read, _ = m.read.Load().(readOnly) + e, ok = read.m[key] + if !ok && read.amended { + e, ok = m.dirty[key] + // Regardless of whether the entry was present, record a miss: this key + // will take the slow path until the dirty map is promoted to the read + // map. + m.missLocked() + } + m.mu.Unlock() + } + if !ok { + return nil, false + } + return e.load() +} + +func (e *entry) load() (value interface{}, ok bool) { + p := atomic.LoadPointer(&e.p) + if p == nil || p == expunged { + return nil, false + } + return *(*interface{})(p), true +} + +// unexpungeLocked ensures that the entry is not marked as expunged. +// +// If the entry was previously expunged, it must be added to the dirty map +// before m.mu is unlocked. +func (e *entry) unexpungeLocked() (wasExpunged bool) { + return atomic.CompareAndSwapPointer(&e.p, expunged, nil) +} + +// LoadOrStore returns the existing value for the key if present. +// Otherwise, it stores and returns the given value. +// The loaded result is true if the value was loaded, false if stored. +func (m *Map) LoadOrStore(k interface{}) (actual interface{}, loaded bool) { + // Avoid locking if it's a clean hit. + read, _ := m.read.Load().(readOnly) + if e, ok := read.m[k]; ok { + actual, loaded, ok = e.tryLoadOrStore(nil, m.New) + if ok { + return actual, loaded + } + } + + m.mu.Lock() + read, _ = m.read.Load().(readOnly) + if e, ok := read.m[k]; ok { + if e.unexpungeLocked() { + m.dirty[k] = e + } + actual, loaded, _ = e.tryLoadOrStore(actual, m.New) + } else if e, ok := m.dirty[k]; ok { + actual, loaded, _ = e.tryLoadOrStore(actual, m.New) + m.missLocked() + } else { + if !read.amended { + // We're adding the first new key to the dirty map. + // Make sure it is allocated and mark the read-only map as incomplete. + m.dirtyLocked() + m.read.Store(readOnly{m: read.m, amended: true}) + } + // This will likely allocate if the first tryLoadOrStore sees an + // expunged value and this else branch is hit. + if actual == nil { + actual = m.New() + } + m.dirty[k] = newEntry(actual) + loaded = false + } + m.mu.Unlock() + + return actual, loaded +} + +// tryLoadOrStore atomically loads or stores a value if the entry is not +// expunged. +// +// If the entry is expunged, tryLoadOrStore leaves the entry unchanged and +// returns with ok==false. +func (e *entry) tryLoadOrStore( + i interface{}, newFn func() interface{}) (actual interface{}, loaded, ok bool) { + + p := atomic.LoadPointer(&e.p) + if p == expunged { + return nil, false, false + } + if p != nil { + return *(*interface{})(p), true, true + } + + if i == nil { + i = newFn() + } + + // Copy the interface after the first load to make this method more amenable + // to escape analysis: if we hit the "load" path or the entry is expunged, we + // shouldn't bother heap-allocating. + ic := i + + for { + if atomic.CompareAndSwapPointer(&e.p, nil, unsafe.Pointer(&ic)) { + return i, false, true + } + p = atomic.LoadPointer(&e.p) + if p == expunged { + return i, false, false + } + if p != nil { + return *(*interface{})(p), true, true + } + } +} + +func (m *Map) missLocked() { + m.misses++ + if m.misses < len(m.dirty) { + return + } + m.read.Store(readOnly{m: m.dirty}) + m.dirty = nil + m.misses = 0 +} + +func (m *Map) dirtyLocked() { + if m.dirty != nil { + return + } + + read, _ := m.read.Load().(readOnly) + m.dirty = make(map[interface{}]*entry, len(read.m)) + for k, e := range read.m { + if !e.tryExpungeLocked() { + m.dirty[k] = e + } + } +} + +func (e *entry) tryExpungeLocked() (isExpunged bool) { + p := atomic.LoadPointer(&e.p) + for p == nil { + if atomic.CompareAndSwapPointer(&e.p, nil, expunged) { + return true + } + p = atomic.LoadPointer(&e.p) + } + return p == expunged +}