Task Queue Data Processor

Build a distributed data processing system that uses a message queue to coordinate work between producer and consumer functions.

Requirements

Producer Function

Create a producer function that:

• Accepts a list of data items as input
• Adds all items to a shared queue in a single operation
• Returns the count of items successfully added

Consumer Function

Create a consumer function that:

• Accepts a count parameter specifying how many items to retrieve
• Retrieves items from the shared queue
• Processes each item by doubling its value
• Returns the list of processed results

Queue Management

• Use a persistent shared queue accessible by both functions
• The queue must preserve FIFO ordering
• Support adding and retrieving multiple items efficiently

Test Cases

Test Case 1: Basic Producer-Consumer Flow

• The producer function adds 5 numbers [1, 2, 3, 4, 5] to the queue @test
• The consumer function retrieves all items from the queue
• Expected: Consumer receives all 5 numbers in the same order [1, 2, 3, 4, 5]

Test Case 2: Batch Operations

• The producer function adds 10 items to the queue in a single batch operation @test
• Expected: All 10 items are successfully added
• The consumer retrieves all 10 items from the queue
• Expected: All items are retrieved in order

Test Case 3: Empty Queue Handling

• The consumer function attempts to retrieve items from an empty queue @test
• Expected: Consumer returns an empty list without errors

Implementation

@generates

API

import modal

app = modal.App("task-queue-processor")

# Shared queue for communication
queue = modal.Queue.from_name("processing-queue", create_if_missing=True)

@app.function()
def producer(items: list) -> int:
    """
    Adds items to the queue for processing.

    Args:
        items: List of items to add to the queue

    Returns:
        Number of items successfully added to the queue
    """
    pass

@app.function()
def consumer(num_items: int) -> list:
    """
    Retrieves and processes items from the queue.

    Args:
        num_items: Maximum number of items to retrieve from the queue

    Returns:
        List of processed items
    """
    pass

@app.local_entrypoint()
def main():
    """
    Example usage demonstrating the producer-consumer pattern.
    """
    pass

Dependencies { .dependencies }

modal { .dependency }

Provides serverless cloud computing platform with queue support.