rust/src/test/run-pass/issue-2718.rs

// Copyright 2012-2014 The Rust Project Developers. See the COPYRIGHT
// file at the top-level directory of this distribution and at
// http://rust-lang.org/COPYRIGHT.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.
//

#![feature(std_misc)]

pub type Task = isize;

// tjc: I don't know why
pub mod pipes {
    use self::state::{empty, full, blocked, terminated};
    use super::Task;
    use std::mem::{forget, transmute};
    use std::mem::{replace, swap};
    use std::mem;
    use std::thread;
    use std::marker::Send;

    pub struct Stuff<T> {
        state: state,
        blocked_task: Option<Task>,
        payload: Option<T>
    }

    #[derive(PartialEq, Debug)]
    #[repr(isize)]
    pub enum state {
        empty,
        full,
        blocked,
        terminated
    }

    pub struct packet<T> {
        state: state,
        blocked_task: Option<Task>,
        payload: Option<T>
    }

    unsafe impl<T:Send> Send for packet<T> {}

    pub fn packet<T:Send>() -> *const packet<T> {
        unsafe {
            let p: *const packet<T> = mem::transmute(Box::new(Stuff{
                state: empty,
                blocked_task: None::<Task>,
                payload: None::<T>
            }));
            p
        }
    }

    mod rusti {
      pub fn atomic_xchg(_dst: &mut isize, _src: isize) -> isize { panic!(); }
      pub fn atomic_xchg_acq(_dst: &mut isize, _src: isize) -> isize { panic!(); }
      pub fn atomic_xchg_rel(_dst: &mut isize, _src: isize) -> isize { panic!(); }
    }

    // We should consider moving this to ::std::unsafe, although I
    // suspect graydon would want us to use void pointers instead.
    pub unsafe fn uniquify<T>(x: *const T) -> Box<T> {
        mem::transmute(x)
    }

    pub fn swap_state_acq(dst: &mut state, src: state) -> state {
        unsafe {
            transmute(rusti::atomic_xchg_acq(transmute(dst), src as isize))
        }
    }

    pub fn swap_state_rel(dst: &mut state, src: state) -> state {
        unsafe {
            transmute(rusti::atomic_xchg_rel(transmute(dst), src as isize))
        }
    }

    pub fn send<T:Send>(mut p: send_packet<T>, payload: T) {
        let p = p.unwrap();
        let mut p = unsafe { uniquify(p) };
        assert!((*p).payload.is_none());
        (*p).payload = Some(payload);
        let old_state = swap_state_rel(&mut (*p).state, full);
        match old_state {
          empty => {
            // Yay, fastpath.

            // The receiver will eventually clean this up.
            unsafe { forget(p); }
          }
          full => { panic!("duplicate send") }
          blocked => {

            // The receiver will eventually clean this up.
            unsafe { forget(p); }
          }
          terminated => {
            // The receiver will never receive this. Rely on drop_glue
            // to clean everything up.
          }
        }
    }

    pub fn recv<T:Send>(mut p: recv_packet<T>) -> Option<T> {
        let p = p.unwrap();
        let mut p = unsafe { uniquify(p) };
        loop {
            let old_state = swap_state_acq(&mut (*p).state,
                                           blocked);
            match old_state {
              empty | blocked => { thread::yield_now(); }
              full => {
                let payload = replace(&mut p.payload, None);
                return Some(payload.unwrap())
              }
              terminated => {
                assert_eq!(old_state, terminated);
                return None;
              }
            }
        }
    }

    pub fn sender_terminate<T:Send>(p: *const packet<T>) {
        let mut p = unsafe { uniquify(p) };
        match swap_state_rel(&mut (*p).state, terminated) {
          empty | blocked => {
            // The receiver will eventually clean up.
            unsafe { forget(p) }
          }
          full => {
            // This is impossible
            panic!("you dun goofed")
          }
          terminated => {
            // I have to clean up, use drop_glue
          }
        }
    }

    pub fn receiver_terminate<T:Send>(p: *const packet<T>) {
        let mut p = unsafe { uniquify(p) };
        match swap_state_rel(&mut (*p).state, terminated) {
          empty => {
            // the sender will clean up
            unsafe { forget(p) }
          }
          blocked => {
            // this shouldn't happen.
            panic!("terminating a blocked packet")
          }
          terminated | full => {
            // I have to clean up, use drop_glue
          }
        }
    }

    pub struct send_packet<T:Send> {
        p: Option<*const packet<T>>,
    }

