Proactor
IO

Unsettled API - An API for multithreaded IO. More...

Collaboration diagram for Proactor:

## Macros

#define | PN_MAX_ADDR
| Size of buffer that can hold the largest connection or listening address.

## Typedefs

typedef struct pn_netaddr_t | pn_netaddr_t
| Unsettled API - The network address of a proactor transport.
typedef struct pn_proactor_t | pn_proactor_t
| A harness for multithreaded IO.

## Functions

PNP_EXTERN int | pn_netaddr_str (const pn_netaddr_t *addr, char *buf, size_t size)
| Format a network address as a human-readable string in `buf`. More...
PNP_EXTERN const pn_netaddr_t * | pn_transport_local_addr (pn_transport_t *t)
| Get the local address of a transport. More...
PNP_EXTERN const pn_netaddr_t * | pn_transport_remote_addr (pn_transport_t *t)
| Get the local address of a transport. More...
PNP_EXTERN const pn_netaddr_t * | pn_listener_addr (pn_listener_t *l)
| Get the listening addresses of a listener. More...
PNP_EXTERN const pn_netaddr_t * | pn_netaddr_next (const pn_netaddr_t *na)
PNP_EXTERN const struct sockaddr * | pn_netaddr_sockaddr (const pn_netaddr_t *na)
| On POSIX or Windows, get the underlying `struct sockaddr`. More...
PNP_EXTERN size_t | pn_netaddr_socklen (const pn_netaddr_t *na)
| On POSIX or Windows, get the size of the underlying `struct sockaddr`. More...
PNP_EXTERN int | pn_netaddr_host_port (const pn_netaddr_t *na, char *host, size_t hlen, char *port, size_t plen)
| Get the host and port name from na as separate strings. More...
PNP_EXTERN int | pn_proactor_addr (char *addr, size_t size, const char *host, const char *port)
| Format a host:port address string for pn_proactor_connect2() or pn_proactor_listen2() More...
PNP_EXTERN pn_proactor_t * | pn_proactor (void)
| Create a proactor. More...
PNP_EXTERN void | pn_proactor_free (pn_proactor_t *proactor)
| Free the proactor. More...
PNP_EXTERN void | pn_proactor_connect2 (pn_proactor_t *proactor, pn_connection_t *connection, pn_transport_t *transport, const char *addr)
| Connect `transport` to `addr` and bind to `connection`. More...
PNP_EXTERN void | pn_proactor_connect (pn_proactor_t *proactor, pn_connection_t *connection, const char *addr)
PNP_EXTERN void | pn_proactor_listen (pn_proactor_t *proactor, pn_listener_t *listener, const char *addr, int backlog)
| Start listening for incoming connections. More...
PNP_EXTERN void | pn_proactor_disconnect (pn_proactor_t *proactor, pn_condition_t *condition)
| Disconnect all connections and listeners belonging to the proactor. More...
PNP_EXTERN pn_event_batch_t * | pn_proactor_wait (pn_proactor_t *proactor)
| Wait until there are Proactor events to handle. More...
PNP_EXTERN pn_event_batch_t * | pn_proactor_get (pn_proactor_t *proactor)
| Return Proactor events if any are available immediately. More...
PNP_EXTERN void | pn_proactor_done (pn_proactor_t *proactor, pn_event_batch_t *events)
| Call when finished handling a batch of events. More...
PNP_EXTERN void | pn_proactor_interrupt (pn_proactor_t *proactor)
| Return a PN_PROACTOR_INTERRUPT event as soon as possible. More...
PNP_EXTERN void | pn_proactor_set_timeout (pn_proactor_t *proactor, pn_millis_t timeout)
| Return a PN_PROACTOR_TIMEOUT after `timeout` milliseconds elapse. More...
PNP_EXTERN void | pn_proactor_cancel_timeout (pn_proactor_t *proactor)
| Cancel the pending timeout set by pn_proactor_set_timeout(). More...
PNP_EXTERN void | pn_proactor_release_connection (pn_connection_t *connection)
| Release ownership of `connection`, disassociate it from its proactor. More...
PNP_EXTERN void | pn_connection_wake (pn_connection_t *connection)
| Return a PN_CONNECTION_WAKE event for `connection` as soon as possible. More...
PNP_EXTERN pn_proactor_t * | pn_connection_proactor (pn_connection_t *connection)
| Return the proactor associated with a connection. More...
PNP_EXTERN pn_proactor_t * | pn_event_proactor (pn_event_t *event)
| Return the proactor associated with an event. More...
PNP_EXTERN pn_millis_t | pn_proactor_now (void)
| Get the real elapsed time since an arbitrary point in the past in milliseconds. More...
PNP_EXTERN const pn_netaddr_t * | pn_netaddr_local (pn_transport_t *t)
PNP_EXTERN const pn_netaddr_t * | pn_netaddr_remote (pn_transport_t *t)
PNP_EXTERN const pn_netaddr_t * | pn_netaddr_listening (pn_listener_t *l)

