attempt 4: chunk lines before sending to channel

.gitignore

@@ -1,3 +1,3 @@
 measurements.txt
 trace*
-.DS_store
+.DS_store

README.md

@@ -3,6 +3,7 @@
 | Attempt Number | Approach | Execution Time | Diff | Commit |
 |-----------------|---|---|---|--|
 |1| Naive Implementation: Read temperatures into a map of cities. Iterate serially over each key (city) in map to find min, max and average temperatures.| 6:13.15 | || 
-|2| Evaluate each city in map concurrently using goroutines.|4:32.80|-100.35| 8bd5f43|
-|3|Remove sorting float64 slices. Calculate min, max and average by iterating.|4:25.59|-7.21|830e5df|
-|4|Decouple reading and processing of file content. A buffered goroutine if used to communicate between the two processes.|5:22.83|+57.24||
+|2| Evaluate each city in map concurrently using goroutines.|4:32.80|-100.35| [8bd5f43](https://github.com/shraddhaag/1brc/commit/8bd5f437e8cc231e3ee18348b83f4dc694137546)|
+|3|Remove sorting float64 slices. Calculate min, max and average by iterating.|4:25.59|-7.21|[830e5df](https://github.com/shraddhaag/1brc/commit/830e5dfacff9fb7a41d12027e21399736bc34701)|
+|4|Decouple reading and processing of file content. A buffered goroutine is used to communicate between the two processes.|5:22.83|+57.24|[2babf7d](https://github.com/shraddhaag/1brc/commit/2babf7dda72d92c72722b220b8b663e747075bd7)|
+|5|Instead of sending each line to the channel, now sending 100 lines chunked together. Also, to minimise garbage collection, not freeing up memory when resetting a slice. |3:41.76|-161.07||

main.go

@@ -4,12 +4,8 @@ import (
 	"bufio"
 	"flag"
 	"fmt"
-	"log"
 	"math"
 	"os"
-	"runtime"
-	"runtime/pprof"
-	"runtime/trace"
 	"sort"
 	"strconv"
 	"strings"
@@ -21,35 +17,36 @@ var memprofile = flag.String("memprofile", "", "write memory profile to `file`")
 
 func main() {
 
-	trace.Start(os.Stderr)
-	defer trace.Stop()
-
-	flag.Parse()
-	if *cpuprofile != "" {
-		f, err := os.Create("./profiles/" + *cpuprofile)
-		if err != nil {
-			log.Fatal("could not create CPU profile: ", err)
-		}
-		defer f.Close() // error handling omitted for example
-		if err := pprof.StartCPUProfile(f); err != nil {
-			log.Fatal("could not start CPU profile: ", err)
-		}
-		defer pprof.StopCPUProfile()
-	}
-
-	fmt.Println(evaluate())
-
-	if *memprofile != "" {
-		f, err := os.Create("./profiles/" + *memprofile)
-		if err != nil {
-			log.Fatal("could not create memory profile: ", err)
-		}
-		defer f.Close() // error handling omitted for example
-		runtime.GC()    // get up-to-date statistics
-		if err := pprof.WriteHeapProfile(f); err != nil {
-			log.Fatal("could not write memory profile: ", err)
-		}
-	}
+	// trace.Start(os.Stderr)
+	// defer trace.Stop()
+
+	// flag.Parse()
+	// if *cpuprofile != "" {
+	// 	f, err := os.Create("./profiles/" + *cpuprofile)
+	// 	if err != nil {
+	// 		log.Fatal("could not create CPU profile: ", err)
+	// 	}
+	// 	defer f.Close() // error handling omitted for example
+	// 	if err := pprof.StartCPUProfile(f); err != nil {
+	// 		log.Fatal("could not start CPU profile: ", err)
+	// 	}
+	// 	defer pprof.StopCPUProfile()
+	// }
+
+	evaluate()
+	// fmt.Println(evaluate())
+
+	// if *memprofile != "" {
+	// 	f, err := os.Create("./profiles/" + *memprofile)
+	// 	if err != nil {
+	// 		log.Fatal("could not create memory profile: ", err)
+	// 	}
+	// 	defer f.Close() // error handling omitted for example
+	// 	runtime.GC()    // get up-to-date statistics
+	// 	if err := pprof.WriteHeapProfile(f); err != nil {
+	// 		log.Fatal("could not write memory profile: ", err)
+	// 	}
+	// }
 }
 
 func evaluate() string {
@@ -109,34 +106,54 @@ func readFileLineByLineIntoAMap(filepath string) (map[string][]float64, error) {
 
 	mapOfTemp := make(map[string][]float64)
 
-	chanOwner := func() <-chan string {
-		resultStream := make(chan string, 100)
+	chanOwner := func() <-chan []string {
+		resultStream := make(chan []string, 100)
+		toSend := make([]string, 100)
 		go func() {
 			defer close(resultStream)
 			scanner := bufio.NewScanner(file)
+			var count int
 			for scanner.Scan() {
-				text := scanner.Text()
-				resultStream <- text
+				if count == 100 {
+					localCopy := make([]string, 100)
+					copy(localCopy, toSend)
+					resultStream <- localCopy
+					count = 0
+				}
+				toSend[count] = scanner.Text()
+				count++
+			}
+			if count != 0 {
+				resultStream <- toSend[:count]
 			}
 		}()
 		return resultStream
 	}
 
 	resultStream := chanOwner()
-
-	for text := range resultStream {
-		index := strings.Index(text, ";")
-		city := text[:index]
-		temp := convertStringToFloat(text[index+1:])
-		if _, ok := mapOfTemp[city]; ok {
-			mapOfTemp[city] = append(mapOfTemp[city], temp)
-		} else {
-			mapOfTemp[city] = []float64{temp}
+	for t := range resultStream {
+		for _, text := range t {
+			index := strings.Index(text, ";")
+			if index == -1 {
+				continue
+			}
+			city := text[:index]
+			temp := convertStringToFloat(text[index+1:])
+			if _, ok := mapOfTemp[city]; ok {
+				mapOfTemp[city] = append(mapOfTemp[city], temp)
+			} else {
+				mapOfTemp[city] = []float64{temp}
+			}
 		}
 	}
 	return mapOfTemp, nil
 }
 
+type cityTemp struct {
+	city string
+	temp float64
+}
+
 func convertStringToFloat(input string) float64 {
 	output, _ := strconv.ParseFloat(input, 64)
 	return output