| Server IP : 88.223.91.39 / Your IP : 216.73.216.122 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/./alt/python311/lib/python3.11/site-packages/pyroute2/dhcp/ |
Upload File : |
'''Timers to manage lease rebinding, renewal & expiration.'''
import asyncio
import dataclasses
from logging import getLogger
from typing import Optional
from pyroute2.dhcp.fsm import AsyncCallback
from pyroute2.dhcp.leases import Lease
LOG = getLogger(__name__)
@dataclasses.dataclass
class LeaseTimers:
'''Manage callbacks associated with DHCP leases.'''
renewal: Optional[asyncio.TimerHandle] = None
rebinding: Optional[asyncio.TimerHandle] = None
expiration: Optional[asyncio.TimerHandle] = None
def cancel(self) -> None:
'''Cancel all current timers.'''
for timer_name in ('renewal', 'rebinding', 'expiration'):
self._reset_timer(timer_name)
def _reset_timer(self, timer_name: str) -> None:
'''Cancel a timer and set it to None.'''
if timer_name not in ('renewal', 'rebinding', 'expiration'):
raise ValueError(f'Unknown timer {timer_name!r}')
if timer := getattr(self, timer_name):
assert isinstance(timer, asyncio.TimerHandle)
if not timer.cancelled():
# FIXME: how do we know a timer wasn't cancelled ?
# this causes spurious logs
LOG.debug('Canceling %s timer', timer_name)
timer.cancel()
setattr(self, timer_name, None)
def arm(self, lease: Lease, **callbacks: AsyncCallback) -> None:
'''Reset & arm timers from a `Lease`.
`callbacks` must be async callables with no arguments
that will be called when the associated timer expires.
'''
self.cancel()
loop = asyncio.get_running_loop()
for timer_name, async_callback in callbacks.items():
self._reset_timer(timer_name)
lease_time = getattr(lease, f'{timer_name}_in')
if not lease_time:
LOG.debug('%s time is infinite', timer_name)
continue
if lease_time < 0.0:
LOG.debug('%s is in the past', timer_name)
continue
LOG.info('Scheduling %s in %.2fs', timer_name, lease_time)
timer = loop.call_later(
lease_time, self._create_timer_task, timer_name, async_callback
)
setattr(self, timer_name, timer)
def _create_timer_task(
self, timer_name: str, async_callback: AsyncCallback
) -> None:
''' 'Internal callback for loop.call_later.
Creates a Task that runs the async_callback.
'''
asyncio.create_task(
async_callback(), name=f"{timer_name} timer callback"
)