<commit_before>use std::sync::{mpsc, Arc};
use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use user_event::UserEvent;

#[cfg(any(target_os = "linux", target_os = "android"))]
use epoll::KernelRegistrar;

#[cfg(any(target_os = "bitrig", target_os = "dragonfly",
          target_os = "freebsd", target_os = "ios", target_os = "macos",
          target_os = "netbsd", target_os = "openbsd"))]
pub use kqueue::KernelRegistrar;

pub fn channel<T>(registrar: &mut KernelRegistrar) -> io::Result<(Sender<T>, Receiver<T>)> {
    let (tx, rx) = mpsc::channel();
    let pending = Arc::new(AtomicUsize::new(0));
    let user_event = Arc::new(registrar.register_user_event().map_err(|e| io::Error::from(e))?);

    let tx = Sender {
        tx: tx,
        user_event: user_event.clone(),
        pending: pending.clone()
    };

    let rx = Receiver {
        rx: rx,
        user_event: user_event,
        pending: pending
    };

    Ok((tx, rx))
}

pub fn sync_channel<T>(registrar: &mut KernelRegistrar,
                       bound: usize) -> io::Result<(SyncSender<T>, Receiver<T>)> {
    let (tx, rx) = mpsc::sync_channel(bound);
    let pending = Arc::new(AtomicUsize::new(0));
    let user_event = Arc::new(registrar.register_user_event().map_err(|e| io::Error::from(e))?);

    let tx = SyncSender {
        tx: tx,
        user_event: user_event.clone(),
        pending: pending.clone()
    };

    let rx = Receiver {
        rx: rx,
        user_event: user_event,
        pending: pending
    };

    Ok((tx, rx))
}


#[derive(Debug, Clone)]
pub struct Sender<T> {
    tx: mpsc::Sender<T>,
    user_event: Arc<UserEvent>,
    pending: Arc<AtomicUsize>
}

impl<T> Sender<T> {
    pub fn send(&self, msg: T) -> Result<(), ChannelError<T>> {
        try!(self.tx.send(msg));
        if self.pending.fetch_add(1, Ordering::SeqCst) == 0 {
            // Notify the kernel poller that a read is ready
            try!(self.user_event.trigger());
        }
        Ok(())
    }

    // Return the poll id for the channel
    pub fn get_id(&self) -> usize {
        self.user_event.get_id()
    }
}

#[derive(Debug, Clone)]
pub struct SyncSender<T> {
    tx: mpsc::SyncSender<T>,
    user_event: Arc<UserEvent>,
    pending: Arc<AtomicUsize>
}<commit_msg>Manually implement Sender clone, so `T` does not have to be clone<commit_after>use std::sync::{mpsc, Arc};
use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use user_event::UserEvent;

#[cfg(any(target_os = "linux", target_os = "android"))]
use epoll::KernelRegistrar;

#[cfg(any(target_os = "bitrig", target_os = "dragonfly",
          target_os = "freebsd", target_os = "ios", target_os = "macos",
          target_os = "netbsd", target_os = "openbsd"))]
pub use kqueue::KernelRegistrar;

pub fn channel<T>(registrar: &mut KernelRegistrar) -> io::Result<(Sender<T>, Receiver<T>)> {
    let (tx, rx) = mpsc::channel();
    let pending = Arc::new(AtomicUsize::new(0));
    let user_event = Arc::new(registrar.register_user_event().map_err(|e| io::Error::from(e))?);

    let tx = Sender {
        tx: tx,
        user_event: user_event.clone(),
        pending: pending.clone()
    };

    let rx = Receiver {
        rx: rx,
        user_event: user_event,
        pending: pending
    };

    Ok((tx, rx))
}

pub fn sync_channel<T>(registrar: &mut KernelRegistrar,
                       bound: usize) -> io::Result<(SyncSender<T>, Receiver<T>)> {
    let (tx, rx) = mpsc::sync_channel(bound);
    let pending = Arc::new(AtomicUsize::new(0));
    let user_event = Arc::new(registrar.register_user_event().map_err(|e| io::Error::from(e))?);

    let tx = SyncSender {
        tx: tx,
        user_event: user_event.clone(),
        pending: pending.clone()
    };

    let rx = Receiver {
        rx: rx,
        user_event: user_event,
        pending: pending
    };

    Ok((tx, rx))
}


#[derive(Debug)]
pub struct Sender<T> {
    tx: mpsc::Sender<T>,
    user_event: Arc<UserEvent>,
    pending: Arc<AtomicUsize>
}

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Sender<T> {
        Sender {
            tx: self.tx.clone(),
            user_event: self.user_event.clone(),
            pending: self.pending.clone()
        }
    }
}

impl<T> Sender<T> {
    pub fn send(&self, msg: T) -> Result<(), ChannelError<T>> {
        try!(self.tx.send(msg));
        if self.pending.fetch_add(1, Ordering::SeqCst) == 0 {
            // Notify the kernel poller that a read is ready
            try!(self.user_event.trigger());
        }
        Ok(())
    }

    // Return the poll id for the channel
    pub fn get_id(&self) -> usize {
        self.user_event.get_id()
    }
}

#[derive(Debug, Clone)]
pub struct SyncSender<T> {
    tx: mpsc::SyncSender<T>,
    user_event: Arc<UserEvent>,
    pending: Arc<AtomicUsize>
}
{'fraction_non_alphanumeric': 0.1643778610070745, 'fraction_numerical': 0.0016645859342488557, 'mean_word_length': 2.882875605815832, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7170306', 'n_tokens_mistral': 1726, 'n_tokens_neox': 1713, 'n_words': 364}