src/libcore/comm.rs

   1 /*
   2 Module: comm
   3
   4 Communication between tasks
   5
   6 Communication between tasks is facilitated by ports (in the receiving task),
   7 and channels (in the sending task). Any number of channels may feed into a
   8 single port.
   9
  10 Ports and channels may only transmit values of unique types; that is,
  11 values that are statically guaranteed to be accessed by a single
  12 'owner' at a time.  Unique types include scalars, vectors, strings,
  13 and records, tags, tuples and unique boxes (~T) thereof. Most notably,
  14 shared boxes (@T) may not be transmitted across channels.
  15
  16 Example:
  17
  18 > import std::{task, comm, io};
  19 >
  20 > let p = comm::port();
  21 > task::spawn(comm::chan(p), fn (c: chan<str>) {
  22 >   comm::send(c, "Hello, World");
  23 > });
  24 >
  25 > io::println(comm::recv(p));
  26
  27 */
  28
  29 import sys;
  30 import task;
  31
  32 export send;
  33 export recv;
  34 export chan;
  35 export port;
  36
  37 #[abi = "cdecl"]
  38 native mod rustrt {
  39     type void;
  40     type rust_port;
  41
  42     fn chan_id_send<T: send>(t: *sys::type_desc,
  43                             target_task: task::task, target_port: port_id,
  44                             data: T) -> ctypes::uintptr_t;
  45
  46     fn new_port(unit_sz: uint) -> *rust_port;
  47     fn del_port(po: *rust_port);
  48     fn rust_port_detach(po: *rust_port);
  49     fn get_port_id(po: *rust_port) -> port_id;
  50     fn rust_port_size(po: *rust_port) -> ctypes::size_t;
  51     fn port_recv(dptr: *uint, po: *rust_port,
  52                  yield: *ctypes::uintptr_t,
  53                  killed: *ctypes::uintptr_t);
  54 }
  55
  56 #[abi = "rust-intrinsic"]
  57 native mod rusti {
  58     fn call_with_retptr<T: send>(&&f: fn@(*uint)) -> T;
  59 }
  60
  61 type port_id = int;
  62
  63 // It's critical that this only have one variant, so it has a record
  64 // layout, and will work in the rust_task structure in task.rs.
  65 /*
  66 Type: chan
  67
  68 A communication endpoint that can send messages. Channels send
  69 messages to ports.
  70
  71 Each channel is bound to a port when the channel is constructed, so
  72 the destination port for a channel must exist before the channel
  73 itself.
  74
  75 Channels are weak: a channel does not keep the port it is bound to alive.
  76 If a channel attempts to send data to a dead port that data will be silently
  77 dropped.
  78
  79 Channels may be duplicated and themselves transmitted over other channels.
  80 */
  81 tag chan<T: send> {
  82     chan_t(task::task, port_id);
  83 }
  84
  85 resource port_ptr<T: send>(po: *rustrt::rust_port) {
  86     // Once the port is detached it's guaranteed not to receive further
  87     // messages
  88     rustrt::rust_port_detach(po);
  89     // Drain the port so that all the still-enqueued items get dropped
  90     while rustrt::rust_port_size(po) > 0u {
  91         // FIXME: For some reason if we don't assign to something here
  92         // we end up with invalid reads in the drop glue.
  93         let _t = recv_::<T>(po);
  94     }
  95     rustrt::del_port(po);
  96 }
  97
  98 /*
  99 Type: port
 100
 101 A communication endpoint that can receive messages. Ports receive
 102 messages from channels.
 103
 104 Each port has a unique per-task identity and may not be replicated or
 105 transmitted. If a port value is copied, both copies refer to the same port.
 106
 107 Ports may be associated with multiple <chan>s.
 108 */
 109 tag port<T: send> { port_t(@port_ptr<T>); }
 110
 111 /*
 112 Function: send
 113
 114 Sends data over a channel.
 115
 116 The sent data is moved into the channel, whereupon the caller loses access
 117 to it.
 118 */
 119 fn send<T: send>(ch: chan<T>, -data: T) {
 120     let chan_t(t, p) = ch;
 121     let res = rustrt::chan_id_send(sys::get_type_desc::<T>(), t, p, data);
 122     if res != 0u unsafe {
 123         // Data sent successfully
 124         unsafe::leak(data);
 125     }
 126     task::yield();
 127 }
 128
 129 /*
 130 Function: port
 131
 132 Constructs a port.
 133 */
 134 fn port<T: send>() -> port<T> {
 135     port_t(@port_ptr(rustrt::new_port(sys::size_of::<T>())))
 136 }
 137
 138 /*
 139 Function: recv
 140
 141 Receive from a port.
 142
 143 If no data is available on the port then the task will block until data
 144 becomes available.
 145 */
 146 fn recv<T: send>(p: port<T>) -> T { recv_(***p) }
 147
 148 // Receive on a raw port pointer
 149 fn recv_<T: send>(p: *rustrt::rust_port) -> T {
 150     // FIXME: Due to issue 1185 we can't use a return pointer when
 151     // calling C code, and since we can't create our own return
 152     // pointer on the stack, we're going to call a little intrinsic
 153     // that will grab the value of the return pointer, then call this
 154     // function, which we will then use to call the runtime.
 155     fn recv(dptr: *uint, port: *rustrt::rust_port,
 156             yield: *ctypes::uintptr_t,
 157             killed: *ctypes::uintptr_t) unsafe {
 158         rustrt::port_recv(dptr, port, yield, killed);
 159     }
 160     let yield = 0u;
 161     let yieldp = ptr::addr_of(yield);
 162     let killed = 0u;
 163     let killedp = ptr::addr_of(killed);
 164     let res = rusti::call_with_retptr(bind recv(_, p, yieldp, killedp));
 165     if killed != 0u {
 166         fail "killed";
 167     }
 168     if yield != 0u {
 169         // Data isn't available yet, so res has not been initialized.
 170         task::yield();
 171     }
 172     ret res;
 173 }
 174
 175 /*
 176 Function: chan
 177
 178 Constructs a channel.
 179
 180 The channel is bound to the port used to construct it.
 181 */
 182 fn chan<T: send>(p: port<T>) -> chan<T> {
 183     chan_t(task::get_task(), rustrt::get_port_id(***p))
 184 }