lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-05-03 21:17:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

change LogBuffer visibility in LocalPartition

Browse Source
This commit is contained in:
chrislu 2024-03-07 10:57:04 -08:00
parent 28f8f33d6a
commit 3b28433cb1
1 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -13,7 +13,7 @@ type LocalPartition struct {
Partition Partition
isLeader bool isLeader bool
FollowerBrokers []pb.ServerAddress FollowerBrokers []pb.ServerAddress
logBuffer *log_buffer.LogBuffer LogBuffer *log_buffer.LogBuffer
ConsumerCount int32 ConsumerCount int32
StopPublishersCh chan struct{} StopPublishersCh chan struct{}
Publishers *LocalPartitionPublishers Publishers *LocalPartitionPublishers
@ -28,7 +28,7 @@ func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.
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), LogBuffer: log_buffer.NewLogBuffer(fmt.Sprintf("%d/%04d-%04d", partition.UnixTimeNs, partition.RangeStart, partition.RangeStop),
2*time.Minute, logFlushFn, readFromDiskFn, func() {}), 2*time.Minute, logFlushFn, readFromDiskFn, func() {}),
Publishers: NewLocalPartitionPublishers(), Publishers: NewLocalPartitionPublishers(),
Subscribers: NewLocalPartitionSubscribers(), Subscribers: NewLocalPartitionSubscribers(),
@ -36,7 +36,7 @@ func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.
} }
func (p *LocalPartition) Publish(message *mq_pb.DataMessage) { func (p *LocalPartition) Publish(message *mq_pb.DataMessage) {
p.logBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano()) p.LogBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano())
} }
func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.MessagePosition, func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.MessagePosition,
@ -47,7 +47,7 @@ func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.M
var isDone bool var isDone bool
for { for {
processedPosition, isDone, readPersistedLogErr = p.logBuffer.ReadFromDiskFn(startPosition, 0, eachMessageFn) processedPosition, isDone, readPersistedLogErr = p.LogBuffer.ReadFromDiskFn(startPosition, 0, eachMessageFn)
if readPersistedLogErr != nil { if readPersistedLogErr != nil {
glog.V(0).Infof("%s read %v persisted log: %v", clientName, p.Partition, readPersistedLogErr) glog.V(0).Infof("%s read %v persisted log: %v", clientName, p.Partition, readPersistedLogErr)
return readPersistedLogErr return readPersistedLogErr
@ -57,7 +57,7 @@ func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.M
} }
startPosition = processedPosition startPosition = processedPosition
processedPosition, isDone, readInMemoryLogErr = p.logBuffer.LoopProcessLogData(clientName, startPosition, 0, onNoMessageFn, eachMessageFn) processedPosition, isDone, readInMemoryLogErr = p.LogBuffer.LoopProcessLogData(clientName, startPosition, 0, onNoMessageFn, eachMessageFn)
startPosition = processedPosition startPosition = processedPosition
if readInMemoryLogErr == log_buffer.ResumeFromDiskError { if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
@ -74,15 +74,15 @@ func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.M
} }
func (p *LocalPartition) GetEarliestMessageTimeInMemory() time.Time { func (p *LocalPartition) GetEarliestMessageTimeInMemory() time.Time {
return p.logBuffer.GetEarliestTime() return p.LogBuffer.GetEarliestTime()
} }
func (p *LocalPartition) HasData() bool { func (p *LocalPartition) HasData() bool {
return !p.logBuffer.GetEarliestTime().IsZero() return !p.LogBuffer.GetEarliestTime().IsZero()
} }
func (p *LocalPartition) GetEarliestInMemoryMessagePosition() log_buffer.MessagePosition { func (p *LocalPartition) GetEarliestInMemoryMessagePosition() log_buffer.MessagePosition {
return p.logBuffer.GetEarliestPosition() return p.LogBuffer.GetEarliestPosition()
} }
func FromPbBrokerPartitionAssignment(self pb.ServerAddress, partition Partition, assignment *mq_pb.BrokerPartitionAssignment, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition { func FromPbBrokerPartitionAssignment(self pb.ServerAddress, partition Partition, assignment *mq_pb.BrokerPartitionAssignment, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition {
@ -113,7 +113,7 @@ func (p *LocalPartition) WaitUntilNoPublishers() {
func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) { func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) {
if p.Publishers.IsEmpty() && p.Subscribers.IsEmpty() { if p.Publishers.IsEmpty() && p.Subscribers.IsEmpty() {
p.logBuffer.ShutdownLogBuffer() p.LogBuffer.ShutdownLogBuffer()
hasShutdown = true hasShutdown = true
} }
return return