### Build Documentation Locally

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Install documentation dependencies and build the HTML documentation. After installation, navigate to the docs directory and run 'make html'.

```text
cd path/to/project/checkout/
pip install -r docs-requirements.txt
cd docs
make html
```

--------------------------------

### Install and Run Pre-commit Hooks

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Install pre-commit to automatically format code before committing. This ensures consistent code style across the project.

```text
pip install -U pre-commit
pre-commit
```

--------------------------------

### NTP Client Example Configuration

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Notes on configuring an NTP client example for IPv6 support. This involves changing the address family and hostname.

```python
# If you want to use IPv6, then:
# - replace AF_INET with AF_INET6 everywhere
# - use the hostname "2.pool.ntp.org"

```

--------------------------------

### SSL/TLS Connection Setup

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Imports necessary libraries for SSL/TLS operations, including `socket`, `ssl`, and `threading`. This snippet serves as a starting point for demonstrating an issue related to SSL/TLS handling in PyPy.

```python
# This demonstrates a PyPy bug:
#   https://bitbucket.org/pypy/pypy/issues/2578/

import socket
import ssl
import threading
```

--------------------------------

### Set Up Development Environment

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Installs Trio and its test dependencies within a virtual environment. Ensure you activate the virtual environment before running commands.

```shell
cd path/to/trio/checkout/
python -m venv .venv
source .venv/bin/activate
pip install -e .
pip install -r test-requirements.txt
```

--------------------------------

### Trio from_thread example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates spawning a child thread and using a memory channel to send messages between the thread and a Trio task. This example requires the 'reference-core/from-thread-example.py' file.

```python
import trio

async def child_task(name, channel):
    print(f"Child task {name} started")
    await channel.send(f"Hello from {name}!")
    print(f"Child task {name} finished")

def parent_task():
    async def main():
        # Create a memory channel for communication
        channel = trio.MemoryChannel(str)

        # Spawn a child task
        async with trio.open_nursery() as nursery:
            nursery.start_soon(child_task, "Task 1", channel)
            nursery.start_soon(child_task, "Task 2", channel)

        # Receive messages from child tasks
        print("Receiving messages:")
        async for message in channel:
            print(f"Received: {message}")

    # Run the Trio main function
    trio.run(main)

if __name__ == "__main__":
    parent_task()
```

--------------------------------

### Starting a background task with nursery.start

Source: https://github.com/python-trio/trio/blob/main/docs/source/history.rst

Use nursery.start to initiate a long-running task in the background. It returns the result of the task after it has initialized and is ready. This is useful for starting services like servers.

```python
port = await nursery.start(http_server_on_random_open_port)
```

--------------------------------

### serve_tcp

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Starts a TCP server that listens for incoming connections.

```APIDOC
## serve_tcp

### Description
Starts a TCP server that listens for incoming connections.

### Parameters

This function is documented via its signature and type hints in the Trio library.
```

--------------------------------

### Async Generator Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates how to create and consume an asynchronous generator that yields values with a delay. Requires `trio` to be imported.

```python
async def range_slowly(*args):
    """Like range(), but adds a 1-second sleep before each value."""
    for value in range(*args):
        await trio.sleep(1)
        yield value

async def use_it():
    async for value in range_slowly(10):
        print(value)

trio.run(use_it)
```

--------------------------------

### Print Task Tree Example

Source: https://github.com/python-trio/trio/wiki/notes-to-self

A placeholder comment indicating a potential alternative for printing task trees, suggesting the 'tree-format' PyPI package.

```python
# NOTE:
# possibly it would be easier to use https://pypi.org/project/tree-format/

```

--------------------------------

### Trio Echo Server Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

This is the complete echo server code. It listens for incoming TCP connections and echoes any data received back to the client. It uses trio.serve_tcp for simplified server setup.

```python
import trio

PORT = 12345

async def main():
    await trio.serve_tcp(echo_server, PORT)

async def echo_server(server_stream):
    try:
        async for data in server_stream:
            await server_stream.send_all(data)
    finally:
        print("Client disconnected")

trio.run(main)

```

--------------------------------

### Basic Trio Execution Example

