commit 635ad2a42f10a5b24f08021782b71b4cf8326e19 parent 86e8e7fd21f8fc3eea3a55a4928ca1a68c8af3f2 Author: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Mon, 20 Dec 2021 09:35:32 +0000 Update codeberg.org/gruf libraries and fix go-store issue (#347) * update codeberg.org/gruf/ libraries Signed-off-by: kim <grufwub@gmail.com> * another update Signed-off-by: kim <grufwub@gmail.com> Diffstat:
16 files changed, 482 insertions(+), 313 deletions(-)
diff --git 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 @@ -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 @@ -12,7 +12,7 @@ import ( 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 @@ -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 @@ -15,7 +15,7 @@ var ( // 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 @@ -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 @@ const ( // 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/clock.go b/vendor/codeberg.org/gruf/go-nowish/clock.go @@ -0,0 +1,132 @@ +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() +} diff --git a/vendor/codeberg.org/gruf/go-nowish/time.go b/vendor/codeberg.org/gruf/go-nowish/time.go @@ -1,141 +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 { - noCopy noCopy //nolint noCopy because a copy will fuck with atomics - - // 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 -func (c *Clock) Start(precision time.Duration) func() { - // Create ticker from duration - tick := time.NewTicker(precision) - - // 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 - b := c.Now().AppendFormat( - make([]byte, 0, len(c.format)), - c.format, - ) - - // Update the stored value and set valid! - atomic.StorePointer(&c.nowfmt, unsafe.Pointer(&b)) - 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)) -} - -// 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() -} diff --git a/vendor/codeberg.org/gruf/go-nowish/timeout.go b/vendor/codeberg.org/gruf/go-nowish/timeout.go @@ -6,113 +6,228 @@ import ( "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 + + // Receive first timeout duration + d := atomic.SwapInt64(&t.next, 0) - // Signal started - t.wg.Done() + // Indicate finished starting, this + // was left locked by t.start(). + t.mu.Unlock() - select { - // Timeout reached - case <-t.tk.C: - if !t.st.stop() /* a sneaky cancel! */ { - t.ch.recv() + 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 - defer t.ch.send() + break } + } - // Ensure timer stopped - if cancelled && !t.tk.Stop() { - <-t.tk.C - } + if cancelled { + // Release the .Cancel() + defer t.cncl.notify() + } + + // Mark as done + t.reset() +} - // Defer reset state - defer t.st.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 + } - // If timed out call hook - if !cancelled { - hook() + // Cancel called + case <-t.cncl.wait(): + cancelled = true + if !t.timer.Stop() { + <-t.timer.C } - }() - 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") + } + + // 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)) } - t.runTimeout(hook) - t.tk.Reset(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() +} + +// 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 } -// timeoutState provides a thread-safe timeout state mechanism -type timeoutState uint32 +// 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) +} -// start attempts to start the state, must be already reset, returns success -func (t *timeoutState) start() bool { - return atomic.CompareAndSwapUint32((*uint32)(t), 0, 1) +// 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 } -// stop attempts to stop the state, must already be started, returns success -func (t *timeoutState) stop() bool { - return atomic.CompareAndSwapUint32((*uint32)(t), 1, 2) +// 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 is fairly self explanatory -func (t *timeoutState) reset() { - atomic.StoreUint32((*uint32)(t), 0) +// 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 +// 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 @@ -2,6 +2,7 @@ package kv import ( "io" + "sync" "codeberg.org/gruf/go-errors" ) @@ -15,9 +16,14 @@ var ErrStateClosed = errors.New("store/kv: state closed") // 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 @@ -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() + return &StateRO{store: st} +} - // Create new store state (defer close) - state := &StateRO{store: st} - defer state.close() +// 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 state - do(state) + // Pass to fn + fn(state) } -// 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 +// 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() + return &StateRW{store: st} +} - // Create new store state (defer close) - state := &StateRW{store: st} - defer state.close() +// 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 state - do(state) + // 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 @@ -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 @@ -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 @@ -2,7 +2,6 @@ package storage import ( "io" - "sync" "codeberg.org/gruf/go-bytes" "codeberg.org/gruf/go-store/util" @@ -12,14 +11,12 @@ import ( // 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 @@ -9,14 +9,22 @@ import ( "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 @@ -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