17140236898

Committed 21 Aug 2025 10:06PM UTC coverage: 66.975% (+6.5%) from 60.477%

Build # 17140236898

Build Type

Pull #10

github

Committed by

kelindar

Commit Message

reduce stress test count from 1,000,000 to 10,000

Pull Request Pull Request #10: Refactored emit package for events

Coverage Stats

127 of 205 new or added lines in 4 files covered. (61.95%)

290 of 433 relevant lines covered (66.97%)

1315.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.1

/emit/queue.go

package emit

import (
        "sync"
        "sync/atomic"
        "time"

        "github.com/kelindar/event"
        "github.com/kelindar/timeline"
)

const segmentSize = 64

var _ timeline.Task = (*queue[fault])(nil).Drain

// segment represents a fixed-size buffer that holds events
type segment[T any] struct {
        next  atomic.Pointer[segment[T]] // next segment in the linked list
        mu    sync.Mutex                 // protects write operations within this segment
        data  [segmentSize]T             // fixed-size array of events
        write atomic.Uint32              // current write position (0-63)
        read  uint32                     // current read position (consumer only)
}

// queue is a multiple-producer, single-consumer queue
type queue[T event.Event] struct {
        head atomic.Pointer[segment[T]] // producers write to head segment
        tail *segment[T]                // consumer reads from tail segment
        pool sync.Pool                  // recycles segments to reduce GC
}

// newQueue creates a new queue
func newQueue[T event.Event]() *queue[T] {
        q := &queue[T]{}
        q.pool.New = func() any {
                return new(segment[T])
        }

        // Initialize with one empty segment
        seg := q.borrow()
        q.head.Store(seg)
        q.tail = seg
        return q
}

// Push is safe for concurrent producers.
func (q *queue[T]) Push(v T) {
        for {
                head := q.head.Load()

                // Try to write to current head segment under lock; revalidate head
                head.mu.Lock()
                if q.head.Load() != head {
                        head.mu.Unlock()
                        continue
                }

                // Space available in current segment
                if writeAt := head.write.Load(); writeAt < segmentSize {
                        head.data[writeAt] = v
                        head.write.Store(writeAt + 1)
                        head.mu.Unlock()
                        return
                }

                // Segment is full; create and link a new head seg while holding lock
                seg := q.borrow()
                seg.data[0] = v
                seg.write.Store(1)
                head.next.Store(seg)
                q.head.Store(seg)
                head.mu.Unlock()
                return
        }
}

// Drain is called by the scheduler to publish events, single-consumer only.
func (q *queue[T]) Drain(now time.Time, elapsed time.Duration) bool {
        var zero T
        for {
                seg := q.tail
                idx := seg.write.Load()

                // Process all items in this segment, until the write index
                for seg.read < idx {
                        val := seg.data[seg.read]
                        seg.data[seg.read] = zero // clear for GC
                        seg.read++

                        // Publish the event
                        event.Publish(event.Default, signal[T]{
                                Data:    val,
                                Time:    tickOf(now),
                                Elapsed: durationOf(elapsed),
                        })
                }

                // Current segment is exhausted
                next := seg.next.Load()
                if next == nil {
                        return true // empty
                }

                // Move to next segment and reset the current one
                q.tail = next
                q.reset(seg)
        }
}

func (q *queue[T]) borrow() *segment[T] {
        seg := q.pool.Get().(*segment[T])
        seg.write.Store(0)
        seg.read = 0
        seg.next.Store(nil)
        return seg
}

func (q *queue[T]) reset(seg *segment[T]) {
        seg.write.Store(0)
        seg.read = 0
        seg.next.Store(nil)
        q.pool.Put(seg)
}

