Managing Migration Depth

The Async Propagation Challenge

In AsyncIO, when a function uses await, it must be defined with async def. Any function that calls an async function must also be async, creating a chain reaction throughout your codebase.

# The await keyword propagates up the call stack
async def low_level_io():
    # Async I/O operation
    return await asyncio.sleep(1)

async def mid_level_function():
    # Must be async to call low_level_io
    return await low_level_io()

async def high_level_function():
    # Must be async to call mid_level_function
    return await mid_level_function()

# Your entire call stack becomes async

What Awaitlet Does

Awaitlet allows you to call asynchronous functions from synchronous code without having to make the calling function async. It acts as a containment mechanism for async/await propagation.

import awaitlet

# A regular synchronous function
def sync_function():
    # Call an async function without making sync_function async
    result = awaitlet.awaitlet(async_function())
    return result

# An async function that can be called from sync code
async def async_function():
    await asyncio.sleep(1)
    return "result"

import awaitlet
import asyncio

# The async function you want to call
async def fetch_data_async(url):
    # Some async operation that replaces an Eventlet call
    await asyncio.sleep(1)  # Simulate network I/O
    return f"Data from {url}"

# Your existing synchronous function
def process_data(url):
    # Instead of using eventlet.spawn, use awaitlet
    data = awaitlet.awaitlet(fetch_data_async(url))
    return data

# Usage remains unchanged
result = process_data("https://example.com")

For concurrent operations (similar to Eventlet's GreenPool), Awaitlet can work with gathered tasks:

import awaitlet
import asyncio
import aiohttp

async def fetch_all_async(urls):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(fetch_one_async(session, url))
        return await asyncio.gather(*tasks)

async def fetch_one_async(session, url):
    async with session.get(url) as response:
        return await response.text()

def process_urls(urls):
    # Single awaitlet call for multiple concurrent async operations
    results = awaitlet.awaitlet(fetch_all_async(urls))
    return results

# Usage remains unchanged
urls = ['http://example.com', 'http://example.org']
results = process_urls(urls)

import awaitlet
import asyncio

# Original deep call chain with Eventlet dependencies
def api_endpoint(request):
    data = service_layer(request.params)
    return {"result": data}

def service_layer(params):
    return business_logic(params)

def business_logic(params):
    return data_access_layer(params)

def data_access_layer(params):
    # This uses Eventlet for I/O
    import eventlet
    return eventlet.spawn(fetch_from_db, params).wait()

def fetch_from_db(params):
    # Simulated DB query
    eventlet.sleep(0.1)
    return {"id": params.get("id"), "value": "data"}

# Refactored with Awaitlet - only changing the lower layers
def api_endpoint(request):
    # Unchanged
    data = service_layer(request.params)
    return {"result": data}

def service_layer(params):
    # Unchanged
    return business_logic(params)

def business_logic(params):
    # Unchanged
    return data_access_layer(params)

def data_access_layer(params):
    # Changed to use awaitlet instead of eventlet
    return awaitlet.awaitlet(fetch_from_db_async(params))

async def fetch_from_db_async(params):
    # Async implementation
    await asyncio.sleep(0.1)
    return {"id": params.get("id"), "value": "data"}

Creating an API Gateway with Awaitlet

# api_gateway.py
import awaitlet

class InternalAPIGateway:
    """Migration interface that abstracts implementation details."""
    
    def fetch_data(self, resource_id):
        # Version 1: During migration, call the async implementation
        # without requiring callers to change
        return awaitlet.awaitlet(self._fetch_data_async(resource_id))
    
    async def _fetch_data_async(self, resource_id):
        # New async implementation that replaces Eventlet
        # This can be gradually enhanced with AsyncIO features
        import asyncio
        await asyncio.sleep(0.1)  # Simulated async I/O
        return {"id": resource_id, "data": "result"}
    
    # More API methods following the same pattern
    def save_data(self, resource_id, data):
        return awaitlet.awaitlet(self._save_data_async(resource_id, data))
        
    async def _save_data_async(self, resource_id, data):
        # Async implementation
        import asyncio
        await asyncio.sleep(0.1)
        return {"status": "saved", "id": resource_id}

import awaitlet
import asyncio

# Original implementation using Eventlet
class RPCClientOriginal:
    def __init__(self):
        import eventlet
        self.pool = eventlet.GreenPool()
    
    def call_remote(self, method, *args, **kwargs):
        def _execute():
            # Simulate RPC call with eventlet
            import eventlet
            eventlet.sleep(0.1)
            return {'result': f"Response from {method}"}
        
        return self.pool.spawn(_execute).wait()

# Migrated implementation with Awaitlet
class RPCClient:
    def call_remote(self, method, *args, **kwargs):
        # Same interface, but using awaitlet internally
        return awaitlet.awaitlet(self._call_remote_async(method, *args, **kwargs))
    
    async def _call_remote_async(self, method, *args, **kwargs):
        # Async implementation
        await asyncio.sleep(0.1)
        return {'result': f"Response from {method}"}
    
    # Optionally expose async API for new consumers
    async def call_remote_async(self, method, *args, **kwargs):
        return await self._call_remote_async(method, *args, **kwargs)

# Usage remains unchanged for existing code
client = RPCClient()
result = client.call_remote("get_data", id=123)

# New async code can use the async API
async def new_consumer():
    client = RPCClient()
    result = await client.call_remote_async("get_data", id=123)
    return result

A large application with multiple services and deep Eventlet usage:

• API Server with multiple endpoints
• Worker services processing jobs in the background
• Shared internal libraries used by all components
• Extensive use of Eventlet for concurrency throughout the stack