## Detailed Description

Unsettled API - An API for multithreaded IO.

The proactor associates an abstract AMQP protocol Connection with a concrete IO Transport implementation for outgoing and incoming connections.

pn_proactor_wait() returns Proactor events to application threads for handling.

The `pn_proactor_*` functions are thread-safe, but to handle Proactor events you must also use the Core APIs, which are not. Core objects associated with different connections can be used concurrently, but objects associated with a single connection can only be used from their own thread.

The proactor serializes Proactor events for each connection

- it never returns Proactor events for the same connection concurrently in different threads. Event-handling code can safely use any Core object obtained from the current event. You can attach application data to Core objects (for example with pn_connection_attachments()).

pn_connection_wake() allows any thread to "wake up" a connection. It causes pn_proactor_wait() to return a PN_CONNECTION_WAKE event that is serialized with the connection's other Proactor events. You can use this to implement communication between different connections, or from non-proactor threads.

Serialization and pn_connection_wake() simplify building applications with a shared thread pool, which serialize work per connection. Many other variations are possible, but you are responsible for any additional synchronization needed.

## Function Documentation

## pn_netaddr_str()

PNP_EXTERN int pn_netaddr_str | ( | const pn_netaddr_t * | addr,
|  | char * | buf,
|  | size_t | size
| ) |  |

Format a network address as a human-readable string in `buf`.

Returns
  the length of the string (excluding trailing '\0'), if >= size then the address was truncated.

## pn_transport_local_addr()

PNP_EXTERN const pn_netaddr_t* pn_transport_local_addr | ( | pn_transport_t * | t | ) |

Get the local address of a transport.

Return `NULL` if not available. Pointer is invalid after the transport closes (PN_TRANSPORT_CLOSED event is handled)

## pn_transport_remote_addr()

PNP_EXTERN const pn_netaddr_t* pn_transport_remote_addr | ( | pn_transport_t * | t | ) |

Get the local address of a transport.

Return `NULL` if not available. Pointer is invalid after the transport closes (PN_TRANSPORT_CLOSED event is handled)

## pn_listener_addr()

PNP_EXTERN const pn_netaddr_t* pn_listener_addr | ( | pn_listener_t * | l | ) |

Get the listening addresses of a listener.

Addresses are only available after the PN_LISTENER_OPEN event for the listener.

A listener can have more than one address for several reasons:

- DNS host records may indicate more than one address
- On a multi-homed host, listening on the default host "" will listen on all local addresses.
- Some IPv4/IPV6 configurations may expand a single address into a v4/v6 pair.

pn_netaddr_next() will iterate over the addresses in the list.

Parameters

  l | points to the listener

Returns
  The first listening address or NULL if there are no addresses are available. Use pn_netaddr_next() to iterate over the list. Pointer is invalid after the listener closes (PN_LISTENER_CLOSED event is handled)

Examples
  broker.c, and direct.c.

## pn_netaddr_next()

PNP_EXTERN const pn_netaddr_t* pn_netaddr_next | ( | const pn_netaddr_t * | na | ) |

Returns
  Pointer to the next address in a list of addresses, NULL if at the end of the list or if this address is not part of a list.

## pn_netaddr_sockaddr()

PNP_EXTERN const struct sockaddr* pn_netaddr_sockaddr | ( | const pn_netaddr_t * | na | ) |

On POSIX or Windows, get the underlying `struct sockaddr`.

Return NULL if not available.

## pn_netaddr_socklen()

PNP_EXTERN size_t pn_netaddr_socklen | ( | const pn_netaddr_t * | na | ) |

On POSIX or Windows, get the size of the underlying `struct sockaddr`.

Return 0 if not available.

## pn_netaddr_host_port()

PNP_EXTERN int pn_netaddr_host_port | ( | const pn_netaddr_t * | na,
|  | char * | host,
|  | size_t | hlen,
|  | char * | port,
|  | size_t | plen
| ) |  |

Get the host and port name from na as separate strings.

Returns 0 if successful, non-0 on error.

Examples
  broker.c, and direct.c.

## pn_netaddr_local()

PNP_EXTERN const pn_netaddr_t* pn_netaddr_local | ( | pn_transport_t * | t | ) |

## pn_proactor_addr()

PNP_EXTERN int pn_proactor_addr | ( | char * | addr,
|  | size_t | size,
|  | const char * | host,
|  | const char * | port
| ) |  |

Format a host:port address string for pn_proactor_connect2() or pn_proactor_listen2()

