lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-09-24 21:23:39 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
chrislu
2024-04-02 16:25:43 -07:00
parent f37c0d0d7a
commit ccdd9cd8de
6 changed files with 16 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
weed/filer/filer_notify.go
View File

@@ -83,7 +83,7 @@ func (f *Filer) logMetaEvent(ctx context.Context, fullpath string, eventNotifica
return return
} }
f.LocalMetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs) f.LocalMetaLogBuffer.AddDataToBuffer([]byte(dir), data, event.TsNs)
} }

2
weed/filer/meta_aggregator.go
View File

@@ -168,7 +168,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
} }
dir := event.Directory dir := event.Directory
// println("received meta change", dir, "size", len(data)) // println("received meta change", dir, "size", len(data))
ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs) ma.MetaLogBuffer.AddDataToBuffer([]byte(dir), data, event.TsNs)
if maybeReplicateMetadataChange != nil { if maybeReplicateMetadataChange != nil {
maybeReplicateMetadataChange(event) maybeReplicateMetadataChange(event)
} }

2
weed/mq/broker/broker_grpc_pub_follow.go
View File

@@ -52,7 +52,7 @@ func (b *MessageQueueBroker) PublishFollowMe(stream mq_pb.SeaweedMessaging_Publi
// TODO: change this to DataMessage // TODO: change this to DataMessage
// log the message // log the message
logBuffer.AddToBuffer(dataMessage.Key, dataMessage.Value, dataMessage.TsNs) logBuffer.AddToBuffer(dataMessage)
// send back the ack // send back the ack
if err := stream.Send(&mq_pb.PublishFollowMeResponse{ if err := stream.Send(&mq_pb.PublishFollowMeResponse{

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

@@ -52,7 +52,7 @@ func NewLocalPartition(partition Partition, logFlushFn log_buffer.LogFlushFuncTy
} }
func (p *LocalPartition) Publish(message *mq_pb.DataMessage) error { func (p *LocalPartition) Publish(message *mq_pb.DataMessage) error {
p.LogBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano()) p.LogBuffer.AddToBuffer(message)
// maybe send to the follower // maybe send to the follower
if p.followerStream != nil { if p.followerStream != nil {

7
weed/util/log_buffer/log_buffer.go
View File

@@ -2,6 +2,7 @@ package log_buffer
import ( import (
"bytes" "bytes"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
@@ -68,7 +69,11 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc
return lb return lb
} }
func (logBuffer *LogBuffer) AddToBuffer(partitionKey, data []byte, processingTsNs int64) { func (logBuffer *LogBuffer) AddToBuffer(message *mq_pb.DataMessage) {
logBuffer.AddDataToBuffer(message.Key, message.Value, message.TsNs)
}
func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processingTsNs int64) {
var toFlush *dataToFlush var toFlush *dataToFlush
logBuffer.Lock() logBuffer.Lock()

7
weed/util/log_buffer/log_buffer_test.go
View File

@@ -3,6 +3,7 @@ package log_buffer
import ( import (
"crypto/rand" "crypto/rand"
"fmt" "fmt"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"io" "io"
"sync" "sync"
"testing" "testing"
@@ -50,7 +51,11 @@ func TestNewLogBufferFirstBuffer(t *testing.T) {
var buf = make([]byte, messageSize) var buf = make([]byte, messageSize)
for i := 0; i < messageCount; i++ { for i := 0; i < messageCount; i++ {
rand.Read(buf) rand.Read(buf)
lb.AddToBuffer(nil, buf, 0) lb.AddToBuffer(&mq_pb.DataMessage{
Key: nil,
Value: buf,
TsNs: 0,
})
} }
wg.Wait() wg.Wait()