Introduce StreamDone in Stream Writer #1061

ashish-goswami · 2019-10-03T09:43:57Z

This PR introduces, a way to tell StreamWriter to close a stream. Previously Streams were always open until Flush is called on StreamWriter. This resulted in memory utilisation, because of underlying TableBuilder to a sortedWriter.

I used below program to test the change.

package main

import (
	"fmt"
	"io/ioutil"
	"os"

	badger "github.com/dgraph-io/badger"
	"github.com/dgraph-io/badger/pb"
)

func main() {
	dir, err := ioutil.TempDir(".", "badger-test")
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := os.RemoveAll(dir); err != nil {
			panic(err)
		}
	}()
	opts := badger.DefaultOptions(dir)
	db, err := badger.Open(opts)
	if err != nil {
		panic("unable to open db " + err.Error())
	}
	defer func() {
		if err := db.Close(); err != nil {
			panic("unable to close db " + err.Error())
		}
	}()

	sw := db.NewStreamWriter()
	if err := sw.Prepare(); err != nil {
		panic("sw.Prepare() failed " + err.Error())
	}

	var val [10]byte
	for i := 0; i < 50000; i++ {
		list := &pb.KVList{}
		kv1 := &pb.KV{
			Key:      []byte(fmt.Sprintf("%d", i)),
			Value:    val[:],
			Version:  1,
			StreamId: uint32(i),
		}
		kv2 := &pb.KV{
			StreamId:   uint32(i),
			StreamDone: true,
		}
		list.Kv = append(list.Kv, kv1, kv2)
		if err := sw.Write(list); err != nil {
			panic("sw.Write() failed " + err.Error())
		}
		fmt.Println("Stream Done: ", i)
	}

	if err := sw.Flush(); err != nil {
		panic("sw.Flush() failed " + err.Error())
	}
}

Memory usage spikes(RES reach >25GB) in case of master branch. While this PR limits memory usage to very low.

I left above program running for a day(on this PR), it completed around ~800K streams without OOM.

This change is

pullrequest

✅ A review job has been created and sent to the PullRequest network.

@ashish-goswami you can click here to see the review status or cancel the code review job.

coveralls · 2019-10-03T09:52:24Z

Coverage increased (+0.2%) to 77.62% when pulling ffb416a on ashish/stream-writer into b20e8b6 on master.

jarifibrahim

Added a few comments.

Reviewable status: 0 of 4 files reviewed, 3 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @jarifibrahim, and @manishrjain)

stream_writer.go, line 33 at r1 (raw file):

var (
	// ErrStreamClosed is returned when a sent is performed on closed stream.

... when data is sent on a closed stream.

stream_writer.go, line 145 at r1 (raw file):

	defer sw.writeLock.Unlock()

	// Check if any of the stream we got records for are closed.

// Check if we've received data on a closed stream.

stream_writer.go, line 171 at r1 (raw file):

	// close any streams if required.
	for streamID := range closedStreams {
		writer, ok := sw.writers[streamID]

If we have a streamID in closedStreamsthat implies that we've recieved some data on that stream. Which also means there should be a writer for it. I would suggest we either log an error or return an error here if we don't find the streamID in sw.writers map.

pullrequest

Overall changes here look pretty good and thanks for showing the tests that you did to validate. I left some feedback and questions inline.

Reviewed with ❤️ by PullRequest

pullrequest · 2019-10-03T18:13:46Z

stream_writer.go

@@ -85,8 +90,23 @@ func (sw *StreamWriter) Write(kvs *pb.KVList) error {
 	if len(kvs.GetKv()) == 0 {
 		return nil
 	}
+
+	closedStreams := make(map[uint32]bool)


minor nitpick: Since you are just using these as basically a set you could use make(map[uint32]struct{}) since struct{} is zero-byte.

pullrequest · 2019-10-03T18:13:46Z

stream_writer.go

+			return err
+		}
+
+		sw.closedStreams[streamID] = true


Just to double check, should these be marked as closed directly after the SignalAndWait() or after the wait.Done()?

pullrequest · 2019-10-03T18:13:46Z

stream_writer.go

+	// Check if any of the stream we got records for are closed.
+	for streamID := range streamWithRecords {
+		if _, ok := sw.closedStreams[streamID]; ok {
+			return ErrStreamClosed


It seems like there may be a race where two threads are calling Write at the same time and one takes the lock first that is closing the stream and the other one still has a streamWithRecords version of that streamID since the above code is run concurrently and unlocked. I just wanted to check if this is okay to return an error in this case? My guess is that its probably fine, but it may be a slight change in behavior from the current behavior in some cases.

pullrequest · 2019-10-03T18:13:46Z

stream_writer.go

+const headStreamID uint32 = math.MaxUint32
+
+var (
+	// ErrStreamClosed is returned when a sent is performed on closed stream.


Should this possibly say "when a write is performed" instead of "when a sent is performed"?

pullrequest · 2019-10-03T18:13:46Z

stream_writer.go

@@ -121,6 +141,14 @@ func (sw *StreamWriter) Write(kvs *pb.KVList) error {

 	sw.writeLock.Lock()
 	defer sw.writeLock.Unlock()
+
+	// Check if any of the stream we got records for are closed.


stream -> streams

pullrequest

Overall this looks good, but it might be worth checking the script from the commit message in as well for future testing and verification

Reviewed with ❤️ by PullRequest

golangcibot · 2019-10-04T10:50:58Z

stream_writer.go

+
+	// Now we can close any streams if required. We will make writer for
+	// the closed streams as nil.
+	for streamId := range closedStreams {


range var streamId should be streamID (from golint)

ashish-goswami

Reviewable status: 0 of 4 files reviewed, 8 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @jarifibrahim, @manishrjain, and @pullrequest[bot])

stream_writer.go, line 33 at r1 (raw file):

Previously, jarifibrahim (Ibrahim Jarif) wrote…

... when data is sent on a closed stream.

I have removed the error.

stream_writer.go, line 94 at r1 (raw file):

Previously, pullrequest[bot] wrote…

minor nitpick: Since you are just using these as basically a set you could use make(map[uint32]struct{}) since struct{} is zero-byte.

Done.

stream_writer.go, line 145 at r1 (raw file):

Previously, jarifibrahim (Ibrahim Jarif) wrote…

// Check if we've received data on a closed stream.

changed the comment.

stream_writer.go, line 145 at r1 (raw file):

Previously, pullrequest[bot] wrote…

stream -> streams

changed the comment.

stream_writer.go, line 148 at r1 (raw file):

Previously, pullrequest[bot] wrote…

It seems like there may be a race where two threads are calling Write at the same time and one takes the lock first that is closing the stream and the other one still has a streamWithRecords version of that streamID since the above code is run concurrently and unlocked. I just wanted to check if this is okay to return an error in this case? My guess is that its probably fine, but it may be a slight change in behavior from the current behavior in some cases.

Right now, we are panicking if sent is done on closed stream. Normally it is expected that single stream is handled by a same thread.

stream_writer.go, line 171 at r1 (raw file):

Previously, jarifibrahim (Ibrahim Jarif) wrote…

If we have a streamID in closedStreamsthat implies that we've recieved some data on that stream. Which also means there should be a writer for it. I would suggest we either log an error or return an error here if we don't find the streamID in sw.writers map.

Not necessarily. We can just get close(StreamDone) without any data also. If it is with data, we will definitely have writer for stream(if not code panics). If no data has been received(just StreamDone) and we don't have any writer for it, we are just logging it.

stream_writer.go, line 181 at r1 (raw file):

Previously, pullrequest[bot] wrote…

Just to double check, should these be marked as closed directly after the SignalAndWait() or after the wait.Done()?

SignalAndWait already waits for it.

manishrjain

Make the above program part of the tests. And a comment.

Reviewed 2 of 4 files at r1, 1 of 2 files at r2, 1 of 1 files at r3.
Reviewable status: all files reviewed, 10 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @jarifibrahim, and @pullrequest[bot])

stream_writer.go, line 140 at r3 (raw file):

	// We are writing all requests to vlog even if some request belongs to already closed stream.
	// It is safe to do because we are panicing while writing to sorted writer, which will be nil

panicking

stream_writer.go, line 282 at r3 (raw file):

		builder:  table.NewTableBuilder(bopts),
		reqCh:    make(chan *request, 3),
		closer:   closer,

closer: y.NewCloser(1)

jarifibrahim

Reviewable status: all files reviewed, 8 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @jarifibrahim, @manishrjain, and @pullrequest[bot])

stream_writer.go, line 140 at r3 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

panicking

This needs to be fixed.

ashish-goswami

Reviewable status: 2 of 4 files reviewed, 8 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @jarifibrahim, @manishrjain, and @pullrequest[bot])

stream_writer.go, line 140 at r3 (raw file):

Previously, jarifibrahim (Ibrahim Jarif) wrote…

This needs to be fixed.

Done.

stream_writer.go, line 282 at r3 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

closer: y.NewCloser(1)

Done.

jarifibrahim

Reviewable status: 2 of 5 files reviewed, 7 unresolved discussions (waiting on @ashish-goswami, @balajijinnah, @manishrjain, and @pullrequest[bot])

This PR introduces, a way to tell StreamWriter to close a stream. Previously Streams were always open until Flush is called on StreamWriter. This resulted in memory utilisation, because of underlying TableBuilder to a sortedWriter. Also closing all sorted writer in single call resulted in more memory allocation(during Flush()). This can be useful in some case such as bulk loader in Dgraph, where only one stream is active at a time. (cherry picked from commit 385da91)

ashish-goswami added 5 commits October 1, 2019 13:21

Add support for stream end in stream writer

55ca739

Fix some issues

7d60e0f

Linting fixes

70335b7

Add tests for Stream Done

eaadf99

Move closers inside sorted writer

73d78bf

ashish-goswami requested review from jarifibrahim, manishrjain and a team as code owners October 3, 2019 09:43

pullrequest bot reviewed Oct 3, 2019

View reviewed changes

ashish-goswami requested a review from poonai October 3, 2019 09:44

jarifibrahim suggested changes Oct 3, 2019

View reviewed changes

pullrequest bot reviewed Oct 3, 2019

View reviewed changes

Address review comments

e50743c

golangcibot reviewed Oct 4, 2019

View reviewed changes

ashish-goswami commented Oct 4, 2019

View reviewed changes

Fix AppVeyor tests

f0381be

manishrjain approved these changes Oct 4, 2019

View reviewed changes

jarifibrahim approved these changes Oct 7, 2019

View reviewed changes

Address review comments

3badcac

ashish-goswami commented Oct 10, 2019

View reviewed changes

ashish-goswami added 4 commits October 16, 2019 15:45

Add -v in travis go test

5a1d188

Decrease no of stream to fix travis build

f065b79

Fix maxhead while closing streams

6c20977

Add test for maxhead update in stream writer

135834d

jarifibrahim approved these changes Oct 17, 2019

View reviewed changes

ashish-goswami added 3 commits October 17, 2019 18:40

Remove test as it is failing with race

c1684fe

Trigger license/cla

eda20f5

Trigger license/cla2

ffb416a

ashish-goswami merged commit 385da91 into master Oct 18, 2019

ashish-goswami deleted the ashish/stream-writer branch October 18, 2019 13:35

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce StreamDone in Stream Writer #1061

Introduce StreamDone in Stream Writer #1061

ashish-goswami commented Oct 3, 2019 •

edited by manishrjain

Loading

pullrequest bot left a comment

coveralls commented Oct 3, 2019 •

edited

Loading

jarifibrahim left a comment

pullrequest bot left a comment

pullrequest bot Oct 3, 2019

pullrequest bot Oct 3, 2019

pullrequest bot Oct 3, 2019

pullrequest bot Oct 3, 2019

pullrequest bot Oct 3, 2019

pullrequest bot left a comment

golangcibot Oct 4, 2019

ashish-goswami left a comment

manishrjain left a comment

jarifibrahim left a comment

ashish-goswami left a comment

jarifibrahim left a comment

Introduce StreamDone in Stream Writer #1061

Introduce StreamDone in Stream Writer #1061

Conversation

ashish-goswami commented Oct 3, 2019 • edited by manishrjain Loading

pullrequest bot left a comment

Choose a reason for hiding this comment

coveralls commented Oct 3, 2019 • edited Loading

jarifibrahim left a comment

Choose a reason for hiding this comment

pullrequest bot left a comment

Choose a reason for hiding this comment

pullrequest bot Oct 3, 2019

Choose a reason for hiding this comment

pullrequest bot Oct 3, 2019

Choose a reason for hiding this comment

pullrequest bot Oct 3, 2019

Choose a reason for hiding this comment

pullrequest bot Oct 3, 2019

Choose a reason for hiding this comment

pullrequest bot Oct 3, 2019

Choose a reason for hiding this comment

pullrequest bot left a comment

Choose a reason for hiding this comment

golangcibot Oct 4, 2019

Choose a reason for hiding this comment

ashish-goswami left a comment

Choose a reason for hiding this comment

manishrjain left a comment

Choose a reason for hiding this comment

jarifibrahim left a comment

Choose a reason for hiding this comment

ashish-goswami left a comment

Choose a reason for hiding this comment

jarifibrahim left a comment

Choose a reason for hiding this comment

ashish-goswami commented Oct 3, 2019 •

edited by manishrjain

Loading

coveralls commented Oct 3, 2019 •

edited

Loading