Bulkloader stage 2 (shuffle)

[peterstace]

• After N sorted files are dumped to disk, the shuffle phase begins.
• Each file is opened in its own goroutine.
• Parse each file for flat postings.
• Push each posting onto a channel (1 channel per posting file).
• A min-heap data structure knows about each of the channels.
• Reads off the channels to provide the final sorted stream of flat postings.
• Completely sorted flat postings are batched up and sent to the reduce phase (currently stubbed out -- to be completed in the next phase).

---

This change is 

[manishrjain]

Reviewed 8 of 8 files at r1.
Review status: all files reviewed at latest revision, 13 unresolved discussions, some commit checks broke.

---

cmd/bulkloader/loader.go, line 37 at r1 (raw file):

	*state
	mappers     []*mapper
	mappedFiles []string

mapOutput

---

cmd/bulkloader/loader.go, line 74 at r1 (raw file):

	go func() {
		ld.mappedFiles = writeMappedFiles(tmpPostingsDir, ld.postingsCh, ld.prog)

writeMapOutput

---

cmd/bulkloader/loader.go, line 113 at r1 (raw file):

func (ld *loader) reduceStage() {
	// Read from map stage.
	flatPostingChs := make([]chan *protos.FlatPosting, len(ld.mappedFiles))

shuffleInputChs

---

cmd/bulkloader/loader.go, line 114 at r1 (raw file):

	// Read from map stage.
	flatPostingChs := make([]chan *protos.FlatPosting, len(ld.mappedFiles))
	uniDirFlatPostingChs := make([]<-chan *protos.FlatPosting, len(ld.mappedFiles))

No need for this.

---

cmd/bulkloader/loader.go, line 122 at r1 (raw file):

	// Shuffle concurrently with reduce.
	batchCh := make(chan []*protos.FlatPosting, 2) // Small buffer size since each element has a lot of data.

3

---

cmd/bulkloader/loader.go, line 122 at r1 (raw file):

	// Shuffle concurrently with reduce.
	batchCh := make(chan []*protos.FlatPosting, 2) // Small buffer size since each element has a lot of data.

reduceCh

---

cmd/bulkloader/loader.go, line 126 at r1 (raw file):

	// Reduce stage.
	counter := make(chan struct{}, ld.opt.numGoroutines)

pending

---

cmd/bulkloader/progress.go, line 44 at r1 (raw file):

func (p *progress) reportOnce() {

remove v space.

---

cmd/bulkloader/write_flat.go, line 71 at r1 (raw file):

}

func readFlatFile(filename string, postingCh chan<- *protos.FlatPosting) {

readMapOutput(filename string, shuffleCh)

---

cmd/bulkloader/write_flat.go, line 90 at r1 (raw file):

		x.Check2(r.Discard(n))

		for len(unmarshalBuf) < int(sz) {

if cap(unmarshalBuf) < int(sz) { create slice of size sz }

---

cmd/bulkloader/write_flat.go, line 102 at r1 (raw file):

}

func shuffleFlatFiles(batchCh chan<- []*protos.FlatPosting,

shufflePostings

---

cmd/bulkloader/write_flat.go, line 139 at r1 (raw file):

type heapNode struct {
	head *protos.FlatPosting

posting

---

cmd/bulkloader/write_flat.go, line 144 at r1 (raw file):

type postingHeap struct {
	data []heapNode

nodes []heapNode

---

Comments from Reviewable

[peterstace]

Review status: 5 of 8 files reviewed at latest revision, 13 unresolved discussions.

---

cmd/bulkloader/loader.go, line 37 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

mapOutput

Done.

---

cmd/bulkloader/loader.go, line 74 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

writeMapOutput

Done.

---

cmd/bulkloader/loader.go, line 113 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

shuffleInputChs

Done.

---

cmd/bulkloader/loader.go, line 114 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

No need for this.

Done.

---

cmd/bulkloader/loader.go, line 122 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

3

Done.

---

cmd/bulkloader/loader.go, line 122 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

reduceCh

Done.

---

cmd/bulkloader/loader.go, line 126 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

pending

Done.

---

cmd/bulkloader/progress.go, line 44 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

remove v space.

Done.

---

cmd/bulkloader/write_flat.go, line 71 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

readMapOutput(filename string, shuffleCh)

Done.

---

cmd/bulkloader/write_flat.go, line 90 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

if cap(unmarshalBuf) < int(sz) { create slice of size sz }

Done.

---

cmd/bulkloader/write_flat.go, line 102 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

shufflePostings

Done.

---

cmd/bulkloader/write_flat.go, line 139 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

posting

Done.

---

cmd/bulkloader/write_flat.go, line 144 at r1 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

nodes []heapNode

Done.

---

Comments from Reviewable

[manishrjain]

---

Reviewed 3 of 3 files at r2.
Review status: all files reviewed at latest revision, all discussions resolved.

---

Comments from Reviewable