seqflow/sequencer_test.go at master · gocronx/seqflow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package seqflow

import "testing"

func TestSingleSequencer_Reserve_Basic(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	consumer.Store(1024)
	s := newSingleSequencer(1024, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	upper := s.reserve(1)
	if upper != 0 {
		t.Errorf("首次 reserve(1) = %d, 期望 0", upper)
	}
}

func TestSingleSequencer_Reserve_Batch(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	consumer.Store(1024)
	s := newSingleSequencer(1024, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	upper := s.reserve(16)
	if upper != 15 {
		t.Errorf("reserve(16) = %d, 期望 15", upper)
	}
}

func TestSingleSequencer_Reserve_InvalidSize(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	s := newSingleSequencer(1024, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	if got := s.reserve(0); got != errReservationSize {
		t.Errorf("reserve(0) = %d, 期望 sentinel %d", got, errReservationSize)
	}
	if got := s.reserve(2048); got != errReservationSize {
		t.Errorf("reserve(2048) = %d, 期望 sentinel %d", got, errReservationSize)
	}
}

func TestSingleSequencer_TryReserve_Success(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	consumer.Store(1024)
	s := newSingleSequencer(1024, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	upper := s.tryReserve(1)
	if upper != 0 {
		t.Errorf("tryReserve(1) = %d, 期望 0", upper)
	}
}

func TestSingleSequencer_TryReserve_Unavailable(t *testing.T) {
	committed := newSequence()
	consumer := newSequence() // 消费者位于 -1
	s := newSingleSequencer(4, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	// 填满缓冲区
	for i := uint32(0); i < 4; i++ {
		s.reserve(1)
	}

	if got := s.tryReserve(1); got != errCapacityUnavailable {
		t.Errorf("缓冲区满时 tryReserve = %d, 期望 sentinel %d", got, errCapacityUnavailable)
	}
}

func TestSingleSequencer_Commit(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	consumer.Store(1024)
	s := newSingleSequencer(1024, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	upper := s.reserve(1)
	s.commit(upper, upper)

	if got := committed.Load(); got != upper {
		t.Errorf("已提交序列 = %d, 期望 %d", got, upper)
	}
}

func TestSingleSequencer_Reserve_WrapAround(t *testing.T) {
	committed := newSequence()
	consumer := newSequence()
	s := newSingleSequencer(4, committed, newAtomicBarrier(consumer), NewSleepingStrategy())

	// 填满缓冲区并推进消费者
	for i := uint32(0); i < 4; i++ {
		upper := s.reserve(1)
		s.commit(upper, upper)
		consumer.Store(upper)
	}

	// 回绕后应成功
	upper := s.reserve(1)
	if upper < 0 {
		t.Fatalf("回绕后 reserve 返回 sentinel: %d", upper)
	}
	if upper != 4 {
		t.Errorf("回绕后 reserve = %d, 期望 4", upper)
	}
}