1	package emit
2
3	import (
4	"sync"
5	"sync/atomic"
6	"time"
7
8	"github.com/kelindar/event"
9	"github.com/kelindar/timeline"
10	)
11
12	const segmentSize = 64
13
14	var _ timeline.Task = (*queue[fault])(nil).Drain
15
16	// segment represents a fixed-size buffer that holds events
17	type segment[T any] struct {
18	next atomic.Pointer[segment[T]] // next segment in the linked list
19	mu sync.Mutex // protects write operations within this segment
20	data [segmentSize]T // fixed-size array of events
21	write atomic.Uint32 // current write position (0-63)
22	read uint32 // current read position (consumer only)
23	}
24
25	// queue is a multiple-producer, single-consumer queue
26	type queue[T event.Event] struct {
27	head atomic.Pointer[segment[T]] // producers write to head segment
28	tail *segment[T] // consumer reads from tail segment
29	pool sync.Pool // recycles segments to reduce GC
30	}
31
32	// newQueue creates a new queue
33	func newQueue[T event.Event]() *queue[T] {	3✔
34	q := &queue[T]{}	3✔
35	q.pool.New = func() any {	90✔
36	return new(segment[T])	87✔
37	}	87✔
38
39	// Initialize with one empty segment
40	seg := q.borrow()	3✔
41	q.head.Store(seg)	3✔
42	q.tail = seg	3✔
43	return q	3✔
44	}
45
46	// Push is safe for concurrent producers.
47	func (q *queue[T]) Push(v T) {	10,004✔
48	for {	20,008✔
49	head := q.head.Load()	10,004✔
50		10,004✔
51	// Try to write to current head segment under lock; revalidate head	10,004✔
52	head.mu.Lock()	10,004✔
53	if q.head.Load() != head {	10,004✔
NEW 54	head.mu.Unlock()	×
NEW 55	continue	×
56	}
57
58	// Space available in current segment
59	if writeAt := head.write.Load(); writeAt < segmentSize {	19,852✔
60	head.data[writeAt] = v	9,848✔
61	head.write.Store(writeAt + 1)	9,848✔
62	head.mu.Unlock()	9,848✔
63	return	9,848✔
64	}	9,848✔
65
66	// Segment is full; create and link a new head seg while holding lock
67	seg := q.borrow()	156✔
68	seg.data[0] = v	156✔
69	seg.write.Store(1)	156✔
70	head.next.Store(seg)	156✔
71	q.head.Store(seg)	156✔
72	head.mu.Unlock()	156✔
73	return	156✔
74	}
75	}
76
77	// Drain is called by the scheduler to publish events, single-consumer only.
78	func (q *queue[T]) Drain(now time.Time, elapsed time.Duration) bool {	36✔
79	var zero T	36✔
80	for {	228✔
81	seg := q.tail	192✔
82	idx := seg.write.Load()	192✔
83		192✔
84	// Process all items in this segment, until the write index	192✔
85	for seg.read < idx {	10,196✔
86	val := seg.data[seg.read]	10,004✔
87	seg.data[seg.read] = zero // clear for GC	10,004✔
88	seg.read++	10,004✔
89		10,004✔
90	// Publish the event	10,004✔
91	event.Publish(event.Default, signal[T]{	10,004✔
92	Data: val,	10,004✔
93	Time: tickOf(now),	10,004✔
94	Elapsed: durationOf(elapsed),	10,004✔
95	})	10,004✔
96	}	10,004✔
97
98	// Current segment is exhausted
99	next := seg.next.Load()	192✔
100	if next == nil {	228✔
101	return true // empty	36✔
102	}	36✔
103
104	// Move to next segment and reset the current one
105	q.tail = next	156✔
106	q.reset(seg)	156✔
107	}
108	}
109
110	func (q queue[T]) borrow() segment[T] {	159✔
111	seg := q.pool.Get().(*segment[T])	159✔
112	seg.write.Store(0)	159✔
113	seg.read = 0	159✔
114	seg.next.Store(nil)	159✔
115	return seg	159✔
116	}	159✔
117
118	func (q queue[T]) reset(seg segment[T]) {	156✔
119	seg.write.Store(0)	156✔
120	seg.read = 0	156✔
121	seg.next.Store(nil)	156✔
122	q.pool.Put(seg)	156✔
123	}	156✔

kelindar / timeline / 17140236898

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous