Pyjo.IOLoop - Minimalistic event loop

import Pyjo.IOLoop

# Listen on port 3000
@Pyjo.IOLoop.server(port=3000)
def server(loop, stream, id):
    @stream.on
    def read(stream, chunk):
        # Process input chunk
        print(chunk)

        # Write response
        stream.write(b'HTTP/1.1 200 OK')

# Connect to port 3000
@Pyjo.IOLoop.client(port=3000)
def id(loop, err, stream):
    @stream.on
    def read(stream, chunk):
        # Process input
        print(f'Input: {chunk}')

    # Write request
    stream.write(b'GET / HTTP/1.1\x0d\x0a\x0d\x0a')

# Add a timer
Pyjo.IOLoop.timer(5, lambda loop: loop.remove(id))

# Start event loop if necessary
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

Pyjo.IOLoop is a very minimalistic event loop based on Pyjo.Reactor, it has been reduced to the absolute minimal feature set required to build solid and scalable non-blocking clients and servers.

Depending on operating system, the default per-process and system-wide file descriptor limits are often very low and need to be tuned for better scalability.

The event loop is resilient to time jumps, as it uses a monotonic clock. A TLS certificate and key are also built right in, to make writing test servers as easy as possible.

What Mojo calls on the class, as Mojo::IOLoop->timer, is a function of the module that works with the singleton, as Pyjo.IOLoop.timer. Each thread has a singleton of its own, as each thread of Perl has a copy of the interpreter, so that the threads run event loops of their own at the same time, and the functions of the module work with the one of the thread that calls them. A method that takes a callback takes it also as the keyword argument cb, and is a decorator without it, and the decorated name is what the method returns.

@Pyjo.IOLoop.timer(3)
def id(loop):
    print('Timeout!')

The end of a with block removes everything, as reset() does.

Events

Pyjo.IOLoop inherits all events from Pyjo.EventEmitter and can emit the following new ones.

finish

loop.on('finish', lambda loop: ...)

Emitted when the event loop wants to shut down gracefully and is just waiting for all existing connections to be closed.

reset

loop.on('reset', lambda loop: ...)

Emitted when the event loop is reset, this usually happens after the process is forked to clean up resources that cannot be shared.

Debugging

You can set the PYJO_IOLOOP_DEBUG environment variable to get some advanced diagnostics information printed to sys.stderr.

PYJO_IOLOOP_DEBUG=1

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::IOLoop of Mojolicious 9.49.

class Pyjo.IOLoop.Pyjo_IOLoop(*, max_accepts: int = 0, max_connections: int = 1000, reactor: Pyjo_Reactor | None = None)

Pyjo.IOLoop inherits all methods from Pyjo.EventEmitter and implements the following new ones.

acceptor(acceptor: Any) → Any

server = Pyjo.IOLoop.acceptor(id)
server = loop.acceptor(id)
id = loop.acceptor(Pyjo.IOLoop.Server.new())

Get Pyjo.IOLoop.Server object for id or turn object into an acceptor.

client(cb: ClientCallback, **options: Any) → str

client(**options: Any) → Callable[[ClientCallback], str]

id = Pyjo.IOLoop.client(cb, address='127.0.0.1', port=3000)
id = loop.client(cb, address='127.0.0.1', port=3000)
id = loop.client(cb=cb, address='127.0.0.1', port=3000)

@loop.client(address='127.0.0.1', port=3000)
def id(loop, err, stream):
    ...

Open a TCP/IP or UNIX domain socket connection with Pyjo.IOLoop.Client and create a stream object (usually Pyjo.IOLoop.Stream), takes the same arguments as Pyjo.IOLoop.Client.connect().

datagram(cb: DatagramCallback, **options: Any) → str

datagram(**options: Any) → Callable[[DatagramCallback], str]

id = Pyjo.IOLoop.datagram(cb, port=9999)
id = loop.datagram(cb, address='127.0.0.1', port=9999)
id = loop.datagram(cb=cb)

@loop.datagram(port=9999)
def id(loop, datagram, id):
    ...

Open a datagram socket with Pyjo.IOLoop.Datagram, bound with the options of Pyjo.IOLoop.Datagram.Pyjo_IOLoop_Datagram.bind() if any are given, which the event loop watches until it is closed. The callback gets the loop, the datagram socket and its id, which remove() takes.

