mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-07-02 15:16:51 +02:00
Code Issues Projects Releases Wiki Activity

continue to read from memory if there is no flush

Browse source
This commit is contained in:
Chris Lu 2021-10-17 13:53:04 -07:00
parent 29fa1b9fdc
commit 3833dac3f7
1 changed files with 10 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
weed/util/log_buffer/log_buffer.go
View file

@ -29,6 +29,7 @@ type LogBuffer struct {
pos int pos int
startTime time.Time startTime time.Time
stopTime time.Time stopTime time.Time
lastFlushTime time.Time
sizeBuf []byte sizeBuf []byte
flushInterval time.Duration flushInterval time.Duration
flushFn func(startTime, stopTime time.Time, buf []byte) flushFn func(startTime, stopTime time.Time, buf []byte)
@ -132,6 +133,8 @@ func (m *LogBuffer) loopFlush() {
// glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, 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()) m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
d.releaseMemory() d.releaseMemory()
// local logbuffer is different from aggregate logbuffer here
m.lastFlushTime = d.stopTime
} }
} }
} }
@ -162,6 +165,9 @@ func (m *LogBuffer) copyToFlush() *dataToFlush {
data: copiedBytes(m.buf[:m.pos]), data: copiedBytes(m.buf[:m.pos]),
} }
// glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime) // glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
} else {
// glog.V(4).Infof("%s removed from memory [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
m.lastFlushTime = m.stopTime
} }
m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos) m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos)
m.startTime = time.Unix(0, 0) m.startTime = time.Unix(0, 0)
@ -203,7 +209,10 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
if tsMemory.IsZero() { // case 2.2 if tsMemory.IsZero() { // case 2.2
return nil, nil return nil, nil
} else if lastReadTime.Before(tsMemory) { // case 2.3 } else if lastReadTime.Before(tsMemory) { // case 2.3
return nil, ResumeFromDiskError if !m.lastFlushTime.IsZero() {
glog.V(0).Infof("resume with last flush time: %v", m.lastFlushTime)
return nil, ResumeFromDiskError
}
} }
// the following is case 2.1 // the following is case 2.1