2 * Main implementation file for interface to Forwarding Plane Manager.
4 * Copyright (C) 2012 by Open Source Routing.
5 * Copyright (C) 2012 by Internet Systems Consortium, Inc. ("ISC")
7 * This file is part of GNU Zebra.
9 * GNU Zebra is free software; you can redistribute it and/or modify it
10 * under the terms of the GNU General Public License as published by the
11 * Free Software Foundation; either version 2, or (at your option) any
14 * GNU Zebra is distributed in the hope that it will be useful, but
15 * WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * General Public License for more details.
19 * You should have received a copy of the GNU General Public License along
20 * with this program; see the file COPYING; if not, write to the Free Software
21 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
34 #include "zebra/rib.h"
35 #include "zebra/zserv.h"
36 #include "zebra/zebra_ns.h"
37 #include "zebra/zebra_vrf.h"
40 #include "zebra_fpm_private.h"
43 * Interval at which we attempt to connect to the FPM.
45 #define ZFPM_CONNECT_RETRY_IVL 5
48 * Sizes of outgoing and incoming stream buffers for writing/reading
51 #define ZFPM_OBUF_SIZE (2 * FPM_MAX_MSG_LEN)
52 #define ZFPM_IBUF_SIZE (FPM_MAX_MSG_LEN)
55 * The maximum number of times the FPM socket write callback can call
56 * 'write' before it yields.
58 #define ZFPM_MAX_WRITES_PER_RUN 10
61 * Interval over which we collect statistics.
63 #define ZFPM_STATS_IVL_SECS 10
66 * Structure that holds state for iterating over all route_node
67 * structures that are candidates for being communicated to the FPM.
69 typedef struct zfpm_rnodes_iter_t_
{
70 rib_tables_iter_t tables_iter
;
71 route_table_iter_t iter
;
77 typedef struct zfpm_stats_t_
{
78 unsigned long connect_calls
;
79 unsigned long connect_no_sock
;
81 unsigned long read_cb_calls
;
83 unsigned long write_cb_calls
;
84 unsigned long write_calls
;
85 unsigned long partial_writes
;
86 unsigned long max_writes_hit
;
87 unsigned long t_write_yields
;
89 unsigned long nop_deletes_skipped
;
90 unsigned long route_adds
;
91 unsigned long route_dels
;
93 unsigned long updates_triggered
;
94 unsigned long redundant_triggers
;
95 unsigned long non_fpm_table_triggers
;
97 unsigned long dests_del_after_update
;
99 unsigned long t_conn_down_starts
;
100 unsigned long t_conn_down_dests_processed
;
101 unsigned long t_conn_down_yields
;
102 unsigned long t_conn_down_finishes
;
104 unsigned long t_conn_up_starts
;
105 unsigned long t_conn_up_dests_processed
;
106 unsigned long t_conn_up_yields
;
107 unsigned long t_conn_up_aborts
;
108 unsigned long t_conn_up_finishes
;
113 * States for the FPM state machine.
118 * In this state we are not yet ready to connect to the FPM. This
119 * can happen when this module is disabled, or if we're cleaning up
120 * after a connection has gone down.
125 * Ready to talk to the FPM and periodically trying to connect to
131 * In the middle of bringing up a TCP connection. Specifically,
132 * waiting for a connect() call to complete asynchronously.
134 ZFPM_STATE_CONNECTING
,
137 * TCP connection to the FPM is up.
139 ZFPM_STATE_ESTABLISHED
144 * Message format to be used to communicate with the FPM.
147 ZFPM_MSG_FORMAT_NONE
,
148 ZFPM_MSG_FORMAT_NETLINK
,
149 ZFPM_MSG_FORMAT_PROTOBUF
,
154 typedef struct zfpm_glob_t_
{
157 * True if the FPM module has been enabled.
162 * Message format to be used to communicate with the fpm.
164 zfpm_msg_format_e message_format
;
166 struct thread_master
*master
;
170 in_addr_t fpm_server
;
172 * Port on which the FPM is running.
177 * List of rib_dest_t structures to be processed
179 TAILQ_HEAD(zfpm_dest_q
, rib_dest_t_
) dest_q
;
182 * Stream socket to the FPM.
187 * Buffers for messages to/from the FPM.
195 struct thread
*t_connect
;
196 struct thread
*t_write
;
197 struct thread
*t_read
;
200 * Thread to clean up after the TCP connection to the FPM goes down
201 * and the state that belongs to it.
203 struct thread
*t_conn_down
;
206 zfpm_rnodes_iter_t iter
;
210 * Thread to take actions once the TCP conn to the FPM comes up, and
211 * the state that belongs to it.
213 struct thread
*t_conn_up
;
216 zfpm_rnodes_iter_t iter
;
219 unsigned long connect_calls
;
220 time_t last_connect_call_time
;
223 * Stats from the start of the current statistics interval up to
224 * now. These are the counters we typically update in the code.
229 * Statistics that were gathered in the last collection interval.
231 zfpm_stats_t last_ivl_stats
;
234 * Cumulative stats from the last clear to the start of the current
235 * statistics interval.
237 zfpm_stats_t cumulative_stats
;
240 * Stats interval timer.
242 struct thread
*t_stats
;
245 * If non-zero, the last time when statistics were cleared.
247 time_t last_stats_clear_time
;
251 static zfpm_glob_t zfpm_glob_space
;
252 static zfpm_glob_t
*zfpm_g
= &zfpm_glob_space
;
254 static int zfpm_trigger_update(struct route_node
*rn
, const char *reason
);
256 static int zfpm_read_cb(struct thread
*thread
);
257 static int zfpm_write_cb(struct thread
*thread
);
259 static void zfpm_set_state(zfpm_state_t state
, const char *reason
);
260 static void zfpm_start_connect_timer(const char *reason
);
261 static void zfpm_start_stats_timer(void);
264 * zfpm_thread_should_yield
266 static inline int zfpm_thread_should_yield(struct thread
*t
)
268 return thread_should_yield(t
);
274 static const char *zfpm_state_to_str(zfpm_state_t state
)
278 case ZFPM_STATE_IDLE
:
281 case ZFPM_STATE_ACTIVE
:
284 case ZFPM_STATE_CONNECTING
:
287 case ZFPM_STATE_ESTABLISHED
:
288 return "established";
296 * zfpm_get_elapsed_time
298 * Returns the time elapsed (in seconds) since the given time.
300 static time_t zfpm_get_elapsed_time(time_t reference
)
304 now
= monotime(NULL
);
306 if (now
< reference
) {
311 return now
- reference
;
315 * zfpm_is_table_for_fpm
317 * Returns TRUE if the the given table is to be communicated to the
320 static inline int zfpm_is_table_for_fpm(struct route_table
*table
)
322 rib_table_info_t
*info
;
324 info
= rib_table_info(table
);
327 * We only send the unicast tables in the main instance to the FPM
330 if (zvrf_id(info
->zvrf
) != 0)
333 if (info
->safi
!= SAFI_UNICAST
)
340 * zfpm_rnodes_iter_init
342 static inline void zfpm_rnodes_iter_init(zfpm_rnodes_iter_t
*iter
)
344 memset(iter
, 0, sizeof(*iter
));
345 rib_tables_iter_init(&iter
->tables_iter
);
348 * This is a hack, but it makes implementing 'next' easier by
349 * ensuring that route_table_iter_next() will return NULL the first
352 route_table_iter_init(&iter
->iter
, NULL
);
353 route_table_iter_cleanup(&iter
->iter
);
357 * zfpm_rnodes_iter_next
359 static inline struct route_node
*zfpm_rnodes_iter_next(zfpm_rnodes_iter_t
*iter
)
361 struct route_node
*rn
;
362 struct route_table
*table
;
365 rn
= route_table_iter_next(&iter
->iter
);
370 * We've made our way through this table, go to the next one.
372 route_table_iter_cleanup(&iter
->iter
);
374 while ((table
= rib_tables_iter_next(&iter
->tables_iter
))) {
375 if (zfpm_is_table_for_fpm(table
))
382 route_table_iter_init(&iter
->iter
, table
);
389 * zfpm_rnodes_iter_pause
391 static inline void zfpm_rnodes_iter_pause(zfpm_rnodes_iter_t
*iter
)
393 route_table_iter_pause(&iter
->iter
);
397 * zfpm_rnodes_iter_cleanup
399 static inline void zfpm_rnodes_iter_cleanup(zfpm_rnodes_iter_t
*iter
)
401 route_table_iter_cleanup(&iter
->iter
);
402 rib_tables_iter_cleanup(&iter
->tables_iter
);
408 * Initialize a statistics block.
410 static inline void zfpm_stats_init(zfpm_stats_t
*stats
)
412 memset(stats
, 0, sizeof(*stats
));
418 static inline void zfpm_stats_reset(zfpm_stats_t
*stats
)
420 zfpm_stats_init(stats
);
426 static inline void zfpm_stats_copy(const zfpm_stats_t
*src
, zfpm_stats_t
*dest
)
428 memcpy(dest
, src
, sizeof(*dest
));
434 * Total up the statistics in two stats structures ('s1 and 's2') and
435 * return the result in the third argument, 'result'. Note that the
436 * pointer 'result' may be the same as 's1' or 's2'.
438 * For simplicity, the implementation below assumes that the stats
439 * structure is composed entirely of counters. This can easily be
440 * changed when necessary.
442 static void zfpm_stats_compose(const zfpm_stats_t
*s1
, const zfpm_stats_t
*s2
,
443 zfpm_stats_t
*result
)
445 const unsigned long *p1
, *p2
;
446 unsigned long *result_p
;
449 p1
= (const unsigned long *)s1
;
450 p2
= (const unsigned long *)s2
;
451 result_p
= (unsigned long *)result
;
453 num_counters
= (sizeof(zfpm_stats_t
) / sizeof(unsigned long));
455 for (i
= 0; i
< num_counters
; i
++) {
456 result_p
[i
] = p1
[i
] + p2
[i
];
463 static inline void zfpm_read_on(void)
465 assert(!zfpm_g
->t_read
);
466 assert(zfpm_g
->sock
>= 0);
468 thread_add_read(zfpm_g
->master
, zfpm_read_cb
, 0, zfpm_g
->sock
,
475 static inline void zfpm_write_on(void)
477 assert(!zfpm_g
->t_write
);
478 assert(zfpm_g
->sock
>= 0);
480 thread_add_write(zfpm_g
->master
, zfpm_write_cb
, 0, zfpm_g
->sock
,
487 static inline void zfpm_read_off(void)
489 THREAD_READ_OFF(zfpm_g
->t_read
);
495 static inline void zfpm_write_off(void)
497 THREAD_WRITE_OFF(zfpm_g
->t_write
);
501 * zfpm_conn_up_thread_cb
503 * Callback for actions to be taken when the connection to the FPM
506 static int zfpm_conn_up_thread_cb(struct thread
*thread
)
508 struct route_node
*rnode
;
509 zfpm_rnodes_iter_t
*iter
;
512 zfpm_g
->t_conn_up
= NULL
;
514 iter
= &zfpm_g
->t_conn_up_state
.iter
;
516 if (zfpm_g
->state
!= ZFPM_STATE_ESTABLISHED
) {
518 "Connection not up anymore, conn_up thread aborting");
519 zfpm_g
->stats
.t_conn_up_aborts
++;
523 while ((rnode
= zfpm_rnodes_iter_next(iter
))) {
524 dest
= rib_dest_from_rnode(rnode
);
527 zfpm_g
->stats
.t_conn_up_dests_processed
++;
528 zfpm_trigger_update(rnode
, NULL
);
534 if (!zfpm_thread_should_yield(thread
))
537 zfpm_g
->stats
.t_conn_up_yields
++;
538 zfpm_rnodes_iter_pause(iter
);
539 zfpm_g
->t_conn_up
= NULL
;
540 thread_add_timer_msec(zfpm_g
->master
, zfpm_conn_up_thread_cb
,
541 NULL
, 0, &zfpm_g
->t_conn_up
);
545 zfpm_g
->stats
.t_conn_up_finishes
++;
548 zfpm_rnodes_iter_cleanup(iter
);
555 * Called when the connection to the FPM comes up.
557 static void zfpm_connection_up(const char *detail
)
559 assert(zfpm_g
->sock
>= 0);
562 zfpm_set_state(ZFPM_STATE_ESTABLISHED
, detail
);
565 * Start thread to push existing routes to the FPM.
567 assert(!zfpm_g
->t_conn_up
);
569 zfpm_rnodes_iter_init(&zfpm_g
->t_conn_up_state
.iter
);
571 zfpm_debug("Starting conn_up thread");
572 zfpm_g
->t_conn_up
= NULL
;
573 thread_add_timer_msec(zfpm_g
->master
, zfpm_conn_up_thread_cb
, NULL
, 0,
575 zfpm_g
->stats
.t_conn_up_starts
++;
581 * Check if an asynchronous connect() to the FPM is complete.
583 static void zfpm_connect_check(void)
592 slen
= sizeof(status
);
593 ret
= getsockopt(zfpm_g
->sock
, SOL_SOCKET
, SO_ERROR
, (void *)&status
,
596 if (ret
>= 0 && status
== 0) {
597 zfpm_connection_up("async connect complete");
602 * getsockopt() failed or indicated an error on the socket.
607 zfpm_start_connect_timer("getsockopt() after async connect failed");
612 * zfpm_conn_down_thread_cb
614 * Callback that is invoked to clean up state after the TCP connection
615 * to the FPM goes down.
617 static int zfpm_conn_down_thread_cb(struct thread
*thread
)
619 struct route_node
*rnode
;
620 zfpm_rnodes_iter_t
*iter
;
623 assert(zfpm_g
->state
== ZFPM_STATE_IDLE
);
625 zfpm_g
->t_conn_down
= NULL
;
627 iter
= &zfpm_g
->t_conn_down_state
.iter
;
629 while ((rnode
= zfpm_rnodes_iter_next(iter
))) {
630 dest
= rib_dest_from_rnode(rnode
);
633 if (CHECK_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
)) {
634 TAILQ_REMOVE(&zfpm_g
->dest_q
, dest
,
638 UNSET_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
);
639 UNSET_FLAG(dest
->flags
, RIB_DEST_SENT_TO_FPM
);
641 zfpm_g
->stats
.t_conn_down_dests_processed
++;
644 * Check if the dest should be deleted.
652 if (!zfpm_thread_should_yield(thread
))
655 zfpm_g
->stats
.t_conn_down_yields
++;
656 zfpm_rnodes_iter_pause(iter
);
657 zfpm_g
->t_conn_down
= NULL
;
658 thread_add_timer_msec(zfpm_g
->master
, zfpm_conn_down_thread_cb
,
659 NULL
, 0, &zfpm_g
->t_conn_down
);
663 zfpm_g
->stats
.t_conn_down_finishes
++;
664 zfpm_rnodes_iter_cleanup(iter
);
667 * Start the process of connecting to the FPM again.
669 zfpm_start_connect_timer("cleanup complete");
674 * zfpm_connection_down
676 * Called when the connection to the FPM has gone down.
678 static void zfpm_connection_down(const char *detail
)
683 assert(zfpm_g
->state
== ZFPM_STATE_ESTABLISHED
);
685 zlog_info("connection to the FPM has gone down: %s", detail
);
690 stream_reset(zfpm_g
->ibuf
);
691 stream_reset(zfpm_g
->obuf
);
693 if (zfpm_g
->sock
>= 0) {
699 * Start thread to clean up state after the connection goes down.
701 assert(!zfpm_g
->t_conn_down
);
702 zfpm_debug("Starting conn_down thread");
703 zfpm_rnodes_iter_init(&zfpm_g
->t_conn_down_state
.iter
);
704 zfpm_g
->t_conn_down
= NULL
;
705 thread_add_timer_msec(zfpm_g
->master
, zfpm_conn_down_thread_cb
, NULL
, 0,
706 &zfpm_g
->t_conn_down
);
707 zfpm_g
->stats
.t_conn_down_starts
++;
709 zfpm_set_state(ZFPM_STATE_IDLE
, detail
);
715 static int zfpm_read_cb(struct thread
*thread
)
722 zfpm_g
->stats
.read_cb_calls
++;
723 zfpm_g
->t_read
= NULL
;
726 * Check if async connect is now done.
728 if (zfpm_g
->state
== ZFPM_STATE_CONNECTING
) {
729 zfpm_connect_check();
733 assert(zfpm_g
->state
== ZFPM_STATE_ESTABLISHED
);
734 assert(zfpm_g
->sock
>= 0);
738 already
= stream_get_endp(ibuf
);
739 if (already
< FPM_MSG_HDR_LEN
) {
742 nbyte
= stream_read_try(ibuf
, zfpm_g
->sock
,
743 FPM_MSG_HDR_LEN
- already
);
744 if (nbyte
== 0 || nbyte
== -1) {
745 zfpm_connection_down("closed socket in read");
749 if (nbyte
!= (ssize_t
)(FPM_MSG_HDR_LEN
- already
))
752 already
= FPM_MSG_HDR_LEN
;
755 stream_set_getp(ibuf
, 0);
757 hdr
= (fpm_msg_hdr_t
*)stream_pnt(ibuf
);
759 if (!fpm_msg_hdr_ok(hdr
)) {
760 zfpm_connection_down("invalid message header");
764 msg_len
= fpm_msg_len(hdr
);
767 * Read out the rest of the packet.
769 if (already
< msg_len
) {
772 nbyte
= stream_read_try(ibuf
, zfpm_g
->sock
, msg_len
- already
);
774 if (nbyte
== 0 || nbyte
== -1) {
775 zfpm_connection_down("failed to read message");
779 if (nbyte
!= (ssize_t
)(msg_len
- already
))
783 zfpm_debug("Read out a full fpm message");
786 * Just throw it away for now.
796 * zfpm_writes_pending
798 * Returns TRUE if we may have something to write to the FPM.
800 static int zfpm_writes_pending(void)
804 * Check if there is any data in the outbound buffer that has not
805 * been written to the socket yet.
807 if (stream_get_endp(zfpm_g
->obuf
) - stream_get_getp(zfpm_g
->obuf
))
811 * Check if there are any prefixes on the outbound queue.
813 if (!TAILQ_EMPTY(&zfpm_g
->dest_q
))
822 * Encode a message to the FPM with information about the given route.
824 * Returns the number of bytes written to the buffer. 0 or a negative
825 * value indicates an error.
827 static inline int zfpm_encode_route(rib_dest_t
*dest
, struct route_entry
*re
,
828 char *in_buf
, size_t in_buf_len
,
829 fpm_msg_type_e
*msg_type
)
837 *msg_type
= FPM_MSG_TYPE_NONE
;
839 switch (zfpm_g
->message_format
) {
841 case ZFPM_MSG_FORMAT_PROTOBUF
:
843 len
= zfpm_protobuf_encode_route(dest
, re
, (uint8_t *)in_buf
,
845 *msg_type
= FPM_MSG_TYPE_PROTOBUF
;
849 case ZFPM_MSG_FORMAT_NETLINK
:
851 *msg_type
= FPM_MSG_TYPE_NETLINK
;
852 cmd
= re
? RTM_NEWROUTE
: RTM_DELROUTE
;
853 len
= zfpm_netlink_encode_route(cmd
, dest
, re
, in_buf
,
855 assert(fpm_msg_align(len
) == len
);
856 *msg_type
= FPM_MSG_TYPE_NETLINK
;
857 #endif /* HAVE_NETLINK */
868 * zfpm_route_for_update
870 * Returns the re that is to be sent to the FPM for a given dest.
872 struct route_entry
*zfpm_route_for_update(rib_dest_t
*dest
)
874 struct route_entry
*re
;
876 RE_DEST_FOREACH_ROUTE(dest
, re
)
878 if (!CHECK_FLAG(re
->status
, ROUTE_ENTRY_SELECTED_FIB
))
885 * We have no route for this destination.
893 * Process the outgoing queue and write messages to the outbound
896 static void zfpm_build_updates(void)
900 unsigned char *buf
, *data
, *buf_end
;
904 struct route_entry
*re
;
905 int is_add
, write_msg
;
906 fpm_msg_type_e msg_type
;
910 assert(stream_empty(s
));
915 * Make sure there is enough space to write another message.
917 if (STREAM_WRITEABLE(s
) < FPM_MAX_MSG_LEN
)
920 buf
= STREAM_DATA(s
) + stream_get_endp(s
);
921 buf_end
= buf
+ STREAM_WRITEABLE(s
);
923 dest
= TAILQ_FIRST(&zfpm_g
->dest_q
);
927 assert(CHECK_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
));
929 hdr
= (fpm_msg_hdr_t
*)buf
;
930 hdr
->version
= FPM_PROTO_VERSION
;
932 data
= fpm_msg_data(hdr
);
934 re
= zfpm_route_for_update(dest
);
940 * If this is a route deletion, and we have not sent the route
942 * the FPM previously, skip it.
944 if (!is_add
&& !CHECK_FLAG(dest
->flags
, RIB_DEST_SENT_TO_FPM
)) {
946 zfpm_g
->stats
.nop_deletes_skipped
++;
950 data_len
= zfpm_encode_route(dest
, re
, (char *)data
,
951 buf_end
- data
, &msg_type
);
955 hdr
->msg_type
= msg_type
;
956 msg_len
= fpm_data_len_to_msg_len(data_len
);
957 hdr
->msg_len
= htons(msg_len
);
958 stream_forward_endp(s
, msg_len
);
961 zfpm_g
->stats
.route_adds
++;
963 zfpm_g
->stats
.route_dels
++;
968 * Remove the dest from the queue, and reset the flag.
970 UNSET_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
);
971 TAILQ_REMOVE(&zfpm_g
->dest_q
, dest
, fpm_q_entries
);
974 SET_FLAG(dest
->flags
, RIB_DEST_SENT_TO_FPM
);
976 UNSET_FLAG(dest
->flags
, RIB_DEST_SENT_TO_FPM
);
980 * Delete the destination if necessary.
982 if (rib_gc_dest(dest
->rnode
))
983 zfpm_g
->stats
.dests_del_after_update
++;
991 static int zfpm_write_cb(struct thread
*thread
)
996 zfpm_g
->stats
.write_cb_calls
++;
997 zfpm_g
->t_write
= NULL
;
1000 * Check if async connect is now done.
1002 if (zfpm_g
->state
== ZFPM_STATE_CONNECTING
) {
1003 zfpm_connect_check();
1007 assert(zfpm_g
->state
== ZFPM_STATE_ESTABLISHED
);
1008 assert(zfpm_g
->sock
>= 0);
1013 int bytes_to_write
, bytes_written
;
1018 * If the stream is empty, try fill it up with data.
1020 if (stream_empty(s
)) {
1021 zfpm_build_updates();
1024 bytes_to_write
= stream_get_endp(s
) - stream_get_getp(s
);
1025 if (!bytes_to_write
)
1029 write(zfpm_g
->sock
, STREAM_PNT(s
), bytes_to_write
);
1030 zfpm_g
->stats
.write_calls
++;
1033 if (bytes_written
< 0) {
1034 if (ERRNO_IO_RETRY(errno
))
1037 zfpm_connection_down("failed to write to socket");
1041 if (bytes_written
!= bytes_to_write
) {
1046 stream_forward_getp(s
, bytes_written
);
1047 zfpm_g
->stats
.partial_writes
++;
1052 * We've written out the entire contents of the stream.
1056 if (num_writes
>= ZFPM_MAX_WRITES_PER_RUN
) {
1057 zfpm_g
->stats
.max_writes_hit
++;
1061 if (zfpm_thread_should_yield(thread
)) {
1062 zfpm_g
->stats
.t_write_yields
++;
1067 if (zfpm_writes_pending())
1076 static int zfpm_connect_cb(struct thread
*t
)
1079 struct sockaddr_in serv
;
1081 zfpm_g
->t_connect
= NULL
;
1082 assert(zfpm_g
->state
== ZFPM_STATE_ACTIVE
);
1084 sock
= socket(AF_INET
, SOCK_STREAM
, 0);
1086 zfpm_debug("Failed to create socket for connect(): %s",
1088 zfpm_g
->stats
.connect_no_sock
++;
1092 set_nonblocking(sock
);
1094 /* Make server socket. */
1095 memset(&serv
, 0, sizeof(serv
));
1096 serv
.sin_family
= AF_INET
;
1097 serv
.sin_port
= htons(zfpm_g
->fpm_port
);
1098 #ifdef HAVE_STRUCT_SOCKADDR_IN_SIN_LEN
1099 serv
.sin_len
= sizeof(struct sockaddr_in
);
1100 #endif /* HAVE_STRUCT_SOCKADDR_IN_SIN_LEN */
1101 if (!zfpm_g
->fpm_server
)
1102 serv
.sin_addr
.s_addr
= htonl(INADDR_LOOPBACK
);
1104 serv
.sin_addr
.s_addr
= (zfpm_g
->fpm_server
);
1107 * Connect to the FPM.
1109 zfpm_g
->connect_calls
++;
1110 zfpm_g
->stats
.connect_calls
++;
1111 zfpm_g
->last_connect_call_time
= monotime(NULL
);
1113 ret
= connect(sock
, (struct sockaddr
*)&serv
, sizeof(serv
));
1115 zfpm_g
->sock
= sock
;
1116 zfpm_connection_up("connect succeeded");
1120 if (errno
== EINPROGRESS
) {
1121 zfpm_g
->sock
= sock
;
1124 zfpm_set_state(ZFPM_STATE_CONNECTING
,
1125 "async connect in progress");
1129 zlog_info("can't connect to FPM %d: %s", sock
, safe_strerror(errno
));
1133 * Restart timer for retrying connection.
1135 zfpm_start_connect_timer("connect() failed");
1142 * Move state machine into the given state.
1144 static void zfpm_set_state(zfpm_state_t state
, const char *reason
)
1146 zfpm_state_t cur_state
= zfpm_g
->state
;
1151 if (state
== cur_state
)
1154 zfpm_debug("beginning state transition %s -> %s. Reason: %s",
1155 zfpm_state_to_str(cur_state
), zfpm_state_to_str(state
),
1160 case ZFPM_STATE_IDLE
:
1161 assert(cur_state
== ZFPM_STATE_ESTABLISHED
);
1164 case ZFPM_STATE_ACTIVE
:
1165 assert(cur_state
== ZFPM_STATE_IDLE
1166 || cur_state
== ZFPM_STATE_CONNECTING
);
1167 assert(zfpm_g
->t_connect
);
1170 case ZFPM_STATE_CONNECTING
:
1171 assert(zfpm_g
->sock
);
1172 assert(cur_state
== ZFPM_STATE_ACTIVE
);
1173 assert(zfpm_g
->t_read
);
1174 assert(zfpm_g
->t_write
);
1177 case ZFPM_STATE_ESTABLISHED
:
1178 assert(cur_state
== ZFPM_STATE_ACTIVE
1179 || cur_state
== ZFPM_STATE_CONNECTING
);
1180 assert(zfpm_g
->sock
);
1181 assert(zfpm_g
->t_read
);
1182 assert(zfpm_g
->t_write
);
1186 zfpm_g
->state
= state
;
1190 * zfpm_calc_connect_delay
1192 * Returns the number of seconds after which we should attempt to
1193 * reconnect to the FPM.
1195 static long zfpm_calc_connect_delay(void)
1200 * Return 0 if this is our first attempt to connect.
1202 if (zfpm_g
->connect_calls
== 0) {
1206 elapsed
= zfpm_get_elapsed_time(zfpm_g
->last_connect_call_time
);
1208 if (elapsed
> ZFPM_CONNECT_RETRY_IVL
) {
1212 return ZFPM_CONNECT_RETRY_IVL
- elapsed
;
1216 * zfpm_start_connect_timer
1218 static void zfpm_start_connect_timer(const char *reason
)
1222 assert(!zfpm_g
->t_connect
);
1223 assert(zfpm_g
->sock
< 0);
1225 assert(zfpm_g
->state
== ZFPM_STATE_IDLE
1226 || zfpm_g
->state
== ZFPM_STATE_ACTIVE
1227 || zfpm_g
->state
== ZFPM_STATE_CONNECTING
);
1229 delay_secs
= zfpm_calc_connect_delay();
1230 zfpm_debug("scheduling connect in %ld seconds", delay_secs
);
1232 thread_add_timer(zfpm_g
->master
, zfpm_connect_cb
, 0, delay_secs
,
1233 &zfpm_g
->t_connect
);
1234 zfpm_set_state(ZFPM_STATE_ACTIVE
, reason
);
1240 * Returns TRUE if the zebra FPM module has been enabled.
1242 static inline int zfpm_is_enabled(void)
1244 return zfpm_g
->enabled
;
1250 * Returns TRUE if the connection to the FPM is up.
1252 static inline int zfpm_conn_is_up(void)
1254 if (zfpm_g
->state
!= ZFPM_STATE_ESTABLISHED
)
1257 assert(zfpm_g
->sock
>= 0);
1263 * zfpm_trigger_update
1265 * The zebra code invokes this function to indicate that we should
1266 * send an update to the FPM about the given route_node.
1268 static int zfpm_trigger_update(struct route_node
*rn
, const char *reason
)
1271 char buf
[PREFIX_STRLEN
];
1274 * Ignore if the connection is down. We will update the FPM about
1275 * all destinations once the connection comes up.
1277 if (!zfpm_conn_is_up())
1280 dest
= rib_dest_from_rnode(rn
);
1283 * Ignore the trigger if the dest is not in a table that we would
1286 if (!zfpm_is_table_for_fpm(rib_dest_table(dest
))) {
1287 zfpm_g
->stats
.non_fpm_table_triggers
++;
1291 if (CHECK_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
)) {
1292 zfpm_g
->stats
.redundant_triggers
++;
1297 zfpm_debug("%s triggering update to FPM - Reason: %s",
1298 prefix2str(&rn
->p
, buf
, sizeof(buf
)), reason
);
1301 SET_FLAG(dest
->flags
, RIB_DEST_UPDATE_FPM
);
1302 TAILQ_INSERT_TAIL(&zfpm_g
->dest_q
, dest
, fpm_q_entries
);
1303 zfpm_g
->stats
.updates_triggered
++;
1306 * Make sure that writes are enabled.
1308 if (zfpm_g
->t_write
)
1316 * zfpm_stats_timer_cb
1318 static int zfpm_stats_timer_cb(struct thread
*t
)
1320 zfpm_g
->t_stats
= NULL
;
1323 * Remember the stats collected in the last interval for display
1326 zfpm_stats_copy(&zfpm_g
->stats
, &zfpm_g
->last_ivl_stats
);
1329 * Add the current set of stats into the cumulative statistics.
1331 zfpm_stats_compose(&zfpm_g
->cumulative_stats
, &zfpm_g
->stats
,
1332 &zfpm_g
->cumulative_stats
);
1335 * Start collecting stats afresh over the next interval.
1337 zfpm_stats_reset(&zfpm_g
->stats
);
1339 zfpm_start_stats_timer();
1345 * zfpm_stop_stats_timer
1347 static void zfpm_stop_stats_timer(void)
1349 if (!zfpm_g
->t_stats
)
1352 zfpm_debug("Stopping existing stats timer");
1353 THREAD_TIMER_OFF(zfpm_g
->t_stats
);
1357 * zfpm_start_stats_timer
1359 void zfpm_start_stats_timer(void)
1361 assert(!zfpm_g
->t_stats
);
1363 thread_add_timer(zfpm_g
->master
, zfpm_stats_timer_cb
, 0,
1364 ZFPM_STATS_IVL_SECS
, &zfpm_g
->t_stats
);
1368 * Helper macro for zfpm_show_stats() below.
1370 #define ZFPM_SHOW_STAT(counter) \
1372 vty_out(vty, "%-40s %10lu %16lu\n", #counter, \
1373 total_stats.counter, zfpm_g->last_ivl_stats.counter); \
1379 static void zfpm_show_stats(struct vty
*vty
)
1381 zfpm_stats_t total_stats
;
1384 vty_out(vty
, "\n%-40s %10s Last %2d secs\n\n", "Counter", "Total",
1385 ZFPM_STATS_IVL_SECS
);
1388 * Compute the total stats up to this instant.
1390 zfpm_stats_compose(&zfpm_g
->cumulative_stats
, &zfpm_g
->stats
,
1393 ZFPM_SHOW_STAT(connect_calls
);
1394 ZFPM_SHOW_STAT(connect_no_sock
);
1395 ZFPM_SHOW_STAT(read_cb_calls
);
1396 ZFPM_SHOW_STAT(write_cb_calls
);
1397 ZFPM_SHOW_STAT(write_calls
);
1398 ZFPM_SHOW_STAT(partial_writes
);
1399 ZFPM_SHOW_STAT(max_writes_hit
);
1400 ZFPM_SHOW_STAT(t_write_yields
);
1401 ZFPM_SHOW_STAT(nop_deletes_skipped
);
1402 ZFPM_SHOW_STAT(route_adds
);
1403 ZFPM_SHOW_STAT(route_dels
);
1404 ZFPM_SHOW_STAT(updates_triggered
);
1405 ZFPM_SHOW_STAT(non_fpm_table_triggers
);
1406 ZFPM_SHOW_STAT(redundant_triggers
);
1407 ZFPM_SHOW_STAT(dests_del_after_update
);
1408 ZFPM_SHOW_STAT(t_conn_down_starts
);
1409 ZFPM_SHOW_STAT(t_conn_down_dests_processed
);
1410 ZFPM_SHOW_STAT(t_conn_down_yields
);
1411 ZFPM_SHOW_STAT(t_conn_down_finishes
);
1412 ZFPM_SHOW_STAT(t_conn_up_starts
);
1413 ZFPM_SHOW_STAT(t_conn_up_dests_processed
);
1414 ZFPM_SHOW_STAT(t_conn_up_yields
);
1415 ZFPM_SHOW_STAT(t_conn_up_aborts
);
1416 ZFPM_SHOW_STAT(t_conn_up_finishes
);
1418 if (!zfpm_g
->last_stats_clear_time
)
1421 elapsed
= zfpm_get_elapsed_time(zfpm_g
->last_stats_clear_time
);
1423 vty_out(vty
, "\nStats were cleared %lu seconds ago\n",
1424 (unsigned long)elapsed
);
1430 static void zfpm_clear_stats(struct vty
*vty
)
1432 if (!zfpm_is_enabled()) {
1433 vty_out(vty
, "The FPM module is not enabled...\n");
1437 zfpm_stats_reset(&zfpm_g
->stats
);
1438 zfpm_stats_reset(&zfpm_g
->last_ivl_stats
);
1439 zfpm_stats_reset(&zfpm_g
->cumulative_stats
);
1441 zfpm_stop_stats_timer();
1442 zfpm_start_stats_timer();
1444 zfpm_g
->last_stats_clear_time
= monotime(NULL
);
1446 vty_out(vty
, "Cleared FPM stats\n");
1450 * show_zebra_fpm_stats
1452 DEFUN (show_zebra_fpm_stats
,
1453 show_zebra_fpm_stats_cmd
,
1454 "show zebra fpm stats",
1456 "Zebra information\n"
1457 "Forwarding Path Manager information\n"
1460 zfpm_show_stats(vty
);
1465 * clear_zebra_fpm_stats
1467 DEFUN (clear_zebra_fpm_stats
,
1468 clear_zebra_fpm_stats_cmd
,
1469 "clear zebra fpm stats",
1471 "Zebra information\n"
1472 "Clear Forwarding Path Manager information\n"
1475 zfpm_clear_stats(vty
);
1480 * update fpm connection information
1482 DEFUN ( fpm_remote_ip
,
1484 "fpm connection ip A.B.C.D port (1-65535)",
1485 "fpm connection remote ip and port\n"
1486 "Remote fpm server ip A.B.C.D\n"
1490 in_addr_t fpm_server
;
1493 fpm_server
= inet_addr(argv
[3]->arg
);
1494 if (fpm_server
== INADDR_NONE
)
1495 return CMD_ERR_INCOMPLETE
;
1497 port_no
= atoi(argv
[5]->arg
);
1498 if (port_no
< TCP_MIN_PORT
|| port_no
> TCP_MAX_PORT
)
1499 return CMD_ERR_INCOMPLETE
;
1501 zfpm_g
->fpm_server
= fpm_server
;
1502 zfpm_g
->fpm_port
= port_no
;
1508 DEFUN ( no_fpm_remote_ip
,
1509 no_fpm_remote_ip_cmd
,
1510 "no fpm connection ip A.B.C.D port (1-65535)",
1511 "fpm connection remote ip and port\n"
1513 "Remote fpm server ip A.B.C.D\n"
1516 if (zfpm_g
->fpm_server
!= inet_addr(argv
[4]->arg
)
1517 || zfpm_g
->fpm_port
!= atoi(argv
[6]->arg
))
1518 return CMD_ERR_NO_MATCH
;
1520 zfpm_g
->fpm_server
= FPM_DEFAULT_IP
;
1521 zfpm_g
->fpm_port
= FPM_DEFAULT_PORT
;
1527 * zfpm_init_message_format
1529 static inline void zfpm_init_message_format(const char *format
)
1531 int have_netlink
, have_protobuf
;
1539 #ifdef HAVE_PROTOBUF
1545 zfpm_g
->message_format
= ZFPM_MSG_FORMAT_NONE
;
1549 zfpm_g
->message_format
= ZFPM_MSG_FORMAT_NETLINK
;
1550 } else if (have_protobuf
) {
1551 zfpm_g
->message_format
= ZFPM_MSG_FORMAT_PROTOBUF
;
1556 if (!strcmp("netlink", format
)) {
1557 if (!have_netlink
) {
1558 zlog_err("FPM netlink message format is not available");
1561 zfpm_g
->message_format
= ZFPM_MSG_FORMAT_NETLINK
;
1565 if (!strcmp("protobuf", format
)) {
1566 if (!have_protobuf
) {
1568 "FPM protobuf message format is not available");
1571 zfpm_g
->message_format
= ZFPM_MSG_FORMAT_PROTOBUF
;
1575 zlog_warn("Unknown fpm format '%s'", format
);
1579 * fpm_remote_srv_write
1581 * Module to write remote fpm connection
1583 * Returns ZERO on success.
1586 static int fpm_remote_srv_write(struct vty
*vty
)
1590 in
.s_addr
= zfpm_g
->fpm_server
;
1592 if ((zfpm_g
->fpm_server
!= FPM_DEFAULT_IP
1593 && zfpm_g
->fpm_server
!= INADDR_ANY
)
1594 || (zfpm_g
->fpm_port
!= FPM_DEFAULT_PORT
1595 && zfpm_g
->fpm_port
!= 0))
1596 vty_out(vty
, "fpm connection ip %s port %d\n", inet_ntoa(in
),
1604 static struct cmd_node zebra_node
= {ZEBRA_NODE
, "", 1};
1610 * One-time initialization of the Zebra FPM module.
1612 * @param[in] port port at which FPM is running.
1613 * @param[in] enable TRUE if the zebra FPM module should be enabled
1614 * @param[in] format to use to talk to the FPM. Can be 'netink' or 'protobuf'.
1616 * Returns TRUE on success.
1618 static int zfpm_init(struct thread_master
*master
)
1622 const char *format
= THIS_MODULE
->load_args
;
1624 memset(zfpm_g
, 0, sizeof(*zfpm_g
));
1625 zfpm_g
->master
= master
;
1626 TAILQ_INIT(&zfpm_g
->dest_q
);
1628 zfpm_g
->state
= ZFPM_STATE_IDLE
;
1630 zfpm_stats_init(&zfpm_g
->stats
);
1631 zfpm_stats_init(&zfpm_g
->last_ivl_stats
);
1632 zfpm_stats_init(&zfpm_g
->cumulative_stats
);
1634 install_node(&zebra_node
, fpm_remote_srv_write
);
1635 install_element(ENABLE_NODE
, &show_zebra_fpm_stats_cmd
);
1636 install_element(ENABLE_NODE
, &clear_zebra_fpm_stats_cmd
);
1637 install_element(CONFIG_NODE
, &fpm_remote_ip_cmd
);
1638 install_element(CONFIG_NODE
, &no_fpm_remote_ip_cmd
);
1640 zfpm_init_message_format(format
);
1643 * Disable FPM interface if no suitable format is available.
1645 if (zfpm_g
->message_format
== ZFPM_MSG_FORMAT_NONE
)
1648 zfpm_g
->enabled
= enable
;
1650 if (!zfpm_g
->fpm_server
)
1651 zfpm_g
->fpm_server
= FPM_DEFAULT_IP
;
1654 port
= FPM_DEFAULT_PORT
;
1656 zfpm_g
->fpm_port
= port
;
1658 zfpm_g
->obuf
= stream_new(ZFPM_OBUF_SIZE
);
1659 zfpm_g
->ibuf
= stream_new(ZFPM_IBUF_SIZE
);
1661 zfpm_start_stats_timer();
1662 zfpm_start_connect_timer("initialized");
1666 static int zebra_fpm_module_init(void)
1668 hook_register(rib_update
, zfpm_trigger_update
);
1669 hook_register(frr_late_init
, zfpm_init
);
1673 FRR_MODULE_SETUP(.name
= "zebra_fpm", .version
= FRR_VERSION
,
1674 .description
= "zebra FPM (Forwarding Plane Manager) module",
1675 .init
= zebra_fpm_module_init
, )