feat: v2.0.0, makefile

Author: RamazanKara

README.md

@@ -13,7 +13,7 @@ This Go program is designed to efficiently process a large dataset of temperatur
 
 Processing Time: 9m21s. Tested with a Ryzen 5800x3d
 
-## Recent Optimizations (v1.1.0)
+## v1.1.0
 
 The program has undergone several optimizations to improve its processing time:
 
@@ -24,6 +24,20 @@ The program has undergone several optimizations to improve its processing time:
 
 Processing Time: 6m53s. Tested with a Ryzen 5800x3d
 
+## v2.0.0
+
+Version 2.0 of the One Billion Row Challenge Processor introduces significant optimizations, leading to a substantial reduction in processing time. This release focuses on enhancing concurrency handling and reducing contention, along with other performance improvements.
+
+## Performance Enhancements
+
+- **Concurrent Map Implementation:** Introduced a sharded concurrent map to reduce lock contention. This allows for more efficient updates to the data structure in a multi-threaded environment.
+- **Hash-Based Sharding:** Implemented hash-based sharding for distributing data across multiple shards, further reducing the chance of lock conflicts.
+- **Optimized String Processing:** Refined the string handling logic to minimize overhead during file parsing.
+- **Buffer Size Adjustments:** Tuned the buffer sizes for channels to balance throughput and memory usage.
+- **Efficient Data Aggregation:** Streamlined the data aggregation process for improved efficiency.
+
+Processing Time 5m19s. Tested with a Ryzen 5800x3d
+
 ## Requirements
 
 - Go Runtime ofc (1.21)

main.go

@@ -1,28 +1,54 @@
 package main
 
 import (
-	"bufio"
-	"fmt"
-	"os"
-	"sort"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
+    "bufio"
+    "fmt"
+    "hash/fnv"
+    "os"
+    "sort"
+    "strconv"
+    "strings"
+    "sync"
+    "time"
 )
 
-// StationData stores min, mean, max, and total sum of temperatures and count of readings for a station
 type StationData struct {
-	min, max, sum, count float64
+    min, max, sum, count float64
 }
 
