<commit_before>//! Broker fuzzer.  See README.md for a full description.

use std::collections::HashMap;
use std::thread;
use std::time::Duration;

use libc::c_void;
use rand::{self, Rng};

use futures::{Future, Stream, Sink, Complete};
use futures::sync::mpsc::{unbounded, Sender};

use tickgrinder_util::strategies::Strategy;
use tickgrinder_util::trading::broker::{Broker, BrokerResult};
use tickgrinder_util::trading::objects::BrokerAction;
use tickgrinder_util::trading::tick::Tick;
use tickgrinder_util::transport::command_server::CommandServer;
use tickgrinder_util::transport::query_server::QueryServer;
use tickgrinder_util::transport::textlog::get_logger_handle;
use tickgrinder_util::conf::CONF;

// link with the libboost_random wrapper
#[link(name="rand_bindings")]
extern {
    fn init_rng(seed: u32) -> *mut c_void;
    fn rand_int_range(void_rng: *mut c_void, min: i32, max: i32) -> u32;
}

pub struct Fuzzer {
    pub gen: *mut c_void,
    pub logger: EventLogger,
    pub pairs: Vec<String>,
}

impl Strategy for Fuzzer {
    fn new(cs: CommandServer, qs: QueryServer, conf: HashMap<String, String>) -> Fuzzer {
        // convert the seed string into an integer we can use to seen the PNRG if deterministic fuzzing is enabled
        let seed: u32 = if CONF.fuzzer_deterministic_rng {
            let mut sum = 0;
            // convert the seed string into an integer for seeding the fuzzer
            for c in CONF.fuzzer_seed.chars() {
                sum += c as u32;
            }
            sum
        } else {
            let mut rng = rand::thread_rng();
            rng.gen()
        };

        // parse the settings HashMap to get the list of pairs to subscribe to
        let pairs_list = conf.get("pairs")
            .expect("This needs a list of pairs to subscribe to from the connected broker, else we can't do anything!");
        let pairs: Vec<String> = (&pairs_list).split(", ").map(|x| String::from(x)).collect();

        Fuzzer {
            gen: unsafe { init_rng(seed)},
            logger: EventLogger::new(),
            pairs: pairs,
        }
    }

    fn init(&mut self, mut broker: Box<Broker>) {
        // subscribe to all the tickstreams as supplied in the configuration and combine the streams
        let (streams_tx, streams_rx) = unbounded();
        let mut symbol_enumeration = Vec::new(); // way to match symbols with their id
        for (i, symbol) in (&self.pairs).iter().enumerate() {
            let streams_tx = &streams_tx;
            symbol_enumeration.push((i, symbol,));
            let rx = broker.sub_ticks(symbol.clone())
                .expect(&format!("Unable to sub ticks for symbol {}", symbol))
                .map(move |t| (i, t));
            streams_tx.send(rx).unwrap();
        }
        let master_rx = streams_rx.flatten();
        self.logger.log_misc(format!("Subscribed to {} tickstreams", symbol_enumeration.len()));

        let pushstream_rx = broker.get_stream().unwrap();
        thread::spawn(move || {
            for msg in pushstream_rx.wait() {
                println!("PUSHTREAM: {:?}", msg.unwrap());
            }
        });

        // start responding to ticks from all the streams.
        self.logger.log_misc(String::from("Initializing fuzzer loop..."));
        for msg in master_rx.wait() {
            let (i, t) = msg.unwrap();
            self.logger.log_tick(t, i);

            match get_action(t, self.gen) {
                Some(action) => {
                    let id = self.logger.log_request(&action);
                    let fut = broker.execute(action);
                    let res = fut.wait().unwrap();
                    self.logger.log_response(&res, id);
                },
                None => (),
            }
        }
    }

    fn exit_now(&mut self, ready: Complete<()>) {
        unimplemented!();
    }
}

