crash copying pipes::SharedChan

[jesse99]

I'm getting crashes after converting oldcomms code over to pipes. The crash happens when I copy a pipes::SharedChan that has been captured by a closure.

Here is a test case:

#[allow(non_implicitly_copyable_typarams)];

extern mod std;
use core::send_map::linear::{LinearMap, linear_map_with_capacity};

enum StateMesg
{
    AddListener(~str, pipes::Chan<int>),    // str is used to identify the listener
    RemoveListener(~str),
    Shutdown,
}

type StatePort = pipes::Port<StateMesg>;
type StateChan = pipes::Chan<StateMesg>;

pub type PushChan = pipes::SharedChan<~str>;
pub type OpenSse = fn~ (channel: PushChan);

pub struct Config
{
    pub sse: LinearMap<~str, OpenSse>,
}

fn main()
{
    let (_state_port, state_chan) : (StatePort, StateChan) = pipes::stream();
    let state_chan = pipes::SharedChan(state_chan);

    // The up closure captures the state_chan SharedChan. The closure
    // is then moved into a LinearMap. When a connection is created
    // config is copied which is when things go south.
    let up: OpenSse = |push| {uptime_sse(state_chan.clone(), push)};
    let config = Config {sse: linear_map_from_vector(~[(~"/uptime", up)])};

    handle_connection(config);
    libc::exit(0);
} 

fn uptime_sse(registrar: pipes::SharedChan<StateMesg>, _push: PushChan)
{
    error!("starting sse client");
    do task::spawn_sched(task::ThreadPerCore) |move registrar|
    {
        let (notify_port, notify_chan) = pipes::stream();

        let key = fmt!("uptime %?", ptr::addr_of(&notify_port));
        registrar.send(AddListener(copy key, notify_chan));
    }
}

pub fn handle_connection(config: Config)
{
    error!("starting connection");

    // If we don't do any of these copies the app exits normally. (Although in 
    // the real app there are multiple connections and each connection got its 
    // own copy of Config which worked fine with oldcomms).
    let config = copy config;
    do task::spawn_sched(task::ManualThreads(1)) |copy config|
    {
        let (_sse_port, sse_chan) = pipes::stream();
        let sse_chan = pipes::SharedChan(sse_chan);

        openSse(&config, ~"/uptime", sse_chan);
    }
}

pub fn openSse(config: &Config, path: ~str, push_data: PushChan)
{
    match config.sse.find(&path)
    {
        option::Some(ref opener) =>
        {
            let sse = (*opener)(push_data);
            error!("created %?", sse);
        }
        option::None =>
        {
            fail fmt!("%s was not found in config.sse", path);
        }
    }
}

pub fn linear_map_from_vector<K: cmp::Eq hash::Hash to_bytes::IterBytes, V>
    (vector: &[(K, V)]) -> LinearMap<K, V>
{
    let mut map = linear_map_with_capacity(vector.len());

    for vector.each |&(key, value)|
    {
        map.insert(key, value);
    }

    map
}

Here is the output:

RUST_LOG=3 && rustc test.rs
warning: no debug symbols in executable (-arch x86_64)
rust: ~"starting connection"
rust: task failed at 'Assertion new_count >= 0 failed', test.rs:1
rust: task failed at 'Assertion new_count >= 0 failed', test.rs:1
test(47582,0x1006de000) malloc: *** error for object 0x7fa94040a990: pointer being freed was not allocated
*** set a breakpoint in malloc_error_break to debug
test(47582,0x1007fb000) malloc: *** error for object 0x7fa94040a990: pointer being freed was not allocated
*** set a breakpoint in malloc_error_break to debug
Abort trap: 6

[jesse99]

I think that this is happening because SharedChan is clonable, not copyable. So this may be a dupe of #2828.

[bblum]

Your assessment is correct (it is a dupe).