dbeel 0.1.0

dbeel is a fast and minimal distributed thread-per-core document database.
Documentation 
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

use dbeel::{
    args::parse_args,
    error::{Error, Result},
    local_shard::LocalShardConnection,
    run_shard::{create_shard, run_shard},
};
use glommio::{enclose, CpuSet, LocalExecutorBuilder, Placement};
use pretty_env_logger::formatted_timed_builder;

extern crate pretty_env_logger;
#[macro_use]
extern crate log;

const DEFAULT_DEBUG_LOG_LEVEL: &str = "dbeel=trace";
const DEFAULT_RELEASE_LOG_LEVEL: &str = "dbeel=info";

fn main() -> Result<()> {
    let default_log_level = if cfg!(debug_assertions) {
        DEFAULT_DEBUG_LOG_LEVEL
    } else {
        DEFAULT_RELEASE_LOG_LEVEL
    };

    let mut log_builder = formatted_timed_builder();
    log_builder.parse_filters(
        &std::env::var("RUST_LOG")
            .unwrap_or_else(|_| default_log_level.to_string()),
    );
    log_builder.try_init().unwrap();

    let args = parse_args();

    if !matches!(args.foreground_tasks_shares, 0..=1000)
        || !matches!(args.background_tasks_shares, 0..=1000)
    {
        panic!("Shares must be between 0..=1000.");
    }

    let cpu_set = CpuSet::online()?;
    assert!(!cpu_set.is_empty());

    let local_connections = cpu_set
        .iter()
        .map(|x| x.cpu as u16)
        .map(LocalShardConnection::new)
        .collect::<Vec<_>>();

    let handles = cpu_set
        .into_iter()
        .map(|x| x.cpu as u16)
        .enumerate()
        .map(|(i, cpu)| {
            LocalExecutorBuilder::new(Placement::Fixed(cpu as usize))
                .name(format!("executor({cpu})").as_str())
                .spawn(enclose!((local_connections.clone() => connections,
                                args.clone() => args) move || async move {
                    let shard = create_shard(args, cpu, connections);
                    if let Err(e) = run_shard(shard, i == 0).await {
                        error!("Failed to start shard {}: {}", cpu, e);
                    }
                }))
                .map_err(Error::GlommioError)
        })
        .collect::<Result<Vec<_>>>()?;

    handles
        .into_iter()
        .map(|h| h.join().map_err(Error::GlommioError))
        .collect::<Result<Vec<_>>>()?;

    Ok(())
}