This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
class Barrier(): | |
"""asyncio version of threading.Barrier (partial). | |
Implements wait() only, without support for timeouts.""" | |
def __init__(self, parties, loop=None): | |
if loop is None: | |
self._loop = asyncio.get_event_loop() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Accessing TCP metrics in Linux using Generic Netlink via gnlpy.""" | |
import array | |
import ipaddress | |
import struct | |
from pprint import pprint | |
from typing import Union | |
import gnlpy.netlink as netlink |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
class CancelledFromOutside(asyncio.CancelledError): | |
pass | |
class CancelledFromInside(asyncio.CancelledError): | |
pass |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import ipaddress | |
import typing | |
import abc | |
IPNetworkType = typing.TypeVar('IPNetworkType', ipaddress.IPv4Network, ipaddress.IPv6Network) | |
class IPSubnetSequence(typing.Sequence[IPNetworkType]): | |
"""An indexable sequence of subnets of a given network. |