aboutsummaryrefslogtreecommitdiff
path: root/weed/util/log_buffer
diff options
context:
space:
mode:
Diffstat (limited to 'weed/util/log_buffer')
-rw-r--r--weed/util/log_buffer/log_buffer.go19
-rw-r--r--weed/util/log_buffer/log_buffer_test.go4
-rw-r--r--weed/util/log_buffer/log_read.go14
3 files changed, 17 insertions, 20 deletions
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)
}
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 07:48:03 +0000