3 #![allow(unused_unsafe)]
4 #![allow(unused_imports)]
5 #![allow(non_camel_case_types)]
9 // tjc: I don't know why
11 use self::state::{empty, full, blocked, terminated};
13 use std::mem::{forget, transmute};
14 use std::mem::{replace, swap};
17 use std::marker::Send;
21 blocked_task: Option<Task>,
25 #[derive(PartialEq, Debug)]
34 pub struct packet<T> {
36 blocked_task: Option<Task>,
40 unsafe impl<T:Send> Send for packet<T> {}
42 pub fn packet<T:Send>() -> *const packet<T> {
44 let p: *const packet<T> = mem::transmute(Box::new(Stuff{
46 blocked_task: None::<Task>,
54 pub fn atomic_xchg(_dst: &mut isize, _src: isize) -> isize { panic!(); }
55 pub fn atomic_xchg_acq(_dst: &mut isize, _src: isize) -> isize { panic!(); }
56 pub fn atomic_xchg_rel(_dst: &mut isize, _src: isize) -> isize { panic!(); }
59 // We should consider moving this to ::std::unsafe, although I
60 // suspect graydon would want us to use void pointers instead.
61 pub unsafe fn uniquify<T>(x: *const T) -> Box<T> {
65 pub fn swap_state_acq(dst: &mut state, src: state) -> state {
67 transmute(rusti::atomic_xchg_acq(transmute(dst), src as isize))
71 pub fn swap_state_rel(dst: &mut state, src: state) -> state {
73 transmute(rusti::atomic_xchg_rel(transmute(dst), src as isize))
77 pub fn send<T:Send>(mut p: send_packet<T>, payload: T) {
79 let mut p = unsafe { uniquify(p) };
80 assert!((*p).payload.is_none());
81 (*p).payload = Some(payload);
82 let old_state = swap_state_rel(&mut (*p).state, full);
87 // The receiver will eventually clean this up.
90 full => { panic!("duplicate send") }
93 // The receiver will eventually clean this up.
97 // The receiver will never receive this. Rely on drop_glue
98 // to clean everything up.
103 pub fn recv<T:Send>(mut p: recv_packet<T>) -> Option<T> {
105 let mut p = unsafe { uniquify(p) };
107 let old_state = swap_state_acq(&mut (*p).state,
110 empty | blocked => { thread::yield_now(); }
112 let payload = replace(&mut p.payload, None);
113 return Some(payload.unwrap())
116 assert_eq!(old_state, terminated);
123 pub fn sender_terminate<T:Send>(p: *const packet<T>) {
124 let mut p = unsafe { uniquify(p) };
125 match swap_state_rel(&mut (*p).state, terminated) {
127 // The receiver will eventually clean up.
131 // This is impossible
132 panic!("you dun goofed")
135 // I have to clean up, use drop_glue
140 pub fn receiver_terminate<T:Send>(p: *const packet<T>) {
141 let mut p = unsafe { uniquify(p) };
142 match swap_state_rel(&mut (*p).state, terminated) {
144 // the sender will clean up
148 // this shouldn't happen.
149 panic!("terminating a blocked packet")
151 terminated | full => {
152 // I have to clean up, use drop_glue
157 pub struct send_packet<T:Send> {
158 p: Option<*const packet<T>>,
161 impl<T:Send> Drop for send_packet<T> {
165 let self_p: &mut Option<*const packet<T>> =
166 mem::transmute(&mut self.p);
167 let p = replace(self_p, None);
168 sender_terminate(p.unwrap())
174 impl<T:Send> send_packet<T> {
175 pub fn unwrap(&mut self) -> *const packet<T> {
176 replace(&mut self.p, None).unwrap()
180 pub fn send_packet<T:Send>(p: *const packet<T>) -> send_packet<T> {
186 pub struct recv_packet<T:Send> {
187 p: Option<*const packet<T>>,
190 impl<T:Send> Drop for recv_packet<T> {
194 let self_p: &mut Option<*const packet<T>> =
195 mem::transmute(&mut self.p);
196 let p = replace(self_p, None);
197 receiver_terminate(p.unwrap())
203 impl<T:Send> recv_packet<T> {
204 pub fn unwrap(&mut self) -> *const packet<T> {
205 replace(&mut self.p, None).unwrap()
209 pub fn recv_packet<T:Send>(p: *const packet<T>) -> recv_packet<T> {
215 pub fn entangle<T:Send>() -> (send_packet<T>, recv_packet<T>) {
217 (send_packet(p), recv_packet(p))
224 pub struct ping(::pipes::send_packet<pong>);
226 unsafe impl Send for ping {}
228 pub struct pong(::pipes::send_packet<ping>);
230 unsafe impl Send for pong {}
232 pub fn liberate_ping(p: ping) -> ::pipes::send_packet<pong> {
234 let _addr : *const ::pipes::send_packet<pong> = match &p {
235 &ping(ref x) => { mem::transmute(x) }
241 pub fn liberate_pong(p: pong) -> ::pipes::send_packet<ping> {
243 let _addr : *const ::pipes::send_packet<ping> = match &p {
244 &pong(ref x) => { mem::transmute(x) }
250 pub fn init() -> (client::ping, server::ping) {
257 pub type ping = ::pipes::send_packet<pingpong::ping>;
258 pub type pong = ::pipes::recv_packet<pingpong::pong>;
260 pub fn do_ping(c: ping) -> pong {
261 let (sp, rp) = ::pipes::entangle();
263 ::pipes::send(c, pingpong::ping(sp));
267 pub fn do_pong(c: pong) -> (ping, ()) {
268 let packet = ::pipes::recv(c);
269 if packet.is_none() {
270 panic!("sender closed the connection")
272 (pingpong::liberate_pong(packet.unwrap()), ())
279 pub type ping = ::pipes::recv_packet<pingpong::ping>;
280 pub type pong = ::pipes::send_packet<pingpong::pong>;
282 pub fn do_ping(c: ping) -> (pong, ()) {
283 let packet = ::pipes::recv(c);
284 if packet.is_none() {
285 panic!("sender closed the connection")
287 (pingpong::liberate_ping(packet.unwrap()), ())
290 pub fn do_pong(c: pong) -> ping {
291 let (sp, rp) = ::pipes::entangle();
292 ::pipes::send(c, pingpong::pong(sp));
298 fn client(chan: pingpong::client::ping) {
299 let chan = pingpong::client::do_ping(chan);
300 println!("Sent ping");
301 let (_chan, _data) = pingpong::client::do_pong(chan);
302 println!("Received pong");
305 fn server(chan: pingpong::server::ping) {
306 let (chan, _data) = pingpong::server::do_ping(chan);
307 println!("Received ping");
308 let _chan = pingpong::server::do_pong(chan);
309 println!("Sent pong");
314 // Commented out because of option::get error
316 let (client_, server_) = pingpong::init();
318 task::spawn {|client_|
319 let client__ = client_.take();
322 task::spawn {|server_|
323 let server__ = server_.take();