chord-dht 0.1.0

A DHT implementation in Rust based on Chord with high peformance and data replication.
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
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

use chord_dht::{
	core::{
		config::*,
		Node,
		NodeServer
	},
	client::setup_client
};
use rand::prelude::*;
use tarpc::context;

// Common mod in tests
mod common;
use common::*;

/// Test kv store operations
/// based on the ring of figure 5a but with a larger range
#[tokio::test]
async fn test_replication() -> anyhow::Result<()> {
	env_logger::init();
	// Node 0
	let n0 = Node {
		addr: "localhost:9800".to_string(),
		id: 0
	};
	// Node 1
	let n1 = Node {
		addr: "localhost:9801".to_string(),
		id: u64::MAX / 4
	};
	// Node 3
	let n3 = Node {
		addr: "localhost:9803".to_string(),
		id: u64::MAX / 4 * 2
	};
	// Node 6
	let n6 = Node {
		addr: "localhost:9806".to_string(),
		id: u64::MAX / 4 * 3
	};

	// With replication factor of 3
	let config = Config {
		fault_tolerance: 2,
		replication_factor: 3,
		fix_finger_interval: 0,
		stabilize_interval: 0,
		..Config::default()
	};
	// Node 1 initializes
	let mut s0 = NodeServer::new(n0.clone(), config.clone());
	let m0 = s0.start(None).await?;
	let c0 = setup_client(&n0.addr).await?;
	s0.stabilize().await;

	// Node 1 joins node 0
	let mut s1 = NodeServer::new(n1.clone(), config.clone());
	let m1 = s1.start(Some(n0.clone())).await?;
	let c1 = setup_client(&n1.addr).await?;
	s1.stabilize().await;
	s0.stabilize().await;

	fix_all_fingers(&mut s0).await;
	fix_all_fingers(&mut s1).await;

	// Node 3 joins node 1
	let mut s3 = NodeServer::new(n3.clone(), config.clone());
	let m3 = s3.start(Some(n1.clone())).await?;
	let c3 = setup_client(&n3.addr).await?;
	s3.stabilize().await;
	s1.stabilize().await;
	s0.stabilize().await;

	// See finger table in Figure 3b
	fix_all_fingers(&mut s0).await;
	fix_all_fingers(&mut s1).await;
	fix_all_fingers(&mut s3).await;

	// Node 6 joins node 0
	let mut s6 = NodeServer::new(n6.clone(), config.clone());
	let m6 = s6.start(Some(n0.clone())).await?;
	let c6 = setup_client(&n6.addr).await?;
	s6.stabilize().await;
	s3.stabilize().await;
	s1.stabilize().await;
	s0.stabilize().await;

	// See finger table in Figure 6a
	fix_all_fingers(&mut s0).await;
	fix_all_fingers(&mut s1).await;
	fix_all_fingers(&mut s3).await;
	fix_all_fingers(&mut s6).await;



	let mut rng = StdRng::seed_from_u64(0);
	// k1 should be placed at n1, n3, n6
	let k1 = generate_key_in_range(&mut rng, n0.id, n1.id);
	let v1 = vec![1u8];
	c0.set_rpc(context::current(), k1.clone(), Some(v1.clone())).await?;

	assert_eq!(c0.get_rpc(context::current(), k1.clone()).await?.unwrap(), v1);
	assert_eq!(c0.get_local_rpc(context::current(), k1.clone()).await?, None);
	assert_eq!(c1.get_local_rpc(context::current(), k1.clone()).await?.unwrap(), v1);
	assert_eq!(c3.get_local_rpc(context::current(), k1.clone()).await?.unwrap(), v1);
	assert_eq!(c6.get_local_rpc(context::current(), k1.clone()).await?.unwrap(), v1);

	// k2 should be placed at n3, n6, n0
	let k2 = generate_key_in_range(&mut rng, n1.id, n3.id);
	let v2 = vec![2u8];
	c6.set_rpc(context::current(), k2.clone(), Some(v2.clone())).await?;

	assert_eq!(c1.get_rpc(context::current(), k2.clone()).await?.unwrap(), v2);
	assert_eq!(c1.get_local_rpc(context::current(), k2.clone()).await?, None);
	assert_eq!(c3.get_rpc(context::current(), k2.clone()).await?.unwrap(), v2);
	assert_eq!(c3.get_local_rpc(context::current(), k2.clone()).await?.unwrap(), v2);
	assert_eq!(c6.get_local_rpc(context::current(), k2.clone()).await?.unwrap(), v2);
	assert_eq!(c0.get_local_rpc(context::current(), k2.clone()).await?.unwrap(), v2);

	// delete k1 from n1, n3, n6
	c3.set_rpc(context::current(), k1.clone(), None).await?;

	assert_eq!(c0.get_rpc(context::current(), k1.clone()).await?, None);
	assert_eq!(c1.get_local_rpc(context::current(), k1.clone()).await?, None);
	assert_eq!(c3.get_local_rpc(context::current(), k1.clone()).await?, None);
	assert_eq!(c6.get_local_rpc(context::current(), k1.clone()).await?, None);

	// Stop servers here or the channel in manager will be dropped before this
	m0.stop().await?;
	m1.stop().await?;
	m3.stop().await?;
	m6.stop().await?;

	Ok(())
}