/* * Main implementation file for interface to Forwarding Plane Manager. * * Copyright (C) 2012 by Open Source Routing. * Copyright (C) 2012 by Internet Systems Consortium, Inc. ("ISC") * * This file is part of GNU Zebra. * * GNU Zebra is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the * Free Software Foundation; either version 2, or (at your option) any * later version. * * GNU Zebra is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * General Public License for more details. * * You should have received a copy of the GNU General Public License * along with GNU Zebra; see the file COPYING. If not, write to the Free * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA * 02111-1307, USA. */ #include #include "log.h" #include "stream.h" #include "thread.h" #include "network.h" #include "command.h" #include "zebra/rib.h" #include "fpm/fpm.h" #include "zebra_fpm.h" #include "zebra_fpm_private.h" /* * Interval at which we attempt to connect to the FPM. */ #define ZFPM_CONNECT_RETRY_IVL 5 /* * Sizes of outgoing and incoming stream buffers for writing/reading * FPM messages. */ #define ZFPM_OBUF_SIZE (2 * FPM_MAX_MSG_LEN) #define ZFPM_IBUF_SIZE (FPM_MAX_MSG_LEN) /* * The maximum number of times the FPM socket write callback can call * 'write' before it yields. */ #define ZFPM_MAX_WRITES_PER_RUN 10 /* * Interval over which we collect statistics. */ #define ZFPM_STATS_IVL_SECS 10 /* * Structure that holds state for iterating over all route_node * structures that are candidates for being communicated to the FPM. */ typedef struct zfpm_rnodes_iter_t_ { rib_tables_iter_t tables_iter; route_table_iter_t iter; } zfpm_rnodes_iter_t; /* * Statistics. */ typedef struct zfpm_stats_t_ { unsigned long connect_calls; unsigned long connect_no_sock; unsigned long read_cb_calls; unsigned long write_cb_calls; unsigned long write_calls; unsigned long partial_writes; unsigned long max_writes_hit; unsigned long t_write_yields; unsigned long nop_deletes_skipped; unsigned long route_adds; unsigned long route_dels; unsigned long updates_triggered; unsigned long redundant_triggers; unsigned long non_fpm_table_triggers; unsigned long dests_del_after_update; unsigned long t_conn_down_starts; unsigned long t_conn_down_dests_processed; unsigned long t_conn_down_yields; unsigned long t_conn_down_finishes; unsigned long t_conn_up_starts; unsigned long t_conn_up_dests_processed; unsigned long t_conn_up_yields; unsigned long t_conn_up_aborts; unsigned long t_conn_up_finishes; } zfpm_stats_t; /* * States for the FPM state machine. */ typedef enum { /* * In this state we are not yet ready to connect to the FPM. This * can happen when this module is disabled, or if we're cleaning up * after a connection has gone down. */ ZFPM_STATE_IDLE, /* * Ready to talk to the FPM and periodically trying to connect to * it. */ ZFPM_STATE_ACTIVE, /* * In the middle of bringing up a TCP connection. Specifically, * waiting for a connect() call to complete asynchronously. */ ZFPM_STATE_CONNECTING, /* * TCP connection to the FPM is up. */ ZFPM_STATE_ESTABLISHED } zfpm_state_t; /* * Globals. */ typedef struct zfpm_glob_t_ { /* * True if the FPM module has been enabled. */ int enabled; struct thread_master *master; zfpm_state_t state; /* * Port on which the FPM is running. */ int fpm_port; /* * List of rib_dest_t structures to be processed */ TAILQ_HEAD (zfpm_dest_q, rib_dest_t_) dest_q; /* * Stream socket to the FPM. */ int sock; /* * Buffers for messages to/from the FPM. */ struct stream *obuf; struct stream *ibuf; /* * Threads for I/O. */ struct thread *t_connect; struct thread *t_write; struct thread *t_read; /* * Thread to clean up after the TCP connection to the FPM goes down * and the state that belongs to it. */ struct thread *t_conn_down; struct { zfpm_rnodes_iter_t iter; } t_conn_down_state; /* * Thread to take actions once the TCP conn to the FPM comes up, and * the state that belongs to it. */ struct thread *t_conn_up; struct { zfpm_rnodes_iter_t iter; } t_conn_up_state; unsigned long connect_calls; time_t last_connect_call_time; /* * Stats from the start of the current statistics interval up to * now. These are the counters we typically update in the code. */ zfpm_stats_t stats; /* * Statistics that were gathered in the last collection interval. */ zfpm_stats_t last_ivl_stats; /* * Cumulative stats from the last clear to the start of the current * statistics interval. */ zfpm_stats_t cumulative_stats; /* * Stats interval timer. */ struct thread *t_stats; /* * If non-zero, the last time when statistics were cleared. */ time_t last_stats_clear_time; } zfpm_glob_t; static zfpm_glob_t zfpm_glob_space; static zfpm_glob_t *zfpm_g = &zfpm_glob_space; static int zfpm_read_cb (struct thread *thread); static int zfpm_write_cb (struct thread *thread); static void zfpm_set_state (zfpm_state_t state, const char *reason); static void zfpm_start_connect_timer (const char *reason); static void zfpm_start_stats_timer (void); /* * zfpm_thread_should_yield */ static inline int zfpm_thread_should_yield (struct thread *t) { return thread_should_yield (t); } /* * zfpm_state_to_str */ static const char * zfpm_state_to_str (zfpm_state_t state) { switch (state) { case ZFPM_STATE_IDLE: return "idle"; case ZFPM_STATE_ACTIVE: return "active"; case ZFPM_STATE_CONNECTING: return "connecting"; case ZFPM_STATE_ESTABLISHED: return "established"; default: return "unknown"; } } /* * zfpm_get_time */ static time_t zfpm_get_time (void) { struct timeval tv; if (quagga_gettime (QUAGGA_CLK_MONOTONIC, &tv) < 0) zlog_warn ("FPM: quagga_gettime failed!!"); return tv.tv_sec; } /* * zfpm_get_elapsed_time * * Returns the time elapsed (in seconds) since the given time. */ static time_t zfpm_get_elapsed_time (time_t reference) { time_t now; now = zfpm_get_time (); if (now < reference) { assert (0); return 0; } return now - reference; } /* * zfpm_is_table_for_fpm * * Returns TRUE if the the given table is to be communicated to the * FPM. */ static inline int zfpm_is_table_for_fpm (struct route_table *table) { rib_table_info_t *info; info = rib_table_info (table); /* * We only send the unicast tables in the main instance to the FPM * at this point. */ if (info->zvrf->vrf_id != 0) return 0; if (info->safi != SAFI_UNICAST) return 0; return 1; } /* * zfpm_rnodes_iter_init */ static inline void zfpm_rnodes_iter_init (zfpm_rnodes_iter_t *iter) { memset (iter, 0, sizeof (*iter)); rib_tables_iter_init (&iter->tables_iter); /* * This is a hack, but it makes implementing 'next' easier by * ensuring that route_table_iter_next() will return NULL the first * time we call it. */ route_table_iter_init (&iter->iter, NULL); route_table_iter_cleanup (&iter->iter); } /* * zfpm_rnodes_iter_next */ static inline struct route_node * zfpm_rnodes_iter_next (zfpm_rnodes_iter_t *iter) { struct route_node *rn; struct route_table *table; while (1) { rn = route_table_iter_next (&iter->iter); if (rn) return rn; /* * We've made our way through this table, go to the next one. */ route_table_iter_cleanup (&iter->iter); while ((table = rib_tables_iter_next (&iter->tables_iter))) { if (zfpm_is_table_for_fpm (table)) break; } if (!table) return NULL; route_table_iter_init (&iter->iter, table); } return NULL; } /* * zfpm_rnodes_iter_pause */ static inline void zfpm_rnodes_iter_pause (zfpm_rnodes_iter_t *iter) { route_table_iter_pause (&iter->iter); } /* * zfpm_rnodes_iter_cleanup */ static inline void zfpm_rnodes_iter_cleanup (zfpm_rnodes_iter_t *iter) { route_table_iter_cleanup (&iter->iter); rib_tables_iter_cleanup (&iter->tables_iter); } /* * zfpm_stats_init * * Initialize a statistics block. */ static inline void zfpm_stats_init (zfpm_stats_t *stats) { memset (stats, 0, sizeof (*stats)); } /* * zfpm_stats_reset */ static inline void zfpm_stats_reset (zfpm_stats_t *stats) { zfpm_stats_init (stats); } /* * zfpm_stats_copy */ static inline void zfpm_stats_copy (const zfpm_stats_t *src, zfpm_stats_t *dest) { memcpy (dest, src, sizeof (*dest)); } /* * zfpm_stats_compose * * Total up the statistics in two stats structures ('s1 and 's2') and * return the result in the third argument, 'result'. Note that the * pointer 'result' may be the same as 's1' or 's2'. * * For simplicity, the implementation below assumes that the stats * structure is composed entirely of counters. This can easily be * changed when necessary. */ static void zfpm_stats_compose (const zfpm_stats_t *s1, const zfpm_stats_t *s2, zfpm_stats_t *result) { const unsigned long *p1, *p2; unsigned long *result_p; int i, num_counters; p1 = (const unsigned long *) s1; p2 = (const unsigned long *) s2; result_p = (unsigned long *) result; num_counters = (sizeof (zfpm_stats_t) / sizeof (unsigned long)); for (i = 0; i < num_counters; i++) { result_p[i] = p1[i] + p2[i]; } } /* * zfpm_read_on */ static inline void zfpm_read_on (void) { assert (!zfpm_g->t_read); assert (zfpm_g->sock >= 0); THREAD_READ_ON (zfpm_g->master, zfpm_g->t_read, zfpm_read_cb, 0, zfpm_g->sock); } /* * zfpm_write_on */ static inline void zfpm_write_on (void) { assert (!zfpm_g->t_write); assert (zfpm_g->sock >= 0); THREAD_WRITE_ON (zfpm_g->master, zfpm_g->t_write, zfpm_write_cb, 0, zfpm_g->sock); } /* * zfpm_read_off */ static inline void zfpm_read_off (void) { THREAD_READ_OFF (zfpm_g->t_read); } /* * zfpm_write_off */ static inline void zfpm_write_off (void) { THREAD_WRITE_OFF (zfpm_g->t_write); } /* * zfpm_conn_up_thread_cb * * Callback for actions to be taken when the connection to the FPM * comes up. */ static int zfpm_conn_up_thread_cb (struct thread *thread) { struct route_node *rnode; zfpm_rnodes_iter_t *iter; rib_dest_t *dest; assert (zfpm_g->t_conn_up); zfpm_g->t_conn_up = NULL; iter = &zfpm_g->t_conn_up_state.iter; if (zfpm_g->state != ZFPM_STATE_ESTABLISHED) { zfpm_debug ("Connection not up anymore, conn_up thread aborting"); zfpm_g->stats.t_conn_up_aborts++; goto done; } while ((rnode = zfpm_rnodes_iter_next (iter))) { dest = rib_dest_from_rnode (rnode); if (dest) { zfpm_g->stats.t_conn_up_dests_processed++; zfpm_trigger_update (rnode, NULL); } /* * Yield if need be. */ if (!zfpm_thread_should_yield (thread)) continue; zfpm_g->stats.t_conn_up_yields++; zfpm_rnodes_iter_pause (iter); zfpm_g->t_conn_up = thread_add_background (zfpm_g->master, zfpm_conn_up_thread_cb, 0, 0); return 0; } zfpm_g->stats.t_conn_up_finishes++; done: zfpm_rnodes_iter_cleanup (iter); return 0; } /* * zfpm_connection_up * * Called when the connection to the FPM comes up. */ static void zfpm_connection_up (const char *detail) { assert (zfpm_g->sock >= 0); zfpm_read_on (); zfpm_write_on (); zfpm_set_state (ZFPM_STATE_ESTABLISHED, detail); /* * Start thread to push existing routes to the FPM. */ assert (!zfpm_g->t_conn_up); zfpm_rnodes_iter_init (&zfpm_g->t_conn_up_state.iter); zfpm_debug ("Starting conn_up thread"); zfpm_g->t_conn_up = thread_add_background (zfpm_g->master, zfpm_conn_up_thread_cb, 0, 0); zfpm_g->stats.t_conn_up_starts++; } /* * zfpm_connect_check * * Check if an asynchronous connect() to the FPM is complete. */ static void zfpm_connect_check () { int status; socklen_t slen; int ret; zfpm_read_off (); zfpm_write_off (); slen = sizeof (status); ret = getsockopt (zfpm_g->sock, SOL_SOCKET, SO_ERROR, (void *) &status, &slen); if (ret >= 0 && status == 0) { zfpm_connection_up ("async connect complete"); return; } /* * getsockopt() failed or indicated an error on the socket. */ close (zfpm_g->sock); zfpm_g->sock = -1; zfpm_start_connect_timer ("getsockopt() after async connect failed"); return; } /* * zfpm_conn_down_thread_cb * * Callback that is invoked to clean up state after the TCP connection * to the FPM goes down. */ static int zfpm_conn_down_thread_cb (struct thread *thread) { struct route_node *rnode; zfpm_rnodes_iter_t *iter; rib_dest_t *dest; assert (zfpm_g->state == ZFPM_STATE_IDLE); assert (zfpm_g->t_conn_down); zfpm_g->t_conn_down = NULL; iter = &zfpm_g->t_conn_down_state.iter; while ((rnode = zfpm_rnodes_iter_next (iter))) { dest = rib_dest_from_rnode (rnode); if (dest) { if (CHECK_FLAG (dest->flags, RIB_DEST_UPDATE_FPM)) { TAILQ_REMOVE (&zfpm_g->dest_q, dest, fpm_q_entries); } UNSET_FLAG (dest->flags, RIB_DEST_UPDATE_FPM); UNSET_FLAG (dest->flags, RIB_DEST_SENT_TO_FPM); zfpm_g->stats.t_conn_down_dests_processed++; /* * Check if the dest should be deleted. */ rib_gc_dest(rnode); } /* * Yield if need be. */ if (!zfpm_thread_should_yield (thread)) continue; zfpm_g->stats.t_conn_down_yields++; zfpm_rnodes_iter_pause (iter); zfpm_g->t_conn_down = thread_add_background (zfpm_g->master, zfpm_conn_down_thread_cb, 0, 0); return 0; } zfpm_g->stats.t_conn_down_finishes++; zfpm_rnodes_iter_cleanup (iter); /* * Start the process of connecting to the FPM again. */ zfpm_start_connect_timer ("cleanup complete"); return 0; } /* * zfpm_connection_down * * Called when the connection to the FPM has gone down. */ static void zfpm_connection_down (const char *detail) { if (!detail) detail = "unknown"; assert (zfpm_g->state == ZFPM_STATE_ESTABLISHED); zlog_info ("connection to the FPM has gone down: %s", detail); zfpm_read_off (); zfpm_write_off (); stream_reset (zfpm_g->ibuf); stream_reset (zfpm_g->obuf); if (zfpm_g->sock >= 0) { close (zfpm_g->sock); zfpm_g->sock = -1; } /* * Start thread to clean up state after the connection goes down. */ assert (!zfpm_g->t_conn_down); zfpm_debug ("Starting conn_down thread"); zfpm_rnodes_iter_init (&zfpm_g->t_conn_down_state.iter); zfpm_g->t_conn_down = thread_add_background (zfpm_g->master, zfpm_conn_down_thread_cb, 0, 0); zfpm_g->stats.t_conn_down_starts++; zfpm_set_state (ZFPM_STATE_IDLE, detail); } /* * zfpm_read_cb */ static int zfpm_read_cb (struct thread *thread) { size_t already; struct stream *ibuf; uint16_t msg_len; fpm_msg_hdr_t *hdr; zfpm_g->stats.read_cb_calls++; assert (zfpm_g->t_read); zfpm_g->t_read = NULL; /* * Check if async connect is now done. */ if (zfpm_g->state == ZFPM_STATE_CONNECTING) { zfpm_connect_check(); return 0; } assert (zfpm_g->state == ZFPM_STATE_ESTABLISHED); assert (zfpm_g->sock >= 0); ibuf = zfpm_g->ibuf; already = stream_get_endp (ibuf); if (already < FPM_MSG_HDR_LEN) { ssize_t nbyte; nbyte = stream_read_try (ibuf, zfpm_g->sock, FPM_MSG_HDR_LEN - already); if (nbyte == 0 || nbyte == -1) { zfpm_connection_down ("closed socket in read"); return 0; } if (nbyte != (ssize_t) (FPM_MSG_HDR_LEN - already)) goto done; already = FPM_MSG_HDR_LEN; } stream_set_getp (ibuf, 0); hdr = (fpm_msg_hdr_t *) stream_pnt (ibuf); if (!fpm_msg_hdr_ok (hdr)) { zfpm_connection_down ("invalid message header"); return 0; } msg_len = fpm_msg_len (hdr); /* * Read out the rest of the packet. */ if (already < msg_len) { ssize_t nbyte; nbyte = stream_read_try (ibuf, zfpm_g->sock, msg_len - already); if (nbyte == 0 || nbyte == -1) { zfpm_connection_down ("failed to read message"); return 0; } if (nbyte != (ssize_t) (msg_len - already)) goto done; } zfpm_debug ("Read out a full fpm message"); /* * Just throw it away for now. */ stream_reset (ibuf); done: zfpm_read_on (); return 0; } /* * zfpm_writes_pending * * Returns TRUE if we may have something to write to the FPM. */ static int zfpm_writes_pending (void) { /* * Check if there is any data in the outbound buffer that has not * been written to the socket yet. */ if (stream_get_endp (zfpm_g->obuf) - stream_get_getp (zfpm_g->obuf)) return 1; /* * Check if there are any prefixes on the outbound queue. */ if (!TAILQ_EMPTY (&zfpm_g->dest_q)) return 1; return 0; } /* * zfpm_encode_route * * Encode a message to the FPM with information about the given route. * * Returns the number of bytes written to the buffer. 0 or a negative * value indicates an error. */ static inline int zfpm_encode_route (rib_dest_t *dest, struct rib *rib, char *in_buf, size_t in_buf_len) { #ifndef HAVE_NETLINK return 0; #else int cmd; cmd = rib ? RTM_NEWROUTE : RTM_DELROUTE; return zfpm_netlink_encode_route (cmd, dest, rib, in_buf, in_buf_len); #endif /* HAVE_NETLINK */ } /* * zfpm_route_for_update * * Returns the rib that is to be sent to the FPM for a given dest. */ static struct rib * zfpm_route_for_update (rib_dest_t *dest) { struct rib *rib; RIB_DEST_FOREACH_ROUTE (dest, rib) { if (!CHECK_FLAG (rib->flags, ZEBRA_FLAG_SELECTED)) continue; return rib; } /* * We have no route for this destination. */ return NULL; } /* * zfpm_build_updates * * Process the outgoing queue and write messages to the outbound * buffer. */ static void zfpm_build_updates (void) { struct stream *s; rib_dest_t *dest; unsigned char *buf, *data, *buf_end; size_t msg_len; size_t data_len; fpm_msg_hdr_t *hdr; struct rib *rib; int is_add, write_msg; s = zfpm_g->obuf; assert (stream_empty (s)); do { /* * Make sure there is enough space to write another message. */ if (STREAM_WRITEABLE (s) < FPM_MAX_MSG_LEN) break; buf = STREAM_DATA (s) + stream_get_endp (s); buf_end = buf + STREAM_WRITEABLE (s); dest = TAILQ_FIRST (&zfpm_g->dest_q); if (!dest) break; assert (CHECK_FLAG (dest->flags, RIB_DEST_UPDATE_FPM)); hdr = (fpm_msg_hdr_t *) buf; hdr->version = FPM_PROTO_VERSION; hdr->msg_type = FPM_MSG_TYPE_NETLINK; data = fpm_msg_data (hdr); rib = zfpm_route_for_update (dest); is_add = rib ? 1 : 0; write_msg = 1; /* * If this is a route deletion, and we have not sent the route to * the FPM previously, skip it. */ if (!is_add && !CHECK_FLAG (dest->flags, RIB_DEST_SENT_TO_FPM)) { write_msg = 0; zfpm_g->stats.nop_deletes_skipped++; } if (write_msg) { data_len = zfpm_encode_route (dest, rib, (char *) data, buf_end - data); assert (data_len); if (data_len) { msg_len = fpm_data_len_to_msg_len (data_len); hdr->msg_len = htons (msg_len); stream_forward_endp (s, msg_len); if (is_add) zfpm_g->stats.route_adds++; else zfpm_g->stats.route_dels++; } } /* * Remove the dest from the queue, and reset the flag. */ UNSET_FLAG (dest->flags, RIB_DEST_UPDATE_FPM); TAILQ_REMOVE (&zfpm_g->dest_q, dest, fpm_q_entries); if (is_add) { SET_FLAG (dest->flags, RIB_DEST_SENT_TO_FPM); } else { UNSET_FLAG (dest->flags, RIB_DEST_SENT_TO_FPM); } /* * Delete the destination if necessary. */ if (rib_gc_dest (dest->rnode)) zfpm_g->stats.dests_del_after_update++; } while (1); } /* * zfpm_write_cb */ static int zfpm_write_cb (struct thread *thread) { struct stream *s; int num_writes; zfpm_g->stats.write_cb_calls++; assert (zfpm_g->t_write); zfpm_g->t_write = NULL; /* * Check if async connect is now done. */ if (zfpm_g->state == ZFPM_STATE_CONNECTING) { zfpm_connect_check (); return 0; } assert (zfpm_g->state == ZFPM_STATE_ESTABLISHED); assert (zfpm_g->sock >= 0); num_writes = 0; do { int bytes_to_write, bytes_written; s = zfpm_g->obuf; /* * If the stream is empty, try fill it up with data. */ if (stream_empty (s)) { zfpm_build_updates (); } bytes_to_write = stream_get_endp (s) - stream_get_getp (s); if (!bytes_to_write) break; bytes_written = write (zfpm_g->sock, STREAM_PNT (s), bytes_to_write); zfpm_g->stats.write_calls++; num_writes++; if (bytes_written < 0) { if (ERRNO_IO_RETRY (errno)) break; zfpm_connection_down ("failed to write to socket"); return 0; } if (bytes_written != bytes_to_write) { /* * Partial write. */ stream_forward_getp (s, bytes_written); zfpm_g->stats.partial_writes++; break; } /* * We've written out the entire contents of the stream. */ stream_reset (s); if (num_writes >= ZFPM_MAX_WRITES_PER_RUN) { zfpm_g->stats.max_writes_hit++; break; } if (zfpm_thread_should_yield (thread)) { zfpm_g->stats.t_write_yields++; break; } } while (1); if (zfpm_writes_pending ()) zfpm_write_on (); return 0; } /* * zfpm_connect_cb */ static int zfpm_connect_cb (struct thread *t) { int sock, ret; struct sockaddr_in serv; assert (zfpm_g->t_connect); zfpm_g->t_connect = NULL; assert (zfpm_g->state == ZFPM_STATE_ACTIVE); sock = socket (AF_INET, SOCK_STREAM, 0); if (sock < 0) { zfpm_debug ("Failed to create socket for connect(): %s", strerror(errno)); zfpm_g->stats.connect_no_sock++; return 0; } set_nonblocking(sock); /* Make server socket. */ memset (&serv, 0, sizeof (serv)); serv.sin_family = AF_INET; serv.sin_port = htons (zfpm_g->fpm_port); #ifdef HAVE_STRUCT_SOCKADDR_IN_SIN_LEN serv.sin_len = sizeof (struct sockaddr_in); #endif /* HAVE_STRUCT_SOCKADDR_IN_SIN_LEN */ serv.sin_addr.s_addr = htonl (INADDR_LOOPBACK); /* * Connect to the FPM. */ zfpm_g->connect_calls++; zfpm_g->stats.connect_calls++; zfpm_g->last_connect_call_time = zfpm_get_time (); ret = connect (sock, (struct sockaddr *) &serv, sizeof (serv)); if (ret >= 0) { zfpm_g->sock = sock; zfpm_connection_up ("connect succeeded"); return 1; } if (errno == EINPROGRESS) { zfpm_g->sock = sock; zfpm_read_on (); zfpm_write_on (); zfpm_set_state (ZFPM_STATE_CONNECTING, "async connect in progress"); return 0; } zlog_info ("can't connect to FPM %d: %s", sock, safe_strerror (errno)); close (sock); /* * Restart timer for retrying connection. */ zfpm_start_connect_timer ("connect() failed"); return 0; } /* * zfpm_set_state * * Move state machine into the given state. */ static void zfpm_set_state (zfpm_state_t state, const char *reason) { zfpm_state_t cur_state = zfpm_g->state; if (!reason) reason = "Unknown"; if (state == cur_state) return; zfpm_debug("beginning state transition %s -> %s. Reason: %s", zfpm_state_to_str (cur_state), zfpm_state_to_str (state), reason); switch (state) { case ZFPM_STATE_IDLE: assert (cur_state == ZFPM_STATE_ESTABLISHED); break; case ZFPM_STATE_ACTIVE: assert (cur_state == ZFPM_STATE_IDLE || cur_state == ZFPM_STATE_CONNECTING); assert (zfpm_g->t_connect); break; case ZFPM_STATE_CONNECTING: assert (zfpm_g->sock); assert (cur_state == ZFPM_STATE_ACTIVE); assert (zfpm_g->t_read); assert (zfpm_g->t_write); break; case ZFPM_STATE_ESTABLISHED: assert (cur_state == ZFPM_STATE_ACTIVE || cur_state == ZFPM_STATE_CONNECTING); assert (zfpm_g->sock); assert (zfpm_g->t_read); assert (zfpm_g->t_write); break; } zfpm_g->state = state; } /* * zfpm_calc_connect_delay * * Returns the number of seconds after which we should attempt to * reconnect to the FPM. */ static long zfpm_calc_connect_delay (void) { time_t elapsed; /* * Return 0 if this is our first attempt to connect. */ if (zfpm_g->connect_calls == 0) { return 0; } elapsed = zfpm_get_elapsed_time (zfpm_g->last_connect_call_time); if (elapsed > ZFPM_CONNECT_RETRY_IVL) { return 0; } return ZFPM_CONNECT_RETRY_IVL - elapsed; } /* * zfpm_start_connect_timer */ static void zfpm_start_connect_timer (const char *reason) { long delay_secs; assert (!zfpm_g->t_connect); assert (zfpm_g->sock < 0); assert(zfpm_g->state == ZFPM_STATE_IDLE || zfpm_g->state == ZFPM_STATE_ACTIVE || zfpm_g->state == ZFPM_STATE_CONNECTING); delay_secs = zfpm_calc_connect_delay(); zfpm_debug ("scheduling connect in %ld seconds", delay_secs); THREAD_TIMER_ON (zfpm_g->master, zfpm_g->t_connect, zfpm_connect_cb, 0, delay_secs); zfpm_set_state (ZFPM_STATE_ACTIVE, reason); } /* * zfpm_is_enabled * * Returns TRUE if the zebra FPM module has been enabled. */ static inline int zfpm_is_enabled (void) { return zfpm_g->enabled; } /* * zfpm_conn_is_up * * Returns TRUE if the connection to the FPM is up. */ static inline int zfpm_conn_is_up (void) { if (zfpm_g->state != ZFPM_STATE_ESTABLISHED) return 0; assert (zfpm_g->sock >= 0); return 1; } /* * zfpm_trigger_update * * The zebra code invokes this function to indicate that we should * send an update to the FPM about the given route_node. */ void zfpm_trigger_update (struct route_node *rn, const char *reason) { rib_dest_t *dest; char buf[PREFIX_STRLEN]; /* * Ignore if the connection is down. We will update the FPM about * all destinations once the connection comes up. */ if (!zfpm_conn_is_up ()) return; dest = rib_dest_from_rnode (rn); /* * Ignore the trigger if the dest is not in a table that we would * send to the FPM. */ if (!zfpm_is_table_for_fpm (rib_dest_table (dest))) { zfpm_g->stats.non_fpm_table_triggers++; return; } if (CHECK_FLAG (dest->flags, RIB_DEST_UPDATE_FPM)) { zfpm_g->stats.redundant_triggers++; return; } if (reason) { zfpm_debug ("%s triggering update to FPM - Reason: %s", prefix2str (&rn->p, buf, sizeof(buf)), reason); } SET_FLAG (dest->flags, RIB_DEST_UPDATE_FPM); TAILQ_INSERT_TAIL (&zfpm_g->dest_q, dest, fpm_q_entries); zfpm_g->stats.updates_triggered++; /* * Make sure that writes are enabled. */ if (zfpm_g->t_write) return; zfpm_write_on (); } /* * zfpm_stats_timer_cb */ static int zfpm_stats_timer_cb (struct thread *t) { assert (zfpm_g->t_stats); zfpm_g->t_stats = NULL; /* * Remember the stats collected in the last interval for display * purposes. */ zfpm_stats_copy (&zfpm_g->stats, &zfpm_g->last_ivl_stats); /* * Add the current set of stats into the cumulative statistics. */ zfpm_stats_compose (&zfpm_g->cumulative_stats, &zfpm_g->stats, &zfpm_g->cumulative_stats); /* * Start collecting stats afresh over the next interval. */ zfpm_stats_reset (&zfpm_g->stats); zfpm_start_stats_timer (); return 0; } /* * zfpm_stop_stats_timer */ static void zfpm_stop_stats_timer (void) { if (!zfpm_g->t_stats) return; zfpm_debug ("Stopping existing stats timer"); THREAD_TIMER_OFF (zfpm_g->t_stats); } /* * zfpm_start_stats_timer */ void zfpm_start_stats_timer (void) { assert (!zfpm_g->t_stats); THREAD_TIMER_ON (zfpm_g->master, zfpm_g->t_stats, zfpm_stats_timer_cb, 0, ZFPM_STATS_IVL_SECS); } /* * Helper macro for zfpm_show_stats() below. */ #define ZFPM_SHOW_STAT(counter) \ do { \ vty_out (vty, "%-40s %10lu %16lu%s", #counter, total_stats.counter, \ zfpm_g->last_ivl_stats.counter, VTY_NEWLINE); \ } while (0) /* * zfpm_show_stats */ static void zfpm_show_stats (struct vty *vty) { zfpm_stats_t total_stats; time_t elapsed; vty_out (vty, "%s%-40s %10s Last %2d secs%s%s", VTY_NEWLINE, "Counter", "Total", ZFPM_STATS_IVL_SECS, VTY_NEWLINE, VTY_NEWLINE); /* * Compute the total stats up to this instant. */ zfpm_stats_compose (&zfpm_g->cumulative_stats, &zfpm_g->stats, &total_stats); ZFPM_SHOW_STAT (connect_calls); ZFPM_SHOW_STAT (connect_no_sock); ZFPM_SHOW_STAT (read_cb_calls); ZFPM_SHOW_STAT (write_cb_calls); ZFPM_SHOW_STAT (write_calls); ZFPM_SHOW_STAT (partial_writes); ZFPM_SHOW_STAT (max_writes_hit); ZFPM_SHOW_STAT (t_write_yields); ZFPM_SHOW_STAT (nop_deletes_skipped); ZFPM_SHOW_STAT (route_adds); ZFPM_SHOW_STAT (route_dels); ZFPM_SHOW_STAT (updates_triggered); ZFPM_SHOW_STAT (non_fpm_table_triggers); ZFPM_SHOW_STAT (redundant_triggers); ZFPM_SHOW_STAT (dests_del_after_update); ZFPM_SHOW_STAT (t_conn_down_starts); ZFPM_SHOW_STAT (t_conn_down_dests_processed); ZFPM_SHOW_STAT (t_conn_down_yields); ZFPM_SHOW_STAT (t_conn_down_finishes); ZFPM_SHOW_STAT (t_conn_up_starts); ZFPM_SHOW_STAT (t_conn_up_dests_processed); ZFPM_SHOW_STAT (t_conn_up_yields); ZFPM_SHOW_STAT (t_conn_up_aborts); ZFPM_SHOW_STAT (t_conn_up_finishes); if (!zfpm_g->last_stats_clear_time) return; elapsed = zfpm_get_elapsed_time (zfpm_g->last_stats_clear_time); vty_out (vty, "%sStats were cleared %lu seconds ago%s", VTY_NEWLINE, (unsigned long) elapsed, VTY_NEWLINE); } /* * zfpm_clear_stats */ static void zfpm_clear_stats (struct vty *vty) { if (!zfpm_is_enabled ()) { vty_out (vty, "The FPM module is not enabled...%s", VTY_NEWLINE); return; } zfpm_stats_reset (&zfpm_g->stats); zfpm_stats_reset (&zfpm_g->last_ivl_stats); zfpm_stats_reset (&zfpm_g->cumulative_stats); zfpm_stop_stats_timer (); zfpm_start_stats_timer (); zfpm_g->last_stats_clear_time = zfpm_get_time(); vty_out (vty, "Cleared FPM stats%s", VTY_NEWLINE); } /* * show_zebra_fpm_stats */ DEFUN (show_zebra_fpm_stats, show_zebra_fpm_stats_cmd, "show zebra fpm stats", SHOW_STR "Zebra information\n" "Forwarding Path Manager information\n" "Statistics\n") { zfpm_show_stats (vty); return CMD_SUCCESS; } /* * clear_zebra_fpm_stats */ DEFUN (clear_zebra_fpm_stats, clear_zebra_fpm_stats_cmd, "clear zebra fpm stats", CLEAR_STR "Zebra information\n" "Clear Forwarding Path Manager information\n" "Statistics\n") { zfpm_clear_stats (vty); return CMD_SUCCESS; } /** * zfpm_init * * One-time initialization of the Zebra FPM module. * * @param[in] port port at which FPM is running. * @param[in] enable TRUE if the zebra FPM module should be enabled * * Returns TRUE on success. */ int zfpm_init (struct thread_master *master, int enable, uint16_t port) { static int initialized = 0; if (initialized) { return 1; } initialized = 1; memset (zfpm_g, 0, sizeof (*zfpm_g)); zfpm_g->master = master; TAILQ_INIT(&zfpm_g->dest_q); zfpm_g->sock = -1; zfpm_g->state = ZFPM_STATE_IDLE; /* * Netlink must currently be available for the Zebra-FPM interface * to be enabled. */ #ifndef HAVE_NETLINK enable = 0; #endif zfpm_g->enabled = enable; zfpm_stats_init (&zfpm_g->stats); zfpm_stats_init (&zfpm_g->last_ivl_stats); zfpm_stats_init (&zfpm_g->cumulative_stats); install_element (ENABLE_NODE, &show_zebra_fpm_stats_cmd); install_element (ENABLE_NODE, &clear_zebra_fpm_stats_cmd); if (!enable) { return 1; } if (!port) port = FPM_DEFAULT_PORT; zfpm_g->fpm_port = port; zfpm_g->obuf = stream_new (ZFPM_OBUF_SIZE); zfpm_g->ibuf = stream_new (ZFPM_IBUF_SIZE); zfpm_start_stats_timer (); zfpm_start_connect_timer ("initialized"); return 1; }