reduce: separate batch calls

1 files changed, 18 insertions, 18 deletions

diff --git a/skate/reduce.go b/skate/reduce.go
index 5b30fdf..fc286d9 100644
--- a/skate/reduce.go
+++ b/skate/reduce.go
@@ -44,7 +44,7 @@ func ZippyExact(releases, refs io.Reader, matchResult MatchResult, w io.Writer)
 	var (
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			var (
 				target *Release
 				ref    *Ref
@@ -80,9 +80,9 @@ func ZippyExact(releases, refs io.Reader, matchResult MatchResult, w io.Writer)
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000) // hard-code for now; on 24 cores 10K take up over 8G of RAM
 	)
-	batcher.Size = 10000 // hard-code for now; on 24 cores 10K take up over 8G of RAM
 	defer batcher.Close()
 	zipper := zipkey.New(releases, refs, keyer, batcher.GroupFunc)
 	return zipper.Run()
@@ -94,7 +94,7 @@ func ZippyExactReleases(olr, releases io.Reader, matchResult MatchResult, w io.W
 	var (
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			var (
 				target, re *Release
 				err        error
@@ -130,9 +130,9 @@ func ZippyExactReleases(olr, releases io.Reader, matchResult MatchResult, w io.W
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000)
 	)
-	batcher.Size = 10000
 	defer batcher.Close()
 	zipper := zipkey.New(olr, releases, keyer, batcher.GroupFunc)
 	return zipper.Run()
@@ -144,7 +144,7 @@ func ZippyExactWiki(releases, wiki io.Reader, mr MatchResult, w io.Writer) error
 	var (
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			var (
 				target *Release
 				wiki   *MinimalCitations
@@ -173,9 +173,9 @@ func ZippyExactWiki(releases, wiki io.Reader, mr MatchResult, w io.Writer) error
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000)
 	)
-	batcher.Size = 10000
 	defer batcher.Close()
 	zipper := zipkey.New(releases, wiki, keyer, batcher.GroupFunc)
 	return zipper.Run()
@@ -188,7 +188,7 @@ func ZippyVerifyRefs(releases, refs io.Reader, w io.Writer) error {
 	var (
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			var (
 				re, pivot *Release
 				err       error
@@ -217,9 +217,9 @@ func ZippyVerifyRefs(releases, refs io.Reader, w io.Writer) error {
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000)
 	)
-	batcher.Size = 10000
 	defer batcher.Close()
 	zipper := zipkey.New(releases, refs, keyer, batcher.GroupFunc)
 	return zipper.Run()
@@ -273,7 +273,7 @@ func ZippyVerifyRefsOpenLibrary(olr, refs io.Reader, w io.Writer) error {
 	var (
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			var (
 				ref, pivot *Release // ref (reference), pivot (open library)
 				err        error
@@ -313,9 +313,9 @@ func ZippyVerifyRefsOpenLibrary(olr, refs io.Reader, w io.Writer) error {
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000)
 	)
-	batcher.Size = 10000
 	defer batcher.Close()
 	zipper := zipkey.New(olr, refs, keyer, batcher.GroupFunc)
 	return zipper.Run()
@@ -336,7 +336,7 @@ func ZippyBrefAugment(bref, raw io.Reader, w io.Writer) error {
 		stats   = statsAugment{}
 		enc     = json.NewEncoder(xio.NewSingleWriter(w))
 		keyer   = makeKeyFunc("\t", 1)
-		batcher = zipkey.NewBatcher(func(g *zipkey.Group) error {
+		grouper = func(g *zipkey.Group) error {
 			// g.G0 contains matched docs for a given work id, g.G1 all raw
 			// refs, with the same work id.
 
@@ -373,9 +373,9 @@ func ZippyBrefAugment(bref, raw io.Reader, w io.Writer) error {
 				}
 			}
 			return nil
-		})
+		}
+		batcher = zipkey.NewBatcherSize(grouper, 10000)
 	)
-	batcher.Size = 10000
 	defer batcher.Close()
 	zipper := zipkey.New(bref, raw, keyer, batcher.GroupFunc)
 	err := zipper.Run()