# Echo server
@Pyjo.IOLoop.datagram(address='127.0.0.1', port=9999)
def id(loop, datagram, id):
    datagram.on('message', lambda datagram, data, peer: datagram.send(data, peer))

property is_running: bool

boolean = Pyjo.IOLoop.is_running()
boolean = loop.is_running

Check if event loop is running.

max_accepts

max = loop.max_accepts
loop.max_accepts = 1000

The maximum number of connections this event loop is allowed to accept, before shutting down gracefully without interrupting existing connections, defaults to 0. Setting the value to 0 will allow this event loop to accept new connections indefinitely. Note that up to half of this value can be subtracted randomly to improve load balancing between multiple server processes, and to make sure that not all of them restart at the same time.

max_connections

max = loop.max_connections
loop.max_connections = 100

The maximum number of accepted connections this event loop is allowed to handle concurrently, before stopping to accept new incoming connections, defaults to 1000.

next_tick(cb: TimerCallback) → None

next_tick() → Callable[[TimerCallback], None]

none = Pyjo.IOLoop.next_tick(lambda loop: ...)
none = loop.next_tick(lambda loop: ...)

Execute callback as soon as possible, but not before returning or other callbacks that have been registered with this method, always returns None.

# Perform operation on next reactor tick
Pyjo.IOLoop.next_tick(lambda loop: ...)

one_tick() → None

Pyjo.IOLoop.one_tick()
loop.one_tick()

Run event loop until an event occurs.

# Don't block longer than 0.5 seconds
id = Pyjo.IOLoop.timer(0.5, lambda loop: None)
Pyjo.IOLoop.one_tick()
Pyjo.IOLoop.remove(id)

property reactor: Pyjo_Reactor

reactor = loop.reactor
loop.reactor = Pyjo.Reactor.Asyncio.new()

Low-level event reactor, usually a Pyjo.Reactor.Asyncio object with a default subscriber to the event error of Pyjo.Reactor.

# Watch if handle becomes readable or writable
Pyjo.IOLoop.singleton.reactor.io(handle, lambda reactor, writable: print(
    'Handle is writable' if writable else 'Handle is readable'
))

# Change to watching only if handle becomes writable
Pyjo.IOLoop.singleton.reactor.watch(handle, False, True)

# Remove handle again
Pyjo.IOLoop.singleton.reactor.remove(handle)

recurring(after: float, cb: TimerCallback) → str

recurring(after: float) → Callable[[TimerCallback], str]

id = Pyjo.IOLoop.recurring(3, lambda loop: ...)
id = loop.recurring(0, lambda loop: ...)
id = loop.recurring(0.25, lambda loop: ...)

Create a new recurring timer, invoking the callback repeatedly after a given amount of time in seconds.

# Perform operation every 5 seconds
Pyjo.IOLoop.recurring(5, lambda loop: ...)

remove(id: str, *, close: bool = True) → None

Pyjo.IOLoop.remove(id)
loop.remove(id)
loop.remove(id, close=False)

Remove anything with an id, connections will be dropped gracefully by allowing them to finish writing all data in their write buffers. An acceptor keeps its listen socket open with close=False, so that it can be turned into an acceptor again.

reset(*, freeze: bool = False) → None

Pyjo.IOLoop.reset()
loop.reset()
loop.reset(freeze=True)

Remove everything and stop the event loop.

These options are currently available:

freeze

freeze=True

Freeze the current state of the event loop in time before resetting it. This will prevent active connections from getting closed immediately, which can help with many unintended side effects when processes are forked.

server(cb: ServerCallback, **options: Any) → str

server(**options: Any) → Callable[[ServerCallback], str]

id = Pyjo.IOLoop.server(cb, port=3000)
id = loop.server(cb, port=3000)
id = loop.server(cb=cb, port=3000)

@loop.server(port=3000)
def id(loop, stream, id):
    ...

Accept TCP/IP and UNIX domain socket connections with Pyjo.IOLoop.Server and create stream objects (usually Pyjo.IOLoop.Stream), takes the same arguments as Pyjo.IOLoop.Server.listen().

# Listen on random port
id = Pyjo.IOLoop.server(lambda loop, stream, id: ..., address='127.0.0.1')
port = Pyjo.IOLoop.acceptor(id).port

