IQ.Lvbs/pycapnp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate the KJ event loop into Python's asyncio event loop (#310)

Browse Source 
* Integrate the KJ event loop into Python's asyncio event loop

Fix #256

This PR attempts to remove the slow and expensive polling behavior for asyncio
in favor of proper linking of the KJ event loop to the asyncio event loop.

* Don't memcopy buffer

* Improve promise cancellation and prepare for timer implementation

* Add attribution for asyncProvider.cpp

* Implement timeout

* Cleanup

* First round of simplifications

* Add more a_wait functions and a shutdown function

* Fix edge-cases with loop shutdown

* Clean up calculator examples

* Cleanup

* Cleanup

* Reformat

* Fix warnings

* Reformat again

* Compatibility with macos

* Inline the asyncio loop in some places where this is feasible

* Add todo

* Fix

* Remove synchronous wait

* Wrap fd listening callbacks in a class

* Remove poll_forever

* Remove the thread-local/thread-global optimization

This will not matter much soon anyway, and simplifies things

* Share promise code by using fused types

* Improve refcounting of python objects in promises

We replace many instances of PyObject* by Own<PyRefCounter> for more automatic
reference management.

* Code wrapPyFunc in a similar way to wrapPyFuncNoArg

* Refactor capabilityHelper, fix several memory bugs for promises and add __await__

* Improve promise ownership, reduce memory leaks