        impl<T:Send> Drop for send_packet<T> {
        fn drop(&mut self) {
            unsafe {
                if self.p != None {
                    let self_p: &mut Option<*const packet<T>> =
                        mem::transmute(&self.p);
                    let p = replace(self_p, None);
                    sender_terminate(p.unwrap())
                }
            }
        }
    }

    impl<T:Send> send_packet<T> {
        pub fn unwrap(&mut self) -> *const packet<T> {
            replace(&mut self.p, None).unwrap()
        }
    }

    pub fn send_packet<T:Send>(p: *const packet<T>) -> send_packet<T> {
        send_packet {
            p: Some(p)
        }
    }

    pub struct recv_packet<T:Send> {
        p: Option<*const packet<T>>,
    }

        impl<T:Send> Drop for recv_packet<T> {
        fn drop(&mut self) {
            unsafe {
                if self.p != None {
                    let self_p: &mut Option<*const packet<T>> =
                        mem::transmute(&self.p);
                    let p = replace(self_p, None);
                    receiver_terminate(p.unwrap())
                }
            }
        }
    }

    impl<T:Send> recv_packet<T> {
        pub fn unwrap(&mut self) -> *const packet<T> {
            replace(&mut self.p, None).unwrap()
        }
    }

    pub fn recv_packet<T:Send>(p: *const packet<T>) -> recv_packet<T> {
        recv_packet {
            p: Some(p)
        }
    }

    pub fn entangle<T:Send>() -> (send_packet<T>, recv_packet<T>) {
        let p = packet();
        (send_packet(p), recv_packet(p))
    }
}

pub mod pingpong {
    use std::mem;

    pub struct ping(::pipes::send_packet<pong>);

    unsafe impl Send for ping {}

    pub struct pong(::pipes::send_packet<ping>);

    unsafe impl Send for pong {}

    pub fn liberate_ping(p: ping) -> ::pipes::send_packet<pong> {
        unsafe {
            let _addr : *const ::pipes::send_packet<pong> = match &p {
              &ping(ref x) => { mem::transmute(x) }
            };
            panic!()
        }
    }

    pub fn liberate_pong(p: pong) -> ::pipes::send_packet<ping> {
        unsafe {
            let _addr : *const ::pipes::send_packet<ping> = match &p {
              &pong(ref x) => { mem::transmute(x) }
            };
            panic!()
        }
    }

    pub fn init() -> (client::ping, server::ping) {
        ::pipes::entangle()
    }

    pub mod client {
        use pingpong;

        pub type ping = ::pipes::send_packet<pingpong::ping>;
        pub type pong = ::pipes::recv_packet<pingpong::pong>;

        pub fn do_ping(c: ping) -> pong {
            let (sp, rp) = ::pipes::entangle();

            ::pipes::send(c, pingpong::ping(sp));
            rp
        }

        pub fn do_pong(c: pong) -> (ping, ()) {
            let packet = ::pipes::recv(c);
            if packet.is_none() {
                panic!("sender closed the connection")
            }
            (pingpong::liberate_pong(packet.unwrap()), ())
        }
    }

    pub mod server {
        use pingpong;

        pub type ping = ::pipes::recv_packet<pingpong::ping>;
        pub type pong = ::pipes::send_packet<pingpong::pong>;

        pub fn do_ping(c: ping) -> (pong, ()) {
            let packet = ::pipes::recv(c);
            if packet.is_none() {
                panic!("sender closed the connection")
            }
            (pingpong::liberate_ping(packet.unwrap()), ())
        }

        pub fn do_pong(c: pong) -> ping {
            let (sp, rp) = ::pipes::entangle();
            ::pipes::send(c, pingpong::pong(sp));
            rp
        }
    }
}

fn client(chan: pingpong::client::ping) {
    let chan = pingpong::client::do_ping(chan);
    println!("Sent ping");
    let (_chan, _data) = pingpong::client::do_pong(chan);
    println!("Received pong");
}

fn server(chan: pingpong::server::ping) {
    let (chan, _data) = pingpong::server::do_ping(chan);
    println!("Received ping");
    let _chan = pingpong::server::do_pong(chan);
    println!("Sent pong");
}

pub fn main() {
  /*
//    Commented out because of option::get error

    let (client_, server_) = pingpong::init();

    task::spawn {|client_|
        let client__ = client_.take();
        client(client__);
    };
    task::spawn {|server_|
        let server__ = server_.take();
        server(server_ˊ);
    };
  */
}