<commit_before>use std::collections::HashMap;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use abstract_ns::HostResolve;
use futures::Stream;
use futures::future::{Future};
use futures::sync::oneshot::{channel as oneshot, Sender, Receiver};
use futures_cpupool;
use ns_router::{self, SubscribeExt};
use ns_std_threaded;
use self_meter_http::Meter;
use tk_http::server::Proto;
use tk_http;
use tk_listen::ListenExt;
use tokio_core::net::TcpListener;
use tokio_core::reactor::{Handle};

use config::{ListenSocket, ConfigCell};
use incoming::Router;
use chat;
use runtime::Runtime;
use http_pools::{HttpPools};
use handlers::files::{DiskPools};
use request_id;


pub struct State {
    http_pools: HttpPools,
    session_pools: chat::SessionPools,
    disk_pools: DiskPools,
    ns: ns_router::Router,
    listener_shutters: HashMap<SocketAddr, Sender<()>>,
    replication_session: chat::ReplicationSession,
    runtime: Arc<Runtime>,
}

pub fn spawn_listener(addr: SocketAddr, handle: &Handle,
    runtime: &Arc<Runtime>, shutter: Receiver<()>)
    -> Result<(), io::Error>
{
    let root = runtime.config.get();
    let r1 = runtime.clone();
    let r2 = runtime.clone();
    let listener = TcpListener::bind(&addr, &handle)?;
    // TODO(tailhook) how to update?
    let hcfg = tk_http::server::Config::new()
        .inflight_request_limit(root.pipeline_depth)
        // TODO(tailhook) make it configurable?
        .inflight_request_prealoc(0)
        .first_byte_timeout(root.first_byte_timeout)
        .keep_alive_timeout(root.keep_alive_timeout)
        .headers_timeout(root.headers_timeout)
        .input_body_byte_timeout(root.input_body_byte_timeout)
        .input_body_whole_timeout(root.input_body_whole_timeout)
        .output_body_byte_timeout(root.output_body_byte_timeout)
        .output_body_whole_timeout(root.output_body_whole_timeout)
        .done();
    let h1 = handle.clone();

    handle.spawn(
        listener.incoming()
        .sleep_on_error(r1.config.get().listen_error_timeout, &r1.handle)
        .map(move |(socket, saddr)| {
             Proto::new(socket, &hcfg,
                Router::new(saddr, r2.clone(), h1.clone()), &h1)
             .map_err(|e| debug!("Http protocol error: {}", e))
        })
        .listen(root.max_connections)
        .select(shutter.map_err(|_| unreachable!()))
        .map(move |(_, _)| info!("Listener {} exited", addr))
        .map_err(move |(_, _)| info!("Listener {} exited", addr))
    );
    Ok(())
}

