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)]
#![feature(test)]
extern crate test;
extern crate tokio;
#[macro_use(try_ready)]
extern crate futures;
extern crate env_logger;
#[macro_use]
extern crate log;
extern crate bytes;
#[macro_use]
extern crate lazy_static;
extern crate btoi;
extern crate itoa;
extern crate net2;
extern crate tokio_codec;
#[macro_use]
extern crate serde_derive;
extern crate hashbrown;
extern crate md5;
extern crate toml;
mod cluster;
mod com;
mod crc;
mod mc;
mod notify;
mod proxy;
mod redis;
use cluster::{start_cluster, Cluster};
pub use com::*;
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 = usize::max(cc.thread, 1);
info!(
"asswecan 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);
}
_ => {
warn!("cache type is not supported");
}
})
.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 ping_fail_limit: Option<usize>,
pub ping_interval: Option<usize>,
pub thread: usize,
pub cache_type: CacheType,
pub servers: Vec<String>,
pub fetch: Option<u64>,
pub dial_timeout: Option<u64>,
pub read_timeout: Option<u64>,
pub write_timeout: Option<u64>,
pub listen_proto: Option<String>,
pub node_connections: Option<String>,
}