Skip to content

Commit f4afed5

Browse files
hwjiangkaihwjiangkai
committed
fix review comments
Signed-off-by: jyjiangkai <[email protected]>
1 parent 8b9f873 commit f4afed5

File tree

12 files changed

+99
-59
lines changed

12 files changed

+99
-59
lines changed

‎client/pkg/eventbus/eventbus.go‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,11 @@ func (b *eventbus) GetLog(ctx context.Context, logID uint64, opts ...api.LogOpti
187187
b.readableMu.RLock()
188188
defer b.readableMu.RUnlock()
189189
if len(b.readableLogs) == 0 {
190-
b.refreshReadableLogs(ctx)
190+
func() {
191+
b.readableMu.RUnlock()
192+
defer b.readableMu.RLock()
193+
b.refreshReadableLogs(ctx)
194+
}()
191195
}
192196
if l, ok := b.readableLogs[logID]; ok {
193197
return l, nil
@@ -197,7 +201,11 @@ func (b *eventbus) GetLog(ctx context.Context, logID uint64, opts ...api.LogOpti
197201
b.writableMu.RLock()
198202
defer b.writableMu.RUnlock()
199203
if len(b.writableLogs) == 0 {
200-
b.refreshWritableLogs(ctx)
204+
func() {
205+
b.writableMu.RUnlock()
206+
defer b.writableMu.RLock()
207+
b.refreshWritableLogs(ctx)
208+
}()
201209
}
202210
if l, ok := b.writableLogs[logID]; ok {
203211
return l, nil
@@ -222,7 +230,11 @@ func (b *eventbus) ListLog(ctx context.Context, opts ...api.LogOption) ([]api.Ev
222230
b.readableMu.RLock()
223231
defer b.readableMu.RUnlock()
224232
if len(b.readableLogs) == 0 {
225-
b.refreshReadableLogs(ctx)
233+
func() {
234+
b.readableMu.RUnlock()
235+
defer b.readableMu.RLock()
236+
b.refreshReadableLogs(ctx)
237+
}()
226238
}
227239
eventlogs := make([]api.Eventlog, 0)
228240
for _, el := range b.readableLogs {
@@ -233,7 +245,11 @@ func (b *eventbus) ListLog(ctx context.Context, opts ...api.LogOption) ([]api.Ev
233245
b.writableMu.RLock()
234246
defer b.writableMu.RUnlock()
235247
if len(b.writableLogs) == 0 {
236-
b.refreshWritableLogs(ctx)
248+
func() {
249+
b.writableMu.RUnlock()
250+
defer b.writableMu.RLock()
251+
b.refreshWritableLogs(ctx)
252+
}()
237253
}
238254
eventlogs := make([]api.Eventlog, 0)
239255
for _, el := range b.writableLogs {

‎client/pkg/eventlog/eventlog_impl.go‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -129,8 +129,8 @@ func (l *eventlog) ID() uint64 {
129129
func (l *eventlog) Close(ctx context.Context) {
130130
l.writableWatcher.Close()
131131
l.readableWatcher.Close()
132-
l.logWriter = nil
133-
l.logReader = nil
132+
l.logWriter.Close(ctx)
133+
l.logReader.Close(ctx)
134134

135135
for _, segment := range l.writableSegments {
136136
segment.Close(ctx)
@@ -222,11 +222,6 @@ func (l *eventlog) QueryOffsetByTime(ctx context.Context, timestamp int64) (int6
222222
// LastEntryStime
223223
// time.UnixMilli
224224
tailSeg := fetchTailSegment(ctx, segs)
225-
if tailSeg.firstEventBornAt.Before(t) {
226-
// the target offset maybe in newer segment, refresh immediately
227-
l.refreshReadableSegments(ctx)
228-
segs = l.fetchReadableSegments(ctx)
229-
}
230225

231226
for idx := range segs {
232227
seg := segs[idx]

‎internal/store/block/raw.go‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -57,18 +57,22 @@ type Raw interface {
5757
Delete(context.Context) error
5858
}
5959

60-
type State string
61-
6260
const (
63-
StateWorking = State("working")
64-
StateArchiving = State("archiving")
65-
StateArchived = State("archived")
61+
working = 0
62+
archived = 1
63+
archiving = 2
64+
)
65+
66+
var (
67+
StateWorking = uint32(working)
68+
StateArchived = uint32(archived)
69+
StateArchiving = uint32(archiving)
6670
)
6771

6872
type Statistics struct {
6973
ID vanus.ID
7074
Capacity uint64
71-
State State
75+
State uint32
7276
EntryNum uint32
7377
EntrySize uint64
7478
// FirstEntryStime is the millisecond timestamp when the first Entry will be written to Block.

‎internal/store/segment/server.go‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -425,14 +425,17 @@ func (s *server) runHeartbeat(_ context.Context) error {
425425
break
426426
}
427427

428-
s.replicas.Range(func(key, value interface{}) bool {
428+
value, ok := s.replicas.Load(info.leader.Uint64())
429+
if ok {
429430
b, _ := value.(replica)
430-
if b.appender.Status().Leader.Equals(vanus.ID(segment.LeaderBlockId)) {
431-
b.appender.Archive(context.Background())
432-
return false
431+
if b.appender.Status().Term != info.term {
432+
break
433433
}
434-
return true
435-
})
434+
if info.leader.Uint64() != segment.LeaderBlockId {
435+
break
436+
}
437+
b.appender.Archive(context.Background())
438+
}
436439
}
437440
}
438441
}()
@@ -904,13 +907,15 @@ func (s *server) checkState() error {
904907
return nil
905908
}
906909

907-
func toSegmentState(state block.State) string {
910+
func toSegmentState(state uint32) string {
908911
switch state {
909-
case block.StateArchiving:
910-
return "freezing"
912+
case block.StateWorking:
913+
return "working"
911914
case block.StateArchived:
912915
return "frozen"
916+
case block.StateArchiving:
917+
return "freezing"
913918
default:
914-
return string(state)
919+
return "working"
915920
}
916921
}

‎internal/store/vsb/block.go‎

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,8 @@ type meta struct {
3838
entryLength int64
3939
// entryNum is the number of persisted entries.
4040
entryNum int64
41-
// archived is the flag indicating Block is archived.
42-
archived bool
41+
// state is the flag indicating Block state.
42+
state uint32
4343
}
4444

4545
// vsBlock is Vanus block file.
@@ -97,7 +97,7 @@ func (b *vsBlock) Close(ctx context.Context) error {
9797
}
9898

9999
// Flush metadata.
100-
if b.fm.archived != m.archived || b.fm.entryLength != m.entryLength {
100+
if b.fm.state != m.state || b.fm.entryLength != m.entryLength {
101101
if err := b.persistHeader(ctx, m); err != nil {
102102
return err
103103
}
@@ -112,15 +112,11 @@ func (b *vsBlock) Delete(context.Context) error {
112112
}
113113

114114
func (b *vsBlock) status() block.Statistics {
115-
state := block.StateWorking
116115
m, indexes := b.makeSnapshot()
117-
if m.archived {
118-
state = block.StateArchiving
119-
}
120-
return b.stat(m, indexes, state)
116+
return b.stat(m, indexes, m.state)
121117
}
122118

123-
func (b *vsBlock) stat(m meta, indexes []index.Index, state block.State) block.Statistics {
119+
func (b *vsBlock) stat(m meta, indexes []index.Index, state uint32) block.Statistics {
124120
s := block.Statistics{
125121
ID: b.id,
126122
Capacity: uint64(b.capacity),
@@ -137,6 +133,14 @@ func (b *vsBlock) stat(m meta, indexes []index.Index, state block.State) block.S
137133
return s
138134
}
139135

136+
func (b *vsBlock) prefull(state uint32) block.Statistics {
137+
s := block.Statistics{
138+
ID: b.id,
139+
State: state,
140+
}
141+
return s
142+
}
143+
140144
func (b *vsBlock) full() bool {
141-
return atomic.LoadUint32(&b.actx.archived) != 0
145+
return atomic.LoadUint32(&b.actx.state) != block.StateWorking
142146
}

‎internal/store/vsb/block_append.go‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -40,9 +40,9 @@ var (
4040
)
4141

4242
type appendContext struct {
43-
seq int64
44-
offset int64
45-
archived uint32
43+
seq int64
44+
offset int64
45+
state uint32
4646
}
4747

4848
// Make sure appendContext implements block.AppendContext.
@@ -56,8 +56,12 @@ func (c *appendContext) WriteOffset() int64 {
5656
return c.offset
5757
}
5858

59+
func (c *appendContext) Archiving() bool {
60+
return c.state == block.StateArchiving
61+
}
62+
5963
func (c *appendContext) Archived() bool {
60-
return c.archived != 0
64+
return c.state == block.StateArchived
6165
}
6266

6367
// Make sure vsBlock implements block.TwoPCAppender.
@@ -70,9 +74,10 @@ func (b *vsBlock) NewAppendContext(last block.Fragment) block.AppendContext {
7074
actx := &appendContext{
7175
seq: seq + 1,
7276
offset: last.EndOffset(),
77+
state: block.StateWorking,
7378
}
7479
if ceschema.EntryType(entry) == ceschema.End {
75-
actx.archived = 1
80+
actx.state = block.StateArchived
7681
}
7782
return actx
7883
}
@@ -110,8 +115,7 @@ func (b *vsBlock) PrepareAppend(
110115

111116
full := actx.size(b.dataOffset) >= b.capacity
112117
if full && b.lis != nil {
113-
m, indexes := makeSnapshot(b.actx, b.indexes)
114-
b.lis.OnArchived(b.stat(m, indexes, block.StateArchiving))
118+
b.lis.OnArchived(b.prefull(block.StateArchiving))
115119
}
116120

117121
return seqs, frag, full, nil
@@ -129,7 +133,7 @@ func (b *vsBlock) PrepareArchive(ctx context.Context, appendCtx block.AppendCont
129133

130134
actx.offset += int64(frag.Size())
131135
actx.seq++
132-
actx.archived = 1
136+
actx.state = block.StateArchiving
133137

134138
return frag, nil
135139
}
@@ -190,7 +194,7 @@ func (b *vsBlock) CommitAppend(ctx context.Context, frag block.Fragment, cb bloc
190194
return
191195
}
192196

193-
atomic.StoreUint32(&b.actx.archived, 1)
197+
atomic.StoreUint32(&b.actx.state, block.StateArchived)
194198

195199
b.wg.Add(1)
196200
b.s.Append(bytes.NewReader(frag.Payload()), func(n int, err error) {

‎internal/store/vsb/block_append_test.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ func TestVSBlock_Append(t *testing.T) {
7474
So(actx.WriteOffset(), ShouldEqual, vsbtest.EndEntryOffset+vsbtest.EndEntrySize)
7575
So(actx.Archived(), ShouldBeTrue)
7676

77-
b.actx.archived = 1
77+
b.actx.state = block.StateArchived
7878
actx = b.NewAppendContext(nil)
7979
So(actx, ShouldNotBeNil)
8080
So(actx.WriteOffset(), ShouldEqual, headerBlockSize)
@@ -102,6 +102,7 @@ func TestVSBlock_Append(t *testing.T) {
102102
dataOffset: headerBlockSize,
103103
actx: appendContext{
104104
offset: headerBlockSize,
105+
state: block.StateWorking,
105106
},
106107
enc: codec.NewEncoder(),
107108
dec: dec,
@@ -288,7 +289,7 @@ func TestVSBlock_Append(t *testing.T) {
288289
<-ch
289290

290291
stat = b.status()
291-
So(stat.State, ShouldEqual, block.StateArchiving)
292+
So(stat.State, ShouldEqual, block.StateArchived)
292293
So(stat.EntryNum, ShouldEqual, 2)
293294
So(stat.EntrySize, ShouldEqual, vsbtest.EntrySize0+vsbtest.EntrySize1)
294295

‎internal/store/vsb/block_header.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"hash/crc32"
2222

2323
// this project.
24+
"github.com/linkall-labs/vanus/internal/store/block"
2425
"github.com/linkall-labs/vanus/internal/store/block/raw"
2526
)
2627

@@ -52,8 +53,10 @@ func (b *vsBlock) persistHeader(ctx context.Context, m meta) error {
5253
binary.LittleEndian.PutUint32(buf[flagsOffset:], 0) // flags
5354
binary.LittleEndian.PutUint32(buf[breakFlagsOffset:], 0) // break flags
5455
binary.LittleEndian.PutUint32(buf[dataOffsetOffset:], uint32(b.dataOffset)) // data offset
55-
if m.archived { // state
56+
if m.state == block.StateArchived { // state
5657
buf[stateOffset] = 1
58+
} else if m.state == block.StateArchiving {
59+
buf[stateOffset] = 2
5760
}
5861
binary.LittleEndian.PutUint16(buf[indexSizeOffset:], b.indexSize) // index size
5962
binary.LittleEndian.PutUint64(buf[capacityOffset:], uint64(b.capacity)) // capacity
@@ -94,8 +97,14 @@ func (b *vsBlock) loadHeader(ctx context.Context) error {
9497
return errIncomplete
9598
}
9699

97-
b.dataOffset = int64(binary.LittleEndian.Uint32(buf[dataOffsetOffset:])) // data offset
98-
b.fm.archived = buf[stateOffset] != 0 // state
100+
b.dataOffset = int64(binary.LittleEndian.Uint32(buf[dataOffsetOffset:])) // data offset
101+
if buf[stateOffset] == 1 { // state
102+
b.fm.state = block.StateArchived
103+
} else if buf[stateOffset] == 2 {
104+
b.fm.state = block.StateArchiving
105+
} else {
106+
b.fm.state = block.StateWorking
107+
}
99108
b.indexSize = binary.LittleEndian.Uint16(buf[indexSizeOffset:]) // index size
100109
b.capacity = int64(binary.LittleEndian.Uint64(buf[capacityOffset:])) // capacity
101110
b.fm.entryLength = int64(binary.LittleEndian.Uint64(buf[entryLengthOffset:])) // entry length

‎internal/store/vsb/block_open.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -86,17 +86,17 @@ func (b *vsBlock) init(ctx context.Context) error {
8686
func (b *vsBlock) repairMeta() error {
8787
off := b.dataOffset + b.fm.entryLength
8888
seq := b.fm.entryNum
89-
full := b.fm.archived
9089

9190
var entry block.Entry
91+
var full bool
9292
var err error
9393
var n, en int
9494

9595
// Scan entries.
9696
indexes := make([]index.Index, 0)
9797
// Note: use math.MaxInt64-off to avoid overflow.
9898
r := io.NewSectionReader(b.f, off, math.MaxInt64-off)
99-
if full {
99+
if b.fm.state == block.StateArchiving || b.fm.state == block.StateArchived {
100100
n, entry, err = b.dec.UnmarshalReader(r)
101101
if err != nil || ceschema.EntryType(entry) != ceschema.End {
102102
return errCorrupted
@@ -161,7 +161,7 @@ SET_META:
161161
b.actx.seq = seq
162162
b.actx.offset = off
163163
if full {
164-
b.actx.archived = 1
164+
b.actx.state = block.StateArchived
165165
}
166166

167167
return nil

‎internal/store/vsb/block_open_test.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ func TestVSBlock_Open(t *testing.T) {
6767
stat := b.status()
6868
So(stat.Capacity, ShouldEqual, vsbtest.EntrySize0+vsbtest.EntrySize1)
6969
So(stat.EntryNum, ShouldEqual, 2)
70-
So(stat.State, ShouldEqual, block.StateArchiving)
70+
So(stat.State, ShouldEqual, block.StateArchived)
7171
So(stat.EntrySize, ShouldEqual, vsbtest.EntrySize0+vsbtest.EntrySize1)
7272

7373
So(b.indexes, ShouldHaveLength, 2)
@@ -133,6 +133,9 @@ func TestVSBlock_Open(t *testing.T) {
133133
So(err, ShouldBeNil)
134134

135135
b := &vsBlock{
136+
actx: appendContext{
137+
state: block.StateWorking,
138+
},
136139
path: f.Name(),
137140
}
138141

0 commit comments

Comments
 (0)