Python中延迟FIFO管道数据流的技术求助

Hey Andrew, let's get that visualizer synced perfectly with your Bluetooth speaker—your original approach was on the right track, but spawning a new process for every single data point is what's killing your system performance. Let's break down the problem and fix it step by step.

Why Your Original Code Crashes #

Your current code creates a new multiprocessing.Process for every line of data from the FIFO. Even if you're getting just 100 data points per second, that's 6,000 processes in a minute! Processes are heavyweight—each one uses significant system resources, and your OS can't keep up with creating/destroying them that fast, hence the freeze.

Better Approach: Queue + Threads (Lightweight & Reliable) #

Instead of spawning processes for every data point, we'll use a thread-safe queue to buffer incoming data, and a single worker thread that waits until each data point is 0.5 seconds old before processing it. This keeps your data flow continuous while adding the exact delay you need, without overwhelming your system.

Step-by-Step Code Solution #

import time
import queue
import threading

def read_fifo(fifo_path, data_queue):
    """Read data from FIFO and add to queue with timestamp"""
    with open(fifo_path, 'rb') as fd1:
        for line in fd1:
            # Capture when we received the data
            receive_time = time.time()
            # Clean up the input string
            stringvals = line.decode("utf-8").strip()
            # Add to queue: (timestamp, data)
            data_queue.put((receive_time, stringvals))
            # Optional: Small sleep to avoid CPU spiking if data comes too fast
            # time.sleep(0.005)

def process_delayed_data(data_queue, delay_seconds=0.5):
    """Process data only after it's been buffered for the required delay"""
    while True:
        if not data_queue.empty():
            # Peek at the oldest data in the queue
            oldest_time, oldest_data = data_queue.queue[0]
            current_time = time.time()
            
            # Check if enough time has passed
            if current_time - oldest_time >= delay_seconds:
                # Remove from queue and process (replace print with your visualizer code)
                data_queue.get()
                print(oldest_data)
            else:
                # Wait a tiny bit before checking again to save CPU
                time.sleep(0.005)
        else:
            # Queue is empty, wait a moment before checking again
            time.sleep(0.01)

if __name__ == "__main__":
    thepath = "/path/to/your/fifo"  # Replace with your actual FIFO path
    # Limit queue size to prevent memory overflow if processing lags
    data_queue = queue.Queue(maxsize=1000)

    # Start thread to read FIFO data
    read_thread = threading.Thread(target=read_fifo, args=(thepath, data_queue))
    read_thread.daemon = True  # Auto-terminate when main thread exits
    read_thread.start()

    # Start thread to handle delayed processing
    delay_thread = threading.Thread(target=process_delayed_data, args=(data_queue, 0.5))
    delay_thread.daemon = True
    delay_thread.start()

    # Keep main thread running (interrupt with Ctrl+C)
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nVisualizer stopped.")

Key Improvements #

• Lightweight Threads: Threads use far less system resources than processes, so you won't overload your system.
• Continuous Data Flow: The queue buffers all incoming data, so you never get gaps from per-data-point sleeps.
• Precise Delay: Using timestamps ensures every data point is delayed exactly 0.5 seconds, even if your FIFO data comes in at variable rates.
• Memory Safety: The queue has a maximum size to prevent it from growing indefinitely if your visualizer can't keep up temporarily.

Optional Optimization for Fixed-Rate Data #

If your FIFO outputs data at a perfectly consistent rate (e.g., 100 times per second), you can skip the timestamps and just buffer a fixed number of data points. For 0.5 seconds of delay at 100Hz, you'd buffer 50 points:

def process_fixed_rate_data(data_queue, buffer_size=50):
    while True:
        if data_queue.qsize() >= buffer_size:
            # Process the oldest data point
            print(data_queue.get())
        else:
            time.sleep(0.005)

This is slightly more efficient since you avoid timestamp checks, but only use it if your data rate is rock-solid.

Final Notes #

• Replace the print(oldest_data) line with your actual visualizer rendering code.
• If your visualizer processing is CPU-heavy, you could offload that to a separate process (but only one, not per data point!) using multiprocessing, but threads should be sufficient for most cases.

内容的提问来源于stack exchange，提问作者Andrew Fedun