Parameters

  [out] | addr | address is copied to this buffer, with trailing '\0'
  [in] | size | size of addr buffer
  [in] | host | network host name, DNS name or IP address
  [in] | port | network service name or decimal port number, e.g. "amqp" or "5672"

Returns
  the length of network address (excluding trailing '\0'), if >= size then the address was truncated

Examples
  broker.c, direct.c, receive.c, and send.c.

## pn_proactor()

PNP_EXTERN pn_proactor_t* pn_proactor | ( | void |  | ) |

Create a proactor.

Must be freed with pn_proactor_free()

Examples
  broker.c, direct.c, receive.c, and send.c.

## pn_proactor_free()

PNP_EXTERN void pn_proactor_free | ( | pn_proactor_t * | proactor | ) |

Free the proactor.

Abort open connections/listeners, clean up all resources.

Examples
  broker.c, direct.c, receive.c, and send.c.

## pn_proactor_connect2()

PNP_EXTERN void pn_proactor_connect2 | ( | pn_proactor_t * | proactor,
|  | pn_connection_t * | connection,
|  | pn_transport_t * | transport,
|  | const char * | addr
| ) |  |

Connect `transport` to `addr` and bind to `connection`.

Errors are returned as PN_TRANSPORT_CLOSED events by pn_proactor_wait().

Note
  Thread-safe

Parameters

  [in] | proactor | the proactor object
  [in] | connection | If NULL a new connection is created. `proactor` takes ownership of `connection` and will automatically call pn_connection_free() after the final PN_TRANSPORT_CLOSED event is handled, or when pn_proactor_free() is called. You can prevent the automatic free with pn_proactor_release_connection()
  [in] | transport | If NULL a new transport is created. `proactor` takes ownership of `transport`, it will be freed even if pn_proactor_release_connection() is called.
  [in] | addr | the "host:port" network address, constructed by pn_proactor_addr() An empty host will connect to the local host via the default protocol (IPV6 or IPV4). An empty port will connect to the standard AMQP port (5672).

Examples
  receive.c, and send.c.

## pn_proactor_connect()

PNP_EXTERN void pn_proactor_connect | ( | pn_proactor_t * | proactor,
|  | pn_connection_t * | connection,
|  | const char * | addr
| ) |  |

## pn_proactor_listen()

PNP_EXTERN void pn_proactor_listen | ( | pn_proactor_t * | proactor,
|  | pn_listener_t * | listener,
|  | const char * | addr,
|  | int | backlog
| ) |  |

Start listening for incoming connections.

pn_proactor_wait() will return a PN_LISTENER_OPEN event when the listener is ready to accept connections, or a PN_LISTENER_CLOSE if the listen operation fails. If the listen failed, pn_listener_condition() will be set.

When the listener is closed by pn_listener_close(), or because of an error, a PN_LISTENER_CLOSE event will be returned and pn_listener_condition() will be set for an error.

Note
  Thread-safe

Parameters

  [in] | proactor | the proactor object
  [in] | listener | `proactor` takes ownership of `listener`, and will automatically call pn_listener_free() after the final PN_LISTENER_CLOSE event is handled, or when pn_proactor_free() is called.
  [in] | addr | the "host:port" network address, constructed by pn_proactor_addr() An empty host will listen for all protocols (IPV6 and IPV4) on all local interfaces. An empty port will listen on the standard AMQP port (5672).
  [in] | backlog | of un-handled connection requests to allow before refusing connections. If `addr` resolves to multiple interface/protocol combinations, the backlog applies to each separately.

Examples
  broker.c, and direct.c.

## pn_proactor_disconnect()

PNP_EXTERN void pn_proactor_disconnect | ( | pn_proactor_t * | proactor,
|  | pn_condition_t * | condition
| ) |  |

Disconnect all connections and listeners belonging to the proactor.

PN_LISTENER_CLOSE, PN_TRANSPORT_CLOSED and other Proactor events are generated as usual.

A PN_PROACTOR_INACTIVE event will be generated when all connections and listeners are disconnected and no timeout is pending. The event will also be generated if there are no listeners, connections or timeout when pn_proactor_disconnect() is called.

Creating new connections and listeners after this call and before the PN_PROACTOR_INACTIVE event may prevent the proactor from becoming inactive. After the PN_PROACTOR_INACTIVE event, the proactor can be used normally.

Note
  Thread-safe

Parameters

  proactor | the proactor
  condition | if not NULL the condition data is copied to each disconnected transports and listener and is available in the close event.

## pn_proactor_wait()

PNP_EXTERN pn_event_batch_t* pn_proactor_wait | ( | pn_proactor_t * | proactor | ) |

Wait until there are Proactor events to handle.

You must call pn_proactor_done() when you are finished with the batch, you must not use the batch pointer after calling pn_proactor_done().

