Thread-Safe Buffered Writer with a Background Flush Thread

0) Problem Restatement

Thousands of producer threads need to append records (opaque bytes, e.g., log events) to one file. Writing to disk directly from every thread is slow (each write is a system call, and fsync is slower still) and threads would fight over the file. Design a component where:

• producers call write(record), which is fast and doesn't wait on disk,
• a background thread collects records and writes them to disk in batches,
• memory is bounded,
• callers can ask for durability ("return only when my record is safely on disk"),
• shutdown flushes everything.

Databricks asked this twice ("durable concurrent event writer" and "thread-safe buffered writer with background flush").

1) Requirements

• write(record): thread-safe and fast. Records from one thread keep their order.
• Optional write_durable(record) or flush(): returns after the data is fsynced.
• Flush when the buffer reaches N bytes or after T milliseconds, whichever comes first.
• A bounded buffer: when full, producers wait (backpressure) instead of using unlimited memory.
• close(): flush remaining data, stop the thread, close the file.
• Report I/O errors to callers.

2) Design

flowchart LR
    P1["Producer threads"] -->|"append under lock"| B["Active buffer"]
    B -->|"swap when full or timer"| F["Flush buffer"]
    F --> BG["Background flush thread"]
    BG -->|"write + fsync"| FILE[("File")]
    BG -->|"notify waiters of flushed sequence"| P1

• Double buffering: producers append to the active buffer. The flush thread swaps it with an empty one (quick, under the lock), then writes the full one to disk without holding the lock. Producers keep writing while disk I/O happens.
• Sequence numbers: each write gets an increasing sequence number. The flusher tracks flushed_seq. A durable writer waits until flushed_seq >= my_seq. This is group commit: one fsync covers many writers.
• Condition variables: not_full (producers wait when the buffer is at its limit), has_data (the flusher waits for work or a timeout), flushed (durable writers wait).

3) Code (Python)

import os, threading, time

class BufferedWriter:
    def __init__(self, path, max_bytes=1 << 20, flush_interval=0.05):
        self.f = open(path, 'ab')
        self.max_bytes, self.interval = max_bytes, flush_interval
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.has_data = threading.Condition(self.lock)
        self.flushed = threading.Condition(self.lock)
        self.buf, self.size = [], 0
        self.seq = 0            # last assigned sequence number
        self.flushed_seq = 0    # last sequence number safely on disk
        self.error, self.closed = None, False
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

    def write(self, record: bytes, durable=False):
        with self.lock:
            if self.closed: raise ValueError("writer closed")
            if self.error: raise self.error
            while self.size + len(record) > self.max_bytes and self.buf:
                self.has_data.notify()          # ask for an early flush
                self.not_full.wait()            # backpressure
            self.buf.append(record); self.size += len(record)
            self.seq += 1; my_seq = self.seq
            if self.size >= self.max_bytes: self.has_data.notify()
            if durable:
                while self.flushed_seq < my_seq and not self.error:
                    self.flushed.wait()
                if self.error: raise self.error
        return my_seq

    def _run(self):
        while True:
            with self.lock:
                if not self.buf and not self.closed:
                    self.has_data.wait(timeout=self.interval)
                if not self.buf and self.closed:
                    return
                batch, upto = self.buf, self.seq          # swap buffers
                self.buf, self.size = [], 0
                self.not_full.notify_all()
            try:
                if batch:
                    self.f.write(b''.join(batch)); self.f.flush(); os.fsync(self.f.fileno())
                err = None
            except OSError as e:
                err = e
            with self.lock:
                if err: self.error = err
                else: self.flushed_seq = upto
                self.flushed.notify_all()

    def close(self):
        with self.lock:
            self.closed = True
            self.has_data.notify()
        self.thread.join()
        self.f.close()

4) Explaining the Key Decisions

• Why not write directly from each thread? Each write and fsync costs a system call and a disk flush. Batching turns thousands of small writes into one big sequential write and one fsync.
• Why swap buffers? So the lock is held only for a pointer swap, never during slow disk I/O.
• Ordering: records are appended in the order they take the lock, so each thread's own records stay in order. There's no global order across threads beyond that, which is usually fine.
• Durability semantics: plain write returns once the record is in memory, so it could be lost on a crash. durable=True waits for the fsync. Make this explicit in the API.
• Backpressure: if disk is slower than producers, the buffer fills and producers wait. Memory stays bounded and the slowdown is visible.
• Errors: if the disk fails, store the error and raise it to later writers and waiting durable writers, so nobody thinks data was saved.

5) Follow-ups

• Lock contention with thousands of threads: shard the active buffer per CPU core or thread (thread-local buffers), and have the flusher collect from all of them. Or use a lock-free multi-producer queue.
• Record framing: write a length prefix and checksum per record, so a reader can detect a torn last record after a crash.
• File rotation: start a new file after X MB, with the flusher handling the switch.

6) Wrap-Up

Producers append to an in-memory buffer under a short lock, and a background thread swaps buffers and writes each batch with one write + fsync (group commit) outside the lock. Sequence numbers let durable writers wait until their record is flushed, a size limit gives backpressure, a timer bounds latency, errors are reported to callers, and close() flushes everything before stopping.