Implement a Concurrent Buffered File Logger

Problem Statement

High-throughput distributed systems frequently emit millions of log events per second across dozens of worker threads. Writing synchronously to a file on every log call incurs unacceptable I/O overhead and lock contention.

Design and implement a thread-safe, buffered file logger class, ConcurrentFileLogger, that buffers log events in memory and flushes them to a single file in batches.

Requirements

1. Concurrent Ingestion: Multiple threads can call log(message) concurrently without deadlocking or corrupting output.
2. Ordering Guarantee: Log messages submitted by the same thread must appear in the file in their relative submission order. Across different threads, logs should be ordered by their insertion timestamp or relative ingestion order.
3. Flushing Conditions: The logger must flush buffered records to disk when:
   • The buffer reaches max_buffer_size messages.
   • A maximum time of flush_interval_ms has passed since the last flush.
   • An explicit call to flush() is made by the caller.
4. Clean Shutdown: Calling close() must flush all remaining in-memory records and safely stop any background threads.

Interface

Implement the following class:

class ConcurrentFileLogger:
    def __init__(self, file_path: str, max_buffer_size: int = 1000, flush_interval_ms: int = 200):
        """
        Initializes the logger with a destination file, buffer limit, and flush interval.
        """
        pass

    def log(self, message: str) -> None:
        """
        Appends a message to the buffer safely from any thread.
        """
        pass

    def flush(self) -> None:
        """
        Synchronously flushes all currently buffered records to disk.
        """
        pass

    def close(self) -> None:
        """
        Flushes remaining records, closes the underlying file, and terminates background workers.
        """
        pass

Example Usage

import threading

logger = ConcurrentFileLogger("app.log", max_buffer_size=10, flush_interval_ms=500)

def worker(thread_id):
    for i in range(5):
        logger.log(f"Thread-{thread_id}: message {i}")

threads = [threading.Thread(target=worker, args=(t,)) for t in range(3)]
for t in threads: t.start()
for t in threads: t.join()

logger.close()
# app.log now contains all 15 formatted records, with Thread-X messages properly ordered relative to each other.

Edge Cases & Invariants

• Backpressure: Define your policy (blocking vs. dropping vs. unbounded memory) when producers outpace consumer disk writes.
• Error Handling: An I/O error during a flush should not corrupt the in-memory state or silently swallow future logs.
• Double Close: Calling close() multiple times or calling log() after close() must be handled gracefully.