pub fn populate_loop(handle: &Handle, cfg: &ConfigCell, verbose: bool)
    -> State
{
    let meter = Meter::new();
    meter.spawn_scanner(handle);
    meter.track_current_thread_by_name();

    let ns_pool = {
        let m1 = meter.clone();
        let m2 = meter.clone();
        futures_cpupool::Builder::new()
        // TODO(tailhook) configure it
        .pool_size(5)
        .name_prefix("ns-resolver-")
        .after_start(move || m1.track_current_thread_by_name())
        .before_stop(move || m2.untrack_current_thread())
        .create()
    };

    let std = ns_std_threaded::ThreadedResolver::use_pool(ns_pool);

    // TODO(tailhook) add config for it, allow update
    let resolver = ns_router::Router::from_config(
        &ns_router::Config::new()
        .set_fallthrough(std.null_service_resolver()
            // TODO(tailhook) allow configure interval
            .interval_subscriber(Duration::new(1, 0), handle))
        .done(),
        handle);

    let server_id = request_id::new();
    let http_pools = HttpPools::new();
    let processor = chat::Processor::new();
    let mut replication_session = chat::ReplicationSession::new(
        processor.clone(), &resolver, handle, &server_id);
    let session_pools = chat::SessionPools::new(
        processor, replication_session.remote_sender.clone());
    let disk_pools = DiskPools::new(&meter);
    let runtime = Arc::new(Runtime {
        config: cfg.clone(),
        handle: handle.clone(),
        http_pools: http_pools.clone(),
        session_pools: session_pools.clone(),
        disk_pools: disk_pools.clone(),
        meter: meter,
        server_id: server_id,
    });
    let root = cfg.get();

    let mut listener_shutters = HashMap::new();

    // TODO(tailhook) do something when config updates
    for sock in &root.listen {
        match sock {<commit_msg>Add a log message on server startup<commit_after>use std::collections::HashMap;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use abstract_ns::HostResolve;
use futures::Stream;
use futures::future::{Future};
use futures::sync::oneshot::{channel as oneshot, Sender, Receiver};
use futures_cpupool;
use ns_router::{self, SubscribeExt};
use ns_std_threaded;
use self_meter_http::Meter;
use tk_http::server::Proto;
use tk_http;
use tk_listen::ListenExt;
use tokio_core::net::TcpListener;
use tokio_core::reactor::{Handle};

use config::{ListenSocket, ConfigCell};
use incoming::Router;
use chat;
use runtime::Runtime;
use http_pools::{HttpPools};
use handlers::files::{DiskPools};
use request_id;


pub struct State {
    http_pools: HttpPools,
    session_pools: chat::SessionPools,
    disk_pools: DiskPools,
    ns: ns_router::Router,
    listener_shutters: HashMap<SocketAddr, Sender<()>>,
    replication_session: chat::ReplicationSession,
    runtime: Arc<Runtime>,
}

pub fn spawn_listener(addr: SocketAddr, handle: &Handle,
    runtime: &Arc<Runtime>, shutter: Receiver<()>)
    -> Result<(), io::Error>
{
    let root = runtime.config.get();
    let r1 = runtime.clone();
    let r2 = runtime.clone();
    let listener = TcpListener::bind(&addr, &handle)?;
    // TODO(tailhook) how to update?
    let hcfg = tk_http::server::Config::new()
        .inflight_request_limit(root.pipeline_depth)
        // TODO(tailhook) make it configurable?
        .inflight_request_prealoc(0)
        .first_byte_timeout(root.first_byte_timeout)
        .keep_alive_timeout(root.keep_alive_timeout)
        .headers_timeout(root.headers_timeout)
        .input_body_byte_timeout(root.input_body_byte_timeout)
        .input_body_whole_timeout(root.input_body_whole_timeout)
        .output_body_byte_timeout(root.output_body_byte_timeout)
        .output_body_whole_timeout(root.output_body_whole_timeout)
        .done();
    let h1 = handle.clone();

    handle.spawn(
        listener.incoming()
        .sleep_on_error(r1.config.get().listen_error_timeout, &r1.handle)
        .map(move |(socket, saddr)| {
             Proto::new(socket, &hcfg,
                Router::new(saddr, r2.clone(), h1.clone()), &h1)
             .map_err(|e| debug!("Http protocol error: {}", e))
        })
        .listen(root.max_connections)
        .select(shutter.map_err(|_| unreachable!()))
        .map(move |(_, _)| info!("Listener {} exited", addr))
        .map_err(move |(_, _)| info!("Listener {} exited", addr))
    );
    Ok(())
}

pub fn populate_loop(handle: &Handle, cfg: &ConfigCell, verbose: bool)
    -> State
{
    let meter = Meter::new();
    meter.spawn_scanner(handle);
    meter.track_current_thread_by_name();

    let ns_pool = {
        let m1 = meter.clone();
        let m2 = meter.clone();
        futures_cpupool::Builder::new()
        // TODO(tailhook) configure it
        .pool_size(5)
        .name_prefix("ns-resolver-")
        .after_start(move || m1.track_current_thread_by_name())
        .before_stop(move || m2.untrack_current_thread())
        .create()
    };

    let std = ns_std_threaded::ThreadedResolver::use_pool(ns_pool);

    // TODO(tailhook) add config for it, allow update
    let resolver = ns_router::Router::from_config(
        &ns_router::Config::new()
        .set_fallthrough(std.null_service_resolver()
            // TODO(tailhook) allow configure interval
            .interval_subscriber(Duration::new(1, 0), handle))
        .done(),
        handle);

    let server_id = request_id::new();
    let http_pools = HttpPools::new();
    let processor = chat::Processor::new();
    let mut replication_session = chat::ReplicationSession::new(
        processor.clone(), &resolver, handle, &server_id);
    let session_pools = chat::SessionPools::new(
        processor, replication_session.remote_sender.clone());
    let disk_pools = DiskPools::new(&meter);
    let runtime = Arc::new(Runtime {
        config: cfg.clone(),
        handle: handle.clone(),
        http_pools: http_pools.clone(),
        session_pools: session_pools.clone(),
        disk_pools: disk_pools.clone(),
        meter: meter,
        server_id: server_id,
    });
    let root = cfg.get();

    warn!("Started with server_id {}", server_id);

    let mut listener_shutters = HashMap::new();

    // TODO(tailhook) do something when config updates
    for sock in &root.listen {
        match sock {
{'fraction_non_alphanumeric': 0.1395112016293279, 'fraction_numerical': 0.0036207286716451684, 'mean_word_length': 3.4799797263051193, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 19, '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': '5781715', 'n_tokens_mistral': 2889, 'n_tokens_neox': 2853, 'n_words': 552}