Runtime support for message passing with protocol enforcement.
Pipes consist of two endpoints. One endpoint can send messages and the other can receive messages. The set of legal messages and which directions they can flow at any given point are determined by a protocol. Below is an example protocol.
proto! pingpong (
ping: send {
ping -> pong
}
pong: recv {
pong -> ping
}
)
The proto!
syntax extension will convert this into a module called pingpong
, which includes a set of types and functions that can be used to write programs that follow the pingpong protocol.
ChanOne
- The send end of a oneshot pipe.PortOne
- The receive end of a oneshot pipe.RecvPacket
- Represents the receive end of a pipeSendPacket
- The sending end of a pipeSharedChan
- A channel that can be shared between many senders.Chan
- An endpoint that can send many messages.Port
- An endpoint that can receive many messages.BufferHeader
BufferResource
PacketHeader
PortSet
- Treat many ports as one.RecvPacketBuffered
SendPacketBuffered
GenericChan
- A trait for things that can send multiple messages.GenericPort
- A trait for things that can receive multiple messages.GenericSmartChan
- Things that can send multiple messages and can detect when the receiver is closedPeekable
- Ports that can peek
Select2
- Receive a message from one of two endpoints.of Eq for State
for PacketHeader
of HasBuffer for Packet<T>
of Peekable<T> for RecvPacketBuffered<T, Tb>
of Selectable for *PacketHeader
for SendPacketBuffered<T, Tbuffer>
for RecvPacketBuffered<T, Tbuffer>
of Selectable for RecvPacketBuffered<T, Tbuffer>
of GenericChan<T> for Chan<T>
of GenericSmartChan<T> for Chan<T>
of GenericPort<T> for Port<T>
of Peekable<T> for Port<T>
of Selectable for Port<T>
for PortSet<T>
of GenericPort<T> for PortSet<T>
of Peekable<T> for PortSet<T>
of GenericChan<T> for SharedChan<T>
of GenericSmartChan<T> for SharedChan<T>
of Select2<T, U> for (Left, Right)
BufferHeader
BufferResource
PacketHeader
PortSet
RecvPacketBuffered
SendPacketBuffered
SharedChan
- Converts a chan
into a shared_chan
.oneshot
- Initialiase a (send-endpoint, recv-endpoint) oneshot pipe pair.peek
- Returns true if messages are available.recv
- Receives a message from a pipe.recv_one
- Receive a message from a oneshot pipe, failing if the connection was closed.select
- Waits on a set of endpointsselect2
- Receives a message from one of two endpoints.select2i
- Returns 0 or 1 depending on which endpoint is ready to receiveselecti
- Returns the index of an endpoint that is ready to receive.send_one
- Send a message on a oneshot pipe, failing if the connection was closed.spawn_service
- Spawn a task to provide a service.spawn_service_recv
- Like spawn_service_recv
, but for protocols that start in the receive state.stream
- Creates a (chan, port)
pair.try_recv
- Attempts to receive a message from a pipe.try_recv_one
- Receive a message from a oneshot pipe unless the connection was closed.try_send_one
- Send a message on a oneshot pipe, or return false if the connection was closed.wait_many
- Returns when one of the packet headers reports data is available.pipes::oneshot
pipes::rt
pipes::streamp
ChanOne
type ChanOne<T: Owned> = oneshot::client::Oneshot<T>
The send end of a oneshot pipe.
PortOne
type PortOne<T: Owned> = oneshot::server::Oneshot<T>
The receive end of a oneshot pipe.
RecvPacket
type RecvPacket<T: Owned> = RecvPacketBuffered<T, Packet<T>>
Represents the receive end of a pipe. It can receive exactly one message.
SendPacket
type SendPacket<T: Owned> = SendPacketBuffered<T, Packet<T>>
The sending end of a pipe. It can be used to send exactly one message.
Chan
An endpoint that can send many messages.
Chan_(Chan_<T>)
Port
An endpoint that can receive many messages.
Port_(Port_<T>)
BufferHeader
pub struct BufferHeader {
mut ref_count: int,
}
BufferResource
struct BufferResource <T: Owned>{
buffer: ~Buffer<T>,
}
PacketHeader
struct PacketHeader {
mut state: State,
mut blocked_task: *rust_task,
mut buffer: *libc::c_void,
}
PortSet
pub struct PortSet <T: Owned>{
mut ports: ~[pipes::Port<T>],
}
Treat many ports as one.
RecvPacketBuffered
pub struct RecvPacketBuffered <T: Owned, Tbuffer: Owned>{
mut p: Option<*Packet<T>>,
mut buffer: Option<BufferResource<Tbuffer>>,
}
SendPacketBuffered
pub struct SendPacketBuffered <T: Owned, Tbuffer: Owned>{
mut p: Option<*Packet<T>>,
mut buffer: Option<BufferResource<Tbuffer>>,
}
GenericChan
A trait for things that can send multiple messages.
send
fn send(x: T)
Sends a message.
GenericPort
A trait for things that can receive multiple messages.
recv
fn recv() -> T
Receives a message, or fails if the connection closes.
try_recv
fn try_recv() -> Option<T>
Receives a message, or returns none
if the connection is closed or closes.
GenericSmartChan
Things that can send multiple messages and can detect when the receiver is closed
try_send
fn try_send(x: T) -> bool
Sends a message, or report if the receiver has closed the connection.
Peekable
Ports that can peek
peek
fn peek() -> bool
Returns true if a message is available
Select2
Receive a message from one of two endpoints.
try_select
fn try_select() -> Either<Option<T>, Option<U>>
Receive a message or return None
if a connection closes.
select
fn select() -> Either<T, U>
Receive a message or fail if a connection closes.
Eq
for State
eq
fn eq(other: &State) -> bool
ne
fn ne(other: &State) -> bool
PacketHeader
mark_blocked
fn mark_blocked(this: *rust_task) -> State
unblock
fn unblock()
buf_header
fn buf_header() -> ~BufferHeader
set_buffer
fn set_buffer<T: Owned>(b: ~Buffer<T>)
HasBuffer
for Packet<T>
set_buffer_
fn set_buffer_(b: *libc::c_void)
Peekable<T>
for RecvPacketBuffered<T, Tb>
peek
fn peek() -> bool
Selectable
for *PacketHeader
header
fn header() -> *PacketHeader
SendPacketBuffered<T, Tbuffer>
unwrap
fn unwrap() -> *Packet<T>
header
fn header() -> *PacketHeader
reuse_buffer
fn reuse_buffer() -> BufferResource<Tbuffer>
RecvPacketBuffered<T, Tbuffer>
unwrap
fn unwrap() -> *Packet<T>
reuse_buffer
fn reuse_buffer() -> BufferResource<Tbuffer>
Selectable
for RecvPacketBuffered<T, Tbuffer>
header
fn header() -> *PacketHeader
GenericChan<T>
for Chan<T>
send
fn send(x: T)
GenericSmartChan<T>
for Chan<T>
try_send
fn try_send(x: T) -> bool
GenericPort<T>
for Port<T>
recv
fn recv() -> T
try_recv
fn try_recv() -> Option<T>
Peekable<T>
for Port<T>
peek
fn peek() -> bool
Selectable
for Port<T>
header
fn header() -> *PacketHeader
PortSet<T>
add
fn add(port: pipes::Port<T>)
chan
fn chan() -> Chan<T>
GenericPort<T>
for PortSet<T>
try_recv
fn try_recv() -> Option<T>
recv
fn recv() -> T
Peekable<T>
for PortSet<T>
peek
fn peek() -> bool
Select2<T, U>
for (Left, Right)
select
fn select() -> Either<T, U>
try_select
fn try_select() -> Either<Option<T>, Option<U>>
BufferHeader
fn BufferHeader() -> BufferHeader
BufferResource
fn BufferResource<T: Owned>(b: ~Buffer<T>) -> BufferResource<T>
PacketHeader
fn PacketHeader() -> PacketHeader
PortSet
fn PortSet<T: Owned>() -> PortSet<T>
RecvPacketBuffered
fn RecvPacketBuffered<T: Owned, Tbuffer: Owned>(p: *Packet<T>) ->
RecvPacketBuffered<T, Tbuffer>
SendPacketBuffered
fn SendPacketBuffered<T: Owned, Tbuffer: Owned>(p: *Packet<T>) ->
SendPacketBuffered<T, Tbuffer>
oneshot
fn oneshot<T: Owned>() -> (ChanOne<T>, PortOne<T>)
Initialiase a (send-endpoint, recv-endpoint) oneshot pipe pair.
peek
fn peek<T: Owned, Tb: Owned>(p: &RecvPacketBuffered<T, Tb>) -> bool
Returns true if messages are available.
recv
fn recv<T: Owned, Tbuffer: Owned>(p: RecvPacketBuffered<T, Tbuffer>) -> T
Receives a message from a pipe.
Fails if the sender closes the connection.
recv_one
fn recv_one<T: Owned>(port: PortOne<T>) -> T
Receive a message from a oneshot pipe, failing if the connection was closed.
select
fn select<T: Owned, Tb: Owned>(endpoints: ~[RecvPacketBuffered<T, Tb>]) ->
(uint, Option<T>, ~[RecvPacketBuffered<T, Tb>])
Waits on a set of endpoints. Returns a message, its index, and a list of the remaining endpoints.
select2
fn select2<A: Owned, Ab: Owned, B: Owned,
Bb: Owned>(a: RecvPacketBuffered<A, Ab>,
b: RecvPacketBuffered<B, Bb>) ->
Either<(Option<A>, RecvPacketBuffered<B, Bb>),
(RecvPacketBuffered<A, Ab>, Option<B>)>
Receives a message from one of two endpoints.
The return value is left
if the first endpoint received something, or right
if the second endpoint receives something. In each case, the result includes the other endpoint as well so it can be used again. Below is an example of using select2
.
match select2(a, b) {
left((none, b)) {
// endpoint a was closed.
}
right((a, none)) {
// endpoint b was closed.
}
left((Some(_), b)) {
// endpoint a received a message
}
right(a, Some(_)) {
// endpoint b received a message.
}
}
Sometimes messages will be available on both endpoints at once. In this case, select2
may return either left
or right
.
select2i
fn select2i<A: Selectable, B: Selectable>(a: &A, b: &B) -> Either<(), ()>
Returns 0 or 1 depending on which endpoint is ready to receive
selecti
fn selecti<T: Selectable>(endpoints: &[T]) -> uint
Returns the index of an endpoint that is ready to receive.
send_one
fn send_one<T: Owned>(chan: ChanOne<T>, data: T)
Send a message on a oneshot pipe, failing if the connection was closed.
spawn_service
fn spawn_service<T: Owned,
Tb: Owned>(init:
extern fn()
->
(SendPacketBuffered<T, Tb>,
RecvPacketBuffered<T, Tb>),
service: ~fn(v: RecvPacketBuffered<T, Tb>)) ->
SendPacketBuffered<T, Tb>
Spawn a task to provide a service.
It takes an initialization function that produces a send and receive endpoint. The send endpoint is returned to the caller and the receive endpoint is passed to the new task.
spawn_service_recv
fn spawn_service_recv<T: Owned,
Tb: Owned>(init:
extern fn()
->
(RecvPacketBuffered<T, Tb>,
SendPacketBuffered<T, Tb>),
service: ~fn(v: SendPacketBuffered<T, Tb>))
-> RecvPacketBuffered<T, Tb>
Like spawn_service_recv
, but for protocols that start in the receive state.
stream
fn stream<T: Owned>() -> (Port<T>, Chan<T>)
Creates a (chan, port)
pair.
These allow sending or receiving an unlimited number of messages.
try_recv
fn try_recv<T: Owned, Tbuffer: Owned>(p: RecvPacketBuffered<T, Tbuffer>) ->
Option<T>
Attempts to receive a message from a pipe.
Returns None
if the sender has closed the connection without sending a message, or Some(T)
if a message was received.
try_recv_one
fn try_recv_one<T: Owned>(port: PortOne<T>) -> Option<T>
Receive a message from a oneshot pipe unless the connection was closed.
try_send_one
fn try_send_one<T: Owned>(chan: ChanOne<T>, data: T) -> bool
Send a message on a oneshot pipe, or return false if the connection was closed.
wait_many
fn wait_many<T: Selectable>(pkts: &[T]) -> uint
Returns when one of the packet headers reports data is available.
This function is primarily intended for building higher level waiting functions, such as select
, select2
, etc.
It takes a vector slice of packet_headers and returns an index into that vector. The index points to an endpoint that has either been closed by the sender or has a message waiting to be received.