classmethod singleton() → Pyjo_IOLoop

loop = Pyjo.IOLoop.object.singleton()
loop = Pyjo.IOLoop.singleton

The global Pyjo.IOLoop singleton, used to access a single shared event loop object from everywhere inside the thread, which is also the singleton of the module. Each thread has one of its own.

# Many methods also allow you to take shortcuts
Pyjo.IOLoop.timer(2, lambda loop: Pyjo.IOLoop.stop())
Pyjo.IOLoop.start()

# Restart active timer
id = Pyjo.IOLoop.timer(3, lambda loop: print('Timeout!'))
Pyjo.IOLoop.singleton.reactor.again(id)

# Watch a file descriptor if it becomes readable
Pyjo.IOLoop.singleton.reactor.io(fd, lambda reactor, writable: print(
    'Handle is writable' if writable else 'Handle is readable'
)).watch(fd, True, False)

spawn(args: Any, **options: Any) → Pyjo_IOLoop_ChildProcess

child = Pyjo.IOLoop.spawn(['ls', '-l'])
child = loop.spawn(['php-cgi', 'index.php'], cwd='/var/www')

Run a program with Pyjo.IOLoop.ChildProcess on the reactor of the event loop, with the options of Pyjo.IOLoop.ChildProcess.Pyjo_IOLoop_ChildProcess.spawn().

# The output of a program, and its exit code
child = Pyjo.IOLoop.spawn(['ls', '-l'])
child.stdout.on('read', lambda stream, chunk: print(chunk))
child.on('close', lambda child, code, signal: print(f'Exited with {code}'))

start() → None

Pyjo.IOLoop.start()
loop.start()

Start the event loop, this will block until stop() is called. Note that some reactors stop automatically if there are no events being watched anymore.

# Start event loop only if it is not running already
if not Pyjo.IOLoop.is_running():
    Pyjo.IOLoop.start()

stop() → None

Pyjo.IOLoop.stop()
loop.stop()

Stop the event loop, this will not interrupt any existing connections and the event loop can be restarted by running start() again.

stop_gracefully() → None

Pyjo.IOLoop.stop_gracefully()
loop.stop_gracefully()

Stop accepting new connections and wait for already accepted connections to be closed, before stopping the event loop.

stream(stream: Any) → Any

stream = Pyjo.IOLoop.stream(id)
stream = loop.stream(id)
id = loop.stream(Pyjo.IOLoop.Stream.new(handle))

Get Pyjo.IOLoop.Stream object for id or turn object into a connection.

# Increase inactivity timeout for connection to 300 seconds
Pyjo.IOLoop.stream(id).timeout = 300

subprocess(child: Child | None = None, parent: Parent | None = None) → Pyjo.IOLoop.Subprocess.object

subprocess = Pyjo.IOLoop.subprocess()
subprocess = loop.subprocess()
subprocess = loop.subprocess(lambda subprocess: ..., lambda subprocess, err, *results: ...)

Build Pyjo.IOLoop.Subprocess object to perform computationally expensive operations in subprocesses, without blocking the event loop. Callbacks will be passed along to Pyjo.IOLoop.Subprocess.Pyjo_IOLoop_Subprocess.run().

# Operation that would block the event loop for 5 seconds
Pyjo.IOLoop.subprocess().run_p(blocking).then(lambda *results: print(f'I {results[0]} {results[1]}!')).catch(
    lambda err: print(f'Subprocess error: {err}')
)

timer(after: float, cb: TimerCallback) → str

timer(after: float) → Callable[[TimerCallback], str]

id = Pyjo.IOLoop.timer(3, lambda loop: ...)
id = loop.timer(0, lambda loop: ...)
id = loop.timer(0.25, lambda loop: ...)

Create a new timer, invoking the callback after a given amount of time in seconds.

# Perform operation in 5 seconds
Pyjo.IOLoop.timer(5, lambda loop: ...)

Pyjo.IOLoop.is_running() → bool

boolean = Pyjo.IOLoop.is_running()

Check if the singleton is running.

Pyjo.IOLoop.singleton: Pyjo_IOLoop = <Pyjo.IOLoop.Pyjo_IOLoop object>

loop = Pyjo.IOLoop.singleton

The global Pyjo.IOLoop singleton, as Pyjo_IOLoop.singleton() gives it, the one of the thread.