Source: https://github.com/python-trio/trio/wiki/notes-to-self

A simple example demonstrating how to run an asynchronous function using `trio.run`. The function `foo` prints a message and returns a value, which is then printed to standard output.

```python
import trio


async def foo():
    print("in foo!")
    return 3


print("running!")
print(trio.run(foo))

```

--------------------------------

### Starting a Trio Guest Run

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

This snippet shows how to start a Trio run as a guest within another event loop. It uses `start_guest_run` and a future to capture the outcome.

```python
def done_callback(trio_main_outcome):
    done_fut.set_result(trio_main_outcome)

trio.lowlevel.start_guest_run(
    trio_main,
    run_sync_soon_threadsafe=run_sync_soon_threadsafe,
    done_callback=done_callback,
)

# Wait for the guest run to finish
trio_main_outcome = await done_fut
# Pass through the return value or exception from the guest run
return trio_main_outcome.unwrap()
```

--------------------------------

### Example output of concurrent tasks

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

This is a sample output demonstrating the execution flow and timing of concurrent tasks spawned within a Trio nursery. It illustrates that tasks can run and complete in parallel.

```none
parent: started!
parent: spawning child1...
parent: spawning child2...
parent: waiting for children to finish...
  child2: started! sleeping now...
  child1: started! sleeping now...
    [... 1 second passes ...]
  child1: exiting!
  child2: exiting!
parent: all done!
```

--------------------------------

### Trio Lock Fairness Demonstration

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates the fairness of Trio's Lock primitive by showing two tasks alternating access. Ensure you have Trio installed to run this example.

```python
# fairness-demo.py

import trio

async def loopy_child(number, lock):
    while True:
        async with lock:
            print(f"Child {number} has the lock!")
            await trio.sleep(0.5)

async def main():
    async with trio.open_nursery() as nursery:
        lock = trio.Lock()
        nursery.start_soon(loopy_child, 1, lock)
        nursery.start_soon(loopy_child, 2, lock)

trio.run(main)
```

--------------------------------

### Interactive Trio Interpreter Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/history.rst

Demonstrates how to use the interactive interpreter for Trio. Allows direct use of 'await' without 'trio.run()'.

```console
$ python -m trio
Trio 0.21.0+dev, Python 3.10.6
Use "await" directly instead of "trio.run()".
Type "help", "copyright", "credits" or "license" for more information.
>>> import trio
>>> await trio.sleep(1); print("hi")  # prints after one second
hi
```

--------------------------------

### Simple Memory Channel Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates basic usage of memory channels for sending and receiving messages between tasks. This example will hang after printing messages as the consumer waits indefinitely for more data.

```python
import trio

async def producer(send_channel):
    for i in range(3):
        await send_channel.send(f"message {i}")

async def consumer(receive_channel):
    async for value in receive_channel:
        print(f"got value {value!r}")

async def main():
    receive_channel, send_channel = trio.open_memory_channel(1)
    async with trio.open_nursery() as nursery:
        await nursery.spawn(producer, send_channel)
        await nursery.spawn(consumer, receive_channel)

trio.run(main)
```

--------------------------------

### Spawning tasks using a passed-in nursery

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates how a task can open a nursery and pass it to another function to start child tasks. This allows tasks to start siblings of themselves.

```python
async def new_connection_listener(handler, nursery):
    while True:
        conn = await get_new_connection()
        nursery.start_soon(handler, conn)

async def server(handler):
    async with trio.open_nursery() as nursery:
        nursery.start_soon(new_connection_listener, handler, nursery)
```

--------------------------------

### Demonstrate contextvars usage

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

This example illustrates the basic usage of Python's `contextvars` module for managing context-local state, which is useful for scenarios like request tracking in concurrent servers.

```python
import contextvars
import trio

request_id = contextvars.ContextVar('request_id', default='<no request>')

def log(message):
    print(f"[{request_id.get()}] {message}")

async def handle_request(req_id):
    request_id.set(req_id)
    log("started")
    await trio.sleep(1)
    log("finished")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(handle_request, 'A')
        nursery.start_soon(handle_request, 'B')

trio.run(main)
```

--------------------------------

### Running a Trio Application

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

