feat: fix review comments

Signed-off-by: jyjiangkai <[email protected]>

Author: hwjiangkai

client/internal/vanus/store/block_store.go

@@ -216,7 +216,6 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 
 	var (
 		err  error
-		wg   sync.WaitGroup
 		resp *segpb.AppendToBlockStreamResponse
 	)
 
@@ -230,8 +229,6 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 	// generate unique RequestId
 	requestID := rand.New(rand.NewSource(time.Now().UnixNano())).Uint64()
 
-	wg.Add(1)
-
 	//TODO(jiangkai): delete the reference of CloudEvents/v2 in Vanus
 	eventpbs := make([]*cepb.CloudEvent, len(events))
 	for idx := range events {
@@ -242,9 +239,10 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 		eventpbs[idx] = eventpb
 	}
 
+	donec := make(chan struct{}, 1)
 	s.appendCallbacks.Store(requestID, appendCallback(func(res *segpb.AppendToBlockStreamResponse) {
 		resp = res
-		wg.Done()
+		donec <- struct{}{}
 	}))
 
 	req := &segpb.AppendToBlockStreamRequest{
@@ -275,7 +273,7 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 		return nil, err
 	}
 
-	wg.Wait()
+	<-donec
 
 	if resp.ResponseCode == errpb.ErrorCode_FULL {
 		log.Warning(ctx, "block append failed cause the segment is full", nil)
@@ -339,7 +337,6 @@ func (s *BlockStore) ReadStream(
 
 	var (
 		err  error
-		wg   sync.WaitGroup
 		resp *segpb.ReadFromBlockStreamResponse
 	)
 
@@ -353,11 +350,10 @@ func (s *BlockStore) ReadStream(
 	// generate unique RequestId
 	requestID := rand.Uint64()
 
-	wg.Add(1)
-
-	s.appendCallbacks.Store(requestID, readCallback(func(res *segpb.ReadFromBlockStreamResponse) {
+	donec := make(chan struct{}, 1)
+	s.readCallbacks.Store(requestID, readCallback(func(res *segpb.ReadFromBlockStreamResponse) {
 		resp = res
-		wg.Done()
+		donec <- struct{}{}
 	}))
 
 	req := &segpb.ReadFromBlockStreamRequest{
@@ -389,7 +385,7 @@ func (s *BlockStore) ReadStream(
 		return []*ce.Event{}, err
 	}
 
-	wg.Wait()
+	<-donec
 
 	if resp.ResponseCode != errpb.ErrorCode_SUCCESS {
 		log.Warning(ctx, "block read failed cause unknown error", nil)

client/pkg/eventbus/eventbus.go

@@ -469,13 +469,7 @@ func (w *busWriter) AppendOne(ctx context.Context, event *ce.Event, opts ...api.
 		return "", err
 	}
 
-	// 3. generate event ID
-	var buf [16]byte
-	binary.BigEndian.PutUint64(buf[0:8], lw.Log().ID())
-	binary.BigEndian.PutUint64(buf[8:16], uint64(off))
-	encoded := base64.StdEncoding.EncodeToString(buf[:])
-
-	return encoded, nil
+	return genEventID(lw.Log().ID(), off), nil
 }
 
 func (w *busWriter) AppendMany(ctx context.Context, events []*ce.Event, opts ...api.WriteOption) (eid []string, err error) {
@@ -504,22 +498,22 @@ func (w *busWriter) AppendMany(ctx context.Context, events []*ce.Event, opts ...
 	}
 
 	// 3. generate event ID
-	genFunc := func(off int64) string {
-		var buf [16]byte
-		binary.BigEndian.PutUint64(buf[0:8], lw.Log().ID())
-		binary.BigEndian.PutUint64(buf[8:16], uint64(off))
-		encoded := base64.StdEncoding.EncodeToString(buf[:])
-		return encoded
-	}
-
 	eventIDs := make([]string, len(offsets))
 	for idx := range offsets {
-		eventIDs[idx] = genFunc(offsets[idx])
+		eventIDs[idx] = genEventID(lw.Log().ID(), offsets[idx])
 	}
 
 	return eventIDs, nil
 }
 
+func genEventID(logID uint64, off int64) string {
+	var buf [16]byte
+	binary.BigEndian.PutUint64(buf[0:8], logID)
+	binary.BigEndian.PutUint64(buf[8:16], uint64(off))
+	encoded := base64.StdEncoding.EncodeToString(buf[:])
+	return encoded
+}
+
 func (w *busWriter) Bus() api.Eventbus {
 	return w.ebus
 }

client/pkg/eventlog/eventlog_impl.go

@@ -354,15 +354,14 @@ func (w *logWriter) Append(ctx context.Context, event *ce.Event) (int64, error)
 		if err == nil {
 			return offset, nil
 		}
-		vlog.Warning(ctx, "failed to Append", map[string]interface{}{
+		vlog.Warning(ctx, "append failed", map[string]interface{}{
 			vlog.KeyError: err,
 			"offset":      offset,
 		})
 		if errors.Is(err, errors.ErrFull) {
 			vlog.Warning(ctx, "segment is full, retry", map[string]interface{}{
 				vlog.KeyError: err,
-				"offset":      offset,
-				"retry_num":   i,
+				"retry_times": i,
 			})
 			if i < retryTimes {
 				continue
@@ -396,11 +395,15 @@ func (w *logWriter) AppendManyStream(ctx context.Context, events []*ce.Event) ([
 		if err == nil {
 			return offsets, nil
 		}
-		vlog.Warning(ctx, "failed to Append", map[string]interface{}{
+		vlog.Warning(ctx, "append failed", map[string]interface{}{
 			vlog.KeyError: err,
 			"offsets":     offsets,
 		})
 		if errors.Is(err, errors.ErrFull) {
+			vlog.Warning(ctx, "segment is full, retry", map[string]interface{}{
+				vlog.KeyError: err,
+				"retry_times": i,
+			})
 			if i < retryTimes {
 				continue
 			}

internal/store/block/block.go

@@ -45,7 +45,7 @@ type Reader interface {
 }
 
 type Appender interface {
-	Append(ctx context.Context, cb func([]int64, error), entries ...Entry)
+	Append(ctx context.Context, entries []Entry, cb func([]int64, error))
 }
 
 type Block interface {

internal/store/block/raft/appender.go

@@ -48,7 +48,6 @@ const (
 	defaultHeartbeatTick   = 3
 	defaultMaxSizePerMsg   = 4096
 	defaultMaxInflightMsgs = 256
-	defaultWaiterWorker    = 8
 )
 
 type Peer struct {
@@ -68,13 +67,17 @@ type peer struct {
 
 type LeaderChangedListener func(block, leader vanus.ID, term uint64)
 
-type commitWaiter struct {
+type CommitWaiter struct {
 	seqs     []int64
 	offset   int64
 	err      error
 	callback func([]int64, error)
 }
 
+func (w CommitWaiter) Do() {
+	w.callback(w.seqs, w.err)
+}
+
 type Appender interface {
 	block.Appender
 
@@ -89,8 +92,8 @@ type appender struct {
 	actx     block.AppendContext
 	appendMu sync.RWMutex
 
-	waiters      []commitWaiter
-	waiterC      chan commitWaiter
+	waiters      []CommitWaiter
+	waiterC      chan CommitWaiter
 	commitIndex  uint64
 	commitOffset int64
 	waitMu       sync.Mutex
@@ -114,14 +117,14 @@ type appender struct {
 var _ Appender = (*appender)(nil)
 
 func NewAppender(
-	ctx context.Context, raw block.Raw, raftLog *raftlog.Log, host transport.Host, listener LeaderChangedListener,
+	ctx context.Context, raw block.Raw, raftLog *raftlog.Log, host transport.Host, listener LeaderChangedListener, waiterC chan CommitWaiter,
 ) Appender {
 	ctx, cancel := context.WithCancel(ctx)
 
 	a := &appender{
 		raw:      raw,
-		waiters:  make([]commitWaiter, 0),
-		waiterC:  make(chan commitWaiter),
+		waiters:  make([]CommitWaiter, 0),
+		waiterC:  waiterC,
 		listener: listener,
 		log:      raftLog,
 		host:     host,
@@ -200,29 +203,11 @@ func (a *appender) Delete(ctx context.Context) {
 	a.log.Delete(ctx)
 }
 
-func (a *appender) runWaiterWorker(ctx context.Context) {
-	for i := 0; i < defaultWaiterWorker; i++ {
-		go func() {
-			for {
-				select {
-				case waiter := <-a.waiterC:
-					waiter.callback(waiter.seqs, waiter.err)
-				case <-ctx.Done():
-					close(a.waiterC)
-					return
-				}
-			}
-		}()
-	}
-}
-
 func (a *appender) run(ctx context.Context) {
 	// TODO(james.yin): reduce Ticker
 	t := time.NewTicker(defaultTickInterval)
 	defer t.Stop()
 
-	a.runWaiterWorker(ctx)
-
 	for {
 		select {
 		case <-t.C:
@@ -482,18 +467,24 @@ func (a *appender) reset(ctx context.Context) {
 }
 
 // Append implements async block.raw.
-func (a *appender) Append(ctx context.Context, cb func([]int64, error), entries ...block.Entry) {
+func (a *appender) Append(ctx context.Context, entries []block.Entry, cb func([]int64, error)) {
 	ctx, span := a.tracer.Start(ctx, "Append")
 	defer span.End()
 
+	span.AddEvent("Acquiring append lock")
+	a.appendMu.Lock()
+	span.AddEvent("Got append lock")
+
+	defer a.appendMu.Unlock()
+
 	seqs, offset, err := a.append(ctx, entries)
 	if err != nil && !errors.Is(err, errors.ErrFull) {
 		cb(nil, err)
 		return
 	}
 
 	// register callback and wait until entries is committed.
-	a.registerCommitWaiter(ctx, commitWaiter{
+	a.registerCommitWaiter(ctx, CommitWaiter{
 		seqs:     seqs,
 		offset:   offset,
 		err:      err,
@@ -505,12 +496,6 @@ func (a *appender) append(ctx context.Context, entries []block.Entry) ([]int64,
 	ctx, span := a.tracer.Start(ctx, "append")
 	defer span.End()
 
-	span.AddEvent("Acquiring append lock")
-	a.appendMu.Lock()
-	span.AddEvent("Got append lock")
-
-	defer a.appendMu.Unlock()
-
 	if !a.isLeader() {
 		return nil, 0, errors.ErrNotLeader.WithMessage("the appender is not leader")
 	}
@@ -548,7 +533,7 @@ func (a *appender) doAppend(ctx context.Context, frags ...block.Fragment) {
 	_, _ = a.raw.CommitAppend(ctx, frags...)
 }
 
-func (a *appender) registerCommitWaiter(ctx context.Context, waiter commitWaiter) {
+func (a *appender) registerCommitWaiter(ctx context.Context, waiter CommitWaiter) {
 	_, span := a.tracer.Start(ctx, "waitCommit")
 	defer span.End()
 

internal/store/segment/api.go

@@ -17,7 +17,6 @@ package segment
 import (
 	// standard libraries.
 	"context"
-	"sync"
 
 	// third-party libraries.
 	cepb "cloudevents.io/genproto/v1"
@@ -124,17 +123,16 @@ func (s *segmentServer) AppendToBlock(
 	var (
 		err     error
 		offsets []int64
-		wg      sync.WaitGroup
 	)
-	wg.Add(1)
+	donec := make(chan struct{})
 	blockID := vanus.NewIDFromUint64(req.BlockId)
 	events := req.Events.GetEvents()
 	s.srv.AppendToBlock(ctx, blockID, events, func(offs []int64, e error) {
 		offsets = offs
 		err = e
-		wg.Done()
+		donec <- struct{}{}
 	})
-	wg.Wait()
+	<-donec
 	return &segpb.AppendToBlockResponse{Offsets: offsets}, err
 }
 
@@ -165,6 +163,8 @@ func (s *segmentServer) AppendToBlockStream(stream segpb.SegmentServer_AppendToB
 				})
 			}
 
+			s.srv.NewMessageArrived(ctx, vanus.ID(request.BlockId))
+
 			err = stream.Send(&segpb.AppendToBlockStreamResponse{
 				ResponseId:   request.RequestId,
 				ResponseCode: errCode,

internal/store/segment/recovery.go

@@ -78,7 +78,7 @@ func (s *server) recoverBlocks(ctx context.Context, logs map[vanus.ID]*raftlog.L
 				return err
 			}
 		}
-		a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged)
+		a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.waiterC)
 		s.replicas.Store(id, &replica{
 			id:       id,
 			idStr:    id.String(),

internal/store/segment/replica.go

@@ -80,8 +80,8 @@ func (r *replica) Read(ctx context.Context, seq int64, num int) ([]block.Entry,
 	return r.raw.Read(ctx, seq, num)
 }
 
-func (r *replica) Append(ctx context.Context, cb func([]int64, error), entries ...block.Entry) {
-	r.appender.Append(ctx, cb, entries...)
+func (r *replica) Append(ctx context.Context, entries []block.Entry, cb func([]int64, error)) {
+	r.appender.Append(ctx, entries, cb)
 }
 
 func (r *replica) Status() *metapb.SegmentHealthInfo {
@@ -116,7 +116,7 @@ func (s *server) createBlock(ctx context.Context, id vanus.ID, size int64) (Repl
 
 	// Create replica.
 	l := raftlog.NewLog(id, s.wal, s.metaStore, s.offsetStore, nil)
-	a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged)
+	a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.waiterC)
 
 	return &replica{
 		id:       id,