diff --git a/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go b/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go index 98db80a05..53cb2f912 100644 --- a/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go +++ b/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go @@ -4,18 +4,20 @@ import ( "context" "flag" "fmt" + "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "google.golang.org/grpc" "io" "strconv" + "time" ) var ( dir = flag.String("dir", "/tmp", "directory to create files") n = flag.Int("n", 100, "the number of metadata") tailFiler = flag.String("filer", "localhost:8888", "the filer address") - isWrite = flag.Bool("write", false, "only write") + isWrite = flag.Bool("write", false, "only write") ) func main() { @@ -33,7 +35,7 @@ func main() { return nil } name := event.EventNotification.NewEntry.Name - fmt.Printf("=> %s\n", name) + glog.V(0).Infof("=> %s ts:%+v", name, time.Unix(0, event.TsNs)) id := name[4:] if x, err := strconv.Atoi(id); err == nil { if x != expected { @@ -43,6 +45,7 @@ func main() { } else { return err } + time.Sleep(10 * time.Millisecond) return nil }) @@ -96,6 +99,7 @@ func startSubscribeMetadata(eachEntryFunc func(event *filer_pb.SubscribeMetadata return listenErr } if err = eachEntryFunc(resp); err != nil { + glog.V(0).Infof("tail last record:%+v", time.Unix(0, lastTsNs)) return err } lastTsNs = resp.TsNs diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 1bcf57fe7..3082d0f55 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -53,7 +53,7 @@ func NewFiler(masters []string, grpcDialOption grpc.DialOption, GrpcDialOption: grpcDialOption, FilerConf: NewFilerConf(), } - f.LocalMetaLogBuffer = log_buffer.NewLogBuffer(LogFlushInterval, f.logFlushFunc, notifyFn) + f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, notifyFn) f.metaLogCollection = collection f.metaLogReplication = replication diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 241e99a1a..913cbd454 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -34,7 +34,7 @@ func NewMetaAggregator(filers []string, grpcDialOption grpc.DialOption) *MetaAgg grpcDialOption: grpcDialOption, } t.ListenersCond = sync.NewCond(&t.ListenersLock) - t.MetaLogBuffer = log_buffer.NewLogBuffer(LogFlushInterval, nil, func() { + t.MetaLogBuffer = log_buffer.NewLogBuffer("aggr", LogFlushInterval, nil, func() { t.ListenersCond.Broadcast() }) return t @@ -118,6 +118,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, peer string } for { + glog.V(4).Infof("subscribing remote %s meta change: %v", peer, time.Unix(0, lastTsNs)) err := pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/weed/messaging/broker/broker_grpc_server_subscribe.go b/weed/messaging/broker/broker_grpc_server_subscribe.go index 3021473e5..d21fb351f 100644 --- a/weed/messaging/broker/broker_grpc_server_subscribe.go +++ b/weed/messaging/broker/broker_grpc_server_subscribe.go @@ -116,7 +116,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs lastReadTime = time.Unix(0, processedTsNs) } - lastReadTime, err = lock.logBuffer.LoopProcessLogData(lastReadTime, func() bool { + lastReadTime, err = lock.logBuffer.LoopProcessLogData("broker", lastReadTime, func() bool { lock.Mutex.Lock() lock.cond.Wait() lock.Mutex.Unlock() diff --git a/weed/messaging/broker/topic_manager.go b/weed/messaging/broker/topic_manager.go index edddca813..c303c29b3 100644 --- a/weed/messaging/broker/topic_manager.go +++ b/weed/messaging/broker/topic_manager.go @@ -68,7 +68,7 @@ func (tm *TopicManager) buildLogBuffer(tl *TopicControl, tp TopicPartition, topi glog.V(0).Infof("log write failed %s: %v", targetFile, err) } } - logBuffer := log_buffer.NewLogBuffer(time.Minute, flushFn, func() { + logBuffer := log_buffer.NewLogBuffer("broker", time.Minute, flushFn, func() { tl.cond.Broadcast() }) diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index ea481f56a..23dc0bd59 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -34,7 +34,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, for { - glog.V(0).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) + glog.V(4).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) processedTsNs, err = fs.filer.ReadPersistedLogBuffer(lastReadTime, eachLogEntryFn) if err != nil { @@ -45,9 +45,9 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, lastReadTime = time.Unix(0, processedTsNs) } - glog.V(0).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) + glog.V(4).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) - lastReadTime, err = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool { + lastReadTime, err = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData("aggMeta:"+clientName, lastReadTime, func() bool { fs.filer.MetaAggregator.ListenersLock.Lock() fs.filer.MetaAggregator.ListenersCond.Wait() fs.filer.MetaAggregator.ListenersLock.Unlock() @@ -101,7 +101,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq glog.V(0).Infof("read in memory %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) - lastReadTime, err = fs.filer.LocalMetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool { + lastReadTime, err = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, func() bool { fs.listenersLock.Lock() fs.listenersCond.Wait() fs.listenersLock.Unlock() diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index f84c674ff..e3153fddd 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -22,6 +22,7 @@ type dataToFlush struct { } type LogBuffer struct { + name string prevBuffers *SealedBuffers buf []byte idx []int @@ -39,8 +40,9 @@ type LogBuffer struct { sync.RWMutex } -func NewLogBuffer(flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer { +func NewLogBuffer(name string, flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer { lb := &LogBuffer{ + name: name, prevBuffers: newSealedBuffers(PreviousBufferCount), buf: make([]byte, BufferSize), sizeBuf: make([]byte, 4), @@ -93,6 +95,7 @@ func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64) { } if m.startTime.Add(m.flushInterval).Before(ts) || len(m.buf)-m.pos < size+4 { + // glog.V(4).Infof("%s copyToFlush1 start time %v, ts %v, remaining %d bytes", m.name, m.startTime, ts, len(m.buf)-m.pos) m.flushChan <- m.copyToFlush() m.startTime = ts if len(m.buf) < size+4 { @@ -127,9 +130,10 @@ func (m *LogBuffer) Shutdown() { func (m *LogBuffer) loopFlush() { for d := range m.flushChan { if d != nil { - // fmt.Printf("flush [%v, %v] size %d\n", d.startTime, d.stopTime, len(d.data.Bytes())) + // glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes())) m.flushFn(d.startTime, d.stopTime, d.data.Bytes()) d.releaseMemory() + // local logbuffer is different from aggregate logbuffer here m.lastFlushTime = d.stopTime } } @@ -143,7 +147,6 @@ func (m *LogBuffer) loopInterval() { m.Unlock() return } - // println("loop interval") toFlush := m.copyToFlush() m.flushChan <- toFlush m.Unlock() @@ -162,7 +165,6 @@ func (m *LogBuffer) copyToFlush() *dataToFlush { data: copiedBytes(m.buf[:m.pos]), } } - // fmt.Printf("flusing [0,%d) with %d entries\n", m.pos, len(m.idx)) m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos) m.pos = 0 m.idx = m.idx[:0] @@ -200,12 +202,9 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu } if lastReadTime.Before(m.startTime) { // println("checking ", lastReadTime.UnixNano()) - for i, buf := range m.prevBuffers.buffers { + for _, buf := range m.prevBuffers.buffers { if buf.startTime.After(lastReadTime) { - if i == 0 { - // println("return the earliest in memory", buf.startTime.UnixNano()) - return copiedBytes(buf.buf[:buf.size]), nil - } + // glog.V(4).Infof("%s return the %d sealed buffer %v", m.name, i, buf.startTime) // println("return the", i, "th in memory", buf.startTime.UnixNano()) return copiedBytes(buf.buf[:buf.size]), nil } @@ -215,7 +214,7 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu return copiedBytes(buf.buf[pos:buf.size]), nil } } - // println("return the current buf", lastReadTime.UnixNano()) + // glog.V(4).Infof("%s return the current buf %v", m.name, lastReadTime) return copiedBytes(m.buf[:m.pos]), nil } diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go index 3d77afb18..7dcfe5f52 100644 --- a/weed/util/log_buffer/log_buffer_test.go +++ b/weed/util/log_buffer/log_buffer_test.go @@ -10,7 +10,7 @@ import ( ) func TestNewLogBufferFirstBuffer(t *testing.T) { - lb := NewLogBuffer(time.Minute, func(startTime, stopTime time.Time, buf []byte) { + lb := NewLogBuffer("test", time.Minute, func(startTime, stopTime time.Time, buf []byte) { }, func() { @@ -27,7 +27,7 @@ func TestNewLogBufferFirstBuffer(t *testing.T) { } receivedmessageCount := 0 - lb.LoopProcessLogData(startTime, func() bool { + lb.LoopProcessLogData("test", startTime, func() bool { // stop if no more messages return false }, func(logEntry *filer_pb.LogEntry) error { diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go index d6917abfe..02f5af274 100644 --- a/weed/util/log_buffer/log_read.go +++ b/weed/util/log_buffer/log_read.go @@ -17,10 +17,7 @@ var ( ResumeFromDiskError = fmt.Errorf("resumeFromDisk") ) -func (logBuffer *LogBuffer) LoopProcessLogData( - startTreadTime time.Time, - waitForDataFn func() bool, - eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) { +func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startTreadTime time.Time, waitForDataFn func() bool, eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) { // loop through all messages var bytesBuf *bytes.Buffer lastReadTime = startTreadTime @@ -39,7 +36,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData( if err == ResumeFromDiskError { return lastReadTime, ResumeFromDiskError } - // fmt.Printf("ReadFromBuffer by %v\n", lastReadTime) + // glog.V(4).Infof("%s ReadFromBuffer by %v", readerName, lastReadTime) if bytesBuf == nil { if waitForDataFn() { continue @@ -49,7 +46,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData( } buf := bytesBuf.Bytes() - // fmt.Printf("ReadFromBuffer by %v size %d\n", lastReadTime, len(buf)) + // fmt.Printf("ReadFromBuffer %s by %v size %d\n", readerName, lastReadTime, len(buf)) batchSize := 0 var startReadTime time.Time @@ -59,7 +56,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData( size := util.BytesToUint32(buf[pos : pos+4]) if pos+4+int(size) > len(buf) { err = ResumeError - glog.Errorf("LoopProcessLogData: read buffer %v read %d [%d,%d) from [0,%d)", lastReadTime, batchSize, pos, pos+int(size)+4, len(buf)) + glog.Errorf("LoopProcessLogData: %s read buffer %v read %d [%d,%d) from [0,%d)", readerName, lastReadTime, batchSize, pos, pos+int(size)+4, len(buf)) return } entryData := buf[pos+4 : pos+4+int(size)] @@ -81,9 +78,10 @@ func (logBuffer *LogBuffer) LoopProcessLogData( pos += 4 + int(size) batchSize++ + } - // fmt.Printf("sent message ts[%d,%d] size %d\n", startReadTime.UnixNano(), lastReadTime.UnixNano(), batchSize) + // glog.V(4).Infof("%s sent messages ts[%+v,%+v] size %d\n", readerName, startReadTime, lastReadTime, batchSize) } }