mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-01-21 16:43:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

notify

Browse Source
This commit is contained in:
chrislu 2024-03-16 23:16:33 -07:00
parent aba934f0b5
commit 8e5068fd2f
2 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
weed/mq/broker/broker_grpc_sub.go
View File

@ -175,7 +175,6 @@ func (b *MessageQueueBroker) FollowInMemoryMessages(req *mq_pb.FollowInMemoryMes
atomic.StoreInt32(&localTopicPartition.FollowerId, followerId) atomic.StoreInt32(&localTopicPartition.FollowerId, followerId)
glog.V(0).Infof("FollowInMemoryMessages %s connected on %v %v", clientName, t, partition) glog.V(0).Infof("FollowInMemoryMessages %s connected on %v %v", clientName, t, partition)
sleepIntervalCount := 0
var counter int64 var counter int64
defer func() { defer func() {
@ -198,11 +197,12 @@ func (b *MessageQueueBroker) FollowInMemoryMessages(req *mq_pb.FollowInMemoryMes
var prevFlushTsNs int64 var prevFlushTsNs int64
_, _, err = localTopicPartition.LogBuffer.LoopProcessLogData(clientName, startPosition, 0, func() bool { _, _, err = localTopicPartition.LogBuffer.LoopProcessLogData(clientName, startPosition, 0, func() bool {
sleepIntervalCount++ // wait for the log buffer to be ready
if sleepIntervalCount > 32 { localTopicPartition.ListenersLock.Lock()
sleepIntervalCount = 32 atomic.AddInt64(&localTopicPartition.ListenersWaits, 1)
} localTopicPartition.ListenersCond.Wait()
time.Sleep(time.Duration(sleepIntervalCount) * 137 * time.Millisecond) atomic.AddInt64(&localTopicPartition.ListenersWaits, -1)
localTopicPartition.ListenersLock.Unlock()
if localTopicPartition.LogBuffer.IsStopping() { if localTopicPartition.LogBuffer.IsStopping() {
newFollowerId := atomic.LoadInt32(&localTopicPartition.FollowerId) newFollowerId := atomic.LoadInt32(&localTopicPartition.FollowerId)
@ -246,8 +246,6 @@ func (b *MessageQueueBroker) FollowInMemoryMessages(req *mq_pb.FollowInMemoryMes
return true return true
}, func(logEntry *filer_pb.LogEntry) (bool, error) { }, func(logEntry *filer_pb.LogEntry) (bool, error) {
// reset the sleep interval count
sleepIntervalCount = 0
// check the follower id // check the follower id
newFollowerId := atomic.LoadInt32(&localTopicPartition.FollowerId) newFollowerId := atomic.LoadInt32(&localTopicPartition.FollowerId)

19
weed/mq/topic/local_partition.go
View File

@ -6,11 +6,18 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"github.com/seaweedfs/seaweedfs/weed/util/log_buffer" "github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
) )
type LocalPartition struct { type LocalPartition struct {
ListenersWaits int64
// notifying clients
ListenersLock sync.Mutex
ListenersCond *sync.Cond
Partition Partition
isLeader bool isLeader bool
FollowerBrokers []pb.ServerAddress FollowerBrokers []pb.ServerAddress
@ -24,15 +31,21 @@ type LocalPartition struct {
var TIME_FORMAT = "2006-01-02-15-04-05" var TIME_FORMAT = "2006-01-02-15-04-05"
func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.ServerAddress, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition { func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.ServerAddress, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition {
return &LocalPartition{ lp := &LocalPartition{
Partition: partition, Partition: partition,
isLeader: isLeader, isLeader: isLeader,
FollowerBrokers: followerBrokers, FollowerBrokers: followerBrokers,
LogBuffer: log_buffer.NewLogBuffer(fmt.Sprintf("%d/%04d-%04d", partition.UnixTimeNs, partition.RangeStart, partition.RangeStop),
2*time.Minute, logFlushFn, readFromDiskFn, func() {}),
Publishers: NewLocalPartitionPublishers(), Publishers: NewLocalPartitionPublishers(),
Subscribers: NewLocalPartitionSubscribers(), Subscribers: NewLocalPartitionSubscribers(),
} }
lp.ListenersCond = sync.NewCond(&lp.ListenersLock)
lp.LogBuffer = log_buffer.NewLogBuffer(fmt.Sprintf("%d/%04d-%04d", partition.UnixTimeNs, partition.RangeStart, partition.RangeStop),
2*time.Minute, logFlushFn, readFromDiskFn, func() {
if atomic.LoadInt64(&lp.ListenersWaits) > 0 {
lp.ListenersCond.Broadcast()
}
})
return lp
} }
func (p *LocalPartition) Publish(message *mq_pb.DataMessage) { func (p *LocalPartition) Publish(message *mq_pb.DataMessage) {