1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

#![deny(warnings)]
#[macro_use(try_ready)]
extern crate futures;
#[macro_use]
extern crate log;
#[macro_use]
extern crate lazy_static;
#[macro_use]
extern crate serde_derive;

mod cluster;
mod com;
mod crc;
mod mc;
mod mcbin;
mod notify;
pub mod proxy;
pub mod redis;

use crate::cluster::{start_cluster, Cluster};
pub use crate::com::*;
// use futures::task::current;
use std::thread;

pub fn run() -> Result<(), std::io::Error> {
    env_logger::init();
    let config = load_config();
    info!("aster has shined with config={:?}", config);
    let ths: Vec<_> = config
        .clusters
        .iter()
        .map(|cc| create_cluster(cc))
        .flatten()
        .collect();

    for th in ths {
        th.join().unwrap();
    }
    Ok(())
}

fn load_config() -> Config {
    use std::env;
    let path = env::var("AS_CFG").unwrap_or_else(|_| "as.toml".to_string());
    use std::fs;
    use std::io::{BufReader, Read};

    let fd = fs::File::open(&path).expect("fail to open config file(default: as.toml)");
    let mut rd = BufReader::new(fd);
    let mut data = String::new();
    rd.read_to_string(&mut data)
        .expect("fail to read config file");

    toml::from_str(&data).expect("fail to parse toml")
}

pub fn create_cluster(cc: &ClusterConfig) -> Vec<thread::JoinHandle<()>> {
    let count = if let Some(&thread) = cc.thread.as_ref() {
        usize::max(thread, 1)
    } else {
        num_cpus::get()
    };
    info!(
        "aster start {} listen at {} with {} thread",
        &cc.name, &cc.listen_addr, count
    );
    (0..count)
        .map(|i| {
            let cc = cc.clone();
            let name = cc.name.clone();
            let thb = thread::Builder::new().name(format!("cluster-{}-{}", name, i + 1));
            thb.spawn(move || match cc.cache_type {
                CacheType::Memcache => {
                    let p = proxy::Proxy::new(cc).unwrap();
                    proxy::start_proxy::<mc::Req>(p);
                }
                CacheType::RedisCluster => {
                    let cluster = Cluster::new(cc);
                    start_cluster(cluster)
                }
                CacheType::Redis => {
                    let p = proxy::Proxy::new(cc).unwrap();
                    proxy::start_proxy::<redis::cmd::Cmd>(p);
                }
                CacheType::MemcacheBinary => {
                    let p = proxy::Proxy::new(cc).unwrap();
                    proxy::start_proxy::<mcbin::Req>(p);
                }
            })
            .unwrap()
        })
        .collect()
}
#[derive(Deserialize, Debug)]
pub struct Config {
    clusters: Vec<ClusterConfig>,
}

#[derive(Deserialize, Debug, Clone, Copy)]
pub enum CacheType {
    #[serde(rename = "redis")]
    Redis,
    #[serde(rename = "memcache")]
    Memcache,
    #[serde(rename = "memcache_binary")]
    MemcacheBinary,
    #[serde(rename = "redis_cluster")]
    RedisCluster,
}

#[derive(Clone, Debug, Deserialize)]
pub struct ClusterConfig {
    pub name: String,
    pub listen_addr: String,
    pub hash_tag: Option<String>,

    pub thread: Option<usize>,
    pub cache_type: CacheType,

    pub read_timeout: Option<u64>,
    pub write_timeout: Option<u64>,

    pub servers: Vec<String>,

    // cluster special
    pub fetch: Option<u64>,
    pub read_from_slave: Option<bool>,

    // proxy special
    pub ping_fail_limit: Option<usize>,
    pub ping_interval: Option<usize>,

    // dead codes

    // command not support now
    pub dial_timeout: Option<u64>,
    // dead option: not support other proto
    pub listen_proto: Option<String>,

    // dead option: always 1
    pub node_connections: Option<usize>,
}