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
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

use std::{rc::Rc, time::Duration};

use glommio::{
    executor, spawn_local_into, timer::sleep, Latency, Shares, Task,
};
use log::{error, info};
use rand::{seq::IteratorRandom, thread_rng};

use crate::{
    error::Result,
    gossip::GossipEvent,
    messages::{ShardEvent, ShardMessage},
    remote_shard_connection::RemoteShardConnection,
    shards::MyShard,
};

async fn run_failure_detector(my_shard: Rc<MyShard>) -> Result<()> {
    let interval =
        Duration::from_millis(my_shard.args.failure_detection_interval);
    let mut should_sleep = true;

    loop {
        if should_sleep {
            sleep(interval).await;
        }
        should_sleep = true;

        let mut rng = thread_rng();
        let node = if let Some(node) = my_shard
            .nodes
            .borrow()
            .iter()
            .map(|(_, node)| node)
            .filter(|node| !node.ids.is_empty())
            .choose(&mut rng)
        {
            node.clone()
        } else {
            continue;
        };

        sleep(interval).await;
        should_sleep = false;

        let connection = RemoteShardConnection::from_args(
            format!(
                "{}:{}",
                node.ip,
                node.ids
                    .iter()
                    .map(|id| node.remote_shard_base_port + id)
                    .choose(&mut rng)
                    .unwrap()
            ),
            &my_shard.args,
        );

        if let Err(e) = connection.ping().await {
            my_shard.clone().handle_dead_node(&node.name).await;

            info!(
                "Notifying cluster that we failed to ping '{}': {}",
                connection.address, e
            );

            let gossip_event = GossipEvent::Dead(node.name);

            if let Err(e) = my_shard
                .clone()
                .broadcast_message_to_local_shards(&ShardMessage::Event(
                    ShardEvent::Gossip(gossip_event.clone()),
                ))
                .await
            {
                error!(
                    "Failed to broadcast to local shards, node death event: {}",
                    e
                );
            }

            if let Err(e) = my_shard.gossip(gossip_event).await {
                error!("Failed to gossip node death event: {}", e);
            }
        }
    }
}

pub fn spawn_failure_detector_task(my_shard: Rc<MyShard>) -> Task<Result<()>> {
    let shares = my_shard.args.background_tasks_shares.into();
    spawn_local_into(
        async move {
            let result = run_failure_detector(my_shard).await;
            if let Err(e) = &result {
                error!("Error starting failure detector: {}", e);
            }
            result
        },
        executor().create_task_queue(
            Shares::Static(shares),
            Latency::NotImportant,
            "failure-detector",
        ),
    )
    .unwrap()
}