def run_producer(seed, num_workers, worker_id, fill_up, clean_dirty, dataset_type):
    """
    Run the producer.
    The producer will first fill up the buffer with samples.
    Then it will keep replacing dirty samples
    (i.e., samples that have been read by the consumer)
    with new samples.
    """
    vla_dataset = VLADataset(seed=seed, dataset_type=dataset_type)
    chunk_start_idx = worker_id * BUF_NUM_CHUNKS // num_workers
    chunk_end_idx = (worker_id + 1) * BUF_NUM_CHUNKS // num_workers
    if fill_up:
        print(f"Worker {worker_id}: Start filling up the buffer...")
    elif clean_dirty:
        # Only refresh the dirty bits
        print(f"Worker {worker_id}: Start refreshing the dirty bits...")
        for chunk_idx in range(chunk_start_idx, chunk_end_idx):
            chunk_dir = os.path.join(BUF_PATH, f"chunk_{chunk_idx}")
            dirty_bit = np.zeros(BUF_CHUNK_SIZE, dtype=np.uint8)
            save_dirty_bit(chunk_dir, dirty_bit)
        print(f"Worker {worker_id}: Refreshed the dirty bits.")

    fill_chunk_idx = chunk_start_idx
    fill_chunk_item_idx = 0
    dirty_chunk_idx = chunk_start_idx
    dirty_chunk_item_idxs = []
    time_stmp = time.time()
    for episode_steps in vla_dataset:
        for step in episode_steps:
            if fill_up and fill_chunk_idx < chunk_end_idx:
                # Fill up the buffer
                chunk_dir = os.path.join(BUF_PATH, f"chunk_{fill_chunk_idx}")
                if fill_chunk_item_idx == 0:
                    # Create a new chunk
                    os.makedirs(chunk_dir, exist_ok=True)
                    # Write the dirty bit of size BUF_CHUNK_SIZE
                    dirty_bit = np.zeros(BUF_CHUNK_SIZE, dtype=np.uint8)
                    save_dirty_bit(chunk_dir, dirty_bit)
                
                # Save the sample
                save_sample(step, chunk_dir, fill_chunk_item_idx)

                # print(f"Filled up chunk {fill_chunk_item_idx+1}/{BUF_CHUNK_SIZE} {fill_chunk_idx+1}/{BUF_NUM_CHUNKS}")
                local_fill_chunk_idx = fill_chunk_idx - chunk_start_idx
                local_num_chunks = chunk_end_idx - chunk_start_idx
                if (local_fill_chunk_idx % 10 == 0 or local_fill_chunk_idx == local_num_chunks - 1) and fill_chunk_item_idx == 0:
                    print(f"Worker {worker_id}: Filled up chunk {local_fill_chunk_idx+1}/{local_num_chunks}")
                fill_chunk_item_idx += 1
                if fill_chunk_item_idx == BUF_CHUNK_SIZE:
                    fill_chunk_idx += 1
                    fill_chunk_item_idx = 0
                if fill_chunk_idx == BUF_NUM_CHUNKS:
                    print(f"Worker {worker_id}: Buffer filled up. Start replacing dirty samples...")

            else:
                # Search for the dirty chunk to replace
                while len(dirty_chunk_item_idxs) == 0:
                    dirty_chunk_dir = os.path.join(BUF_PATH, f"chunk_{dirty_chunk_idx}")
                    dirty_chunk_item_idxs = get_dirty_item(dirty_chunk_dir)
                    # Print the dirty ratio
                    if time.time() - time_stmp > 2.0:
                        dirty_ratio = len(dirty_chunk_item_idxs) / BUF_CHUNK_SIZE
                        print(f"Worker {worker_id}: Dirty Ratio for Chunk {dirty_chunk_idx}: {dirty_ratio:.2f}")
                        time_stmp = time.time()

                    if len(dirty_chunk_item_idxs) > 0:
                        # Lock the chunk
                        dirty_bit = np.ones(BUF_CHUNK_SIZE, dtype=np.uint8)
                        save_dirty_bit(dirty_chunk_dir, dirty_bit)
                    
                    # Iterate over the chunks
                    dirty_chunk_idx += 1
                    if dirty_chunk_idx == chunk_end_idx:
                        dirty_chunk_idx = chunk_start_idx

                # Replace the dirty item
                dirty_item_idx = dirty_chunk_item_idxs.pop()
                chunk_dir = os.path.join(BUF_PATH, f"chunk_{dirty_chunk_idx}")
                # Save the sample
                save_sample(step, chunk_dir, dirty_item_idx)

                # If we have replaced all dirty items in the chunk
                if len(dirty_chunk_item_idxs) == 0:
                    # Unlock the chunk
                    dirty_bit = np.zeros(BUF_CHUNK_SIZE, dtype=np.uint8)
                    save_dirty_bit(dirty_chunk_dir, dirty_bit)
                    print(f"Worker {worker_id}: Replaced dirty chunk {dirty_chunk_idx}.")

    Run the producer.
    The producer will first fill up the buffer with samples.
    Then it will keep replacing dirty samples
    (i.e., samples that have been read by the consumer)
    with new samples.
    
policy/RDT/data/producer.py