Let's Write a Threaded File Compression Tool with Memory Control
This weekend I was diving deep into concurrent programming. To learn by building, I decided to write a file compressor that can compress files concurrently while also limiting memory usage. I have implemented this in Go, but these concepts can be implemented in any language.
What I’ve Developed?
I designed a tool that reads files in chunks and limits the overall memory used by introducing a memory-aware semaphore system. It also uses a pool of reusable buffers to make things even faster.
Each file is compressed using gzip, and the work is handled concurrently by worker goroutines. The goal is to improve throughput while keeping memory usage bounded.
Let’s Write Some Code
So I started by defining a simple configuration struct called CompressionConfig.
1type CompressionConfig struct {
2 NumWorkers int
3 MaxMemoryUsage int // MegaBytes
4 CompressionLevel int
5 ChunkSize int // MegaBytes
6}
Let Me Break It Down:
- NumWorkers — This defines how many worker goroutines will run. More workers can improve throughput when there is enough CPU and I/O capacity, but too many workers can also add contention.
- MaxMemoryUsage — This is the upper memory limit for the compression process. It ensures we don’t overload the system while handling large files.
- CompressionLevel — Standard gzip compression level (from 1 to 9).
- ChunkSize — Instead of loading entire files into memory, I read them in chunks. This defines how big each chunk should be.
1type Job struct {
2 InputPath string
3 OutputPath string
4 FileSize int64
5 Ctx context.Context
6}
Once I have the configuration ready, the next step was to define what a compression job actually looks like.
Here’s What Each Field Does:
- InputPath — The path of the file that needs to be compressed.
- OutputPath — Where the compressed file should be saved.
- FileSize — This can be useful for tracking and logging, although I didn’t use it heavily in this version.
- Ctx — A context.Context that helps with cancelling jobs cleanly (for example, if the user wants to stop the process midway).
1type CompressionResult struct {
2 Job *Job
3 Error error
4 OriginalSize int64
5 CompressedSize int64
6 Duration time.Duration
7 OriginalChecksum string
8 CompressedChecksum string
9}
After compressing each file, I wanted to collect some useful information about the outcome — so I created a CompressionResult struct:
- Job — A pointer back to the original job, so I know which file this result belongs to.
- Error — If something goes wrong during compression, this will capture it.
- OriginalSize — Size of the file before compression.
- CompressedSize — Size after compression — helps measure how effective the compression was.
- Duration — How long the compression took.
- OriginalChecksum / CompressedChecksum — (Not implemented yet in my version, but placeholders are ready!) These would help verify integrity — making sure the content isn’t corrupted during the process.
1type ParallelCompressor struct {
2 config *CompressionConfig
3 jobQueue chan *Job
4 resultQueue chan *CompressionResult
5 workerWG sync.WaitGroup
6 resultWG sync.WaitGroup
7 ctx context.Context
8 cancel context.CancelFunc
9 bufferPool sync.Pool
10 semaphore chan struct{}
11}
Now comes the heart of the whole system — the ParallelCompressor.
This struct holds everything needed to manage workers, jobs, memory, and results. Here’s how it looks:
- config — Holds the compression settings we defined earlier (like memory limit, chunk size, etc.).
- jobQueue — A channel where all compression jobs are sent. Workers pick jobs from here.
- resultQueue — After processing, workers push results here to be handled separately.
- workerWG / resultWG — WaitGroups help us wait for all workers and result processors to finish cleanly.
- ctx / cancel — Context used to cancel the whole operation (for example, on error or manual stop).
- bufferPool — A pool of byte slices reused across jobs to avoid allocating memory repeatedly — very useful for performance.
- semaphore — It limits how many chunk-sized buffers can be in use at once, using the configured memory budget as a simple accounting model.
1func NewParallelCompressor(config *CompressionConfig) *ParallelCompressor {
2 if config.NumWorkers == 0 {
3 config.NumWorkers = runtime.NumCPU()
4 }
5
6 ctx, cancel := context.WithCancel(context.Background())
7
8 return &ParallelCompressor{
9 config: config,
10 jobQueue: make(chan *Job, 100),
11 resultQueue: make(chan *CompressionResult, 100),
12 ctx: ctx,
13 cancel: cancel,
14 bufferPool: sync.Pool{
15 New: func() any {
16 buf := make([]byte, config.ChunkSize)
17 return &buf
18 },
19 },
20 semaphore: make(chan struct{}, config.MaxMemoryUsage/config.ChunkSize),
21 }
22}
To wire everything together, I created a constructor function called NewParallelCompressor. This sets up all the internals for running parallel compression with memory limits.
If the user doesn’t set a number of workers, I default it to the number of CPU cores — which usually works well for concurrent workloads.
Next, I create a cancellable context. This gives me a clean way to cancel all running jobs in case of an error or user interruption.
The job and result queues are buffered channels to prevent blocking when producing or consuming tasks.
The bufferPool here is super useful. Instead of allocating a new byte slice every time I read a chunk, I reuse buffers from a pool — which reduces memory pressure and garbage collection overhead.
Finally, the semaphore bounds how many chunk-sized buffers can be in use at the same time. If MaxMemoryUsage is 32 MB and ChunkSize is 2 MB, the semaphore can allow at most 16 chunk-sized buffers under that simplified accounting model. This is not a hard process-memory or RSS limit because the process has other allocations as well.
1func (pc *ParallelCompressor) Start() {
2 for i := 0; i < pc.config.NumWorkers; i++ {
3 pc.workerWG.Add(1)
4 go pc.worker()
5 }
6
7 pc.resultWG.Add(1)
8 go pc.processResult()
9}
Once the ParallelCompressor is initialized, I needed a way to kick off the actual work — spinning up workers and setting up result processing.
That’s what the Start() method is for:
- I launch multiple worker goroutines based on the number of CPUs (or user-defined workers). Each worker picks up jobs from the jobQueue and compresses them.
- pc.workerWG.Add(1) increments the WaitGroup counter, so we can later wait for all workers to finish before exiting.
- Then I start one goroutine to handle results — basically, to consume from the resultQueue and print any errors or logs.
1func (pc *ParallelCompressor) worker() {
2 defer pc.workerWG.Done()
3
4 for {
5 select {
6 case job := <-pc.jobQueue:
7 if job == nil {
8 return
9 }
10
11 result := pc.processJob(job)
12
13 select {
14 case pc.resultQueue <- result:
15 case <-pc.ctx.Done():
16 return
17 }
18
19 case <-pc.ctx.Done():
20 return
21 }
22 }
23}
Let’s talk about the worker — the actual unit that performs the compression. Each worker runs in its own goroutine.
First, I make sure that when a worker exits, it signals the WaitGroup so we know it’s done.
The worker continuously listens for jobs from the jobQueue. If it receives nil, that means we’re done — time to exit.
Once it gets a job, it processes it using processJob(), which handles reading the file, compressing it, and storing the result.
The result (success or error) is then pushed to the resultQueue. If the context is cancelled before that, the worker exits early.
The outer select block ensures the worker is also listening to the cancellation signal. This way, if something goes wrong, all workers can exit cleanly without finishing the remaining tasks.
1func (pc *ParallelCompressor) processJob(job *Job) *CompressionResult {
2 result := &CompressionResult{Job: job}
3
4 startTime := time.Now()
5 defer func() { result.Duration = time.Since(startTime) }()
6
7 inputFile, err := os.Open(job.InputPath)
8 if err != nil {
9 return setError(result, err)
10 }
11 defer inputFile.Close()
12
13 result.OriginalSize = getFileSize(inputFile)
14
15 outputFile, err := os.Create(job.OutputPath)
16 if err != nil {
17 return setError(result, err)
18 }
19 defer outputFile.Close()
20
21 if err := pc.compress(inputFile, outputFile); err != nil {
22 return setError(result, err)
23 }
24
25 result.CompressedSize = getFileSize(outputFile)
26 return result
27}
The actual compression logic for each file lives inside the processJob() function. Each worker calls this when it picks up a job.
I start by creating a new CompressionResult tied to the job we’re processing.
I track how long the compression takes. Once the function finishes, the duration is recorded.
Next, I open the input file. If it fails, I capture the error in the result and return early. I store the original file size so I can compare it later with the compressed version.
Now, I prepare the destination file where the compressed data will go and send both files to compress method for compression.
1func (pc *ParallelCompressor) compress(inputFile io.Reader, outputFile io.Writer) error {
2 select {
3 case pc.semaphore <- struct{}{}:
4 case <-pc.ctx.Done():
5 return fmt.Errorf("context cancelled while waiting for memory slot")
6 }
7
8 defer func() {
9 <-pc.semaphore
10 }()
11
12 gzipWriter, err := gzip.NewWriterLevel(outputFile, pc.config.CompressionLevel)
13 if err != nil {
14 return err
15 }
16 defer gzipWriter.Close()
17
18 bufferPtr := pc.bufferPool.Get().(*[]byte)
19 buffer := *bufferPtr
20 defer pc.bufferPool.Put(bufferPtr)
21
22 for {
23 n, err := inputFile.Read(buffer)
24 if n > 0 {
25 if _, writeErr := gzipWriter.Write(buffer[:n]); writeErr != nil {
26 return writeErr
27 }
28 }
29
30 if err == io.EOF {
31 break
32 }
33
34 if err != nil {
35 return err
36 }
37 }
38
39 return err
40}
41
42func setError(r *CompressionResult, err error) *CompressionResult {
43 r.Error = err
44 return r
45}
46
47func getFileSize(file *os.File) int64 {
48 stat, err := file.Stat()
49 if err != nil {
50 return 0
51 }
52 return stat.Size()
53}
The actual compression logic happens inside the compress() function. This is where chunks are read from the input file, compressed using gzip, and written to the output file — all while respecting memory limits.
Before I start compression, I acquire a memory slot from the semaphore. This ensures that only a safe number of jobs are allowed to allocate buffer memory at a time, based on the total memory cap.
If the context is cancelled (maybe due to error or shutdown), the function exits early.
Once the job is done, I release the slot back so others can use it. Simple and effective memory control.
I use Go’s built-in gzip writer to handle the compression. The compression level is customizable through the config.
I pull a buffer from the pool instead of allocating a new slice. This avoids extra allocations and speeds things up. After the work is done, the buffer is returned to the pool.
I read the input file chunk-by-chunk and write the compressed data to the output. The loop continues until we hit EOF (end of file). If any read/write error occurs, we return early.
1func (pc *ParallelCompressor) processResult() {
2 defer pc.resultWG.Done()
3
4 for {
5 select {
6 case result := <-pc.resultQueue:
7 if result == nil {
8 return
9 }
10
11 if result.Error != nil {
12 fmt.Printf("Result Error: %s\n", result.Error.Error())
13 }
14
15 case <-pc.ctx.Done():
16 return
17 }
18 }
19}
Once the files are compressed, I needed a way to handle the results — whether successful or failed. That’s what the processResult() function is built for.
As with the workers, I use a WaitGroup to track when the result processor finishes. This defer ensures it’s marked done at the end.
The result processor runs in a loop, constantly listening on the resultQueue. If it receives nil, it knows the queue has been closed and it’s time to exit.
Right now, I only print errors — but this is the perfect place to extend things later. I could log stats, show compression ratios, update a UI, or store results in a database.
1func printMemUsage(label string) {
2 var m runtime.MemStats
3 runtime.ReadMemStats(&m)
4
5 fmt.Printf("\n--- %s ---\n", label)
6 fmt.Printf("Alloc = %v MiB", m.Alloc/1024/1024)
7 fmt.Printf("\tTotalAlloc = %v MiB", m.TotalAlloc/1024/1024)
8 fmt.Printf("\tSys = %v MiB", m.Sys/1024/1024)
9
10 fmt.Println()
11 fmt.Println()
12}
Here is a simple function that prints how much memory our program used for execution.
1func main() {
2 printMemUsage("Before")
3
4 config := &CompressionConfig{
5 NumWorkers: 8,
6 MaxMemoryUsage: 32 * 1024 * 1024,
7 CompressionLevel: 9,
8 ChunkSize: 2 * 1024 * 1024,
9 }
10
11 compressor := NewParallelCompressor(config)
12
13 compressor.Start()
14
15 files := []string{}
16
17 for i := 1; i <= 13; i++ {
18 files = append(files, fmt.Sprintf("testdata/book%d.pdf", i))
19 }
20
21 now := time.Now()
22
23 for _, inputPath := range files {
24 outputPath := inputPath + "_compressed"
25
26 job := &Job{
27 InputPath: inputPath,
28 OutputPath: outputPath,
29 Ctx: context.Background(),
30 }
31
32 compressor.jobQueue <- job
33 }
34
35 close(compressor.jobQueue)
36 compressor.workerWG.Wait()
37
38 close(compressor.resultQueue)
39 compressor.resultWG.Wait()
40
41 t := time.Since(now)
42
43 fmt.Printf("Time Taken: %d milliseconds\n\n", t.Milliseconds())
44
45 printMemUsage("After")
46
47}
Finally, here’s the main() function — where everything comes together and the tool actually runs.
I start by printing the memory usage before compression. This helps me compare how much memory is used during the process.
Here, I define the configuration:
1 * 8 workers
2 * 32 MB total memory limit
3 * gzip level 9 for max compression
4 * reading files in 2MB chunks
I initialize and start the compressor — this spins up the workers and the result processor.
Next, I prepare a list of files to compress. In my case, they’re just test PDFs named book1.pdf to book13.pdf.
I start a timer to track how long the whole compression process takes.
For each file, I create a Job and push it to the job queue. The workers will pick them up automatically and process them.
Once all jobs are sent, I close the job queue and wait for all workers to finish.
Same thing for the result queue — once results are all pushed, I close the queue and wait for the result processor to finish.
Now I print how long the total compression process took.
What Did I Observe?
After building everything, I ran the tool using:
1go run main.go
And here’s the output I got:
Let’s break it down:
- Before starting, the memory allocation was practically zero — expected since nothing heavy had kicked off yet.
- After compression, the memory allocation was 28 MiB, with total allocated memory reaching 30 MiB.
- The system (Go runtime + buffers + allocations) used 35 MiB in total.
This was close to the MaxMemoryUsage target I had set (32 MB); Go runtime overhead and other allocations mean the process memory figures are not expected to match the limit exactly. The memory cap + buffer reuse through sync.Pool+ semaphore control all worked together as intended.
The compression of 13 PDF files finished in just about 2.2 seconds, which is pretty fast considering everything was processed concurrently and memory was strictly controlled.