To start a Trio application, you must call the `trio.run` function. This is the entry point for all Trio programs.

```python
import trio

async def main():
    print("Hello from Trio!")

trio.run(main)
```

--------------------------------

### serve_ssl_over_tcp

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Starts a TCP server that listens for incoming SSL/TLS connections.

```APIDOC
## serve_ssl_over_tcp

### Description
Starts a TCP server that listens for incoming SSL/TLS connections.

### Parameters

This function is documented via its signature and type hints in the Trio library.
```

--------------------------------

### Minimal Trio Guest Run within Asyncio

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

Demonstrates the basic integration of a Trio program within an existing asyncio event loop using `trio.lowlevel.start_guest_run`. This example shows how to provide the required `run_sync_soon_threadsafe` and `done_callback` functions for seamless operation.

```python
import asyncio
import trio

# A tiny Trio program
async def trio_main():
    for _ in range(5):
        print("Hello from Trio!")
        # This is inside Trio, so we have to use Trio APIs
        await trio.sleep(1)
    return "trio done!"

# The code to run it as a guest inside asyncio
async def asyncio_main():
    asyncio_loop = asyncio.get_running_loop()

    def run_sync_soon_threadsafe(fn):
        asyncio_loop.call_soon_threadsafe(fn)

    def done_callback(trio_main_outcome):
        print(f"Trio program ended with: {trio_main_outcome}")

    # This is where the magic happens:
    trio.lowlevel.start_guest_run(
        trio_main,
        run_sync_soon_threadsafe=run_sync_soon_threadsafe,
        done_callback=done_callback,
    )

    # Let the host loop run for a while to give trio_main time to
    # finish. (WARNING: This is a hack. See below for better
    # approaches.)
    #
    # This function is in asyncio, so we have to use asyncio APIs.
    await asyncio.sleep(10)

asyncio.run(asyncio_main())
```

--------------------------------

### Trio concurrent task execution

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

An example demonstrating how to run multiple asynchronous functions concurrently using Trio's task management.

```python
import trio

async def child1():
    print("child1: starting")
    await trio.sleep(1)
    print("child1: exiting")

async def child2():
    print("child2: starting")
    await trio.sleep(2)
    print("child2: exiting")

async def parent():
    print("parent: starting")
    async with trio.open_nursery() as nursery:
        nursery.start_soon(child1)
        nursery.start_soon(child2)
    print("parent: exiting")

trio.run(parent)

```

--------------------------------

### Memory Channel Cloning Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates using MemorySendChannel.clone and MemoryReceiveChannel.clone to create independent channel endpoints. The underlying channel remains open until all clones are closed.

```python
# Also works, but is more finicky:
send_channel, receive_channel = trio.open_memory_channel(0)
nursery.start_soon(producer, "A", send_channel.clone())
nursery.start_soon(producer, "B", send_channel)
nursery.start_soon(consumer, "X", receive_channel.clone())
nursery.start_soon(consumer, "Y", receive_channel)
```

--------------------------------

### Running a Shell Command with Quoting

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Example of running a command using the shell, demonstrating the need for argument quoting to prevent shell injection vulnerabilities. Use shlex.quote for safety.

```python
Process("ls | grep some_string", shell=True)
```

--------------------------------

### Trio Execution Log Output (Parent Task)

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

Examine the log output as the parent task starts, spawns child tasks, and waits for them. This shows the sequence of task spawning and scheduling.

```none
>>> about to run one step of task: __main__.parent
parent: started!
parent: spawning child1...
### new task spawned: __main__.child1
### task scheduled: __main__.child1
parent: spawning child2...
### new task spawned: __main__.child2
### task scheduled: __main__.child2
parent: waiting for children to finish...
<<< task step finished: __main__.parent
```

--------------------------------

### Install Pre-commit for Git Hooks

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Enable pre-commit to automatically run before each git commit. This helps catch formatting issues early in the development process.

```text
pre-commit install
```

--------------------------------

### Start Child Tasks in a Nursery

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Once a nursery is open, use `nursery.start_soon()` to schedule new child tasks to run concurrently. The parent task will not exit until all children in the nursery have completed.

