123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403 |
- // Copyright (c) 2018 Shivaram Lingamneni <slingamn@cs.stanford.edu>
- // released under the MIT license
-
- package history
-
- import (
- "github.com/oragono/oragono/irc/utils"
- "sync"
- "sync/atomic"
- "time"
- )
-
- type ItemType uint
-
- const (
- uninitializedItem ItemType = iota
- Privmsg
- Notice
- Join
- Part
- Kick
- Quit
- Mode
- Tagmsg
- Nick
- )
-
- const (
- initialAutoSize = 32
- )
-
- // a Tagmsg that consists entirely of transient tags is not stored
- var transientTags = map[string]bool{
- "+draft/typing": true,
- "+typing": true, // future-proofing
- }
-
- // Item represents an event (e.g., a PRIVMSG or a JOIN) and its associated data
- type Item struct {
- Type ItemType
-
- Nick string
- // this is the uncasefolded account name, if there's no account it should be set to "*"
- AccountName string
- // for non-privmsg items, we may stuff some other data in here
- Message utils.SplitMessage
- Tags map[string]string
- Params [1]string
- }
-
- // HasMsgid tests whether a message has the message id `msgid`.
- func (item *Item) HasMsgid(msgid string) bool {
- if item.Message.Msgid == msgid {
- return true
- }
- for _, pair := range item.Message.Wrapped {
- if pair.Msgid == msgid {
- return true
- }
- }
- return false
- }
-
- func (item *Item) isStorable() bool {
- if item.Type == Tagmsg {
- for name := range item.Tags {
- if !transientTags[name] {
- return true
- }
- }
- return false // all tags were blacklisted
- } else {
- return true
- }
- }
-
- type Predicate func(item Item) (matches bool)
-
- // Buffer is a ring buffer holding message/event history for a channel or user
- type Buffer struct {
- sync.RWMutex
-
- // ring buffer, see irc/whowas.go for conventions
- buffer []Item
- start int
- end int
- maximumSize int
- window time.Duration
-
- lastDiscarded time.Time
-
- enabled uint32
-
- nowFunc func() time.Time
- }
-
- func NewHistoryBuffer(size int, window time.Duration) (result *Buffer) {
- result = new(Buffer)
- result.Initialize(size, window)
- return
- }
-
- func (hist *Buffer) Initialize(size int, window time.Duration) {
- initialSize := size
- if window != 0 {
- initialSize = initialAutoSize
- if size < initialSize {
- initialSize = size // min(initialAutoSize, size)
- }
- }
- hist.buffer = make([]Item, initialSize)
- hist.start = -1
- hist.end = -1
- hist.window = window
- hist.maximumSize = size
- hist.nowFunc = time.Now
-
- hist.setEnabled(size)
- }
-
- func (hist *Buffer) setEnabled(size int) {
- var enabled uint32
- if size != 0 {
- enabled = 1
- }
- atomic.StoreUint32(&hist.enabled, enabled)
- }
-
- // Enabled returns whether the buffer is currently storing messages
- // (a disabled buffer blackholes everything it sees)
- func (list *Buffer) Enabled() bool {
- return atomic.LoadUint32(&list.enabled) != 0
- }
-
- // Add adds a history item to the buffer
- func (list *Buffer) Add(item Item) {
- // fast path without a lock acquisition for when we are not storing history
- if !list.Enabled() {
- return
- }
-
- if !item.isStorable() {
- return
- }
-
- if item.Message.Time.IsZero() {
- item.Message.Time = time.Now().UTC()
- }
-
- list.Lock()
- defer list.Unlock()
-
- list.maybeExpand()
-
- var pos int
- if list.start == -1 { // empty
- pos = 0
- list.start = 0
- list.end = 1 % len(list.buffer)
- } else if list.start != list.end { // partially full
- pos = list.end
- list.end = (list.end + 1) % len(list.buffer)
- } else if list.start == list.end { // full
- pos = list.end
- list.end = (list.end + 1) % len(list.buffer)
- list.start = list.end // advance start as well, overwriting first entry
- // record the timestamp of the overwritten item
- if list.lastDiscarded.Before(list.buffer[pos].Message.Time) {
- list.lastDiscarded = list.buffer[pos].Message.Time
- }
- }
-
- list.buffer[pos] = item
- }
-
- // Reverse reverses an []Item, in-place.
- func Reverse(results []Item) {
- for i, j := 0, len(results)-1; i < j; i, j = i+1, j-1 {
- results[i], results[j] = results[j], results[i]
- }
- }
-
- // Between returns all history items with a time `after` <= time <= `before`,
- // with an indication of whether the results are complete or are missing items
- // because some of that period was discarded. A zero value of `before` is considered
- // higher than all other times.
- func (list *Buffer) Between(after, before time.Time, ascending bool, limit int) (results []Item, complete bool) {
- if !list.Enabled() {
- return
- }
-
- list.RLock()
- defer list.RUnlock()
-
- complete = after.Equal(list.lastDiscarded) || after.After(list.lastDiscarded)
-
- satisfies := func(item Item) bool {
- return (after.IsZero() || item.Message.Time.After(after)) && (before.IsZero() || item.Message.Time.Before(before))
- }
-
- return list.matchInternal(satisfies, ascending, limit), complete
- }
-
- // Match returns all history items such that `predicate` returns true for them.
- // Items are considered in reverse insertion order if `ascending` is false, or
- // in insertion order if `ascending` is true, up to a total of `limit` matches
- // if `limit` > 0 (unlimited otherwise).
- // `predicate` MAY be a closure that maintains its own state across invocations;
- // it MUST NOT acquire any locks or otherwise do anything weird.
- // Results are always returned in insertion order.
- func (list *Buffer) Match(predicate Predicate, ascending bool, limit int) (results []Item) {
- if !list.Enabled() {
- return
- }
-
- list.RLock()
- defer list.RUnlock()
-
- return list.matchInternal(predicate, ascending, limit)
- }
-
- // you must be holding the read lock to call this
- func (list *Buffer) matchInternal(predicate Predicate, ascending bool, limit int) (results []Item) {
- if list.start == -1 {
- return
- }
-
- var pos, stop int
- if ascending {
- pos = list.start
- stop = list.prev(list.end)
- } else {
- pos = list.prev(list.end)
- stop = list.start
- }
-
- for {
- if predicate(list.buffer[pos]) {
- results = append(results, list.buffer[pos])
- }
- if pos == stop || (limit != 0 && len(results) == limit) {
- break
- }
- if ascending {
- pos = list.next(pos)
- } else {
- pos = list.prev(pos)
- }
- }
-
- // TODO sort by time instead?
- if !ascending {
- Reverse(results)
- }
- return
- }
-
- // Latest returns the items most recently added, up to `limit`. If `limit` is 0,
- // it returns all items.
- func (list *Buffer) Latest(limit int) (results []Item) {
- matchAll := func(item Item) bool { return true }
- return list.Match(matchAll, false, limit)
- }
-
- // LastDiscarded returns the latest time of any entry that was evicted
- // from the ring buffer.
- func (list *Buffer) LastDiscarded() time.Time {
- list.RLock()
- defer list.RUnlock()
-
- return list.lastDiscarded
- }
-
- func (list *Buffer) prev(index int) int {
- switch index {
- case 0:
- return len(list.buffer) - 1
- default:
- return index - 1
- }
- }
-
- func (list *Buffer) next(index int) int {
- switch index {
- case len(list.buffer) - 1:
- return 0
- default:
- return index + 1
- }
- }
-
- // return n such that v <= n and n == 2**i for some i
- func roundUpToPowerOfTwo(v int) int {
- // http://graphics.stanford.edu/~seander/bithacks.html
- v -= 1
- v |= v >> 1
- v |= v >> 2
- v |= v >> 4
- v |= v >> 8
- v |= v >> 16
- return v + 1
- }
-
- func (list *Buffer) maybeExpand() {
- if list.window == 0 {
- return // autoresize is disabled
- }
-
- length := list.length()
- if length < len(list.buffer) {
- return // we have spare capacity already
- }
-
- if len(list.buffer) == list.maximumSize {
- return // cannot expand any further
- }
-
- wouldDiscard := list.buffer[list.start].Message.Time
- if list.window < list.nowFunc().Sub(wouldDiscard) {
- return // oldest element is old enough to overwrite
- }
-
- newSize := roundUpToPowerOfTwo(length + 1)
- if list.maximumSize < newSize {
- newSize = list.maximumSize
- }
- list.resize(newSize)
- }
-
- // Resize shrinks or expands the buffer
- func (list *Buffer) Resize(maximumSize int, window time.Duration) {
- list.Lock()
- defer list.Unlock()
-
- if list.maximumSize == maximumSize && list.window == window {
- return // no-op
- }
-
- list.maximumSize = maximumSize
- list.window = window
-
- // if we're not autoresizing, we need to resize now;
- // if we are autoresizing, we may need to shrink the buffer down to maximumSize,
- // but we don't need to grow it now (we can just grow it on the next Add)
- // TODO make it possible to shrink the buffer so that it only contains `window`
- if window == 0 || maximumSize < len(list.buffer) {
- list.resize(maximumSize)
- }
- }
-
- func (list *Buffer) resize(size int) {
- newbuffer := make([]Item, size)
-
- list.setEnabled(size)
-
- if list.start == -1 {
- // indices are already correct and nothing needs to be copied
- } else if size == 0 {
- // this is now the empty list
- list.start = -1
- list.end = -1
- } else {
- currentLength := list.length()
- start := list.start
- end := list.end
- // if we're truncating, keep the latest entries, not the earliest
- if size < currentLength {
- start = list.end - size
- if start < 0 {
- start += len(list.buffer)
- }
- // update lastDiscarded for discarded entries
- for i := list.start; i != start; i = (i + 1) % len(list.buffer) {
- if list.lastDiscarded.Before(list.buffer[i].Message.Time) {
- list.lastDiscarded = list.buffer[i].Message.Time
- }
- }
- }
- if start < end {
- copied := copy(newbuffer, list.buffer[start:end])
- list.start = 0
- list.end = copied % size
- } else {
- lenInitial := len(list.buffer) - start
- copied := copy(newbuffer, list.buffer[start:])
- copied += copy(newbuffer[lenInitial:], list.buffer[:end])
- list.start = 0
- list.end = copied % size
- }
- }
-
- list.buffer = newbuffer
- }
-
- func (hist *Buffer) length() int {
- if hist.start == -1 {
- return 0
- } else if hist.start < hist.end {
- return hist.end - hist.start
- } else {
- return len(hist.buffer) - (hist.start - hist.end)
- }
- }
|