<commit_before>use iron::middleware::Handler;
use iron::prelude::*;
use iron::status;
use std::process;
use std::error::Error;

use std::ffi::OsString;

use rustc_serialize::json;

use rand::Rng;
use rand;

use std::sync::{Mutex, RwLock, Barrier, Arc};
use std::thread;
use std::io::{BufReader, BufRead};

#[derive(Debug,RustcDecodable,RustcEncodable,Clone,PartialEq)]
struct StartSuccessful {
    id: String
}

#[derive(Debug,RustcDecodable,RustcEncodable,Clone,PartialEq)]
struct LogOutput {
    id: String,
    lines: Vec<String>,
    next_line: usize
}

struct RunnerState {
    /// Current process identifier, process will stop if identifier is changed
    id: Mutex<String>,

    /// The output of the current process
    output: RwLock<Vec<String>>,

    /// Barrier used to check if starting the subprocess (owned by a thread) went ok
    start_barrier: Barrier,

    /// Any error
    error: Mutex<Option<Result<(), ::std::io::Error>>>
}

pub struct ProcessRunner {
    pub program: OsString,
    state: Arc<RunnerState>
}

impl ProcessRunner {
    pub fn new<S: Into<String>>(s: S) -> ProcessRunner {
        ProcessRunner {
            program: OsString::from(s.into()),
            state: Arc::new(RunnerState {
                id: Mutex::new("".to_owned()),
                output: RwLock::new(vec![]),
                start_barrier: Barrier::new(2),
                error: Mutex::new(None)
            })
        }
    }

    fn start(&self) -> Result<(), ::std::io::Error> {
        let id : String = rand::thread_rng().gen_ascii_chars().take(10).collect();
        *self.state.id.lock().unwrap() = id.clone();
        self.state.output.write().unwrap().clear();

        let state = self.state.clone();
        let program = self.program.clone();

        thread::spawn(move || {
            let child_result = process::Command::new(&program)
                    .stdout(process::Stdio::piped())
                    .spawn();
            if child_result.is_err() {
                {
                    *state.error.lock().unwrap() = Some(child_result.map(|_| ()));
                }
                state.start_barrier.wait();
                return;
            }

            {
                *state.error.lock().unwrap() = Some(Ok(()));
            }
            state.start_barrier.wait();

            if let Ok(mut child) = child_result {
                {
                    let mut reader = BufReader::new(child.stdout.as_mut().unwrap());
                    let mut s = String::new();
                    loop {
                        {
                            if *state.id.lock().unwrap() != id { break; } // End of the party boys, ID changed
                        }

                        match reader.read_line(&mut s) {
                            Ok(n) if n > 0 => (),
                            _ => break
                        };

                        {
                            state.output.write().unwrap().push(s.clone());
                        }
                        s.clear();
                    }
                }

                if let Some(err) = child.kill().err() {
                    println!("Error killing subprocess: {}", err);
                } else {
                    println!("Subprocess stopped");
                }
            }
        });

        self.state.start_barrier.wait();
        {
            self.state.error.lock().unwrap().take().unwrap()
        }
    }

    fn stop(&self) -> Result<(), Box<Error>> {
        // This will cause the background thread to terminate
        self.state.id.lock().unwrap().clear();
        Ok(())
    }

    fn read(&self, line: usize) -> LogOutput {
        let output = self.state.output.read().unwrap();

        let len = output.len();

        LogOutput {
            id: self.state.id.lock().unwrap().to_owned(),
            lines: output[::std::cmp::min(line, len)..].iter().map(|x| x.to_owned()).collect(),
            next_line: len
        }
    }

    fn do_stop(&self) -> Result<String, Box<Error>> {
        try!(self.stop());
        Ok("ok".to_owned())<commit_msg>Make terminating started processes more reliable<commit_after>use iron::middleware::Handler;
use iron::prelude::*;
use iron::status;
use std::process;
use std::error::Error;

use std::ffi::OsString;

use rustc_serialize::json;

use rand::Rng;
use rand;

use std::sync::{Mutex, RwLock, Arc};
use std::thread;
use std::io::{BufReader, BufRead};

#[derive(Debug,RustcDecodable,RustcEncodable,Clone,PartialEq)]
struct StartSuccessful {
    id: String
}

#[derive(Debug,RustcDecodable,RustcEncodable,Clone,PartialEq)]
struct LogOutput {
    id: String,
    lines: Vec<String>,
    next_line: usize
}

struct RunnerState {
    /// Current process identifier, process will stop if identifier is changed
    id: Mutex<String>,

    /// The output of the current process
    output: RwLock<Vec<String>>,

    /// Handle to the child process
    child: Mutex<Option<process::Child>>
}

pub struct ProcessRunner {
    pub program: OsString,
    state: Arc<RunnerState>
}

impl ProcessRunner {
    pub fn new<S: Into<String>>(s: S) -> ProcessRunner {
        ProcessRunner {
            program: OsString::from(s.into()),
            state: Arc::new(RunnerState {
                id: Mutex::new("".to_owned()),
                output: RwLock::new(vec![]),
                child: Mutex::new(None)
            })
        }
    }

    fn start(&self) -> Result<(), ::std::io::Error> {
        let id : String = rand::thread_rng().gen_ascii_chars().take(10).collect();
        *self.state.id.lock().unwrap() = id.clone();
        self.state.output.write().unwrap().clear();

        let state = self.state.clone();
        let program = self.program.clone();

        let mut child = try!(process::Command::new(&program)
                .stdout(process::Stdio::piped())
                .spawn());

        let out = child.stdout.take().expect("No stdout on child");

        *state.child.lock().unwrap() = Some(child);

        thread::spawn(move || {
            let mut reader = BufReader::new(out);
            let mut s = String::new();
            loop {
                {
                    if *state.id.lock().unwrap() != id { break; } // End of the party boys, ID changed
                }

                match reader.read_line(&mut s) {
                    Ok(n) if n > 0 => (),
                    _ => break
                };

                {
                    state.output.write().unwrap().push(s.clone());
                }
                s.clear();
            }
        });

        Ok(())
    }

    fn stop(&self) -> Result<(), Box<Error>> {
        // This will cause the background thread to terminate
        self.state.id.lock().unwrap().clear();
        for child in self.state.child.lock().unwrap().iter_mut() {
            if let Some(err) = child.kill().err() {
                println!("Error killing subprocess: {}", err);
            } else {
                println!("Subprocess stopped");
            }
        }
        Ok(())
    }

    fn read(&self, line: usize) -> LogOutput {
        let output = self.state.output.read().unwrap();

        let len = output.len();

        LogOutput {
            id: self.state.id.lock().unwrap().to_owned(),
            lines: output[::std::cmp::min(line, len)..].iter().map(|x| x.to_owned()).collect(),
            next_line: len
        }
    }

    fn do_stop(&self) -> Result<String, Box<Error>> {
        try!(self.stop());
        Ok("ok".to_owned())
{'fraction_non_alphanumeric': 0.1550006640988179, 'fraction_numerical': 0.0009297383450657458, 'mean_word_length': 1.7796234772978958, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 48, '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': '7099363', 'n_tokens_mistral': 2227, 'n_tokens_neox': 2171, 'n_words': 453}