```python
async def child():
    ...

async def parent():
    async with trio.open_nursery() as nursery:
        # Make two concurrent calls to child()
        nursery.start_soon(child)
        nursery.start_soon(child)
```

--------------------------------

### Trio Echo Client Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

This is the complete echo client code. It requires a running TCP server to function correctly. The client connects to a server, sends data, and receives responses.

```python
import trio

PORT = 12345

async def parent():
    async with await trio.open_tcp_stream("127.0.0.1", PORT) as client_stream:
        await trio.open_nursery()
        async with nursery:
            nursery.start_soon(sender, client_stream)
            nursery.start_soon(receiver, client_stream)

async def sender(client_stream):
    for i in range(10):
        await client_stream.send_all(f"message {i}\n".encode())
        await trio.sleep(1)

async def receiver(client_stream):
    async for data in client_stream:
        print(f"Received: {data.decode()!r}")

trio.run(parent)

```

--------------------------------

### Bounded Memory Channel Backpressure Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates using a memory channel with a buffer size of 0. This enforces backpressure, forcing producers to slow down when the buffer is full, matching the consumer's speed.

```python
import math
import trio

async def producer(name, channel):
    for i in range(100):
        await channel.send(f"Message {i} from {name}")
        await trio.sleep(0.1)

async def consumer(name, channel):
    for _ in range(100):
        message = await channel.receive()
        print(f"Received {message} by {name}")
        await trio.sleep(1)

async def main():
    send_channel, receive_channel = trio.open_memory_channel(0)
    async with trio.open_nursery() as nursery:
        nursery.start_soon(producer, "P1", send_channel)
        nursery.start_soon(consumer, "C1", receive_channel)

trio.run(main)
```

--------------------------------

### Spawning a Child Task with a Nursery

Source: https://github.com/python-trio/trio/blob/main/docs/source/design.rst

Demonstrates how to start a child task within a parent coroutine using a Trio nursery. This pattern ensures exceptions from child tasks are propagated.

```python
async def parent():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(child)
```

--------------------------------

### Run Multiple Looping Tasks Concurrently

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Demonstrates how to start multiple asynchronous tasks that run indefinitely using a loop. Includes a synchronous sleep to prevent high CPU usage and a checkpoint for task yielding.

```python
import time

import trio


async def loopy():
    try:
        while True:
            # synchronous sleep to avoid maxing out CPU
            time.sleep(0.01)  # noqa: ASYNC251
            await trio.lowlevel.checkpoint()
    except KeyboardInterrupt:
        print("KI!")


async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(loopy)
        nursery.start_soon(loopy)
        nursery.start_soon(loopy)

trio.run(main)
```

--------------------------------

### Thread Contextvars Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

Demonstrates how context variables are preserved and copied when using Trio's thread-related functions. Shows how to modify mutable objects within context variables from child threads.

```python
import trio
import contextvars

user_id = contextvars.ContextVar('user_id', default=None)
message = contextvars.ContextVar('message', default='')

def process_user(user_data):
    user_id.set(user_data['id'])
    message.set(user_data['message'])
    print(f"Processed user {user_id.get()} with message {message.get()} in a thread worker")

async def main():
    users = [
        {'id': 0, 'message': 'Hello 0'},
        {'id': 1, 'message': 'Hello 1'},
        {'id': 2, 'message': 'Hello 2'},
    ]

    # Use a mutable object (dict) to share state across threads
    shared_state = {}

    async with trio.open_nursery() as nursery:
        for user_data in users:
            # Pass the shared_state dictionary to the thread function
            await nursery.spawn(trio.to_thread.run_sync, process_user, user_data, shared_state=shared_state)

    # After threads complete, access modified shared_state
    for user_id_val, message_val in shared_state.items():
        print(f"New contextvar value from worker thread for user {user_id_val}: {message_val}")

trio.run(main)

```

--------------------------------

### Trio Echo Server Main Function

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

The main function starts the TCP server using trio.serve_tcp, which listens on the specified port and handles incoming connections by running the echo_server coroutine for each.

```python
async def main():
    await trio.serve_tcp(echo_server, PORT)

```

--------------------------------

### Trio Execution Log Output (Initial)

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