-const numWorkers = 16 // Number of worker goroutines
+const numWorkers = 16
+const numShards = 32 // Number of shards in the concurrent map
+
+type ConcurrentMap struct {
+    shards [numShards]map[string]*StationData
+    locks  [numShards]*sync.Mutex
+}
+
+func NewConcurrentMap() *ConcurrentMap {
+    cMap := &ConcurrentMap{}
+    for i := 0; i < numShards; i++ {
+        cMap.shards[i] = make(map[string]*StationData)
+        cMap.locks[i] = &sync.Mutex{}
+    }
+    return cMap
+}
+
+func (cMap *ConcurrentMap) GetShard(key string) (shard map[string]*StationData, lock *sync.Mutex) {
+    hash := fnv.New32()
+    hash.Write([]byte(key))
+    shardIndex := hash.Sum32() % numShards
+    return cMap.shards[shardIndex], cMap.locks[shardIndex]
+}
 
 func main() {
     startTime := time.Now()
 
-	// Adjust this to the path of your data file
-    fileName := "./data/measurements.txt"
+    if len(os.Args) < 2 {
+        fmt.Println("Usage: brc <file_path>")
+        os.Exit(1)
+    }
+    fileName := os.Args[1]
+
     stationData := processFile(fileName)
 
     printResults(stationData)
@@ -31,18 +57,15 @@ func main() {
     fmt.Printf("Processing completed in %s\n", duration)
 }
 
-func processFile(fileName string) map[string]*StationData {
+func processFile(fileName string) *ConcurrentMap {
     linesCh := make(chan string, 1000)
-
     var wg sync.WaitGroup
     wg.Add(numWorkers)
 
-    stationData := make(map[string]*StationData)
-    var mu sync.Mutex
+    cMap := NewConcurrentMap()
 
-    // Worker pool pattern
     for i := 0; i < numWorkers; i++ {
-        go worker(&wg, linesCh, stationData, &mu)
+        go worker(&wg, linesCh, cMap)
     }
 
     file, err := os.Open(fileName)
@@ -58,17 +81,17 @@ func processFile(fileName string) map[string]*StationData {
     close(linesCh)
     wg.Wait()
 
-    return stationData
+    return cMap
 }
 
-func worker(wg *sync.WaitGroup, lines <-chan string, data map[string]*StationData, mu *sync.Mutex) {
+func worker(wg *sync.WaitGroup, lines <-chan string, cMap *ConcurrentMap) {
     defer wg.Done()
     for line := range lines {
-        processLine(line, data, mu)
+        processLine(line, cMap)
     }
 }
 
-func processLine(line string, data map[string]*StationData, mu *sync.Mutex) {
+func processLine(line string, cMap *ConcurrentMap) {
     parts := strings.Split(line, ";")
     if len(parts) != 2 {
         return
@@ -80,40 +103,50 @@ func processLine(line string, data map[string]*StationData, mu *sync.Mutex) {
         return
     }
 
-    mu.Lock()
-    defer mu.Unlock()
-
-    if sd, exists := data[station]; exists {
-        sd.sum += temp
-        sd.count++
-        if temp < sd.min {
-            sd.min = temp
+    shard, lock := cMap.GetShard(station)
+    lock.Lock()
+    data, exists := shard[station]
+    if !exists {
+        data = &StationData{min: temp, max: temp, sum: temp, count: 1}
+        shard[station] = data
+    } else {
+        data.sum += temp
+        data.count++
+        if temp < data.min {
+            data.min = temp
         }
-        if temp > sd.max {
-            sd.max = temp
+        if temp > data.max {
+            data.max = temp
         }
-    } else {
-        data[station] = &StationData{min: temp, max: temp, sum: temp, count: 1}
     }
+    lock.Unlock()
 }
 
-func printResults(stationData map[string]*StationData) {
-	// Extract keys and sort them
-	keys := make([]string, 0, len(stationData))
-	for key := range stationData {
-		keys = append(keys, key)
-	}
-	sort.Strings(keys)
-
-	// Generate output
-	fmt.Print("{")
-	for i, key := range keys {
-		data := stationData[key]
-		mean := data.sum / data.count
-		fmt.Printf("%s=%.1f/%.1f/%.1f", key, data.min, mean, data.max)
-		if i < len(keys)-1 {
-			fmt.Print(", ")
-		}
-	}
-	fmt.Println("}")
+func printResults(cMap *ConcurrentMap) {
+    // Consolidate data from shards
+    consolidatedData := make(map[string]*StationData)
+    for _, shard := range cMap.shards {
+        for station, data := range shard {
+            consolidatedData[station] = data
+        }
+    }
+
+    // Sort the station names
+    keys := make([]string, 0, len(consolidatedData))
+    for station := range consolidatedData {
+        keys = append(keys, station)
+    }
+    sort.Strings(keys)
+
+    // Print sorted results
+    fmt.Print("{")
+    for i, key := range keys {
+        data := consolidatedData[key]
+        mean := data.sum / data.count
+        fmt.Printf("%s=%.1f/%.1f/%.1f", key, data.min, mean, data.max)
+        if i < len(keys)-1 {
+            fmt.Print(", ")
+        }
+    }
+    fmt.Println("}")
 }

makefile

@@ -0,0 +1,36 @@
+
+BINARY = brc
+# Default rule
+all: build run clean
+
+# Build
+build:
+    go build -o ${BINARY}
+
+# Run
+run:
+    @echo "Usage: make run FILE=<file_path>"
+    @if [ "${FILE}" = "" ]; then \
+        echo "Error: FILE not specified. Use 'make run FILE=<file_path>' to run the project."; \
+    else \
+        ./${BINARY} ${FILE}; \
+    fi
+
+# Clean up
+clean:
+    if [ -f ${BINARY} ] ; then rm ${BINARY} ; fi
+
+# Test
+test:
+    go test
+
+# Help
+help:
+    @echo "make: build and run the project"
+    @echo "make build: build the project"
+    @echo "make run FILE=<file_path>: run the project with specified file"
+    @echo "make clean: clean up the binary"
+    @echo "make test: run tests"
+    @echo "make help: show this message"
+
+.PHONY: all build run clean test help