this repo has no description
gdiazlo.tngl.sh
1type node_id = Node_id of string [@@unboxed]
2
3let node_id_to_string (Node_id s) = s
4let node_id_of_string s = Node_id s
5let equal_node_id (Node_id a) (Node_id b) = String.equal a b
6let compare_node_id (Node_id a) (Node_id b) = String.compare a b
7
8type incarnation = Incarnation of int [@@unboxed]
9
10let incarnation_to_int (Incarnation i) = i
11let incarnation_of_int i = Incarnation i
12let zero_incarnation = Incarnation 0
13let compare_incarnation (Incarnation a) (Incarnation b) = Int.compare a b
14let incr_incarnation (Incarnation i) = Incarnation (i + 1)
15
16type addr = Eio.Net.Sockaddr.datagram
17type node_info = { id : node_id; addr : addr; meta : string }
18
19let make_node_info ~id ~addr ~meta = { id; addr; meta }
20
21type member_state = Alive | Suspect | Dead
22
23let member_state_to_string = function
24 | Alive -> "alive"
25 | Suspect -> "suspect"
26 | Dead -> "dead"
27
28type member_snapshot = {
29 node : node_info;
30 state : member_state;
31 incarnation : incarnation;
32 state_change : Mtime.span;
33}
34
35type protocol_msg =
36 | Ping of { seq : int; sender : node_info }
37 | Ping_req of { seq : int; target : node_id; sender : node_info }
38 | Ack of { seq : int; responder : node_info; payload : string option }
39 | Alive of { node : node_info; incarnation : incarnation }
40 | Suspect of {
41 node : node_id;
42 incarnation : incarnation;
43 suspector : node_id;
44 }
45 | Dead of { node : node_id; incarnation : incarnation; declarator : node_id }
46 | User_msg of { topic : string; payload : string; origin : node_id }
47
48type packet = {
49 cluster : string;
50 primary : protocol_msg;
51 piggyback : protocol_msg list;
52}
53
54type decode_error =
55 | Invalid_magic
56 | Unsupported_version of int
57 | Truncated_message
58 | Invalid_tag of int
59 | Decryption_failed
60
61let decode_error_to_string = function
62 | Invalid_magic -> "invalid magic bytes"
63 | Unsupported_version v -> Printf.sprintf "unsupported version: %d" v
64 | Truncated_message -> "truncated message"
65 | Invalid_tag t -> Printf.sprintf "invalid tag: %d" t
66 | Decryption_failed -> "decryption failed"
67
68type send_error = Node_unreachable | Timeout | Connection_reset
69
70let send_error_to_string = function
71 | Node_unreachable -> "node unreachable"
72 | Timeout -> "timeout"
73 | Connection_reset -> "connection reset"
74
75type node_event =
76 | Join of node_info
77 | Leave of node_info
78 | Update of node_info
79 | Suspect_event of node_info
80 | Alive_event of node_info
81
82type config = {
83 bind_addr : string;
84 bind_port : int;
85 node_name : string option;
86 protocol_interval : float;
87 probe_timeout : float;
88 indirect_checks : int;
89 suspicion_mult : int;
90 suspicion_max_timeout : float;
91 retransmit_mult : int;
92 udp_buffer_size : int;
93 tcp_timeout : float;
94 send_buffer_count : int;
95 recv_buffer_count : int;
96 secret_key : string;
97 cluster_name : string;
98}
99
100let default_config =
101 {
102 bind_addr = "0.0.0.0";
103 bind_port = 7946;
104 node_name = None;
105 protocol_interval = 1.0;
106 probe_timeout = 0.5;
107 indirect_checks = 3;
108 suspicion_mult = 4;
109 suspicion_max_timeout = 60.0;
110 retransmit_mult = 4;
111 udp_buffer_size = 1400;
112 tcp_timeout = 10.0;
113 send_buffer_count = 16;
114 recv_buffer_count = 16;
115 secret_key = String.make 32 '\x00';
116 cluster_name = "default";
117 }
118
119type 'a env = {
120 stdenv : 'a;
121 sw : Eio.Switch.t;
122}
123 constraint
124 'a =
125 < net : _ Eio.Net.t
126 ; clock : _ Eio.Time.clock
127 ; mono_clock : _ Eio.Time.Mono.t
128 ; secure_random : _ Eio.Flow.source
129 ; .. >
130
131type stats = {
132 nodes_alive : int;
133 nodes_suspect : int;
134 nodes_dead : int;
135 msgs_sent : int;
136 msgs_received : int;
137 msgs_dropped : int;
138 queue_depth : int;
139 buffers_available : int;
140 buffers_total : int;
141}
142
143let empty_stats =
144 {
145 nodes_alive = 0;
146 nodes_suspect = 0;
147 nodes_dead = 0;
148 msgs_sent = 0;
149 msgs_received = 0;
150 msgs_dropped = 0;
151 queue_depth = 0;
152 buffers_available = 0;
153 buffers_total = 0;
154 }