Observe the initial output from Trio when running with instrumentation. This includes events like run start, task spawning, scheduling, and I/O checks.

```none
$ python3 tutorial/tasks-with-trace.py
!!! run started
### new task spawned: <init>
### task scheduled: <init>
### doing a quick check for I/O
### finished I/O check (took 1.1122087016701698e-05 seconds)
>>> about to run one step of task: <init>
### new task spawned: <call soon task>
### task scheduled: <call soon task>
### new task spawned: __main__.parent
### task scheduled: __main__.parent
<<< task step finished: <init>
### doing a quick check for I/O
### finished I/O check (took 6.4980704337358475e-06 seconds)
```

--------------------------------

### Demonstrating Blocking Sleep in Trio

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

This example shows how using `time.sleep` instead of `trio.sleep` within Trio tasks can lead to unresponsive behavior, as it blocks the entire event loop. Observe the significant pauses between task outputs.

```none
parent: started!
parent: spawning child1...
parent: spawning child2...
parent: waiting for children to finish...
  child2 started! sleeping now...
    [... pauses for 1 second ...]
  child2 exiting!
  child1: started! sleeping now...
    [... pauses for 1 second ...]
  child1: exiting!
parent: all done!
```

--------------------------------

### FreeBSD Pipe Close Notification with kqueue

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Investigates pipe behavior on FreeBSD, specifically how closing the read end affects the write end's writability notification via `select` and `kqueue`. This example highlights potential issues with pipe handling in older FreeBSD versions.

```python
# This script completes correctly on macOS and FreeBSD 13.0-CURRENT, but hangs
# on FreeBSD 12.1. I'm told the fix will be backported to 12.2 (which is due
# out in October 2020).
#
# Upstream bug: https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=246350

import os
import select

r, w = os.pipe()

os.set_blocking(w, False)

print("filling pipe buffer")
try:
    while True:
        os.write(w, b"x")
except BlockingIOError:
    pass

_, wfds, _ = select.select([], [w], [], 0)
print("select() says the write pipe is", "writable" if w in wfds else "NOT writable")

kq = select.kqueue()
event = select.kevent(w, select.KQ_FILTER_WRITE, select.KQ_EV_ADD)
kq.control([event], 0)

print("closing read end of pipe")
os.close(r)

_, wfds, _ = select.select([], [w], [], 0)
print("select() says the write pipe is", "writable" if w in wfds else "NOT writable")

print("waiting for kqueue to report the write end is writable")
got = kq.control([], 1)
print("done!")
print(got)

```

--------------------------------

### Nested SSLStream for HTTPS Proxy Connections

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

This example shows how to handle SSL-on-top-of-SSL connections, typically for accessing HTTPS servers through an HTTPS proxy. It involves wrapping an existing SSLStream with another SSLStream to establish a secure connection through the proxy to the target website.

```python
# Get a raw SocketStream connection to the proxy:
s0 = await open_tcp_stream("proxy", 443)

# Set up SSL connection to proxy:
s1 = SSLStream(s0, proxy_ssl_context, server_hostname="proxy")
# Request a connection to the website
await s1.send_all(b"CONNECT website:443 / HTTP/1.0\r\n\r\n")
await check_CONNECT_response(s1)

# Set up SSL connection to the real website. Notice that s1 is
# already an SSLStream object, and here we're wrapping a second
# SSLStream object around it.
s2 = SSLStream(s1, website_ssl_context, server_hostname="website")
# Make our request
await s2.send_all(b"GET /index.html HTTP/1.0\r\n\r\n")
...
```

--------------------------------

### Parent Task Launching Broken Child Tasks

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-core.rst

An asynchronous function that uses a Trio nursery to start two child tasks, broken1 and broken2. This setup is designed to trigger an ExceptionGroup when both child tasks fail.

```python
async def parent():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(broken1)
        nursery.start_soon(broken2)
```

--------------------------------

### Serve Built Documentation Locally

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Serve the built HTML documentation using Python's http.server. This allows you to browse the documentation locally before deploying.

```bash
python -m http.server 8000 --bind 127.0.0.1 --directory build/html
```

--------------------------------

### Get service name from port

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