/// Called during each iteration of the fuzzer loop.  Picks a random action to take based on the
/// internally held PRNG and executes it.
pub fn get_action(t: Tick, gen: *mut c_void) -> Option<BrokerAction> {
    let rand = unsafe { rand_int_range(gen, 0, 5) };
    let action_opt: Option<BrokerAction> = match rand {
        0 => Some(BrokerAction::Ping),
        1 => None, // TODO
        // sleep for a few milliseconds, then do either do nothing or perform an action
        2 => {
            let sleep_time = unsafe { rand_int_range(gen, 0, 25) };
            thread::sleep(Duration::from_millis(sleep_time as u64));
            let action_or_no = unsafe { rand_int_range(gen, 0, 5) };
            if action_or_no > 3 {
                get_action(t, gen)
            } else {
                None
            }
        },
        // do nothing at all in response to the tick
        _ => None,
    };

    action_opt
}

pub struct EventLogger {
    tx: Option<Sender<String>>,
    i: usize, // incremented every time an event is logged in order to record order
}
<commit_msg>Implemented beginnings of the callback handler for Fuzzer  - All oneshot receivers returned by broker actions are pushed into a queue where they are processed in the order they are fulfilled  - Looks like we got some order debugging to do still but things look pretty ok atm.  - Integrated the events stream into the fuzzer loop and tested  - Deadlocks are gone!  Testing works as expected. (HURRAY)<commit_after>//! Broker fuzzer.  See README.md for a full description.

use std::collections::HashMap;
use std::thread;
use std::time::Duration;
use std::sync::mpsc;

use libc::c_void;
use rand::{self, Rng};

use futures::{Future, Stream, Sink, Complete, Canceled, BoxFuture};
use futures::stream::{BufferUnordered, MapErr, BoxStream};
use futures::sync::mpsc::{unbounded, channel, Sender, Receiver};
use futures::sync::oneshot;

use tickgrinder_util::strategies::Strategy;
use tickgrinder_util::trading::broker::{Broker, BrokerResult};
use tickgrinder_util::trading::objects::BrokerAction;
use tickgrinder_util::trading::tick::Tick;
use tickgrinder_util::transport::command_server::CommandServer;
use tickgrinder_util::transport::query_server::QueryServer;
use tickgrinder_util::transport::textlog::get_logger_handle;
use tickgrinder_util::conf::CONF;

// link with the libboost_random wrapper
#[link(name="rand_bindings")]
extern {
    fn init_rng(seed: u32) -> *mut c_void;
    fn rand_int_range(void_rng: *mut c_void, min: i32, max: i32) -> u32;
}

pub struct Fuzzer {
    pub gen: *mut c_void,
    pub logger: EventLogger,
    pub pairs: Vec<String>,
    pub events_tx: Option<Sender<oneshot::Receiver<BrokerResult>>>,
    pub events_rx: mpsc::Receiver<BrokerResult>,
}

impl Strategy for Fuzzer {
    fn new(cs: CommandServer, qs: QueryServer, conf: HashMap<String, String>) -> Fuzzer {
        // convert the seed string into an integer we can use to seen the PNRG if deterministic fuzzing is enabled
        let seed: u32 = if CONF.fuzzer_deterministic_rng {
            let mut sum = 0;
            // convert the seed string into an integer for seeding the fuzzer
            for c in CONF.fuzzer_seed.chars() {
                sum += c as u32;
            }
            sum
        } else {
            let mut rng = rand::thread_rng();
            rng.gen()
        };

        // parse the settings HashMap to get the list of pairs to subscribe to
        let pairs_list = conf.get("pairs")
            .expect("This needs a list of pairs to subscribe to from the connected broker, else we can't do anything!");
        let pairs: Vec<String> = (&pairs_list).split(", ").map(|x| String::from(x)).collect();

        // create the stream over which we receive callbacks from the broker
        let buffer_size = 8;//pairs.len() * 2;
        let (tx, rx) = channel(buffer_size);

        // map the output of the events buffer strea into a stdlib mpsc channel so we can try_get it
        let (mpsc_tx, mpsc_rx) = mpsc::sync_channel::<BrokerResult>(0);
        thread::spawn(move || {
            let mod_rx: BoxStream<oneshot::Receiver<BrokerResult>, Canceled> = rx.map_err(|()| Canceled).boxed();
            let buf_rx = mod_rx.buffer_unordered(buffer_size);
            for msg in buf_rx.wait() {
                println!("Mapping message through mpsc...");
                mpsc_tx.send(msg.unwrap()).unwrap();
            }
        });

        Fuzzer {
            gen: unsafe { init_rng(seed)},
            logger: EventLogger::new(),
            pairs: pairs,
            events_tx: Some(tx),
            events_rx: mpsc_rx,
        }
    }

