DEV Community: Manish

Learning, Experimenting - Concurrency in Go - Part 2

Manish — Sun, 07 Jun 2026 01:51:10 +0000

Refresher - I'm building a distributed chunked filestore in Go, and I setup a post for Part 1 here. That part dealt with uploading a file - this post is about downloads.

Setup

Requirements

User hits our endpoint with the filename/fileid
We use this fileid to get a list of chunks
Our retrieve mechanism only depends on this list of chunks
We want to be able to retrieve the associated chunks in parallel
We want correctness (duh, but I'll explain why I explicitly stated this)

The Test Suite

First up, a simple test suite:

type fakeWriter struct {
    hdr http.Header
}

func (w *fakeWriter) Header() http.Header {
    if w.hdr == nil {
        w.hdr = make(http.Header)
    }
    return w.hdr
}

func (w *fakeWriter) WriteHeader(int) {}
func (w *fakeWriter) Write(p []byte) (int, error) {
    return len(p), nil
}

func BenchmarkRetrieveChunk(b *testing.B) {
    dir := b.TempDir()
    const numChunks = 12
    const chunkBytes = 64 << 10

    chunks := make([]Chunk, numChunks)
    for i:= range(numChunks) {
        body := bytes.Repeat([]byte{byte('A'+i)}, chunkBytes)
        path := filepath.Join(dir, fmt.Sprintf("%v.%v", "chunk", i))
        if err := os.WriteFile(path, body, 0o600); err != nil {
            b.Fatal(fmt.Errorf("Error while setting up benchmark - %v", err))
        }
        chunks[i] = Chunk{ChunkIndex: i, ChunkSize: chunkBytes, StoreLoc: path}
    }
    s := &Server{
    config: Config{
        Store: b.TempDir(), 
        Chunksize: 10,
    }, 
    concurrencyConfig: ConcurrencyConfig{
        MaxUploadConcurrency: 5, 
        MaxDownloadConcurrencyPerFile: 1,
    }}

    b.SetBytes(int64(numChunks)*chunkBytes)
    b.ResetTimer()
    for b.Loop() {
        s.retrieveChunks(&fakeWriter{}, "chunk", &chunks)
    }
}

A bit more involved than our storing mechanism.

Why define a fakeWriter? The files are usually large, which means we'd have to stream the chunks as we read them. In such a case, as soon as a chunk is read, we want to immediately write it to the upstream http writer while the other chunks download in the background in parallel.

The Concurrent Chunk retrieval

With the concurrent file upload, we are able to reuse a global chan. This is because once the chunk info is fed, it does not matter in which order the goroutines complete - they just need to get stored anyway. However, with our concurrent file download, we want to have them fetched in order. For example, for a file split into 5 chunks, we have to stream the file back in order of 1..5. However, if chunk 5 finishes before chunk 4, we have to wait until chunk 4 is finished before streaming it back.

But that just means we have to fetch them in order. Why would it mean we can't reuse a global chan? We could, but it would be a bit more complex than what we want to explore at the moment. With a global chan, we also have to handle ordering in such a way that it doesn't overlap across multiple file download requests i.e. separate requests must be isolated from their ordering. I will be exploring ways to do it eventually, but for now we shall move on.

Implementing ordering

How do we go about implementing ordering? We know that chunks are fetched from the bufpool. We can start by doing spawning them in order like this:

queue := make(chan []byte, s.concurrencyConfig.MaxDownloadConcurrencyPerFile)
g.Go(func() error {
    defer close(queue)
    for i := 0; i < numChunks; i++ {
        g.Go(func() error {
            return s.readChunk(ctx, queue, (*chunkInfo)[i]) 
        })
    }
    return nil
})
w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%v", filename))
for res := range(queue) {
    if _, err := w.Write(res); err != nil {
        return
    }
}
if err := g.Wait(); err != nil{
    http.Error(w, "Stream disconnected", http.StatusInternalServerError)
    return
}

This would spawn the goroutine for chunk i before i+1. But that is simply ordered spawning, not ordered completion. A chan simply says - give me the next value that was sent. In that case, what happens if chunk i+1 finishes before chunk i? When looping through the chan to retrieve data, we get chunk i+1 before chunk i, which we then write back the upstream client - which violates correctness.

So what we want is to retrieve chunks in parallel but write them back sequentially. One naive way to do so would be something like this - since we know the number of chunks in advance, we can have a var readmask bool of size numchunks and block our write loop on the ith element until chunk i completes:

queue := make(chan []byte, s.concurrencyConfig.MaxDownloadConcurrencyPerFile)
readMask := make([]bool, numChunks)
g.Go(func() error {
    defer close(queue)
    for i := 0; i < numChunks; i++ {
        g.Go(func() error {
                        defer func() { readMask[i] = true }
            return s.readChunk(ctx, queue, (*chunkInfo)[i]) 
        })
    }
    return nil
})
w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%v", filename))
for res := range(queue) {
        while !readMask[i] {}
    if _, err := w.Write(res); err != nil {
        return
    }
}
if err := g.Wait(); err != nil{
    http.Error(w, "Stream disconnected", http.StatusInternalServerError)
    return
}

We have a readmask defined that waits and runs an inf loop until the ith chunk is read. Seems simple, right? But this leads to a new problem: we are modifying the readmask within multiple goroutines, which means we have to wrap a mutex to prevent simultaneous data access i.e. data race conditions.

What we could do to solve this problem, is introduce an array of chans, like so:

queue := make([]chan []byte, numChunks)
g.Go(func() error {
...
    for i := 0; i < numChunks; i++ {
        g.Go(func() error {
            return s.readChunk(ctx, queue[i], (*chunkInfo)[i]) 
...
})
for i := range queue {
        data, ok := <- queue[i]
        // handle data and ok
    if _, err := w.Write(data); err != nil {
        return
    }
}

queue[i] is closed at readChunk, and so, our <-queue[i] blocks until this closure. How does this our mutex problem, you ask? Well, we are not mutating the slice after setup. Every subsequent operation is read only. queue[i] simply points to the chan at that index, which was already initialized. The readChunk op just writes to the same underlying chan, not the queue[i] object - which means queue itself doesn't need a mutex.

This solves our concurrency problem. But what about boundedness? We could encounter a situation where while the 1st chunk is being streamed, all 100 chunks are read and sit in memory - because nothing is stopping us from limiting the number of goroutines we spawn. We could do something like holding a sem with capacity and releasing it. I chose a different approach - make our queue a chan of chans!

queue := make(chan chan []byte, s.concurrencyConfig.MaxDownloadConcurrencyPerFile)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
    defer close(queue)
    for i := 0; i < numChunks; i++ {
        res := make(chan []byte, 1)
        select {
            case queue <- res:
            case <- ctx.Done():
                return ctx.Err()
        }
        g.Go(func() error { 
            defer close(res)
            return s.readChunk(ctx, res, (*chunkInfo)[i]) 
        })
    }
    return nil
})
w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%v", filename))
for res := range(queue) {
    data, ok := <- res
    if !ok {
        break
    }
    if _, err := w.Write(data); err != nil {
        return
    }
}

Now we have everything we want! Concurrent workers and correctness in writes.

Benchmarking

I tried running the benchmark as is, and ran into similar results (N = number of concurrent workers):

N=1:    246297966 ns/op 510.88 MB/s 125860644 B/op  188 allocs/op
N=5:    253591228 ns/op 496.19 MB/s 125853408 B/op  201 allocs/op

Interesting, we would expect concurrency to speed things up. But its just slowing things down. The main culprit - page cache.

Looking at my benchmark - I am writing my file to tempDir and reading it immediately. The read is served fresh from the page cache, which means it is not read-bound at all. We see no benefits from concurrency since we don't simulate cache misses and disk fetch.

What I tried next is to introduce an artifical lag in my read routine, like so:

func (s *Server) readChunk(ctx context.Context, cache chan[]byte, chunk Chunk) error {
    time.Sleep(50*time.Millisecond)
...

This delays every goroutine by 50ms - cache misses are in ms range, so this should be a decent approximation for cold fetches.

And so, the results - read with 50ms delay in readChunk, 12 chunks × 10MB each:

N=1:    464712476 ns/op 270.77 MB/s 125873928 B/op  231 allocs/op
N=5:    306165247 ns/op 410.98 MB/s 125854244 B/op  217 allocs/op
N=10:   314576026 ns/op 400.00 MB/s 125857106 B/op  232 allocs/op

This tells us where concurrency wins out - when our reads from disk are slow and data encounters a cache-miss, which means they are fetched cold from our disk.

What next?

I don't know, honestly. On the concurrency front, I think I've explored a decent amount. I can work on making this a working product but that requires auth, ui, storing file info etc. I am working on them, but they are more trivial than the above experiments. If I do learn something new, I will be sure to post them. So, until next time!

Learning, Experimenting - Concurrency in Go

Manish — Sun, 31 May 2026 04:02:09 +0000

Torn between writing a toy project I'd use vs learning Go, I decided to try my hand at writing a simple distributed chunked filestore. Here are a few things I learned and how I moved from a naive implementation to a parallelized implementation.

There are a lot of filestore papers already that involve much more sophisticated distribution/replication strategies, and I will be building upon my project as I continue - I've got lots to learn. Anyhow, let's get started!

The Test Suite

First up, a simple test suite:

type fakeReader struct {
    remaining int
}

func (r *fakeReader) Read(p []byte) (int, error) {
    if r.remaining <= 0 {
        return 0, io.EOF
    }
    n := len(p)
    if n > r.remaining {
        n = r.remaining
    }
    for i := 0; i < n; i++ {
        p[i] = 'x'
    }
    r.remaining -= n
    return n, nil
}

func BenchmarkStoreChunk(b *testing.B) {
    const size = 1000<<20
    s := &Server{config: Config{Store: b.TempDir(), Chunksize: 10, MaxConcurrency: 1}}

    b.ResetTimer()
    b.SetBytes(int64(size))

    for b.Loop(){
        r := &fakeReader{remaining: size}
        if _, err := s.storeChunks(r, "bench"); err != nil {
            b.Fatal(err)
        }
    }
}

Pretty straightforward. Initialize a 1000 MB byte array, call my main storeChunks function to handle the write to disk. Note that Chunksize (in MB) and MaxConcurrency (num of concurrent workers) is a configurable parameter at server initialization (for the parallelized implementation). Running it is simply:

go test -bench=. -benchmem -count=3

The option count parameter runs it 3 times in this example.

Note - I will be skipping the server handling and the input read part, and will be going over the core storage logic only.

1. The Naive Write

numChunks := 0
chunkSize := 5 << 20
buf := make([]byte, chunkSize)
for {
    part, err := reader.NextPart()
    if err == io.EOF {
        break
    }
    if err != nil {
        // handle
    }
    for i:=0;;i++{
        n, err := io.ReadFull(part, buf)
        if err != io.EOF && err != io.ErrUnexpectedEOF && err != nil {
            // handle
        }
        if n == 0 {
            break
        }
        chunkPath := fmt.Sprintf("%v/%v.%v", s.config.Store, part.FileName(), numChunks)
        chunk, createErr := os.Create(chunkPath)
        if createErr != nil {
            // handle
        }
        _, writeErr := chunk.Write(buf[:n])
        if writeErr != nil {
            // handle
        }
        chunk.Close()
        numChunks += 1
        if err == io.EOF || err == io.ErrUnexpectedEOF {
            break
        }
    }
    // handle
}

Simple enough. Read from the incoming part in full and write into a buf of size 5 MB. Each buf chunk is then sequentially written to disk. Taking a look at the benchmark:

3   339820142 ns/op 3085.68 MB/s       10514901 B/op     608 allocs/op

Not bad. 3 runs, 3GB/s, 10.5MB/op, 600 allocs/op. We will come back to these numbers later.

2. The Naive Concurrent rewrite

The main issue with the naive write is that concurrency is used nowhere. The incoming part is read in 5MB chunks and written to disk sequentially. This means there is a lot of idle CPU just waiting for the write to finish before reading the next 5MB chunk. Situations like these call for concurrency.

When spawning concurrent workers, it is inevitable that they will expect to share a common variable. This introduces the need for mutexes to prevent corruption/race-conditions. And so:

var wg sync.WaitGroup
var mu sync.Mutex
var firstErr error
numChunks := 0
chunkSize := s.config.Chunksize << 20

for {
    buf := make([]byte, chunkSize)          // Comment 1 - fresh allocation every chunk
    n, err := io.ReadFull(part, buf)
    if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
        // handle
    }
    if n == 0 {
        break
    }

    chunkPath := fmt.Sprintf("%v/%v.%v", s.config.Store, fileName, numChunks)
    b, nn, c := buf, n, chunkPath
    wg.Add(1)
    go func() {                              // Comment 2 - unbounded: one goroutine per chunk, no limit
        defer wg.Done()
        chunk, err := os.Create(c)
        if err != nil {
            mu.Lock()
            if firstErr == nil { firstErr = err }
            mu.Unlock()
            return
        }
        defer chunk.Close()
        if _, err := chunk.Write(b[:nn]); err != nil {
            mu.Lock()
            if firstErr == nil { firstErr = err }
            mu.Unlock()
        }
    }()
    numChunks++
    if err == io.EOF || err == io.ErrUnexpectedEOF {
        break
    }
}

wg.Wait()
if firstErr != nil {
    // handle
}
return numChunks, nil

Okay. Now we have a concurrent implementation that spawns multiple workers to write in parallel. This should improve our benchmark - let's test it:

3       340605556 ns/op 3078.56 MB/s    1059307005 B/op 1028 allocs/op

So - 3 runs, 3GB/s, 1GB/op, 1028 allocs/op.
Wait - 1 GB/op?

I've let the goroutines be created unbounded, which means that every goroutine uses a freshly initialized buf. 100 goroutines with 10MB chunks -> 1 GB. As pointed out in the Comments #1 and #2.

This is a problem. I've regressed - no improvements over my disk writes while eating up my RAM, inviting pesky OOM-Killed issues as well. We can do better.

3. The Better Concurrent rewrite

With the current concurrent implementation, we would need a way to limit the number of goroutines that can be spawned. One way is to simply initialize n buffers, spawn n goroutines at once, and wait for them to complete before moving on to the next iteration(s). While this approach would cap RAM for a single file, it wouldn't work effectively for highly parallelized workloads - the number of buffers is simply number of incoming requests * n. We could also implement a global cap, which would eventually be useful to handle multiple incoming requests.

The inherent problem here is that memory is held on to for far longer than needed. At n=50, we wait for all 50 goroutines to finish. But by the time we start n=25, the goroutines that were spawned for n=1..10 might've completed. A better way would be to have some sort of a resizable record that can provide already processed buffers for reuse. And hence, channels. With that in mind:

var wg sync.WaitGroup
var mu sync.Mutex
var Err error = nil
numChunks := 0
chunkSize := s.config.Chunksize << 20
bufPool := make(chan []byte, s.config.MaxConcurrency)
for range s.config.MaxConcurrency {
    bufPool <- make([]byte, chunkSize)
}
for {
    buf := <-bufPool 
    n, err := io.ReadFull(part, buf)
    if err != io.EOF && err != io.ErrUnexpectedEOF && err != nil {
        // handle
    }
    if n == 0 {
        break
    }
    chunkPath := fmt.Sprintf("%v/%v.%v", s.config.Store, fileName, numChunks)
    wg.Add(1)
    go func(chunkPath string, buf []byte, n int) {
        defer wg.Done()
        defer func() { bufPool <- buf }() 
        chunk, err := os.Create(chunkPath)
        if err != nil {
            mu.Lock()
            if Err == nil { Err = err }
            mu.Unlock()
            return
        }
        defer chunk.Close()
        _, err = chunk.Write(buf[:n])
        if err != nil {
            mu.Lock()
            if Err == nil { Err = err }
            mu.Unlock()
            return
        }
    }(chunkPath, buf, n)
    numChunks += 1

    if err == io.EOF || err == io.ErrUnexpectedEOF {
        fmt.Println("Reached end of input!")
        break
    }
}
wg.Wait()
if Err != nil {
    return numChunks, Err
}
return numChunks, nil

Note that we have an additional configurable parameter now - the number of concurrent workers to spawn. Here's a few benchmarks for different values of that parameter:

N=1:   3    336508657 ns/op 3116.04 MB/s       10525466 B/op     816 allocs/op
N=4:   5    217452331 ns/op 4822.10 MB/s       42000254 B/op     858 allocs/op
N=6:   5    215637466 ns/op 4862.68 MB/s       62963259 B/op     842 allocs/op

At 1 worker - basically sequential - we see 3.1GB/s with 10.5MB/op
At 4 workers, we see 4.8GB/s with 42MB/op.
At 6 workers, we see 4.8GB/s with 63MB/op.

The number of bytes per op is simply number of workers * chunk size, as you can see. Which gives us predictable bound to stay within.

The N=1 case is an interesting datapoint here. It is just as fast as the sequential write and naive concurrent rewrite (3.1 GB/s) while saving 100x the memory (10.5MB/op vs 1GB/op) as compared to the naive write-up.

Simulating environments

The measurements above were taken from a 64GB RAM machine. Target workloads might be different, such as kubernetes pods with a limits.memory of 2GB, or a memory-capped systemd service. K8s pods use cGroups anyway, so the aforementioned scenarios are identical. Luckily, there is an easy way to simulate this cGroup behaviour:

systemd-run --scope -p MemoryMax=$MEM go test -bench=BenchmarkStoreChunk -benchmem

Simulating for $MEM=2G and $MEM=512M gives:

Version	2GB cap (MB/s)	512M cap (MB/s)
sequential	2991	1571
naive	2869	1858
N=1	3023	1627
N=4	4762	3382
N=6	4889	3294

Multiple factors come into play with such restrictions - Go Runtime Memory limits, Garbage collection parameters etc.

What next?

This was just a simple experiment I'm documenting to come back to when I need to refresh concurrency. As for the project, I hope to integrate it with my homelab and start using it properly. I haven't even written a download handler yet, so I suppose that would be next. And a UI, a store to maintain records of file uploads/locations, and then run it as a service/container, expose to relevant devices over LAN or TailScale / Reverse-Proxy - there's so much left to do, and I'm sure I'll learn more as I progress. If you see another blog post in a few months, know that I have succeeded. If not, I've probably procrastinated on it. Anyhow, until next time!