This snippet shows how to get a service name from a port number using getnameinfo, which is the recommended approach over the obsolete socket.getservbyport.

```python
_, service_name = await getnameinfo(('127.0.0.1', port), NI_NUMERICHOST)
```

--------------------------------

### Establish SSL/TLS Connection with Sockets

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Demonstrates setting up a secure client-server connection using SSL sockets. Requires certificate files for authentication.

```python
# client_sock, server_sock = socket.socketpair() 
listen_sock = socket.socket()
listen_sock.bind(("127.0.0.1", 0))
listen_sock.listen(1)
client_sock = socket.socket()
client_sock.connect(listen_sock.getsockname())
server_sock, _ = listen_sock.accept()

server_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
server_ctx.load_cert_chain("trio-test-1.pem")
server = server_ctx.wrap_socket(
    server_sock,
    server_side=True,
    suppress_ragged_eofs=False,
    do_handshake_on_connect=False,
)

client_ctx = ssl.create_default_context(cafile="trio-test-CA.pem")
client = client_ctx.wrap_socket(
    client_sock,
    server_hostname="trio-test-1.example.org",
    suppress_ragged_eofs=False,
    do_handshake_on_connect=False,
)

server_handshake_thread = threading.Thread(target=server.do_handshake)
server_handshake_thread.start()
client_handshake_thread = threading.Thread(target=client.do_handshake)
client_handshake_thread.start()

server_handshake_thread.join()
client_handshake_thread.join()

# Now we have two SSLSockets that have established an encrypted connection
# with each other

assert client.getpeercert() is not None
client.sendall(b"x")
assert server.recv(10) == b"x"

# A few different ways to make attempts to read/write the socket's fd return
# weird failures at the operating system level

# Attempting to send on a socket after shutdown should raise EPIPE or similar
server.shutdown(socket.SHUT_WR)

# Attempting to read/write to the fd after it's closed should raise EBADF
# os.close(server.fileno())

# Attempting to read/write to an fd opened with O_DIRECT raises EINVAL in most
# cases (unless you're very careful with alignment etc. which openssl isn't)
# os.dup2(os.open("/tmp/blah-example-file", os.O_RDWR | os.O_CREAT | os.O_DIRECT), server.fileno())

# Sending or receiving
server.sendall(b"hello")
# server.recv(10)

```

--------------------------------

### Parent function starting child tasks

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

Defines an asynchronous 'parent' function that uses a nursery to concurrently start 'child1' and 'child2' tasks. This demonstrates Trio's concurrency model.

```python
async def parent():
    print("parent: starting")
    async with trio.open_nursery() as nursery:
        nursery.start_soon(child1)
        nursery.start_soon(child2)
    print("parent: exiting")

```

--------------------------------

### checkpoint

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

Ensures that Trio's scheduler gets a chance to run. This is a fundamental operation for cooperative multitasking.

```APIDOC
## checkpoint()

### Description
Ensures that Trio's scheduler gets a chance to run. This is a fundamental operation for cooperative multitasking.
```

--------------------------------

### start_guest_run

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

Integrates Trio into an existing event loop, allowing Trio to run within a guest event loop context. It supports optional arguments for fine-tuning the integration, such as specifying a custom function for scheduling callbacks in the guest loop.

```APIDOC
## start_guest_run

### Description
Starts a Trio guest run within an existing event loop. This function is intended for advanced use cases where Trio needs to be embedded within another event loop.

### Parameters

#### Keyword Arguments

*   **run_sync_soon_not_threadsafe** (callable, optional) - A function that behaves like `loop.call_soon_threadsafe` but is intended to be called from the event loop's thread. If not provided, Trio will use the thread-safe version.

### Returns

This function does not return a value directly; it runs the Trio event loop until completion within the guest context.
```

--------------------------------

### Windows CMD.EXE Escaping Example

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Demonstrates nested escaping required for special characters in Windows CMD.EXE pipelines.

```sh
echo ^^^&x | find "x" | find "x"          # prints: &x
```

--------------------------------

### Trio Task Execution and Scheduling

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

This snippet demonstrates the execution of a task step, including when a task is about to run, its output, when it exits, and when the parent task is scheduled as a result.