    fn init(&mut self, mut broker: Box<Broker>) {
        // subscribe to all the tickstreams as supplied in the configuration and combine the streams
        let (streams_tx, streams_rx) = unbounded();
        let mut symbol_enumeration = Vec::new(); // way to match symbols with their id
        for (i, symbol) in (&self.pairs).iter().enumerate() {
            let streams_tx = &streams_tx;
            symbol_enumeration.push((i, symbol,));
            let rx = broker.sub_ticks(symbol.clone())
                .expect(&format!("Unable to sub ticks for symbol {}", symbol))
                .map(move |t| (i, t));
            streams_tx.send(rx).unwrap();
        }
        let master_rx = streams_rx.flatten();
        self.logger.log_misc(format!("Subscribed to {} tickstreams", symbol_enumeration.len()));

        let pushstream_rx = broker.get_stream().unwrap();
        thread::spawn(move || {
            for msg in pushstream_rx.wait() {
                println!("PUSHTREAM: {:?}", msg.unwrap());
                // TODO: log/handle
            }
        });

        // start responding to ticks from all the streams.
        self.logger.log_misc(String::from("Initializing fuzzer loop..."));
        for msg in master_rx.wait() {
            while let Ok(msg) = self.events_rx.try_recv() {
                self.logger.log_misc(format!("EVENT: {:?}", msg));
            }

            let (i, t) = msg.unwrap();
            self.logger.log_tick(t, i);

            match get_action(t, self.gen) {
                Some(action) => {
                    let id = self.logger.log_request(&action);
                    let fut = broker.execute(action);
                    // store the pending future into the buffered queue
                    let mut tx = self.events_tx.take().unwrap();
                    tx = tx.send(fut).wait().unwrap();
                    self.events_tx = Some(tx);
                    // self.logger.log_response(&res, id);
                },
                None => (),
            }
        }
    }

    fn exit_now(&mut self, ready: Complete<()>) {
        unimplemented!();
    }
}

/// Called during each iteration of the fuzzer loop.  Picks a random action to take based on the
/// internally held PRNG and executes it.
pub fn get_action(t: Tick, gen: *mut c_void) -> Option<BrokerAction> {
    let rand = unsafe { rand_int_range(gen, 0, 5) };
    let action_opt: Option<BrokerAction> = match rand {
        0 => Some(BrokerAction::Ping),
        1 => None, // TODO
        // sleep for a few milliseconds, then do either do nothing or perform an action
        2 => {
            let action_or_no = unsafe { rand_int_range(gen, 0, 5) };
            if action_or_no > 3 {
                get_action(t, gen)
            } else {
                None
            }
        },
        // do nothing at all in response to the tick
        _ => None,
    };

    action_opt
}

pub struct EventLogger {
    tx: Option<Sender<String>>,
    i: usize, // incremented every time an event is logged in order to record order
}

{'fraction_non_alphanumeric': 0.1189404432132964, 'fraction_numerical': 0.00432825484764543, 'mean_word_length': 2.3198275862068964, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 50, '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': '4806552', 'n_tokens_mistral': 3500, 'n_tokens_neox': 3393, 'n_words': 1013}