Normally it is most efficient to handle the entire batch in the calling thread and then call pn_proactor_done(), but see pn_proactor_done() for more options.

pn_proactor_get() is a non-blocking version of this call.

Note
  Thread-safe

Returns
  a non-empty batch of events that must be processed in sequence.

Examples
  broker.c, direct.c, receive.c, and send.c.

## pn_proactor_get()

PNP_EXTERN pn_event_batch_t* pn_proactor_get | ( | pn_proactor_t * | proactor | ) |

Return Proactor events if any are available immediately.

If not, return NULL. If the return value is not NULL, the behavior is the same as pn_proactor_wait()

Note
  Thread-safe

## pn_proactor_done()

PNP_EXTERN void pn_proactor_done | ( | pn_proactor_t * | proactor,
|  | pn_event_batch_t * | events
| ) |  |

Call when finished handling a batch of events.

Must be called exactly once to match each call to pn_proactor_wait().

Note
  Thread-safe: May be called from any thread provided the exactly once rule is respected.

Examples
  broker.c, direct.c, receive.c, and send.c.

## pn_proactor_interrupt()

PNP_EXTERN void pn_proactor_interrupt | ( | pn_proactor_t * | proactor | ) |

Return a PN_PROACTOR_INTERRUPT event as soon as possible.

At least one PN_PROACTOR_INTERRUPT event will be returned after this call. Interrupts can be "coalesced" - if several pn_proactor_interrupt() calls happen close together, there may be only one PN_PROACTOR_INTERRUPT event that occurs after all of them.

Note
  Thread-safe and async-signal-safe: can be called in a signal handler. This is the only pn_proactor function that is async-signal-safe.

Examples
  broker.c.

## pn_proactor_set_timeout()

PNP_EXTERN void pn_proactor_set_timeout | ( | pn_proactor_t * | proactor,
|  | pn_millis_t | timeout
| ) |  |

Return a PN_PROACTOR_TIMEOUT after `timeout` milliseconds elapse.

If no threads are blocked in pn_proactor_wait() when the timeout elapses, the event will be delivered to the next available thread.

Calling pn_proactor_set_timeout() again before the PN_PROACTOR_TIMEOUT is delivered will cancel the previous timeout and deliver an event only after the new timeout.

Note
  Thread-safe

## pn_proactor_cancel_timeout()

PNP_EXTERN void pn_proactor_cancel_timeout | ( | pn_proactor_t * | proactor | ) |

Cancel the pending timeout set by pn_proactor_set_timeout().

Does nothing if no timeout is set.

Note
  Thread-safe

## pn_proactor_release_connection()

PNP_EXTERN void pn_proactor_release_connection | ( | pn_connection_t * | connection | ) |

Release ownership of `connection`, disassociate it from its proactor.

The connection and related objects (sessions, links and so on) remain intact, but the transport is closed and unbound. The proactor will not return any more events for this connection. The caller must call pn_connection_free(), either directly or indirectly by re-using `connection` in another call to pn_proactor_connect2() or pn_proactor_listen2().

Note
  Not thread-safe. Call this function from a connection event handler.
  If `connection` does not belong to a proactor, this call does nothing.
  This has nothing to do with pn_connection_release().

## pn_connection_wake()

PNP_EXTERN void pn_connection_wake | ( | pn_connection_t * | connection | ) |

Return a PN_CONNECTION_WAKE event for `connection` as soon as possible.

At least one wake event will be returned, serialized with other Proactor events for the same connection. Wakes can be "coalesced" - if several pn_connection_wake() calls happen close together, there may be only one PN_CONNECTION_WAKE event that occurs after all of them.

Note
  If `connection` does not belong to a proactor, this call does nothing.
  Thread-safe

Examples
  broker.c, and direct.c.

## pn_connection_proactor()

PNP_EXTERN pn_proactor_t* pn_connection_proactor | ( | pn_connection_t * | connection | ) |

Return the proactor associated with a connection.

Note
  Not thread-safe

Returns
  the proactor or NULL if the connection does not belong to a proactor.

## pn_event_proactor()

PNP_EXTERN pn_proactor_t* pn_event_proactor | ( | pn_event_t * | event | ) |

Return the proactor associated with an event.

Note
  Not thread-safe

Returns
  the proactor or NULL if the connection does not belong to a proactor.

## pn_proactor_now()

PNP_EXTERN pn_millis_t pn_proactor_now | ( | void |  | ) |

Get the real elapsed time since an arbitrary point in the past in milliseconds.

This may be used as a portable way to get a process-local timestamp for the current time. It is monotonically increasing and will never go backwards.

Note: this is not a suitable value for an AMQP timestamp to be sent as part of a message. Such a timestamp should use the real time in milliseconds since the epoch.

Note
  Thread-safe