2 * mc.h: vlib reliable sequenced multicast distributed applications
4 * Copyright (c) 2010 Cisco and/or its affiliates.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at:
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
18 #ifndef included_vlib_mc_h
19 #define included_vlib_mc_h
21 #include <vppinfra/elog.h>
22 #include <vppinfra/fifo.h>
23 #include <vppinfra/mhash.h>
24 #include <vlib/node.h>
26 #ifndef MC_EVENT_LOGGING
27 #define MC_EVENT_LOGGING 1
31 mc_need_byte_swap (void)
33 return CLIB_ARCH_IS_LITTLE_ENDIAN;
37 * Used to uniquely identify hosts.
38 * For IP4 this would be ip4_address plus tcp/udp port.
46 always_inline mc_peer_id_t
47 mc_byte_swap_peer_id (mc_peer_id_t i)
49 /* Peer id is already in network byte order. */
54 mc_peer_id_compare (mc_peer_id_t a, mc_peer_id_t b)
56 return memcmp (a.as_u8, b.as_u8, sizeof (a.as_u8));
59 /* Assert mastership. Lowest peer_id amount all peers wins mastership.
60 Only sent/received over mastership channel (MC_TRANSPORT_MASTERSHIP).
61 So, we don't need a message opcode. */
62 typedef CLIB_PACKED (struct
64 /* Peer id asserting mastership. */
66 /* Global sequence number asserted. */
67 u32 global_sequence;}) mc_msg_master_assert_t;
70 mc_byte_swap_msg_master_assert (mc_msg_master_assert_t * r)
72 if (mc_need_byte_swap ())
74 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
75 r->global_sequence = clib_byte_swap_u32 (r->global_sequence);
79 #define foreach_mc_msg_type \
81 _ (join_or_leave_request) \
90 #define _(f) MC_MSG_TYPE_##f,
93 } mc_relay_msg_type_t;
95 /* Request to join a given stream. Multicast over MC_TRANSPORT_JOIN. */
96 typedef CLIB_PACKED (struct
98 mc_peer_id_t peer_id; mc_relay_msg_type_t type:32;
99 /* MC_MSG_TYPE_join_or_leave_request */
100 /* Stream to join or leave. */
102 /* join = 1, leave = 0 */
103 u8 is_join;}) mc_msg_join_or_leave_request_t;
106 mc_byte_swap_msg_join_or_leave_request (mc_msg_join_or_leave_request_t * r)
108 if (mc_need_byte_swap ())
110 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
111 r->type = clib_byte_swap_u32 (r->type);
112 r->stream_index = clib_byte_swap_u32 (r->stream_index);
116 /* Join reply. Multicast over MC_TRANSPORT_JOIN. */
117 typedef CLIB_PACKED (struct
119 mc_peer_id_t peer_id; mc_relay_msg_type_t type:32;
120 /* MC_MSG_TYPE_join_reply */
122 /* Peer ID to contact to catchup with this stream. */
123 mc_peer_id_t catchup_peer_id;}) mc_msg_join_reply_t;
126 mc_byte_swap_msg_join_reply (mc_msg_join_reply_t * r)
128 if (mc_need_byte_swap ())
130 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
131 r->type = clib_byte_swap_u32 (r->type);
132 r->stream_index = clib_byte_swap_u32 (r->stream_index);
133 r->catchup_peer_id = mc_byte_swap_peer_id (r->catchup_peer_id);
137 /* Generic (application) request. Multicast over MC_TRANSPORT_USER_REQUEST_TO_RELAY and then
138 relayed by relay master after filling in global sequence number. */
139 typedef CLIB_PACKED (struct
141 mc_peer_id_t peer_id; u32 stream_index;
142 /* Global sequence number as filled in by relay master. */
144 /* Local sequence number as filled in by peer sending message. */
146 /* Size of request data. */
148 /* Opaque request data. */
149 u8 data[0];}) mc_msg_user_request_t;
152 mc_byte_swap_msg_user_request (mc_msg_user_request_t * r)
154 if (mc_need_byte_swap ())
156 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
157 r->stream_index = clib_byte_swap_u32 (r->stream_index);
158 r->global_sequence = clib_byte_swap_u32 (r->global_sequence);
159 r->local_sequence = clib_byte_swap_u32 (r->local_sequence);
160 r->n_data_bytes = clib_byte_swap_u32 (r->n_data_bytes);
164 /* Sent unicast over ACK channel. */
165 typedef CLIB_PACKED (struct
167 mc_peer_id_t peer_id;
168 u32 global_sequence; u32 stream_index;
170 i32 seq_cmp_result;}) mc_msg_user_ack_t;
173 mc_byte_swap_msg_user_ack (mc_msg_user_ack_t * r)
175 if (mc_need_byte_swap ())
177 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
178 r->stream_index = clib_byte_swap_u32 (r->stream_index);
179 r->global_sequence = clib_byte_swap_u32 (r->global_sequence);
180 r->local_sequence = clib_byte_swap_u32 (r->local_sequence);
181 r->seq_cmp_result = clib_byte_swap_i32 (r->seq_cmp_result);
185 /* Sent/received unicast over catchup channel (e.g. using TCP). */
186 typedef CLIB_PACKED (struct
188 mc_peer_id_t peer_id;
189 u32 stream_index;}) mc_msg_catchup_request_t;
192 mc_byte_swap_msg_catchup_request (mc_msg_catchup_request_t * r)
194 if (mc_need_byte_swap ())
196 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
197 r->stream_index = clib_byte_swap_u32 (r->stream_index);
201 /* Sent/received unicast over catchup channel. */
202 typedef CLIB_PACKED (struct
204 mc_peer_id_t peer_id; u32 stream_index;
205 /* Last global sequence number included in catchup data. */
206 u32 last_global_sequence_included;
207 /* Size of catchup data. */
210 u8 data[0];}) mc_msg_catchup_reply_t;
213 mc_byte_swap_msg_catchup_reply (mc_msg_catchup_reply_t * r)
215 if (mc_need_byte_swap ())
217 r->peer_id = mc_byte_swap_peer_id (r->peer_id);
218 r->stream_index = clib_byte_swap_u32 (r->stream_index);
219 r->last_global_sequence_included =
220 clib_byte_swap_u32 (r->last_global_sequence_included);
221 r->n_data_bytes = clib_byte_swap_u32 (r->n_data_bytes);
225 typedef struct _mc_serialize_msg
227 /* Name for this type. */
230 /* Functions to serialize/unserialize data. */
231 serialize_function_t *serialize;
232 serialize_function_t *unserialize;
234 /* Maximum message size in bytes when serialized.
235 If zero then this will be set to the largest sent message. */
236 u32 max_n_bytes_serialized;
238 /* Opaque to use for first argument to serialize/unserialize function. */
241 /* Index in global message vector. */
244 /* Registration list */
245 struct _mc_serialize_msg *next_registration;
246 } mc_serialize_msg_t;
250 /* Index into global message vector. */
252 } mc_serialize_stream_msg_t;
254 #define MC_SERIALIZE_MSG(x,...) \
255 __VA_ARGS__ mc_serialize_msg_t x; \
256 static void __mc_serialize_msg_registration_##x (void) \
257 __attribute__((__constructor__)) ; \
258 static void __mc_serialize_msg_registration_##x (void) \
260 vlib_main_t * vm = vlib_get_main(); \
261 x.next_registration = vm->mc_msg_registrations; \
262 vm->mc_msg_registrations = &x; \
264 static void __mc_serialize_msg_unregistration_##x (void) \
265 __attribute__((__destructor__)) ; \
266 static void __mc_serialize_msg_unregistration_##x (void) \
268 vlib_main_t * vm = vlib_get_main(); \
269 VLIB_REMOVE_FROM_LINKED_LIST (vm->mc_msg_registrations, &x, \
270 next_registration); \
272 __VA_ARGS__ mc_serialize_msg_t x
276 MC_TRANSPORT_MASTERSHIP,
278 MC_TRANSPORT_USER_REQUEST_TO_RELAY,
279 MC_TRANSPORT_USER_REQUEST_FROM_RELAY,
281 } mc_transport_type_t;
285 clib_error_t *(*tx_buffer) (void *opaque, mc_transport_type_t type,
288 clib_error_t *(*tx_ack) (void *opaque, mc_peer_id_t peer_id,
291 /* Returns catchup opaque. */
292 uword (*catchup_request_fun) (void *opaque, u32 stream_index,
293 mc_peer_id_t catchup_peer_id);
295 void (*catchup_send_fun) (void *opaque, uword catchup_opaque,
298 /* Opaque passed to callbacks. */
301 mc_peer_id_t our_ack_peer_id;
302 mc_peer_id_t our_catchup_peer_id;
304 /* Max packet size (MTU) for this transport.
305 For IP this is interface MTU less IP + UDP header size. */
308 format_function_t *format_peer_id;
313 /* Count of messages received from this peer from the past/future
314 (with seq_cmp != 0). */
315 u64 n_msgs_from_past;
316 u64 n_msgs_from_future;
317 } mc_stream_peer_stats_t;
321 /* ID of this peer. */
324 /* The last sequence we received from this peer. */
325 u32 last_sequence_received;
327 mc_stream_peer_stats_t stats, stats_last_clear;
334 /* Cached copy of local sequence number from buffer. */
337 /* Number of times this buffer has been sent (retried). */
340 /* Previous/next retries in doubly-linked list. */
341 u32 prev_index, next_index;
343 /* Bitmap of all peers which have acked this msg */
344 uword *unacked_by_peer_bitmap;
346 /* Message send or resend time */
352 /* Number of retries sent for this stream. */
364 /* Number of outstanding messages. */
367 /* Retry interval, in seconds */
373 /* User rx buffer callback */
374 void (*rx_buffer) (struct mc_main_t * mc_main,
375 struct mc_stream_t * stream,
376 mc_peer_id_t peer_id, u32 buffer_index);
378 /* User callback to create a snapshot */
379 u8 *(*catchup_snapshot) (struct mc_main_t * mc_main,
380 u8 * snapshot_vector,
381 u32 last_global_sequence_included);
383 /* User callback to replay a snapshot */
384 void (*catchup) (struct mc_main_t * mc_main,
385 u8 * snapshot_data, u32 n_snapshot_data_bytes);
387 /* Callback to save a snapshot for offline replay */
388 void (*save_snapshot) (struct mc_main_t * mc_main,
390 u8 * snapshot_data, u32 n_snapshot_data_bytes);
392 /* Called when a peer dies */
393 void (*peer_died) (struct mc_main_t * mc_main,
394 struct mc_stream_t * stream, mc_peer_id_t peer_id);
395 } mc_stream_config_t;
397 #define foreach_mc_stream_state \
400 _ (join_in_progress) \
406 #define _(f) MC_STREAM_STATE_##f,
407 foreach_mc_stream_state
411 typedef struct mc_stream_t
413 mc_stream_config_t config;
415 mc_stream_state_t state;
417 /* Index in stream pool. */
420 /* Stream index 0 is always for MC internal use. */
421 #define MC_STREAM_INDEX_INTERNAL 0
423 mc_retry_t *retry_pool;
425 /* Head and tail index of retry pool. */
426 u32 retry_head_index, retry_tail_index;
429 * Country club for recently retired messages
430 * If the set of peers is expanding and a new peer
431 * misses a message, we can easily retire the FIFO
432 * element before we even know about the new peer
434 mc_retry_t *retired_fifo;
436 /* Hash mapping local sequence to retry pool index. */
437 uword *retry_index_by_local_sequence;
439 /* catch-up fifo of VLIB buffer indices.
440 start recording when catching up. */
443 mc_stream_stats_t stats, stats_last_clear;
446 mc_stream_peer_t *peers;
448 /* Bitmap with ones for all peers in peer pool. */
449 uword *all_peer_bitmap;
451 /* Map of 64 bit id to index in stream pool. */
452 mhash_t peer_index_by_id;
454 /* Timeout, in case we're alone in the world */
457 vlib_one_time_waiting_process_t *procs_waiting_for_join_done;
459 vlib_one_time_waiting_process_t *procs_waiting_for_open_window;
461 /* Next sequence number to use */
462 u32 our_local_sequence;
465 * Last global sequence we processed.
466 * When supplying catchup data, we need to tell
467 * the client precisely where to start replaying
469 u32 last_global_sequence_processed;
471 /* Vector of unique messages we've sent on this stream. */
472 mc_serialize_stream_msg_t *stream_msgs;
474 /* Vector global message index into per stream message index. */
475 u32 *stream_msg_index_by_global_index;
477 /* Hashed by message name. */
478 uword *stream_msg_index_by_name;
480 u64 user_requests_sent;
481 u64 user_requests_received;
485 mc_stream_free (mc_stream_t * s)
487 pool_free (s->retry_pool);
488 hash_free (s->retry_index_by_local_sequence);
489 clib_fifo_free (s->catchup_fifo);
490 pool_free (s->peers);
491 mhash_free (&s->peer_index_by_id);
492 vec_free (s->procs_waiting_for_join_done);
493 vec_free (s->procs_waiting_for_open_window);
497 mc_stream_init (mc_stream_t * s)
499 memset (s, 0, sizeof (s[0]));
500 s->retry_head_index = s->retry_tail_index = ~0;
507 u8 *catchup_snapshot;
508 } mc_catchup_process_arg_t;
512 MC_RELAY_STATE_NEGOTIATE,
513 MC_RELAY_STATE_MASTER,
514 MC_RELAY_STATE_SLAVE,
519 mc_peer_id_t peer_id;
521 f64 time_last_master_assert_received;
522 } mc_mastership_peer_t;
528 } mc_stream_and_buffer_t;
530 typedef struct mc_main_t
532 mc_relay_state_t relay_state;
535 u32 we_can_be_relay_master;
537 u64 relay_master_peer_id;
539 mc_mastership_peer_t *mastership_peers;
541 /* Map of 64 bit id to index in stream pool. */
542 mhash_t mastership_peer_index_by_id;
544 /* The transport we're using. */
545 mc_transport_t transport;
547 /* Last-used global sequence number. */
548 u32 relay_global_sequence;
550 /* Vector of streams. */
551 mc_stream_t *stream_vector;
553 /* Hash table mapping stream name to pool index. */
554 uword *stream_index_by_name;
556 uword *procs_waiting_for_stream_name_by_name;
558 vlib_one_time_waiting_process_t **procs_waiting_for_stream_name_pool;
560 int joins_in_progress;
562 mc_catchup_process_arg_t *catchup_process_args;
564 /* Node indices for mastership, join ager,
565 retry and catchup processes. */
566 u32 mastership_process;
567 u32 join_ager_process;
570 u32 unserialize_process;
572 /* Global vector of messages. */
573 mc_serialize_msg_t **global_msgs;
575 /* Hash table mapping message name to index. */
576 uword *global_msg_index_by_name;
578 /* Shared serialize/unserialize main. */
579 serialize_main_t serialize_mains[VLIB_N_RX_TX];
581 vlib_serialize_buffer_main_t serialize_buffer_mains[VLIB_N_RX_TX];
583 /* Convenience variables */
584 struct vlib_main_t *vlib_main;
585 elog_main_t *elog_main;
587 /* Maps 64 bit peer id to elog string table offset for this formatted peer id. */
588 mhash_t elog_id_by_peer_id;
590 uword *elog_id_by_msg_name;
592 /* For mc_unserialize. */
593 mc_stream_and_buffer_t *mc_unserialize_stream_and_buffers;
596 always_inline mc_stream_t *
597 mc_stream_by_name (mc_main_t * m, char *name)
599 uword *p = hash_get (m->stream_index_by_name, name);
600 return p ? vec_elt_at_index (m->stream_vector, p[0]) : 0;
603 always_inline mc_stream_t *
604 mc_stream_by_index (mc_main_t * m, u32 i)
606 return i < vec_len (m->stream_vector) ? m->stream_vector + i : 0;
610 mc_clear_stream_stats (mc_main_t * m)
614 vec_foreach (s, m->stream_vector)
616 s->stats_last_clear = s->stats;
618 pool_foreach (p, s->peers, ({
619 p->stats_last_clear = p->stats;
625 /* Declare all message handlers. */
626 #define _(f) void mc_msg_##f##_handler (mc_main_t * mcm, mc_msg_##f##_t * msg, u32 buffer_index);
629 u32 mc_stream_join (mc_main_t * mcm, mc_stream_config_t *);
631 void mc_stream_leave (mc_main_t * mcm, u32 stream_index);
633 void mc_wait_for_stream_ready (mc_main_t * m, char *stream_name);
635 u32 mc_stream_send (mc_main_t * mcm, u32 stream_index, u32 buffer_index);
637 void mc_main_init (mc_main_t * mcm, char *tag);
639 void mc_enable_disable_mastership (mc_main_t * mcm, int we_can_be_master);
641 void *mc_get_vlib_buffer (struct vlib_main_t *vm, u32 n_bytes,
644 format_function_t format_mc_main;
646 clib_error_t *mc_serialize_internal (mc_main_t * mc,
648 u32 multiple_messages_per_vlib_buffer,
649 mc_serialize_msg_t * msg, ...);
651 clib_error_t *mc_serialize_va (mc_main_t * mc,
653 u32 multiple_messages_per_vlib_buffer,
654 mc_serialize_msg_t * msg, va_list * va);
656 #define mc_serialize_stream(mc,si,msg,args...) \
657 mc_serialize_internal((mc),(si),(0),(msg),(msg)->serialize,args)
659 #define mc_serialize(mc,msg,args...) \
660 mc_serialize_internal((mc),(~0),(0),(msg),(msg)->serialize,args)
662 #define mc_serialize2(mc,add,msg,args...) \
663 mc_serialize_internal((mc),(~0),(add),(msg),(msg)->serialize,args)
665 void mc_unserialize (mc_main_t * mcm, mc_stream_t * s, u32 buffer_index);
666 uword mc_unserialize_message (mc_main_t * mcm, mc_stream_t * s,
667 serialize_main_t * m);
669 serialize_function_t serialize_mc_main, unserialize_mc_main;
672 mc_max_message_size_in_bytes (mc_main_t * mcm)
674 return mcm->transport.max_packet_size - sizeof (mc_msg_user_request_t);
678 mc_serialize_n_bytes_left (mc_main_t * mcm, serialize_main_t * m)
680 return mc_max_message_size_in_bytes (mcm) -
681 serialize_vlib_buffer_n_bytes (m);
684 void unserialize_mc_stream (serialize_main_t * m, va_list * va);
685 void mc_stream_join_process_hold (void);
687 #endif /* included_vlib_mc_h */
690 * fd.io coding-style-patch-verification: ON
693 * eval: (c-set-style "gnu")