From 635ad2a42f10a5b24f08021782b71b4cf8326e19 Mon Sep 17 00:00:00 2001 From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Mon, 20 Dec 2021 09:35:32 +0000 Subject: [PATCH] Update codeberg.org/gruf libraries and fix go-store issue (#347) * update codeberg.org/gruf/ libraries Signed-off-by: kim * another update Signed-off-by: kim --- go.mod | 8 +- go.sum | 14 +- vendor/codeberg.org/gruf/go-errors/data.go | 2 +- vendor/codeberg.org/gruf/go-logger/README.md | 6 +- vendor/codeberg.org/gruf/go-logger/clock.go | 2 +- .../gruf/go-logger/format_text.go | 52 +++- .../gruf/go-nowish/{time.go => clock.go} | 33 +-- vendor/codeberg.org/gruf/go-nowish/timeout.go | 247 +++++++++++++----- vendor/codeberg.org/gruf/go-store/kv/state.go | 63 ++++- vendor/codeberg.org/gruf/go-store/kv/store.go | 50 ++-- .../gruf/go-store/storage/block.go | 8 +- .../gruf/go-store/storage/disk.go | 8 +- .../gruf/go-store/storage/memory.go | 42 --- vendor/codeberg.org/gruf/go-store/util/fs.go | 20 +- vendor/modules.txt | 8 +- 15 files changed, 366 insertions(+), 197 deletions(-) rename vendor/codeberg.org/gruf/go-nowish/{time.go => clock.go} (75%) diff --git a/go.mod b/go.mod index 3f6dfbb2c..019607977 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/superseriousbusiness/gotosocial go 1.17 require ( - codeberg.org/gruf/go-store v1.1.2 + codeberg.org/gruf/go-store v1.1.5 github.com/ReneKroon/ttlcache v1.7.0 github.com/buckket/go-blurhash v1.1.0 github.com/coreos/go-oidc/v3 v3.1.0 @@ -45,12 +45,12 @@ require ( require ( codeberg.org/gruf/go-bytes v1.0.2 // indirect - codeberg.org/gruf/go-errors v1.0.3 // indirect + codeberg.org/gruf/go-errors v1.0.4 // indirect codeberg.org/gruf/go-fastpath v1.0.2 // indirect codeberg.org/gruf/go-hashenc v1.0.1 // indirect - codeberg.org/gruf/go-logger v1.3.1 // indirect + codeberg.org/gruf/go-logger v1.3.2 // indirect codeberg.org/gruf/go-mutexes v1.0.1 // indirect - codeberg.org/gruf/go-nowish v1.0.2 // indirect + codeberg.org/gruf/go-nowish v1.1.0 // indirect codeberg.org/gruf/go-pools v1.0.2 // indirect github.com/aymerick/douceur v0.2.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect diff --git a/go.sum b/go.sum index f49ca195c..5f09d42f1 100644 --- a/go.sum +++ b/go.sum @@ -51,25 +51,27 @@ codeberg.org/gruf/go-bytes v1.0.1/go.mod h1:1v/ibfaosfXSZtRdW2rWaVrDXMc9E3bsi/M9 codeberg.org/gruf/go-bytes v1.0.2 h1:malqE42Ni+h1nnYWBUAJaDDtEzF4aeN4uPN8DfMNNvo= codeberg.org/gruf/go-bytes v1.0.2/go.mod h1:1v/ibfaosfXSZtRdW2rWaVrDXMc9E3bsi/M9Ekx39cg= codeberg.org/gruf/go-cache v1.1.2/go.mod h1:/Dbc+xU72Op3hMn6x2PXF3NE9uIDFeS+sXPF00hN/7o= -codeberg.org/gruf/go-errors v1.0.3 h1:R0Scg9hStLejjN7+x7IWKn5I3nOI+y7J0Oc2gBcUpDY= -codeberg.org/gruf/go-errors v1.0.3/go.mod h1:rJ08LdIE79Jg8vZ2TGylz/I+tZ1UuMJkGK5mNambIfQ= +codeberg.org/gruf/go-errors v1.0.4 h1:jOJCn/GMb6ELLRVlnmpimGRC2CbTreH5/CBZNWh9GZA= +codeberg.org/gruf/go-errors v1.0.4/go.mod h1:rJ08LdIE79Jg8vZ2TGylz/I+tZ1UuMJkGK5mNambIfQ= codeberg.org/gruf/go-fastpath v1.0.1/go.mod h1:edveE/Kp3Eqi0JJm0lXYdkVrB28cNUkcb/bRGFTPqeI= codeberg.org/gruf/go-fastpath v1.0.2 h1:O3nuYPMXnN89dsgAwVFU5iCGINtPJdITWmbRe2an/iQ= codeberg.org/gruf/go-fastpath v1.0.2/go.mod h1:edveE/Kp3Eqi0JJm0lXYdkVrB28cNUkcb/bRGFTPqeI= codeberg.org/gruf/go-hashenc v1.0.1 h1:EBvNe2wW8IPMUqT1XihB6/IM6KMJDLMFBxIUvmsy1f8= codeberg.org/gruf/go-hashenc v1.0.1/go.mod h1:IfHhPCVScOiYmJLqdCQT9bYVS1nxNTV4ewMUvFWDPtc= -codeberg.org/gruf/go-logger v1.3.1 h1:1f10GQAkVbd3gNdpfSNHOVfaTFLLS8ebuA7IRXd8n90= codeberg.org/gruf/go-logger v1.3.1/go.mod h1:tBduUc+Yb9vqGRxY9/FB0ZlYznSteLy/KmIANo7zFjA= +codeberg.org/gruf/go-logger v1.3.2 h1:/2Cg8Tmu6H10lljq/BvHE+76O2d4tDNUDwitN6YUxxk= +codeberg.org/gruf/go-logger v1.3.2/go.mod h1:q4xmTSdaxPzfndSXVF1X2xcyCVk7Nd/PIWCDs/4biMg= codeberg.org/gruf/go-mutexes v1.0.1 h1:X9bZW74YSEplWWdCrVXAvue5ztw3w5hh+INdXTENu88= codeberg.org/gruf/go-mutexes v1.0.1/go.mod h1:y2hbGLkWVHhNyxBOIVsA3/y2QMm6RSrYsC3sLVZ4EXM= codeberg.org/gruf/go-nowish v1.0.0/go.mod h1:70nvICNcqQ9OHpF07N614Dyk7cpL5ToWU1K1ZVCec2s= -codeberg.org/gruf/go-nowish v1.0.2 h1:/y8g38x44sD8JeqBPCkzqLoe0pReR1CTF8p6jXCOG1s= codeberg.org/gruf/go-nowish v1.0.2/go.mod h1:70nvICNcqQ9OHpF07N614Dyk7cpL5ToWU1K1ZVCec2s= +codeberg.org/gruf/go-nowish v1.1.0 h1:rj1z0AXDhLvnxs/DazWFxYAugs6rv5vhgWJkRCgrESg= +codeberg.org/gruf/go-nowish v1.1.0/go.mod h1:70nvICNcqQ9OHpF07N614Dyk7cpL5ToWU1K1ZVCec2s= codeberg.org/gruf/go-pools v1.0.2 h1:B0X6yoCL9FVmnvyoizb1SYRwMYPWwEJBjPnBMM5ILos= codeberg.org/gruf/go-pools v1.0.2/go.mod h1:MjUV3H6IASyBeBPCyCr7wjPpSNu8E2N87LG4r4TAyok= codeberg.org/gruf/go-runners v1.1.1/go.mod h1:9gTrmMnO3d+50C+hVzcmGBf+zTuswReS278E2EMvnmw= -codeberg.org/gruf/go-store v1.1.2 h1:yf7osOqSOlJ9WNsFdp8e6IZCfnoT6VoI66d5SuP4Nsg= -codeberg.org/gruf/go-store v1.1.2/go.mod h1:1CVRMdBbR0drn5pwz01aDT1Yls9W66u7E5kBiP9F9jw= +codeberg.org/gruf/go-store v1.1.5 h1:fp28vzGD15OsAF51CCwi7woH+Y3vb0aMl4OFh9JSjA0= +codeberg.org/gruf/go-store v1.1.5/go.mod h1:Q6ev500ddKghDQ8KS4IstL/W9fptDKa2T9oeHP+tXsI= dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= diff --git a/vendor/codeberg.org/gruf/go-errors/data.go b/vendor/codeberg.org/gruf/go-errors/data.go index c8c8f72e3..3b242f03c 100644 --- a/vendor/codeberg.org/gruf/go-errors/data.go +++ b/vendor/codeberg.org/gruf/go-errors/data.go @@ -12,7 +12,7 @@ var logfmt = logger.TextFormat{ Strict: false, Verbose: true, - MaxDepth: 10, + MaxDepth: 5, } // KV is a structure for setting key-value pairs in ErrorData. diff --git a/vendor/codeberg.org/gruf/go-logger/README.md b/vendor/codeberg.org/gruf/go-logger/README.md index a255a2ea9..57410ea87 100644 --- a/vendor/codeberg.org/gruf/go-logger/README.md +++ b/vendor/codeberg.org/gruf/go-logger/README.md @@ -6,4 +6,8 @@ Supports logging in 2 modes: Running without locks isn't likely to cause you any issues*, but if it does, you can wrap your `io.Writer` using `AddSafety()` when instantiating your new Logger. Even when running the benchmarks, this library has no printing issues without locks, so in most cases you'll be fine, but the safety is there if you need it. -*most logging libraries advertising high speeds are likely not performing mutex locks, which is why with this library you have the option to opt-in/out of them. \ No newline at end of file +*most logging libraries advertising high speeds are likely not performing mutex locks, which is why with this library you have the option to opt-in/out of them. + +Note there are 2 uses of the unsafe package: +- safer interface nil value checks, uses similar logic to reflect package to check if the value in the internal fat pointer is nil +- casting a byte slice to string to allow sharing of similar byte and string methods, performs same logic as `strings.Builder{}.String()` \ No newline at end of file diff --git a/vendor/codeberg.org/gruf/go-logger/clock.go b/vendor/codeberg.org/gruf/go-logger/clock.go index aed97c339..cc7d7ed0c 100644 --- a/vendor/codeberg.org/gruf/go-logger/clock.go +++ b/vendor/codeberg.org/gruf/go-logger/clock.go @@ -15,7 +15,7 @@ // startClock starts the global nowish clock. func startClock() { clockOnce.Do(func() { - clock.Start(time.Millisecond * 10) + clock.Start(time.Millisecond * 100) clock.SetFormat("2006-01-02 15:04:05") }) } diff --git a/vendor/codeberg.org/gruf/go-logger/format_text.go b/vendor/codeberg.org/gruf/go-logger/format_text.go index 50912d025..f9c90f887 100644 --- a/vendor/codeberg.org/gruf/go-logger/format_text.go +++ b/vendor/codeberg.org/gruf/go-logger/format_text.go @@ -45,7 +45,8 @@ func (f TextFormat) fmt(buf *bytes.Buffer) format { } return format{ flags: flags, - depth: uint16(f.MaxDepth) << 8, + curd: 0, + maxd: f.MaxDepth, buf: buf, } } @@ -179,7 +180,9 @@ func (f TextFormat) AppendMsgf(buf *bytes.Buffer, s string, a ...interface{}) { // format is the object passed among the append___ formatting functions type format struct { flags uint8 // 'isKey' and 'verbose' flags - depth uint16 // encoded as 0b(maxDepth)(curDepth) + drefs uint8 // current value deref count + curd uint8 // current depth + maxd uint8 // maximum depth buf *bytes.Buffer // out buffer } @@ -191,7 +194,12 @@ type format struct { // AtMaxDepth returns whether format is currently at max depth. func (f format) AtMaxDepth() bool { - return uint8(f.depth) >= uint8(f.depth>>8) + return f.curd >= f.maxd +} + +// Derefs returns no. times current value has been dereferenced. +func (f format) Derefs() uint8 { + return f.drefs } // IsKey returns whether the isKey flag is set. @@ -214,7 +222,9 @@ func (f format) SetIsKey(is bool) format { } return format{ flags: flags, - depth: f.depth, + drefs: f.drefs, + curd: f.curd, + maxd: f.maxd, buf: f.buf, } } @@ -223,16 +233,37 @@ func (f format) SetIsKey(is bool) format { func (f format) IncrDepth() format { return format{ flags: f.flags, - depth: (f.depth & 0b1111111100000000) | uint16(uint8(f.depth)+1), + drefs: f.drefs, + curd: f.curd + 1, + maxd: f.maxd, buf: f.buf, } } +// IncrDerefs returns format instance with dereference count incremented. +func (f format) IncrDerefs() format { + return format{ + flags: f.flags, + drefs: f.drefs + 1, + curd: f.curd, + maxd: f.maxd, + buf: f.buf, + } +} + +// appendType appends a type using supplied type str. +func appendType(fmt format, t string) { + for i := uint8(0); i < fmt.Derefs(); i++ { + fmt.buf.WriteByte('*') + } + fmt.buf.WriteString(t) +} + // appendNilType writes nil to buf, type included if verbose. func appendNilType(fmt format, t string) { if fmt.Verbose() { fmt.buf.WriteByte('(') - fmt.buf.WriteString(t) + appendType(fmt, t) fmt.buf.WriteString(`)(nil)`) } else { fmt.buf.WriteString(`nil`) @@ -243,7 +274,7 @@ func appendNilType(fmt format, t string) { func appendNilIface(fmt format, i interface{}) { if fmt.Verbose() { fmt.buf.WriteByte('(') - fmt.buf.WriteString(reflect.TypeOf(i).String()) + appendType(fmt, reflect.TypeOf(i).String()) fmt.buf.WriteString(`)(nil)`) } else { fmt.buf.WriteString(`nil`) @@ -254,7 +285,7 @@ func appendNilIface(fmt format, i interface{}) { func appendNilRValue(fmt format, v reflect.Value) { if fmt.Verbose() { fmt.buf.WriteByte('(') - fmt.buf.WriteString(v.Type().String()) + appendType(fmt, v.Type().String()) fmt.buf.WriteString(`)(nil)`) } else { fmt.buf.WriteString(`nil`) @@ -272,7 +303,8 @@ func appendBytes(fmt format, b []byte) { // Values CAN be nil formatted appendNilType(fmt, `[]byte`) } else { - appendString(fmt, bytes.BytesToString(b)) + // unsafe cast as string to prevent reallocation + appendString(fmt, *(*string)(unsafe.Pointer(&b))) } } @@ -707,7 +739,7 @@ func appendRValue(fmt format, v reflect.Value) { if v.IsNil() { appendNilRValue(fmt, v) } else { - appendRValue(fmt, v.Elem()) + appendRValue(fmt.IncrDerefs(), v.Elem()) } case reflect.UnsafePointer: fmt.buf.WriteString("(unsafe.Pointer)") diff --git a/vendor/codeberg.org/gruf/go-nowish/time.go b/vendor/codeberg.org/gruf/go-nowish/clock.go similarity index 75% rename from vendor/codeberg.org/gruf/go-nowish/time.go rename to vendor/codeberg.org/gruf/go-nowish/clock.go index 674ff5669..781e59f18 100644 --- a/vendor/codeberg.org/gruf/go-nowish/time.go +++ b/vendor/codeberg.org/gruf/go-nowish/clock.go @@ -16,8 +16,6 @@ func Start(precision time.Duration) (*Clock, func()) { } type Clock struct { - noCopy noCopy //nolint noCopy because a copy will fuck with atomics - // format stores the time formatting style string format string @@ -35,11 +33,13 @@ type Clock struct { now unsafe.Pointer } -// Start starts the clock with the provided precision, the -// returned function is the stop function for the underlying timer +// 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) + tick := time.NewTicker(precision / 10) // Set initial time t := time.Now() @@ -63,7 +63,7 @@ func (c *Clock) Start(precision time.Duration) func() { return tick.Stop } -// run is the internal clock ticking loop +// run is the internal clock ticking loop. func (c *Clock) run(tick *time.Ticker) { for { // Wait on tick @@ -83,12 +83,12 @@ func (c *Clock) run(tick *time.Ticker) { } } -// Now returns a good (ish) estimate of the current 'now' time +// 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 +// 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 { @@ -105,27 +105,18 @@ func (c *Clock) NowFormat() string { } // Calculate time format - b := c.Now().AppendFormat( - make([]byte, 0, len(c.format)), - c.format, - ) + nowfmt := c.Now().Format(c.format) // Update the stored value and set valid! - atomic.StorePointer(&c.nowfmt, unsafe.Pointer(&b)) + atomic.StorePointer(&c.nowfmt, unsafe.Pointer(&nowfmt)) atomic.StoreUint32(&c.valid, 1) // Unlock and return c.mutex.Unlock() - - // Note: - // it's safe to do this conversion here - // because this byte slice will never change. - // and we have the direct pointer to it, we're - // not requesting it atomicly via c.Format - return *(*string)(unsafe.Pointer(&b)) + return nowfmt } -// SetFormat sets the time format string used by .NowFormat() +// SetFormat sets the time format string used by .NowFormat(). func (c *Clock) SetFormat(format string) { // Get mutex lock c.mutex.Lock() diff --git a/vendor/codeberg.org/gruf/go-nowish/timeout.go b/vendor/codeberg.org/gruf/go-nowish/timeout.go index 1097f5d3d..7fe3e1d1d 100644 --- a/vendor/codeberg.org/gruf/go-nowish/timeout.go +++ b/vendor/codeberg.org/gruf/go-nowish/timeout.go @@ -6,113 +6,228 @@ "time" ) -// Timeout provides a reusable structure for enforcing timeouts with a cancel +// Timeout provides a reusable structure for enforcing timeouts with a cancel. type Timeout struct { - noCopy noCopy //nolint noCopy because a copy will mess with atomics - - tk *time.Timer // tk is the underlying timeout-timer - ch syncer // ch is the cancel synchronization channel - wg sync.WaitGroup // wg is the waitgroup to hold .Start() until timeout goroutine started - st timeoutState // st stores the current timeout state (and protects concurrent use) + 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 +// NewTimeout returns a new Timeout instance. func NewTimeout() Timeout { - tk := time.NewTimer(time.Minute) - tk.Stop() // don't keep it running + timer := time.NewTimer(time.Minute) + timer.Stop() // don't keep it running return Timeout{ - tk: tk, - ch: make(syncer), + timer: timer, + cncl: make(syncer), } } -func (t *Timeout) runTimeout(hook func()) { - t.wg.Add(1) - go func() { - cancelled := false +// 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 - // Signal started - t.wg.Done() + // Receive first timeout duration + d := atomic.SwapInt64(&t.next, 0) - select { - // Timeout reached - case <-t.tk.C: - if !t.st.stop() /* a sneaky cancel! */ { - t.ch.recv() + // 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 - defer t.ch.send() + break } + } - // Cancel called - case <-t.ch: + 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 - defer t.ch.send() } - // Ensure timer stopped - if cancelled && !t.tk.Stop() { - <-t.tk.C + // Cancel called + case <-t.cncl.wait(): + cancelled = true + if !t.timer.Stop() { + <-t.timer.C } + } - // Defer reset state - defer t.st.reset() - - // If timed out call hook - if !cancelled { - hook() - } - }() - t.wg.Wait() + return cancelled } // Start starts the timer with supplied timeout. If timeout is reached before -// cancel then supplied timeout hook will be called. Error may be called if -// Timeout is already running when this function is called +// 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.st.start() { - panic("nowish: timeout already started") + if !t.start() { + t.mu.Unlock() // need to unlock + panic("timeout already started") } - t.runTimeout(hook) - t.tk.Reset(d) + + // 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 +// this function will return after the timeout goroutine is finished. func (t *Timeout) Cancel() { - if !t.st.stop() { + if !t.cancel() { return } - t.ch.send() - t.ch.recv() + t.cncl.notify() + <-t.cncl.wait() } -// timeoutState provides a thread-safe timeout state mechanism -type timeoutState uint32 +// possible timeout states. +const ( + stopped = 0 + started = 1 + timingOut = 2 + cancelled = 3 + timedOut = 4 +) -// start attempts to start the state, must be already reset, returns success -func (t *timeoutState) start() bool { - return atomic.CompareAndSwapUint32((*uint32)(t), 0, 1) +// 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 } -// stop attempts to stop the state, must already be started, returns success -func (t *timeoutState) stop() bool { - return atomic.CompareAndSwapUint32((*uint32)(t), 1, 2) +// 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 } -// reset is fairly self explanatory -func (t *timeoutState) reset() { - atomic.StoreUint32((*uint32)(t), 0) +// 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) } -// syncer provides helpful receiver methods for a synchronization channel +// 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{}) -// send blocks on sending an empty value down channel -func (s syncer) send() { +// notify blocks on sending an empty value down channel. +func (s syncer) notify() { s <- struct{}{} } -// recv blocks on receiving (and dropping) empty value from channel -func (s syncer) recv() { - <-s +// wait returns the underlying channel for blocking until '.notify()'. +func (s syncer) wait() <-chan struct{} { + return s } diff --git a/vendor/codeberg.org/gruf/go-store/kv/state.go b/vendor/codeberg.org/gruf/go-store/kv/state.go index a8c1b9c82..20a3e951d 100644 --- a/vendor/codeberg.org/gruf/go-store/kv/state.go +++ b/vendor/codeberg.org/gruf/go-store/kv/state.go @@ -2,6 +2,7 @@ import ( "io" + "sync" "codeberg.org/gruf/go-errors" ) @@ -15,9 +16,14 @@ // then the state has zero guarantees type StateRO struct { store *KVStore + mutex sync.RWMutex } 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 @@ -28,6 +34,10 @@ func (st *StateRO) Get(key string) ([]byte, error) { } 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 @@ -38,6 +48,10 @@ func (st *StateRO) GetStream(key string) (io.ReadCloser, error) { } 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 @@ -47,8 +61,16 @@ func (st *StateRO) Has(key string) (bool, error) { return st.store.has(key) } -func (st *StateRO) close() { - st.store = nil +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 + } } // StateRW provides a read-write window to the store. While this @@ -58,9 +80,14 @@ func (st *StateRO) close() { // then the state has zero guarantees type StateRW struct { store *KVStore + mutex sync.RWMutex } 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 @@ -71,6 +98,10 @@ func (st *StateRW) Get(key string) ([]byte, error) { } 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 @@ -81,6 +112,10 @@ func (st *StateRW) GetStream(key string) (io.ReadCloser, error) { } 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 @@ -91,6 +126,10 @@ func (st *StateRW) Put(key string, value []byte) error { } 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 @@ -101,6 +140,10 @@ func (st *StateRW) PutStream(key string, r io.Reader) error { } 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 @@ -111,6 +154,10 @@ func (st *StateRW) Has(key string) (bool, error) { } 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 @@ -120,6 +167,14 @@ func (st *StateRW) Delete(key string) error { return st.store.delete(key) } -func (st *StateRW) close() { - st.store = nil +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 + } } diff --git a/vendor/codeberg.org/gruf/go-store/kv/store.go b/vendor/codeberg.org/gruf/go-store/kv/store.go index 148e3f33d..34fe91987 100644 --- a/vendor/codeberg.org/gruf/go-store/kv/store.go +++ b/vendor/codeberg.org/gruf/go-store/kv/store.go @@ -212,32 +212,34 @@ func (st *KVStore) Iterator(matchFn func(string) bool) (*KVIterator, error) { }, nil } -// Read provides a read-only window to the store, holding it in a read-locked state until -// the supplied function returns -func (st *KVStore) Read(do func(*StateRO)) { - // Get store read lock +// 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() - defer st.mutex.RUnlock() - - // Create new store state (defer close) - state := &StateRO{store: st} - defer state.close() - - // Pass state - do(state) + return &StateRO{store: st} } -// Update provides a read-write window to the store, holding it in a read-write-locked state -// until the supplied functions returns -func (st *KVStore) Update(do func(*StateRW)) { - // Get store lock +// ReadFn provides a read-only window to the store, holding it in a read-locked state until fn return. +func (st *KVStore) ReadFn(fn func(*StateRO)) { + // Acquire read-only state + state := st.Read() + defer state.Release() + + // Pass to fn + fn(state) +} + +// 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() - defer st.mutex.Unlock() - - // Create new store state (defer close) - state := &StateRW{store: st} - defer state.close() - - // Pass state - do(state) + return &StateRW{store: st} +} + +// UpdateFn provides a read-write window to the store, holding it in a write-locked state until fn return. +func (st *KVStore) UpdateFn(fn func(*StateRW)) { + // Acquire read-write state + state := st.Update() + defer state.Release() + + // Pass to fn + fn(state) } diff --git a/vendor/codeberg.org/gruf/go-store/storage/block.go b/vendor/codeberg.org/gruf/go-store/storage/block.go index 112b67d80..9a8c4dc7d 100644 --- a/vendor/codeberg.org/gruf/go-store/storage/block.go +++ b/vendor/codeberg.org/gruf/go-store/storage/block.go @@ -585,8 +585,8 @@ func (st *BlockStorage) WalkKeys(opts WalkKeysOptions) error { // nodePathForKey calculates the node file path for supplied key func (st *BlockStorage) nodePathForKey(key string) (string, error) { - // Path separators are illegal - if strings.Contains(key, "/") { + // Path separators are illegal, as directory paths + if strings.Contains(key, "/") || key == "." || key == ".." { return "", ErrInvalidKey } @@ -594,6 +594,10 @@ func (st *BlockStorage) nodePathForKey(key string) (string, error) { pb := util.GetPathBuilder() defer util.PutPathBuilder(pb) + // Append the nodepath to key + pb.AppendString(st.nodePath) + pb.AppendString(key) + // Return joined + cleaned node-path return pb.Join(st.nodePath, key), nil } diff --git a/vendor/codeberg.org/gruf/go-store/storage/disk.go b/vendor/codeberg.org/gruf/go-store/storage/disk.go index cbc365eed..060d56688 100644 --- a/vendor/codeberg.org/gruf/go-store/storage/disk.go +++ b/vendor/codeberg.org/gruf/go-store/storage/disk.go @@ -69,7 +69,6 @@ func getDiskConfig(cfg *DiskConfig) DiskConfig { // DiskStorage is a Storage implementation that stores directly to a filesystem type DiskStorage struct { path string // path is the root path of this store - dots int // dots is the "dotdot" count for the root store path bufp pools.BufferPool // bufp is the buffer pool for this DiskStorage config DiskConfig // cfg is the supplied configuration for this store } @@ -120,7 +119,6 @@ func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) { // Return new DiskStorage return &DiskStorage{ path: path, - dots: util.CountDotdots(path), bufp: pools.NewBufferPool(config.WriteBufSize), config: config, }, nil @@ -282,10 +280,10 @@ func (st *DiskStorage) filepath(key string) (string, error) { pb.AppendString(st.path) pb.AppendString(key) - // If path is dir traversal, and traverses FURTHER - // than store root, this is an error - if util.CountDotdots(pb.StringPtr()) > st.dots { + // Check for dir traversal outside of root + if util.IsDirTraversal(st.path, pb.StringPtr()) { return "", ErrInvalidKey } + return pb.String(), nil } diff --git a/vendor/codeberg.org/gruf/go-store/storage/memory.go b/vendor/codeberg.org/gruf/go-store/storage/memory.go index e30aef571..be60fa464 100644 --- a/vendor/codeberg.org/gruf/go-store/storage/memory.go +++ b/vendor/codeberg.org/gruf/go-store/storage/memory.go @@ -2,7 +2,6 @@ import ( "io" - "sync" "codeberg.org/gruf/go-bytes" "codeberg.org/gruf/go-store/util" @@ -12,14 +11,12 @@ // pairs in a Go map in-memory. The map is protected by a mutex. type MemoryStorage struct { fs map[string][]byte - mu sync.Mutex } // OpenMemory opens a new MemoryStorage instance with internal map of 'size'. func OpenMemory(size int) *MemoryStorage { return &MemoryStorage{ fs: make(map[string][]byte, size), - mu: sync.Mutex{}, } } @@ -30,33 +27,19 @@ func (st *MemoryStorage) Clean() error { // ReadBytes implements Storage.ReadBytes(). func (st *MemoryStorage) ReadBytes(key string) ([]byte, error) { - // Safely check store - st.mu.Lock() b, ok := st.fs[key] - st.mu.Unlock() - - // Return early if not exist if !ok { return nil, ErrNotFound } - - // Create return copy return bytes.Copy(b), nil } // ReadStream implements Storage.ReadStream(). func (st *MemoryStorage) ReadStream(key string) (io.ReadCloser, error) { - // Safely check store - st.mu.Lock() b, ok := st.fs[key] - st.mu.Unlock() - - // Return early if not exist if !ok { return nil, ErrNotFound } - - // Create io.ReadCloser from 'b' copy b = bytes.Copy(b) r := bytes.NewReader(b) return util.NopReadCloser(r), nil @@ -64,68 +47,43 @@ 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 - st.mu.Lock() _, ok := st.fs[key] - - // Check for already exist if ok { - st.mu.Unlock() return ErrAlreadyExists } - - // Write + unlock st.fs[key] = bytes.Copy(b) - st.mu.Unlock() return nil } // WriteStream implements Storage.WriteStream(). func (st *MemoryStorage) WriteStream(key string, r io.Reader) error { - // Read all from reader b, err := io.ReadAll(r) if err != nil { return err } - - // Write to storage return st.WriteBytes(key, b) } // Stat implements Storage.Stat(). func (st *MemoryStorage) Stat(key string) (bool, error) { - st.mu.Lock() _, ok := st.fs[key] - st.mu.Unlock() return ok, nil } // Remove implements Storage.Remove(). func (st *MemoryStorage) Remove(key string) error { - // Safely check store - st.mu.Lock() _, ok := st.fs[key] - - // Check in store if !ok { - st.mu.Unlock() return ErrNotFound } - - // Delete + unlock delete(st.fs, key) - st.mu.Unlock() return nil } // WalkKeys implements Storage.WalkKeys(). func (st *MemoryStorage) WalkKeys(opts WalkKeysOptions) error { - // Safely walk storage keys - st.mu.Lock() for key := range st.fs { opts.WalkFn(entry(key)) } - st.mu.Unlock() - return nil } diff --git a/vendor/codeberg.org/gruf/go-store/util/fs.go b/vendor/codeberg.org/gruf/go-store/util/fs.go index fa6a9d2c4..93b37a261 100644 --- a/vendor/codeberg.org/gruf/go-store/util/fs.go +++ b/vendor/codeberg.org/gruf/go-store/util/fs.go @@ -9,14 +9,22 @@ "codeberg.org/gruf/go-fastpath" ) -var dotdot = "../" +// IsDirTraversal will check if rootPlusPath is a dir traversal outside of root, +// assuming that both are cleaned and that rootPlusPath is path.Join(root, somePath) +func IsDirTraversal(root string, rootPlusPath string) bool { + switch { + // Root is $PWD, check for traversal out of + case root == ".": + return strings.HasPrefix(rootPlusPath, "../") -// CountDotdots returns the number of "dot-dots" (../) in a cleaned filesystem path -func CountDotdots(path string) int { - if !strings.HasSuffix(path, dotdot) { - return 0 + // The path MUST be prefixed by root + case !strings.HasPrefix(rootPlusPath, root): + return true + + // In all other cases, check not equal + default: + return len(root) == len(rootPlusPath) } - return strings.Count(path, dotdot) } // WalkDir traverses the dir tree of the supplied path, performing the supplied walkFn on each entry diff --git a/vendor/modules.txt b/vendor/modules.txt index 9c1ce34a9..051463c5e 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,7 +1,7 @@ # codeberg.org/gruf/go-bytes v1.0.2 ## explicit; go 1.14 codeberg.org/gruf/go-bytes -# codeberg.org/gruf/go-errors v1.0.3 +# codeberg.org/gruf/go-errors v1.0.4 ## explicit; go 1.15 codeberg.org/gruf/go-errors # codeberg.org/gruf/go-fastpath v1.0.2 @@ -10,19 +10,19 @@ codeberg.org/gruf/go-fastpath # codeberg.org/gruf/go-hashenc v1.0.1 ## explicit; go 1.16 codeberg.org/gruf/go-hashenc -# codeberg.org/gruf/go-logger v1.3.1 +# codeberg.org/gruf/go-logger v1.3.2 ## explicit; go 1.14 codeberg.org/gruf/go-logger # codeberg.org/gruf/go-mutexes v1.0.1 ## explicit; go 1.14 codeberg.org/gruf/go-mutexes -# codeberg.org/gruf/go-nowish v1.0.2 +# codeberg.org/gruf/go-nowish v1.1.0 ## explicit; go 1.14 codeberg.org/gruf/go-nowish # codeberg.org/gruf/go-pools v1.0.2 ## explicit; go 1.16 codeberg.org/gruf/go-pools -# codeberg.org/gruf/go-store v1.1.2 +# codeberg.org/gruf/go-store v1.1.5 ## explicit; go 1.14 codeberg.org/gruf/go-store/kv codeberg.org/gruf/go-store/storage