Using Goroutines and Channels for High-Throughput S3 Object Hashing

Problem

In my previous post, I detailed a Rust solution to hash a large number of objects in an S3 bucket. The challenge was to process a massive dataset (12 TB) efficiently, as sequential processing would be prohibitively time-consuming. The Rust implementation leveraged async capabilities to stream and hash multiple objects concurrently, achieving impressive throughput of over 3 GB/s (within an AWS VPC in the same region as the s3 bucket).

Given Go’s reputation for simplicity and powerful built-in concurrency primitives, I was curious to see how it would stack up against Rust in this scenario. This post will focus on implementing the same solution in Go, highlighting the differences in approach and comparing the results.

Approach

I rewrote the Rust logic in Go, maintaining the same overall structure:

1. An input task to stream S3 keys from a file
2. Multiple worker tasks to fetch S3 objects and compute hashes
3. An output task to write results to a file

The key difference lies in how concurrency is handled. Instead of Rust’s async/await and Tokio, we’ll use Go’s goroutines and channels. Here’s a high-level overview of the components:

1. Input goroutine: Reads keys from the input file and sends them to a channel
2. Worker goroutines: Receive keys, fetch S3 objects, compute hashes, and send results to an output channel
3. Output goroutine: Receives results and writes them to the output file

The complete source code for this Go implementation is available on GitHub: go-s3hash

Let’s dive into the implementation details.

Implementation

Main Structure

The main Run function sets up the goroutines and channels:

func Run(appName, bucket, keysFile string, numThreads uint16) {
    // ... (logger setup omitted for brevity)

    chanKeys := make(chan string, numThreads)
    chanOutput := make(chan Output)
    statsChan := make(chan Stats, 1)

    // Spawn goroutines
    go writeOutputFile(outputFile, chanOutput, statsChan, logChan)

    go streamKeys(keysFile, chanKeys, logChan)

    for i := 0; i < int(numThreads); i++ {
        go fetchAndHash(bucket, chanKeys, chanOutput, logChan)
    }

    // ... (wait for goroutines to finish and log results)
}

This structure is similar to our Rust implementation, but uses goroutines and channels instead of async tasks.

Input Stream

The streamKeys function reads S3 keys from a file and sends them to a channel:

func streamKeys(keysFile string, chanKeys chan<- string, logChan chan<- LogMsg) {
    defer close(chanKeys)
    keysFh, err := os.Open(keysFile)
    if err != nil {
        logChan <- LogMsg{
            msg:   fmt.Sprintf("unable to open keys file: %v", err.Error()),
            level: slog.LevelError,
        }
    }
    defer keysFh.Close()

    scanner := bufio.NewScanner(keysFh)
    for scanner.Scan() {
        key := strings.TrimSpace(scanner.Text())
        chanKeys <- key
    }

    // ... (error handling omitted for brevity)
}

Worker Tasks

The fetchAndHash function is our worker, fetching S3 objects and computing hashes:

func fetchAndHash(s3bucket string, chanKeys <-chan string, chanOutput chan<- Output, logChan chan<- LogMsg) {
    s3agent := NewS3Agent(s3bucket)

    for key := range chanKeys {
        bytes, err := s3agent.GetObjectBytes(key)
        if err != nil {
            logChan <- LogMsg{
                msg:   fmt.Sprintf("unable to fetch s3 object '%v', err: %v", key, err.Error()),
                level: slog.LevelError,
            }
        } else {
            hash := sha256Hash(bytes)
            output := Output{key: key, hash: hash, byteLen: uint(len(bytes))}
            chanOutput <- output
        }
    }
}

Output Writer

The writeOutputFile function receives results and writes them to a file:

func writeOutputFile(outputFilename string, chanOutput <-chan Output, chanStats chan<- Stats, logChan chan<- LogMsg) {
    outFh, err := os.OpenFile(outputFilename, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
    if err != nil {
        logChan <- LogMsg{
            msg:   fmt.Sprintf("unable to open output file: %v", err.Error()),
            level: slog.LevelError,
        }
    }
    defer outFh.Close()

    totalBytesHashed := uint(0)
    countHashesWritten := uint(0)
    for output := range chanOutput {
        outStr := fmt.Sprintf("%s,\\\\x%x\n", output.key, output.hash)
        _, err := outFh.WriteString(outStr)
        if err != nil {
            logChan <- LogMsg{
                msg:   fmt.Sprintf("error writing to output file: %v", err.Error()),
                level: slog.LevelError,
            }
        } else {
            countHashesWritten += 1
            totalBytesHashed += output.byteLen
        }
    }

    chanStats <- Stats{countHashesWritten, totalBytesHashed}
}

S3 Interaction

We use the AWS SDK for Go to interact with S3:

type s3agent struct {
    client *s3.Client
    bucket string
}

func NewS3Agent(bucket string) *s3agent {
    cfg, err := config.LoadDefaultConfig(context.TODO())
    if err != nil {
        log.Fatal(err)
    }

    return &s3agent{
        client: s3.NewFromConfig(cfg),
        bucket: bucket,
    }
}

func (s *s3agent) GetObjectBytes(key string) ([]byte, error) {
    input := &s3.GetObjectInput{
        Bucket: aws.String(s.bucket),
        Key:    aws.String(key),
    }

    result, err := s.client.GetObject(context.Background(), input)
    if err != nil {
        return nil, err
    }

    defer result.Body.Close()

    return io.ReadAll(result.Body)
}

Results

I ran the Go implementation on the same test dataset and EC2 instances as the Rust version. Here are the key findings:

1. Performance: The Go version achieved a peak processing rate of 2.98 GB/second on a c7g.metal instance, which is remarkably close to the Rust version’s 3.125 GB/second.
2. Optimal thread count: Similar to Rust, the best performance was achieved at 8x the number of available cores.
3. Resource utilization: Go’s garbage collector resulted in slightly higher memory usage compared to Rust, but CPU utilization was comparable.
4. Development experience: Writing the Go version felt noticeably faster and required less boilerplate code, especially around error handling and concurrency (having just done it in Rust probably affected this timeline a bit, as I had the design hammered out).

Here’s a comparison of processing times for our 46 GB test dataset:

While Rust maintains a slight edge in raw performance, the difference is minimal for most practical purposes.

Reflections

Implementing the S3 hasher in Go was an impressively smooth experience. Go’s concurrency model with goroutines and channels made it straightforward to parallelize our workload. The simplicity of Go’s error handling and the built-in garbage collector allowed for rapid development and testing.

Key takeaways:

1. Go’s performance is impressive, nearly matching Rust in this I/O-bound task.
2. Go’s simplicity and built-in concurrency features make it an excellent choice for tasks requiring high concurrency.
3. The difference in performance between Go and Rust for this specific task is negligible in most real-world scenarios.

Ultimately, both Go and Rust are excellent choices for this kind of high-throughput, concurrent processing task. The decision between them might come down to team expertise, ecosystem compatibility, or specific performance requirements.

For our use case of processing a 12 TB dataset, the Go implementation would complete the task in about 67 minutes, just 3 minutes longer than the Rust version. Given the ease of development and maintenance, Go presents a compelling option for many teams tackling similar problems, especially given the one-off nature of this task. Use whichever language you’re most comfortable with.