```none
>>> about to run one step of task: __main__.child1
  child1: exiting!
### task scheduled: __main__.parent
### task exited: __main__.child1
<<< task step finished: __main__.child1

>>> about to run one step of task: __main__.child2
  child2 exiting!
### task exited: __main__.child2
<<< task step finished: __main__.child2
```

--------------------------------

### Integrating Trio with Qt using Guest Mode

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

Demonstrates how Trio code can interact with Qt objects and how Qt signals can trigger Trio's cancellation scopes within the same thread. This is useful for GUI applications.

```python
# Trio code can create Qt objects without any special ceremony...
my_cancel_button = QPushButton("Cancel")
# ...and Qt can call back to Trio just as easily
my_cancel_button.clicked.connect(my_cancel_scope.cancel)
```

--------------------------------

### Set and Wait on a Windows Waitable Timer with Trio

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Demonstrates how to create, set, and wait for a Windows waitable timer using Trio's asynchronous primitives. This involves converting Python datetimes to FILETIME and using `trio.hazmat.WaitForSingleObject`.

```python
async def main():
    h = kernel32.CreateWaitableTimerW(ffi.NULL, True, ffi.NULL)
    if not h:
        raise_winerror()
    print(h)

    SECONDS = 2

    wakeup = datetime.now(timezone.utc) + timedelta(seconds=SECONDS)
    wakeup_filetime = py_datetime_to_win_filetime(wakeup)
    wakeup_cffi = ffi.new("LARGE_INTEGER *")
    wakeup_cffi[0] = wakeup_filetime

    print(wakeup_filetime, wakeup_cffi)

    print(f"Sleeping for {SECONDS} seconds (until {wakeup})")

    if not kernel32.SetWaitableTimer(
        h,
        wakeup_cffi,
        0,
        ffi.NULL,
        ffi.NULL,
        False,
    ):
        raise_winerror()

    await trio.hazmat.WaitForSingleObject(h)

    print(f"Current FILETIME: {now_as_filetime()}")
    set_leap_seconds_enabled(False)
    print(f"Current FILETIME: {now_as_filetime()}")
    set_leap_seconds_enabled(True)
    print(f"Current FILETIME: {now_as_filetime}")
    set_leap_seconds_enabled(False)
    print(f"Current FILETIME: {now_as_filetime()}")

trio.run(main)
```

--------------------------------

### Using trio.sleep in an async function

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

An example of an async function that utilizes `trio.sleep` for asynchronous delays. This function must be run using `trio.run`.

```python
import trio

async def double_sleep(x):
    await trio.sleep(2 * x)

trio.run(double_sleep, 3)  # does nothing for 6 seconds then returns
```

--------------------------------

### Open a Process with Trio Low-level

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-lowlevel.rst

Use `trio.lowlevel.open_process` to spawn a new process. This function provides a low-level interface for process creation.

```python
await trio.lowlevel.open_process(cmd, ...) 
```

--------------------------------

### Start Numerous Tasks Concurrently

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Launches a specified number of Trio sleep tasks concurrently. The number of tasks is determined by a command-line argument.

```python
import sys

import trio

(COUNT_STR,) = sys.argv[1:]
COUNT = int(COUNT_STR)


async def main():
    async with trio.open_nursery() as nursery:
        for _ in range(COUNT):
            nursery.start_soon(trio.sleep, 1)

trio.run(main)
```

--------------------------------

### Wrapping Socket with wrap_socket (do_handshake=False)

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Demonstrates using ssl.SSLContext.wrap_socket with do_handshake_on_connect=False, requiring manual invocation of do_handshake.

```python
def wrap_socket_via_wrap_socket(ctx, sock, **kwargs):
    return ctx.wrap_socket(sock, do_handshake_on_connect=False, **kwargs)
```

--------------------------------

### Mark Lines for No Coverage

Source: https://github.com/python-trio/trio/blob/main/docs/source/contributing.rst

Example of using '# pragma: no cover' to exclude specific lines from coverage reporting when lack of coverage is intentional and not worth fixing.

```python
if ...:
    ...
else:  # pragma: no cover
    raise AssertionError("this can't happen!")
```

--------------------------------