Promise wrappers now hold a Own<Promise<Own<PyRefCounter>>> object. This might
seem like excessive nesting of objects (which to some degree it is, but with
good reason):
- The outer Own is needed because Cython cannot allocate objects without a
  nullary constructor on the stack (Promise doesn't have a nullary constructor).
  Additionally, I believe it would be difficult or impossible to detect when a
  promise is cancelled/moved if we use a bare Promise.
- Every promise returns a Owned PyRefCounter. PyRefCounter makes sure that a
  reference to the returned object keeps existing until the promise is fulfilled
  or cancelled. Previously, this was attempted using attach, which is redundant
  and makes reasoning about PyINCREF and PyDECREF very difficult.
- Because a promise holds a Own<Promise<...>>, when we perform any kind of
  action on that promise (a_wait, then, ...), we have to explicitly move() the
  ownership around. This will leave the original promise with a NULL-pointer,
  which we can easily detect as a cancelled promise.

Promises now only hold references to their 'parents' when strictly needed. This
should reduce memory pressure.

* Simplify and test the promise joining functionality

* Attach forgotten parent

* Catch exceptions in add_reader and friends

* Further cleanup of memory leaks

* Get rid of a_wait() in examples

* Cancel all fd read operations when the python asyncio loop is closed

* Formatting

* Remove support for capnp < 7000

* Bring asyncProvider.cpp more in line with upstream async-io-unix.c++

It was originally copied from the nodejs implementation, which in turn copied
from async-io-unix.c++. But that copy is pretty old.

* Fix a bug that caused file descriptors to never be closed

* Implement AsyncIoStream based on Python transports and protocols

* Get rid of asyncProvider

All asyncio now goes through _AsyncIoStream

* Formatting

* Add __dict__ to  PyAsyncIoStreamProtocol for python 3.7

* Reintroduce strange ipv4/ipv6 selection code to make ci happy

* Extra pause_reading()

* Work around more python bugs

* Be careful to only close transport when this is still possible

* Move pause_reading() workaround
This commit is contained in:
Lasse Blaauwbroek
2023-06-06 20:08:15 +02:00
committed by GitHub
parent ed894304a3
commit d32854eb00
25 changed files with 1060 additions and 1263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
examples/async_calculator_client.py
View File

@@ -2,7 +2,6 @@
import argparse
import asyncio
import socket
import capnp
import calculator_capnp
@@ -24,19 +23,6 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server):
return pow(params[0], params[1])
async def myreader(client, reader):
while True:
data = await reader.read(4096)
client.write(data)
async def mywriter(client, writer):
while True:
data = await client.read(4096)
writer.write(data.tobytes())
await writer.drain()
def parse_args():
parser = argparse.ArgumentParser(
usage="Connects to the Calculator server \
@@ -48,27 +34,9 @@ at the given address and does some RPCs"
async def main(host):
host = host.split(":")
addr = host[0]
port = host[1]
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
reader, writer = await asyncio.open_connection(
addr, port, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
reader, writer = await asyncio.open_connection(
addr, port, family=socket.AF_INET6
)
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient()
# Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)]
asyncio.gather(*coroutines, return_exceptions=True)
host, port = parse_args().host.split(":")
connection = await capnp.AsyncIoStream.create_connection(host=host, port=port)
client = capnp.TwoPartyClient(connection)
# Bootstrap the Calculator interface
calculator = client.bootstrap().cast_as(calculator_capnp.Calculator)
@@ -106,7 +74,7 @@ async def main(host):
# Now that we've sent all the requests, wait for the response. Until this
# point, we haven't waited at all!
response = await read_promise.a_wait()
response = await read_promise
assert response.value == 123
print("PASS")
@@ -144,7 +112,7 @@ async def main(host):
eval_promise = request.send()
read_promise = eval_promise.value.read()
response = await read_promise.a_wait()
response = await read_promise
assert response.value == 101
print("PASS")
@@ -208,8 +176,8 @@ async def main(host):
add_5_promise = add_5_request.send().value.read()
# Now wait for the results.
assert (await add_3_promise.a_wait()).value == 27
assert (await add_5_promise.a_wait()).value == 29
assert (await add_3_promise).value == 27
assert (await add_5_promise).value == 29
print("PASS")
@@ -290,8 +258,8 @@ async def main(host):
g_eval_promise = g_eval_request.send().value.read()
# Wait for the results.
assert (await f_eval_promise.a_wait()).value == 1234
assert (await g_eval_promise.a_wait()).value == 4244
assert (await f_eval_promise).value == 1234
assert (await g_eval_promise).value == 4244
print("PASS")
@@ -329,7 +297,7 @@ async def main(host):
add_params[1].literal = 5
# Send the request and wait.
response = await request.send().value.read().a_wait()
response = await request.send().value.read()
assert response.value == 512
print("PASS")

84
examples/async_calculator_server.py
View File

@@ -3,7 +3,6 @@
import argparse
import asyncio
import logging
import socket
import capnp
import calculator_capnp
@@ -13,60 +12,6 @@ logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class Server:
async def myreader(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.reader.read(4096), timeout=0.1)
except asyncio.TimeoutError:
logger.debug("myreader timeout.")
continue
except Exception as err:
logger.error("Unknown myreader err: %s", err)
return False
await self.server.write(data)
logger.debug("myreader done.")
return True
async def mywriter(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.server.read(4096), timeout=0.1)
self.writer.write(data.tobytes())
except asyncio.TimeoutError:
logger.debug("mywriter timeout.")
continue
except Exception as err:
logger.error("Unknown mywriter err: %s", err)
return False
logger.debug("mywriter done.")
return True
async def myserver(self, reader, writer):
# Start TwoPartyServer using TwoWayPipe (only requires bootstrap)
self.server = capnp.TwoPartyServer(bootstrap=CalculatorImpl())
self.reader = reader
self.writer = writer
self.retry = True
# Assemble reader and writer tasks, run in the background
coroutines = [self.myreader(), self.mywriter()]
tasks = asyncio.gather(*coroutines, return_exceptions=True)
while True:
self.server.poll_once()
# Check to see if reader has been sent an eof (disconnect)
if self.reader.at_eof():
self.retry = False
break
await asyncio.sleep(0.01)
# Make wait for reader/writer to finish (prevent possible resource leaks)
await tasks
def read_value(value):
"""Helper function to asynchronously call read() on a Calculator::Value and
return a promise for the result. (In the future, the generated code might
@@ -180,6 +125,11 @@ class CalculatorImpl(calculator_capnp.Calculator.Server):
return OperatorImpl(op)
async def new_connection(stream):
server = capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl())
await server.on_disconnect()
def parse_args():
parser = argparse.ArgumentParser(
usage="""Runs the server bound to the\
@@ -191,29 +141,9 @@ given address/port ADDRESS. """
return parser.parse_args()
async def new_connection(reader, writer):
server = Server()
await server.myserver(reader, writer)
async def main():
address = parse_args().address
host = address.split(":")
addr = host[0]
port = host[1]
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
server = await asyncio.start_server(
new_connection, addr, port, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
server = await asyncio.start_server(
new_connection, addr, port, family=socket.AF_INET6
)
host, port = parse_args().address.split(":")
server = await capnp.AsyncIoStream.create_server(new_connection, host, port)
async with server:
await server.serve_forever()

60
examples/async_client.py
View File

@@ -4,13 +4,9 @@ import asyncio
import argparse
import time
import capnp
import socket
import thread_capnp
capnp.remove_event_loop()
capnp.create_event_loop(threaded=True)
def parse_args():
parser = argparse.ArgumentParser(
@@ -29,61 +25,35 @@ class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server):
print("status: {}".format(time.time()))
async def myreader(client, reader):
while True:
data = await reader.read(4096)
client.write(data)
async def mywriter(client, writer):
while True:
data = await client.read(4096)
writer.write(data.tobytes())
async def background(cap):
subscriber = StatusSubscriber()
promise = cap.subscribeStatus(subscriber)
await promise.a_wait()
await cap.subscribeStatus(subscriber)
async def main(host):
host = host.split(":")
addr = host[0]
port = host[1]
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
reader, writer = await asyncio.open_connection(
addr, port, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
reader, writer = await asyncio.open_connection(
addr, port, family=socket.AF_INET6
)
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient()
host, port = host.split(":")
connection = await capnp.AsyncIoStream.create_connection(host=host, port=port)
client = capnp.TwoPartyClient(connection)
cap = client.bootstrap().cast_as(thread_capnp.Example)
# Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)]
asyncio.gather(*coroutines, return_exceptions=True)
# Start background task for subscriber
tasks = [background(cap)]
asyncio.gather(*tasks, return_exceptions=True)
asyncio.create_task(background(cap))
# Run blocking tasks
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
if __name__ == "__main__":
asyncio.run(main(parse_args().host))
args = parse_args()
asyncio.run(main(args.host))
# Test that we can run multiple asyncio loops in sequence. This is particularly tricky, because
# main contains a background task that we never cancel. The entire loop gets cleaned up anyways,
# and we can start a new loop.
asyncio.run(main(args.host))

76
examples/async_reconnecting_ssl_client.py
View File

@@ -4,16 +4,14 @@ import asyncio
import argparse
import os
import time
import socket
import ssl
import socket
import capnp
import thread_capnp
this_dir = os.path.dirname(os.path.abspath(__file__))
capnp.remove_event_loop()
capnp.create_event_loop(threaded=True)
def parse_args():
@@ -33,32 +31,10 @@ class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server):
print("status: {}".format(time.time()))
async def myreader(client, reader):
while True:
try:
# Must be a wait_for in order to give watch_connection a slot
# to try again
data = await asyncio.wait_for(reader.read(4096), timeout=1.0)
except asyncio.TimeoutError:
continue
client.write(data)
async def mywriter(client, writer):
while True:
try:
# Must be a wait_for in order to give watch_connection a slot
# to try again
data = await asyncio.wait_for(client.read(4096), timeout=1.0)
writer.write(data.tobytes())
except asyncio.TimeoutError:
continue
async def watch_connection(cap):
while True:
try:
await asyncio.wait_for(cap.alive().a_wait(), timeout=5)
await asyncio.wait_for(cap.alive(), timeout=5)
await asyncio.sleep(1)
except asyncio.TimeoutError:
print("Watch timeout!")
@@ -68,14 +44,11 @@ async def watch_connection(cap):
async def background(cap):
subscriber = StatusSubscriber()
promise = cap.subscribeStatus(subscriber)
await promise.a_wait()
await cap.subscribeStatus(subscriber)
async def main(host):
host = host.split(":")
addr = host[0]
port = host[1]
addr, port = host.split(":")
# Setup SSL context
ctx = ssl.create_default_context(
@@ -85,46 +58,33 @@ async def main(host):
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
reader, writer = await asyncio.open_connection(
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET
)
except OSError:
except Exception:
print("Try IPv6")
try:
reader, writer = await asyncio.open_connection(
addr, port, ssl=ctx, family=socket.AF_INET6
)
except OSError:
return False
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET6
)
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient()
client = capnp.TwoPartyClient(stream)
cap = client.bootstrap().cast_as(thread_capnp.Example)
# Start watcher to restart socket connection if it is lost
overalltasks = []
watcher = [watch_connection(cap)]
overalltasks.append(asyncio.gather(*watcher, return_exceptions=True))
# Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)]
overalltasks.append(asyncio.gather(*coroutines, return_exceptions=True))
# Start background task for subscriber
tasks = [background(cap)]
overalltasks.append(asyncio.gather(*tasks, return_exceptions=True))
# Start watcher to restart socket connection if it is lost and subscriber background task
background_tasks = asyncio.gather(
background(cap), watch_connection(cap), return_exceptions=True
)
# Run blocking tasks
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
for task in overalltasks:
task.cancel()
background_tasks.cancel()
return True

82
examples/async_server.py
View File

@@ -3,7 +3,6 @@
import argparse
import asyncio
import logging
import socket
import capnp
import thread_capnp
@@ -25,61 +24,12 @@ class ExampleImpl(thread_capnp.Example.Server):
)
def longRunning(self, **kwargs):
return capnp.getTimer().after_delay(1 * 10**9)
return capnp.getTimer().after_delay(11 * 10**8)
class Server:
async def myreader(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.reader.read(4096), timeout=0.1)
except asyncio.TimeoutError:
logger.debug("myreader timeout.")
continue
except Exception as err:
logger.error("Unknown myreader err: %s", err)
return False
await self.server.write(data)
logger.debug("myreader done.")
return True
async def mywriter(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.server.read(4096), timeout=0.1)
self.writer.write(data.tobytes())
except asyncio.TimeoutError:
logger.debug("mywriter timeout.")
continue
except Exception as err:
logger.error("Unknown mywriter err: %s", err)
return False
logger.debug("mywriter done.")
return True
async def myserver(self, reader, writer):
# Start TwoPartyServer using TwoWayPipe (only requires bootstrap)
self.server = capnp.TwoPartyServer(bootstrap=ExampleImpl())
self.reader = reader
self.writer = writer
self.retry = True
# Assemble reader and writer tasks, run in the background
coroutines = [self.myreader(), self.mywriter()]
tasks = asyncio.gather(*coroutines, return_exceptions=True)
while True:
self.server.poll_once()
# Check to see if reader has been sent an eof (disconnect)
if self.reader.at_eof():
self.retry = False
break
await asyncio.sleep(0.01)
# Make wait for reader/writer to finish (prevent possible resource leaks)
await tasks
async def new_connection(stream):
server = capnp.TwoPartyServer(stream, bootstrap=ExampleImpl())
await server.on_disconnect()
def parse_args():
@@ -93,29 +43,9 @@ given address/port ADDRESS. """
return parser.parse_args()
async def new_connection(reader, writer):
server = Server()
await server.myserver(reader, writer)
async def main():
address = parse_args().address
host = address.split(":")
addr = host[0]
port = host[1]
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
server = await asyncio.start_server(
new_connection, addr, port, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
server = await asyncio.start_server(
new_connection, addr, port, family=socket.AF_INET6
)
host, port = parse_args().address.split(":")
server = await capnp.AsyncIoStream.create_server(new_connection, host, port)
async with server:
await server.serve_forever()

44
examples/async_ssl_calculator_client.py
View File

@@ -3,8 +3,8 @@
import argparse
import asyncio
import os
import socket
import ssl
import socket
import capnp
import calculator_capnp
@@ -29,19 +29,6 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server):
return pow(params[0], params[1])
async def myreader(client, reader):
while True:
data = await reader.read(4096)
client.write(data)
async def mywriter(client, writer):
while True:
data = await client.read(4096)
writer.write(data.tobytes())
await writer.drain()
def parse_args():
parser = argparse.ArgumentParser(
usage="Connects to the Calculator server \
@@ -53,9 +40,7 @@ at the given address and does some RPCs"
async def main(host):
host = host.split(":")
addr = host[0]
port = host[1]
addr, port = host.split(":")
# Setup SSL context
ctx = ssl.create_default_context(
@@ -65,21 +50,16 @@ async def main(host):
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
reader, writer = await asyncio.open_connection(
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
reader, writer = await asyncio.open_connection(
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET6
)
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient()
# Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)]
asyncio.gather(*coroutines, return_exceptions=True)
client = capnp.TwoPartyClient(stream)
# Bootstrap the Calculator interface
calculator = client.bootstrap().cast_as(calculator_capnp.Calculator)
@@ -117,7 +97,7 @@ async def main(host):
# Now that we've sent all the requests, wait for the response. Until this
# point, we haven't waited at all!
response = await read_promise.a_wait()
response = await read_promise
assert response.value == 123
print("PASS")
@@ -155,7 +135,7 @@ async def main(host):
eval_promise = request.send()
read_promise = eval_promise.value.read()
response = await read_promise.a_wait()
response = await read_promise
assert response.value == 101
print("PASS")
@@ -219,8 +199,8 @@ async def main(host):
add_5_promise = add_5_request.send().value.read()
# Now wait for the results.
assert (await add_3_promise.a_wait()).value == 27
assert (await add_5_promise.a_wait()).value == 29
assert (await add_3_promise).value == 27
assert (await add_5_promise).value == 29
print("PASS")
@@ -301,8 +281,8 @@ async def main(host):
g_eval_promise = g_eval_request.send().value.read()
# Wait for the results.
assert (await f_eval_promise.a_wait()).value == 1234
assert (await g_eval_promise.a_wait()).value == 4244
assert (await f_eval_promise).value == 1234
assert (await g_eval_promise).value == 4244
print("PASS")
@@ -340,7 +320,7 @@ async def main(host):
add_params[1].literal = 5
# Send the request and wait.
response = await request.send().value.read().a_wait()
response = await request.send().value.read()
assert response.value == 512
print("PASS")

75
examples/async_ssl_calculator_server.py
View File

@@ -4,8 +4,8 @@ import argparse
import asyncio
import logging
import os
import socket
import ssl
import socket
import capnp
import calculator_capnp
@@ -17,60 +17,6 @@ logger.setLevel(logging.DEBUG)
this_dir = os.path.dirname(os.path.abspath(__file__))
class Server:
async def myreader(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.reader.read(4096), timeout=0.1)
except asyncio.TimeoutError:
logger.debug("myreader timeout.")
continue
except Exception as err:
logger.error("Unknown myreader err: %s", err)
return False
await self.server.write(data)
logger.debug("myreader done.")
return True
async def mywriter(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.server.read(4096), timeout=0.1)
self.writer.write(data.tobytes())
except asyncio.TimeoutError:
logger.debug("mywriter timeout.")
continue
except Exception as err:
logger.error("Unknown mywriter err: %s", err)
return False
logger.debug("mywriter done.")
return True
async def myserver(self, reader, writer):
# Start TwoPartyServer using TwoWayPipe (only requires bootstrap)
self.server = capnp.TwoPartyServer(bootstrap=CalculatorImpl())
self.reader = reader
self.writer = writer
self.retry = True
# Assemble reader and writer tasks, run in the background
coroutines = [self.myreader(), self.mywriter()]
tasks = asyncio.gather(*coroutines, return_exceptions=True)
while True:
self.server.poll_once()
# Check to see if reader has been sent an eof (disconnect)
if self.reader.at_eof():
self.retry = False
break
await asyncio.sleep(0.01)
# Make wait for reader/writer to finish (prevent possible resource leaks)
await tasks
def read_value(value):
"""Helper function to asynchronously call read() on a Calculator::Value and
return a promise for the result. (In the future, the generated code might
@@ -195,16 +141,13 @@ given address/port ADDRESS. """
return parser.parse_args()
async def new_connection(reader, writer):
server = Server()
await server.myserver(reader, writer)
async def new_connection(stream):
server = capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl())
await server.on_disconnect()
async def main():
address = parse_args().address
host = address.split(":")
addr = host[0]
port = host[1]
host, port = parse_args().address.split(":")
# Setup SSL context
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
@@ -216,13 +159,13 @@ async def main():
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
server = await asyncio.start_server(
new_connection, addr, port, ssl=ctx, family=socket.AF_INET
server = await capnp.AsyncIoStream.create_server(
new_connection, host, port, ssl=ctx, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
server = await asyncio.start_server(
new_connection, addr, port, ssl=ctx, family=socket.AF_INET6
server = await capnp.AsyncIoStream.create_server(
new_connection, host, port, ssl=ctx, family=socket.AF_INET6
)
async with server:

42
examples/async_ssl_client.py
View File

@@ -3,9 +3,9 @@
import argparse
import asyncio
import os
import socket
import ssl
import time
import socket
import capnp
import thread_capnp
@@ -30,29 +30,13 @@ class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server):
print("status: {}".format(time.time()))
async def myreader(client, reader):
while True:
data = await reader.read(4096)
client.write(data)
async def mywriter(client, writer):
while True:
data = await client.read(4096)
writer.write(data.tobytes())
await writer.drain()
async def background(cap):
subscriber = StatusSubscriber()
promise = cap.subscribeStatus(subscriber)
await promise.a_wait()
await cap.subscribeStatus(subscriber)
async def main(host):
host = host.split(":")
addr = host[0]
port = host[1]
addr, port = host.split(":")
# Setup SSL context
ctx = ssl.create_default_context(
@@ -62,34 +46,28 @@ async def main(host):
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
reader, writer = await asyncio.open_connection(
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
reader, writer = await asyncio.open_connection(
stream = await capnp.AsyncIoStream.create_connection(
addr, port, ssl=ctx, family=socket.AF_INET6
)
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient()
client = capnp.TwoPartyClient(stream)
cap = client.bootstrap().cast_as(thread_capnp.Example)
# Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)]
asyncio.gather(*coroutines, return_exceptions=True)
# Start background task for subscriber
tasks = [background(cap)]
asyncio.gather(*tasks, return_exceptions=True)
asyncio.create_task(background(cap))
# Run blocking tasks
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))
await cap.longRunning().a_wait()
await cap.longRunning()
print("main: {}".format(time.time()))

88
examples/async_ssl_server.py
View File

@@ -4,8 +4,8 @@ import argparse
import asyncio
import logging
import os
import socket
import ssl
import socket
import capnp
import thread_capnp
@@ -35,81 +35,21 @@ class ExampleImpl(thread_capnp.Example.Server):
return True
class Server:
async def myreader(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.reader.read(4096), timeout=0.1)
except asyncio.TimeoutError:
logger.debug("myreader timeout.")
continue
except Exception as err:
logger.error("Unknown myreader err: %s", err)
return False
await self.server.write(data)
logger.debug("myreader done.")
return True
async def mywriter(self):
while self.retry:
try:
# Must be a wait_for so we don't block on read()
data = await asyncio.wait_for(self.server.read(4096), timeout=0.1)
self.writer.write(data.tobytes())
except asyncio.TimeoutError:
logger.debug("mywriter timeout.")
continue
except Exception as err:
logger.error("Unknown mywriter err: %s", err)
return False
logger.debug("mywriter done.")
return True
async def myserver(self, reader, writer):
# Start TwoPartyServer using TwoWayPipe (only requires bootstrap)
self.server = capnp.TwoPartyServer(bootstrap=ExampleImpl())
self.reader = reader
self.writer = writer
self.retry = True
# Assemble reader and writer tasks, run in the background
coroutines = [self.myreader(), self.mywriter()]
tasks = asyncio.gather(*coroutines, return_exceptions=True)
while True:
self.server.poll_once()
# Check to see if reader has been sent an eof (disconnect)
if self.reader.at_eof():
self.retry = False
break
await asyncio.sleep(0.01)
# Make wait for reader/writer to finish (prevent possible resource leaks)
await tasks
async def new_connection(reader, writer):
server = Server()
await server.myserver(reader, writer)
async def new_connection(stream):
server = capnp.TwoPartyServer(stream, bootstrap=ExampleImpl())
await server.on_disconnect()
def parse_args():
parser = argparse.ArgumentParser(
usage="""Runs the server bound to the\
given address/port ADDRESS. """
usage="""Runs the server bound to the given address/port ADDRESS. """
)
parser.add_argument("address", help="ADDRESS:PORT")
return parser.parse_args()
async def main():
address = parse_args().address
host = address.split(":")
addr = host[0]
port = host[1]
host, port = parse_args().address.split(":")
# Setup SSL context
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
@@ -121,21 +61,13 @@ async def main():
# Handle both IPv4 and IPv6 cases
try:
print("Try IPv4")
server = await asyncio.start_server(
new_connection,
addr,
port,
ssl=ctx,
family=socket.AF_INET,
server = await capnp.AsyncIoStream.create_server(
new_connection, host, port, ssl=ctx, family=socket.AF_INET
)
except Exception:
print("Try IPv6")
server = await asyncio.start_server(
new_connection,
addr,
port,
ssl=ctx,
family=socket.AF_INET6,
server = await capnp.AsyncIoStream.create_server(
new_connection, host, port, ssl=ctx, family=socket.AF_INET6
)
async with server:

3
examples/thread_client.py
View File

@@ -7,9 +7,6 @@ import capnp
import thread_capnp
capnp.remove_event_loop()
capnp.create_event_loop(threaded=True)
def parse_args():
parser = argparse.ArgumentParser(

5
examples/thread_server.py
View File

@@ -2,7 +2,6 @@
import argparse
import capnp
import time
import thread_capnp
@@ -38,9 +37,7 @@ def main():
address = parse_args().address
server = capnp.TwoPartyServer(address, bootstrap=ExampleImpl())
while True:
server.poll_once()
time.sleep(0.001)
server.run_forever()
if __name__ == "__main__":