Issue 33 - Exploring io_uring with python

Exploring io_uring with python, part 1

Another weekend, another weekend read, this time about Improving Deterministic Simulation Testing

---

Python excels as a language and an environment to explore concepts and mechanics. In this issue of the Weekend Read, we will implement a conceptual version of Linux’s io_uring in a few lines of code.

What is io_uring?

Introduced in Linux 5.1, io_uring is a high performance, asynchronous I/O interface that allows applications to submit I/O operations and eventually receive their results. io_uring can further enhance the performance of your application by batching.

Submission and Completion Queues

At its core, io_uring uses two shared memory queues between user space and kernel space:

• the Submission Queue (SQ), where the application posts I/O requests, and

• the Completion Queue (CQ), where the kernel posts results.

Simulating io_uring in Python

To get a better grasp of these concepts, let's create a simple simulation of io_uring in Python. This won't be a full implementation, but it will help us understand the core mechanics.

I like to model the subsystems of a system with a step method, so that a simulator can pseudo randomly choose a component to take one step. That way, the system evolves turn based over time, with some subsystem taking a step at every tick.

from dataclasses import dataclass
from typing import Any, Optional

@dataclass
class SQE:
  code: int
  data: Any
  user_data: Any  # User-provided data for identification

@dataclass
class CQE:
  data: Any
  user_data: Any  # Matches the user_data from SQE

class IOUring:
  def __init__(self, queue_size=32):
    self.submission_queue = []
    self.completion_queue = []
    self.queue_size = queue_size

  def enqueueSQE(self, sqe: SQE):
    """Add an SQE to the submission queue."""
    if len(self.submission_queue) < self.queue_size:
      self.submission_queue.append(sqe)
      return True
    return False

  def enqueueCQE(self, cqe: CQE):
    """Add a CQE to the completion queue."""
    self.completion_queue.append(cqe)

  def dequeueSQE(self) -> Optional[SQE]:
    """Dequeue and return a single SQE from the submission queue."""
    if self.submission_queue:
      return self.submission_queue.pop(0)
    return None

  def dequeueCQE(self) -> Optional[CQE]:
    """Dequeue and return a single CQE from the completion queue."""
    if self.completion_queue:
      return self.completion_queue.pop(0)
    return None

  def step(self):
    """Process a single step in the io uring event loop."""
    sqe = self.dequeueSQE()
    if sqe:
      # Simulate I/O operation
      cqe = CQE(data=f"Processed: {sqe.data}", user_data=sqe.user_data)
      self.enqueueCQE(cqe)
      return True
    return False

Let's break down the key components of this simulation:

• SQE (Submission Queue Entry): This represents an I/O operation request.

  • code: An integer representing the type of io operation.

  • data: Any data associated with the io operation.

  • user_data: A way for the application to identify and track the request.

• CQE (Completion Queue Entry): This represents a completed I/O operation.

  • data: The result of the operation.

  • user_data: The same identifier provided in the corresponding SQE.

• IOUring: This class simulates the io_uring interface.

  • enqueueSQE: Adds an SQE to the submission queue, simulating an application request

  • dequeueSQE: Removes an SQE from the submission queue, simulating kernel consumption of the application request.

  • enqueueCQE: Adds a CQE to the completion queue, simulating kernel completion.

  • dequeueCQE: Removes a CQE from the completion queue, simulation application consumption of the kernel completion.

  • step: Simulates processing of an I/O operation.

Putting It All Together

Now, let's see how we might use this in a simple simulation:

def run_simulation(steps):
  
  io_uring = IOUring(queue_size=5)
  
  for i in range(steps):
    print(f"Step {i + 1}") 
    # Simulate adding new SQEs
    if i % 3 == 0:  # Every 3 steps, add a new SQE
      sqe = SQE(
        code=i % 2,
        data=f"data_{i}",
        user_data=f"request_{i}"
      )
      print(f"Submit SQE: user_data={sqe.user_data}")
      io_uring.enqueueSQE(sqe)
    
    # Process an SQE
    io_uring.step()
    
    # Simulate consuming CQEs
    if i % 2 == 0:  # Every 2 steps, try to dequeue a CQE
      cqe = io_uring.dequeueCQE()
      if cqe:
        print(f"Consumed CQE: user_data={cqe.user_data}")
    

# Run the simulation for 10 steps
run_simulation(10)

This simulation demonstrates the core io_uring concepts:

1. Submitting I/O requests (enqueueSQE)

2. Processing those requests (step)

3. Retrieving completed I/O operations (dequeueCQE)

Conclusion

While this Python implementation is a simplification, the code captures the essence of io_uring operates well:

• Asynchronous Operations: The separation of submission and completion allows for asynchronous I/O. Your application can submit multiple requests without waiting for each to complete.

• Efficient Communication: In the actual io_uring setup, the queues are shared memory between the kernel and user space, minimizing context switches and copying of data.

• Batching: Although not explicitly shown in our simulation, io_uring allows for submitting and processing multiple operations in batches, further improving efficiency.

• Request Tracking: The user_data field demonstrates how applications can correlate submissions with completions, crucial for managing asynchronous operations.

However, so far, we did not explore how to use io_uring. Next week, we will explore how we can provide a delightful developer experience on top of io_uring.

Happy Reading