mailbox.go

package log

import (
	"runtime"
	"sync"
	"sync/atomic"
)

// mailbox is a parking MPSC batch-swap queue: producers append under a
// mutex and the consumer swaps the whole batch out under that same mutex —
// one lock acquisition per batch instead of one channel receive per
// record. The consumer parks when idle (zero idle CPU); a producer that
// finds it sleeping sends one token.
type mailbox struct {
	mu       sync.Mutex
	notFull  sync.Cond
	active   []*[]byte // producers append; guarded by mu
	spare    []*[]byte // consumer's recycled backing array
	enqueued uint64    // guarded by mu; Flush target
	sleeping bool      // consumer parked; guarded by mu
	closed   bool      // set by the consumer on exit; guarded by mu
	capacity int
	wake     chan struct{} // cap 1; one token per park cycle
	// consumerID is the consumer goroutine's id, set once at start-up. A
	// push from the consumer itself (a Transform fn that logs) must never
	// wait for space: the consumer is the only thing that drains the
	// queue, so waiting would self-deadlock. Checked only on the
	// queue-full slow path.
	consumerID atomic.Uint64
	_          [64]byte
	written    atomic.Uint64 // consumer-only writes; Flush polls
}

func newMailbox(capacity int) *mailbox {
	m := &mailbox{capacity: capacity, wake: make(chan struct{}, 1)}
	m.notFull.L = &m.mu
	return m
}

func (m *mailbox) push(sp *[]byte) {
	m.mu.Lock()
	if len(m.active) >= m.capacity && !m.closed && m.consumerID.Load() != goid() {
		for len(m.active) >= m.capacity && !m.closed {
			m.notFull.Wait()
		}
	}
	if m.closed {
		// Close happens only at program exit; a straggler goroutine's log
		// is dropped (it missed the final flush anyway) — never hang, and
		// don't turn a clean shutdown into a panic either.
		m.mu.Unlock()
		scratchPool.Put(sp)
		return
	}
	m.active = append(m.active, sp)
	m.enqueued++
	wake := m.sleeping
	if wake {
		m.sleeping = false
	}
	m.mu.Unlock()
	if wake {
		m.wake <- struct{}{}
	}
}

// runMailbox is the consumer loop: swap out the queued batch (parked when
// idle), sort by timestamp key, decode to JSON and write with one writev.
func (c *Core) runMailbox() {
	defer close(c.done)
	if useTSC {
		c.clk.init()
	}
	m := c.mbx
	m.consumerID.Store(goid())
	var bufs [][]byte
	for {
		m.mu.Lock()
		// Promote Close as soon as it is seen — even mid-load — so new
		// pushes start dropping and the remaining drain is finite (a
		// steady producer stream must not livelock Close).
		if !m.closed && c.closed.Load() {
			m.closed = true
		}
		for len(m.active) == 0 {
			if m.closed {
				m.mu.Unlock()
				m.notFull.Broadcast() // release producers blocked on a full queue
				return
			}
			m.sleeping = true
			m.mu.Unlock()
			select {
			case <-m.wake:
			case <-c.stop:
			}
			m.mu.Lock()
			m.sleeping = false
			if c.closed.Load() {
				m.closed = true
			}
		}
		batch := m.active
		m.active = m.spare
		m.mu.Unlock()
		m.notFull.Broadcast()
		for off := 0; off < len(batch); off += maxBatch {
			end := min(off+maxBatch, len(batch))
			c.processRecs(batch[off:end], &bufs)
			m.written.Add(uint64(end - off))
		}
		clear(batch) // don't pin pooled buffers via the spare array
		m.spare = batch[:0]
	}
}

// goid returns the current goroutine's id, parsed from the runtime.Stack
// header ("goroutine 123 [..."). Only called off the hot path: once at
// consumer start-up and on push's queue-full branch.
func goid() uint64 {
	var buf [32]byte
	n := runtime.Stack(buf[:], false)
	var id uint64
	for _, c := range buf[len("goroutine "):n] {
		if c < '0' || c > '9' {
			break
		}
		id = id*10 + uint64(c-'0')
	}
	return id
}