### Opening a file asynchronously with trio.open_file

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Use `trio.open_file` to open a file by its name and get an asynchronous file object. This is suitable for direct file access operations.

```python
async with trio.open_file('my_file.txt', mode='w') as f:
    await f.write('Hello\n')
    await f.write('World\n')

```

--------------------------------

### Running an async function with trio.run

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

Illustrates how to use `trio.run` to execute a simple asynchronous function and retrieve its result.

```python
import trio

async def async_double(x):
    return 2 * x

trio.run(async_double, 3)  # returns 6
```

--------------------------------

### Reopen Pipe and FIFO with File Descriptors

Source: https://github.com/python-trio/trio/wiki/notes-to-self

Demonstrates reopening read and write ends of pipes and FIFOs using file descriptors. Shows how to set file descriptors to non-blocking mode and verify blocking behavior.

```Python
import os
import tempfile
import threading
import time


def check_reopen(r1, w):
    try:
        print("Reopening read end")
        r2 = os.open(f"/proc/self/fd/{r1}", os.O_RDONLY)

        print(f"r1 is {r1}, r2 is {r2}")

        print("checking they both can receive from w...")

        os.write(w, b"a")
        assert os.read(r1, 1) == b"a"

        os.write(w, b"b")
        assert os.read(r2, 1) == b"b"

        print("...ok")

        print("setting r2 to non-blocking")
        os.set_blocking(r2, False)

        print("os.get_blocking(r1) ==", os.get_blocking(r1))
        print("os.get_blocking(r2) ==", os.get_blocking(r2))

        # Check r2 is really truly non-blocking
        try:
            os.read(r2, 1)
        except BlockingIOError:
            print("r2 definitely seems to be in non-blocking mode")

        # Check that r1 is really truly still in blocking mode
        def sleep_then_write():
            time.sleep(1)
            os.write(w, b"c")

        threading.Thread(target=sleep_then_write, daemon=True).start()
        assert os.read(r1, 1) == b"c"
        print("r1 definitely seems to be in blocking mode")
    except Exception as exc:
        print(f"ERROR: {exc!r}")


print("-- testing anonymous pipe --")
check_reopen(*os.pipe())

print("-- testing FIFO --")
with tempfile.TemporaryDirectory() as tmpdir:
    fifo = tmpdir + "/" + "myfifo"
    os.mkfifo(fifo)
    # "A process can open a FIFO in nonblocking mode.  In this case, opening
    # for read-only will succeed even if no-one has opened on the write side
    # yet and opening for write-only will fail with ENXIO (no such device or
    # address) unless the other end has already been opened." -- Linux fifo(7)
    r = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)
    assert not os.get_blocking(r)
    os.set_blocking(r, True)
    assert os.get_blocking(r)
    w = os.open(fifo, os.O_WRONLY)
    check_reopen(r, w)

print("-- testing socketpair --")
import socket

rs, ws = socket.socketpair()
check_reopen(rs.fileno(), ws.fileno())

```

--------------------------------

### Get address info for a service name

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

This snippet demonstrates how to use getaddrinfo to resolve a service name, replacing the obsolete socket.getservbyname function. It's useful for network address resolution.

```python
await getaddrinfo(None, service_name)
```

--------------------------------

### Defining Regular and Async Functions

Source: https://github.com/python-trio/trio/blob/main/docs/source/tutorial.rst

Illustrates the syntax for defining both regular Python functions and asynchronous functions using 'async def'. Async functions require the 'await' keyword for invocation.

```python
def regular_double(x):
    return 2 * x

async def async_double(x):
    return 2 * x
```

--------------------------------

### trio.lowlevel.open_process

Source: https://github.com/python-trio/trio/blob/main/docs/source/reference-io.rst

Low-level API for spawning subprocesses, offering more flexibility than `trio.run_process`, such as spawning a child process that can outlive the parent.

```APIDOC
## trio.lowlevel.open_process

### Description
Provides a lower-level interface for spawning subprocesses, enabling advanced scenarios like orphaned processes.

### Method
`trio.lowlevel.open_process`

### Parameters
Accepts similar keyword arguments to `trio.run_process` and `subprocess.Popen`.
```