Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Flush vlog buffer if it grows beyond threshold #1067

Merged
merged 4 commits into from
Oct 10, 2019
Merged

Flush vlog buffer if it grows beyond threshold #1067

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1ad5160
Flush vlog buffer if it grows beyond threshold
Oct 4, 2019
e58721b
Remove unreliable test
Oct 7, 2019
60c77f0
Address review comments
Oct 9, 2019
bd1c2b1
Address review comments
Oct 10, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 19 additions & 5 deletions value.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1260,11 +1260,11 @@ func (vlog *valueLog) write(reqs []*request) error {
vlog.filesLock.RUnlock()

var buf bytes.Buffer
toDisk := func() error {
flushWrites := func() error {
if buf.Len() == 0 {
return nil
}
vlog.elog.Printf("Flushing %d blocks of total size: %d", len(reqs), buf.Len())
vlog.elog.Printf("Flushing buffer of size %d to vlog", buf.Len())
n, err := curlf.fd.Write(buf.Bytes())
if err != nil {
return errors.Wrapf(err, "Unable to write to value log file: %q", curlf.path)
Expand All @@ -1274,11 +1274,15 @@ func (vlog *valueLog) write(reqs []*request) error {
y.NumBytesWritten.Add(int64(n))
vlog.elog.Printf("Done")
atomic.AddUint32(&vlog.writableLogOffset, uint32(n))

return nil
}
toDisk := func() error {
if err := flushWrites(); err != nil {
return err
}
if vlog.woffset() > uint32(vlog.opt.ValueLogFileSize) ||
vlog.numEntriesWritten > vlog.opt.ValueLogMaxEntries {
var err error
if err = curlf.doneWriting(vlog.woffset()); err != nil {
if err := curlf.doneWriting(vlog.woffset()); err != nil {
return err
}

Expand Down Expand Up @@ -1315,6 +1319,16 @@ func (vlog *valueLog) write(reqs []*request) error {
p.Len = uint32(plen)
b.Ptrs = append(b.Ptrs, p)
written++

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can the buffer length be stored since it is retrieved multiple times?

bl := buf.Len()

// It is possible that the size of the buffer grows beyond the max size of the value
// log (this happens when a transaction contains entries with large value sizes) and
// badger might run into out of memory errors. We flush the buffer here if it's size
// grows beyond the max value log size.
if int64(buf.Len()) > vlog.db.opt.ValueLogFileSize {
if err := flushWrites(); err != nil {
return err
}
}
}
vlog.numEntriesWritten += uint32(written)
// We write to disk here so that all entries that are part of the same transaction are
Expand Down