<commit_before>// Copyright 2015 The Servo Project Developers. See the COPYRIGHT
// file at the top-level directory of this distribution.
//
// 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.

use bincode::serde::DeserializeError;
use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
use libc::{self, MAP_SHARED, PROT_READ, PROT_WRITE, c_char, c_int, c_short, c_ulong};
use libc::{c_ushort, c_void, mode_t, off_t, size_t, sockaddr, sockaddr_un, socklen_t, ssize_t};
use std::cmp;
use std::collections::HashSet;
use std::ffi::{CStr, CString};
use std::fmt::{self, Debug, Formatter};
use std::io::{Error, Write};
use std::mem;
use std::ops::Deref;
use std::ptr;
use std::slice;
use std::sync::atomic::{ATOMIC_USIZE_INIT, AtomicUsize, Ordering};
use std::thread;

const MAX_FDS_IN_CMSG: u32 = 64;

// Yes, really!
const MAP_FAILED: *mut u8 = (!0usize) as *mut u8;

static LAST_FRAGMENT_ID: AtomicUsize = ATOMIC_USIZE_INIT;

pub fn channel() -> Result<(UnixSender, UnixReceiver),UnixError> {
    let mut results = [0, 0];
    unsafe {
        if socketpair(libc::AF_UNIX, SOCK_SEQPACKET, 0, &mut results[0]) >= 0 {
            Ok((UnixSender::from_fd(results[0]), UnixReceiver::from_fd(results[1])))
        } else {
            Err(UnixError::last())
        }
    }
}

#[derive(PartialEq, Debug)]
pub struct UnixReceiver {
    fd: c_int,
}

impl Drop for UnixReceiver {
    fn drop(&mut self) {
        unsafe {
            //assert!(libc::close(self.fd) == 0)
            libc::close(self.fd);
        }
    }
}

impl UnixReceiver {
    fn from_fd(fd: c_int) -> UnixReceiver {
        UnixReceiver {
            fd: fd,
        }
    }

    fn consume_fd(&self) -> c_int {
        unsafe {
            libc::dup(self.fd)
        }
    }

    pub fn consume(&self) -> UnixReceiver {
        UnixReceiver::from_fd(self.consume_fd())
    }

    pub fn recv(&self)
                -> Result<(Vec<u8>, Vec<OpaqueUnixChannel>, Vec<UnixSharedMemory>),UnixError> {
        recv(self.fd, BlockingMode::Blocking)
    }

    pub fn try_recv(&self)
                    -> Result<(Vec<u8>, Vec<OpaqueUnixChannel>, Vec<UnixSharedMemory>),UnixError> {
        recv(self.fd, BlockingMode::Nonblocking)
    }
}

#[derive(PartialEq, Debug)]
pub struct UnixSender {
    fd: c_int,
}

impl Drop for UnixSender {
    fn drop(&mut self) {
        unsafe {
            let result = libc::close(self.fd);
            assert!(thread::panicking() || result == 0);
        }
    }
}

impl Clone for UnixSender {
    fn clone(&self) -> UnixSender {
        unsafe {
            UnixSender {
                fd: libc::dup(self.fd)
            }
        }
    }
}

impl UnixSender {
    fn from_fd(fd: c_int) -> UnixSender {
        UnixSender {
            fd: fd,
        }
    }

    /// Maximum size of the kernel buffer used for transfers over this channel.
    ///
    /// Note: This is *not* the actual maximal packet size we are allowed to use...
    /// Some of it is reserved by the kernel for bookkeeping.
    pub fn get_system_sendbuf_size(&self) -> Result<usize,UnixError> {
        unsafe {
            let mut socket_sendbuf_size: usize = 0;
            let mut socket_sendbuf_size_len = mem::size_of::<usize>() as socklen_t;
            if getsockopt(self.fd,
                          libc::SOL_SOCKET,
                          libc::SO_SNDBUF,
                          &mut socket_sendbuf_size as *mut usize as *mut c_void,
                          &mut socket_sendbuf_size_len as *mut socklen_t) < 0 {
                return Err(UnixError::last())
            }
            Ok(socket_sendbuf_size)
        }
    }

