Event notification with asyncio

[niksabaldun]

I want to respond to IP address and route changes in asyncio application. As pyroute2 uses blocking code, I run the monitor function in executor thread. When the application is closing, I use a threading event to tell the function that it needs to exit. However, the function will still block until some network event is received. So my question is: is there some safe way to generate a network event - any event will do. I thought perhaps to create and immediately delete a virtual link, but I would like to know is there a better way. Some action that won't do any harm to the system and is guaranteed to succeed.

The code follows.

def monitor_network(exit_event):
    change_events = (
        'RTM_NEWLINK', 'RTM_DELLINK', 'RTM_SETLINK', 'RTM_NEWADDR', 'RTM_DELADDR',
        'RTM_NEWROUTE', 'RTM_DELROUTE', 'RTM_NEWNETCONF', 'RTM_DELNETCONF'
    )
    with IPRoute() as ipr:
        ipr.bind()
        while True:
            messages = ipr.get()
            if exit_event.is_set():
                return False
            if any(message.get('event') in change_events for message in messages):
                return True


async def network_watcher():
    loop = asyncio.get_running_loop()
    exit_event = threading.Event()
    try:
        while True:
            if await loop.run_in_executor(None, monitor_network, exit_event):
                log.info('Network change detected, reloading network data...')
                get_local_ip_routes(True)
            else:
                return
    except asyncio.CancelledError:
        exit_event.set() # Signal to monitor_network function that it needs to exit
        # Generate a network event to unblock monitor_network function 
        ipr = IPRoute()
        # What to do here?

[svinota]

You can simply close the socket with ipr.close() and the next ipr.get() will raise OSError with errno.EBADF.

But for that purpose you have to share the IPRoute() instance between monitor_network() and network_watcher().

[niksabaldun]

I have tried that, but OSError is not raised. The get() call just hangs indefinitely.

def monitor_network(ip_route):
    change_events = (
        'RTM_NEWLINK', 'RTM_DELLINK', 'RTM_SETLINK', 'RTM_NEWADDR', 'RTM_DELADDR',
        'RTM_NEWROUTE', 'RTM_DELROUTE', 'RTM_NEWNETCONF', 'RTM_DELNETCONF'
    )
    try:
        while True:
            messages = ip_route.get()
            if any(message.get('event') in change_events for message in messages):
                return True
    except OSError:
        return False

async def network_watcher():
    loop = asyncio.get_running_loop()
    while True:
        with IPRoute() as ip_route:
            try:
                if await loop.run_in_executor(None, monitor_network, ip_route):
                    log.info('Network change detected, reloading network data...')
                    get_local_ip_routes(True)
            except asyncio.CancelledError:
                ip_route.close()
                return

[svinota]

Yep, I've got that. Let me play a bit.

But as a temporary workaround — yes, you can trigger some network event. For example, you can create/delete a route in a separate table, or a rule:

# create/delete a route in a separate table (please check if it doesn't exist)
spec = {'dst': '127.0.1.0/24', 'gateway': '127.0.0.1', 'table': 404}
ipr.route('add', **spec)
ipr.route('del', **spec)

# OR

# create/remove a rule that matches some unused fwmark
spec = {'fwmark': 0x404, 'table': 404, 'priority': 404}
ipr.rule('add', **spec)
ipr.rule('del', **spec)

[niksabaldun]

Thanks. Closing the socket would have been a cleaner solution indeed, but this works flawlessly so it's good enough for me.