    pub fn send(&self,
                data: &[u8],
                channels: Vec<UnixChannel>,
                shared_memory_regions: Vec<UnixSharedMemory>)
                -> Result<(),UnixError> {

        let mut fds = Vec::new();
        for channel in channels.iter() {
            fds.push(channel.fd());
        }
        for shared_memory_region in shared_memory_regions.iter() {
            fds.push(shared_memory_region.fd);
        }

        let mut data_buffer = vec![0; data.len() + mem::size_of::<u32>() * 2];
        {
            let mut data_buffer = &mut data_buffer[..];
            data_buffer.write_u32::<LittleEndian>(0u32).unwrap();
            data_buffer.write_u32::<LittleEndian>(0u32).unwrap();
            data_buffer.write(data).unwrap();
        }

        unsafe {
            unsafe fn construct_header(fds: &[c_int], data_buffer: &[u8]) -> (msghdr, Box<iovec>) {
                let cmsg_length = fds.len() * mem::size_of::<c_int>();
                let cmsg_buffer = libc::malloc(CMSG_SPACE(cmsg_length)) as *mut cmsghdr;
                (*cmsg_buffer).cmsg_len = CMSG_LEN(cmsg_length);
                (*cmsg_buffer).cmsg_level = libc::SOL_SOCKET;
                (*cmsg_buffer).cmsg_type = SCM_RIGHTS;

                ptr::copy_nonoverlapping(fds.as_ptr(),
                                         cmsg_buffer.offset(1) as *mut _ as *mut c_int,
                                         fds.len());

                // Put this on the heap so address remains stable across function return.
                let mut iovec = Box::new(iovec {
                    iov_base: data_buffer.as_ptr() as *const c_char as *mut c_char,
                    iov_len: data_buffer.len(),
                });
<commit_msg>Linux: Cleanup: Use mem::size_of_val() instead of manual calculation<commit_after>// Copyright 2015 The Servo Project Developers. See the COPYRIGHT
// file at the top-level directory of this distribution.
//
// 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.

use bincode::serde::DeserializeError;
use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
use libc::{self, MAP_SHARED, PROT_READ, PROT_WRITE, c_char, c_int, c_short, c_ulong};
use libc::{c_ushort, c_void, mode_t, off_t, size_t, sockaddr, sockaddr_un, socklen_t, ssize_t};
use std::cmp;
use std::collections::HashSet;
use std::ffi::{CStr, CString};
use std::fmt::{self, Debug, Formatter};
use std::io::{Error, Write};
use std::mem;
use std::ops::Deref;
use std::ptr;
use std::slice;
use std::sync::atomic::{ATOMIC_USIZE_INIT, AtomicUsize, Ordering};
use std::thread;

const MAX_FDS_IN_CMSG: u32 = 64;

// Yes, really!
const MAP_FAILED: *mut u8 = (!0usize) as *mut u8;

static LAST_FRAGMENT_ID: AtomicUsize = ATOMIC_USIZE_INIT;

pub fn channel() -> Result<(UnixSender, UnixReceiver),UnixError> {
    let mut results = [0, 0];
    unsafe {
        if socketpair(libc::AF_UNIX, SOCK_SEQPACKET, 0, &mut results[0]) >= 0 {
            Ok((UnixSender::from_fd(results[0]), UnixReceiver::from_fd(results[1])))
        } else {
            Err(UnixError::last())
        }
    }
}

#[derive(PartialEq, Debug)]
pub struct UnixReceiver {
    fd: c_int,
}

impl Drop for UnixReceiver {
    fn drop(&mut self) {
        unsafe {
            //assert!(libc::close(self.fd) == 0)
            libc::close(self.fd);
        }
    }
}

impl UnixReceiver {
    fn from_fd(fd: c_int) -> UnixReceiver {
        UnixReceiver {
            fd: fd,
        }
    }

    fn consume_fd(&self) -> c_int {
        unsafe {
            libc::dup(self.fd)
        }
    }

    pub fn consume(&self) -> UnixReceiver {
        UnixReceiver::from_fd(self.consume_fd())
    }

    pub fn recv(&self)
                -> Result<(Vec<u8>, Vec<OpaqueUnixChannel>, Vec<UnixSharedMemory>),UnixError> {
        recv(self.fd, BlockingMode::Blocking)
    }

    pub fn try_recv(&self)
                    -> Result<(Vec<u8>, Vec<OpaqueUnixChannel>, Vec<UnixSharedMemory>),UnixError> {
        recv(self.fd, BlockingMode::Nonblocking)
    }
}

#[derive(PartialEq, Debug)]
pub struct UnixSender {
    fd: c_int,
}

impl Drop for UnixSender {
    fn drop(&mut self) {
        unsafe {
            let result = libc::close(self.fd);
            assert!(thread::panicking() || result == 0);
        }
    }
}

impl Clone for UnixSender {
    fn clone(&self) -> UnixSender {
        unsafe {
            UnixSender {
                fd: libc::dup(self.fd)
            }
        }
    }
}

impl UnixSender {
    fn from_fd(fd: c_int) -> UnixSender {
        UnixSender {
            fd: fd,
        }
    }

    /// Maximum size of the kernel buffer used for transfers over this channel.
    ///
    /// Note: This is *not* the actual maximal packet size we are allowed to use...
    /// Some of it is reserved by the kernel for bookkeeping.
    pub fn get_system_sendbuf_size(&self) -> Result<usize,UnixError> {
        unsafe {
            let mut socket_sendbuf_size: usize = 0;
            let mut socket_sendbuf_size_len = mem::size_of::<usize>() as socklen_t;
            if getsockopt(self.fd,
                          libc::SOL_SOCKET,
                          libc::SO_SNDBUF,
                          &mut socket_sendbuf_size as *mut usize as *mut c_void,
                          &mut socket_sendbuf_size_len as *mut socklen_t) < 0 {
                return Err(UnixError::last())
            }
            Ok(socket_sendbuf_size)
        }
    }

    pub fn send(&self,
                data: &[u8],
                channels: Vec<UnixChannel>,
                shared_memory_regions: Vec<UnixSharedMemory>)
                -> Result<(),UnixError> {

        let mut fds = Vec::new();
        for channel in channels.iter() {
            fds.push(channel.fd());
        }
        for shared_memory_region in shared_memory_regions.iter() {
            fds.push(shared_memory_region.fd);
        }

        let mut data_buffer = vec![0; data.len() + mem::size_of::<u32>() * 2];
        {
            let mut data_buffer = &mut data_buffer[..];
            data_buffer.write_u32::<LittleEndian>(0u32).unwrap();
            data_buffer.write_u32::<LittleEndian>(0u32).unwrap();
            data_buffer.write(data).unwrap();
        }

        unsafe {
            unsafe fn construct_header(fds: &[c_int], data_buffer: &[u8]) -> (msghdr, Box<iovec>) {
                let cmsg_length = mem::size_of_val(fds);
                let cmsg_buffer = libc::malloc(CMSG_SPACE(cmsg_length)) as *mut cmsghdr;
                (*cmsg_buffer).cmsg_len = CMSG_LEN(cmsg_length);
                (*cmsg_buffer).cmsg_level = libc::SOL_SOCKET;
                (*cmsg_buffer).cmsg_type = SCM_RIGHTS;

                ptr::copy_nonoverlapping(fds.as_ptr(),
                                         cmsg_buffer.offset(1) as *mut _ as *mut c_int,
                                         fds.len());

                // Put this on the heap so address remains stable across function return.
                let mut iovec = Box::new(iovec {
                    iov_base: data_buffer.as_ptr() as *const c_char as *mut c_char,
                    iov_len: data_buffer.len(),
                });

{'fraction_non_alphanumeric': 0.12607806526184762, 'fraction_numerical': 0.008002133902373967, 'mean_word_length': 2.1427773120983513, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 68, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1164879', 'n_tokens_mistral': 3728, 'n_tokens_neox': 3589, 'n_words': 826}