/* Zebra Mlag Code. * Copyright (C) 2018 Cumulus Networks, Inc. * Donald Sharp * * This file is part of FRR. * * FRR 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. * * FRR 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 FRR; see the file COPYING. If not, write to the Free * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA * 02111-1307, USA. */ #include "zebra.h" #include "command.h" #include "hook.h" #include "frr_pthread.h" #include "mlag.h" #include "zebra/zebra_mlag.h" #include "zebra/zebra_mlag_private.h" #include "zebra/zebra_router.h" #include "zebra/zebra_memory.h" #include "zebra/zapi_msg.h" #include "zebra/debug.h" #ifndef VTYSH_EXTRACT_PL #include "zebra/zebra_mlag_clippy.c" #endif #define ZEBRA_MLAG_METADATA_LEN 4 #define ZEBRA_MLAG_MSG_BCAST 0xFFFFFFFF uint8_t mlag_wr_buffer[ZEBRA_MLAG_BUF_LIMIT]; uint8_t mlag_rd_buffer[ZEBRA_MLAG_BUF_LIMIT]; uint32_t mlag_rd_buf_offset; static bool test_mlag_in_progress; static int zebra_mlag_signal_write_thread(void); static int zebra_mlag_terminate_pthread(struct thread *event); static int zebra_mlag_post_data_from_main_thread(struct thread *thread); static void zebra_mlag_publish_process_state(struct zserv *client, zebra_message_types_t msg_type); /**********************MLAG Interaction***************************************/ /* * API to post the Registration to MLAGD * MLAG will not process any messages with out the registration */ void zebra_mlag_send_register(void) { struct stream *s = NULL; s = stream_new(sizeof(struct mlag_msg)); stream_putl(s, MLAG_REGISTER); stream_putw(s, MLAG_MSG_NULL_PAYLOAD); stream_putw(s, MLAG_MSG_NO_BATCH); stream_fifo_push_safe(zrouter.mlag_info.mlag_fifo, s); zebra_mlag_signal_write_thread(); if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("%s: Enqueued MLAG Register to MLAG Thread ", __func__); } /* * API to post the De-Registration to MLAGD * MLAG will not process any messages after the de-registration */ void zebra_mlag_send_deregister(void) { struct stream *s = NULL; s = stream_new(sizeof(struct mlag_msg)); stream_putl(s, MLAG_DEREGISTER); stream_putw(s, MLAG_MSG_NULL_PAYLOAD); stream_putw(s, MLAG_MSG_NO_BATCH); stream_fifo_push_safe(zrouter.mlag_info.mlag_fifo, s); zebra_mlag_signal_write_thread(); if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("%s: Enqueued MLAG De-Register to MLAG Thread ", __func__); } /* * API To handle MLAG Received data * Decodes the data using protobuf and enqueue to main thread * main thread publish this to clients based on client subscription */ void zebra_mlag_process_mlag_data(uint8_t *data, uint32_t len) { struct stream *s = NULL; struct stream *s1 = NULL; int msg_type = 0; s = stream_new(ZEBRA_MAX_PACKET_SIZ); msg_type = zebra_mlag_protobuf_decode_message(&s, data, len); if (msg_type <= 0) { /* Something went wrong in decoding */ stream_free(s); zlog_err("%s: failed to process mlag data-%d, %u", __func__, msg_type, len); return; } /* * additional four bytes are for message type */ s1 = stream_new(stream_get_endp(s) + ZEBRA_MLAG_METADATA_LEN); stream_putl(s1, msg_type); stream_put(s1, s->data, stream_get_endp(s)); thread_add_event(zrouter.master, zebra_mlag_post_data_from_main_thread, s1, 0, NULL); stream_free(s); } /**********************End of MLAG Interaction********************************/ /************************MLAG Thread Processing*******************************/ /* * after posting every 'ZEBRA_MLAG_POST_LIMIT' packets, MLAG Thread will be * yielded to give CPU for other threads */ #define ZEBRA_MLAG_POST_LIMIT 100 /* * This thread reads the clients data from the Global queue and encodes with * protobuf and pass on to the MLAG socket. */ static int zebra_mlag_client_msg_handler(struct thread *event) { struct stream *s; uint32_t wr_count = 0; uint32_t msg_type = 0; int len = 0; wr_count = stream_fifo_count_safe(zrouter.mlag_info.mlag_fifo); if (IS_ZEBRA_DEBUG_MLAG) zlog_debug(":%s: Processing MLAG write, %d messages in queue", __func__, wr_count); zrouter.mlag_info.t_write = NULL; for (wr_count = 0; wr_count < ZEBRA_MLAG_POST_LIMIT; wr_count++) { /* FIFO is empty,wait for teh message to be add */ if (stream_fifo_count_safe(zrouter.mlag_info.mlag_fifo) == 0) break; s = stream_fifo_pop_safe(zrouter.mlag_info.mlag_fifo); if (!s) { zlog_debug(":%s: Got a NULL Messages, some thing wrong", __func__); break; } zebra_mlag_reset_write_buffer(); /* * Encode the data now */ len = zebra_mlag_protobuf_encode_client_data(s, &msg_type); /* * write to MCLAGD */ if (len > 0) zebra_mlag_private_write_data(mlag_wr_buffer, len); /* * If message type is De-register, send a signal to main thread, * so that necessary cleanup will be done by main thread. */ if (msg_type == MLAG_DEREGISTER) { thread_add_event(zrouter.master, zebra_mlag_terminate_pthread, NULL, 0, NULL); } stream_free(s); } if (IS_ZEBRA_DEBUG_MLAG) zlog_debug(":%s: Posted %d messages to MLAGD", __func__, wr_count); /* * Currently there is only message write task is enqueued to this * thread, yielding was added for future purpose, so that this thread * can server other tasks also and in case FIFO is empty, this task will * be schedule when main thread adds some messages */ if (wr_count >= ZEBRA_MLAG_POST_LIMIT) zebra_mlag_signal_write_thread(); return 0; } /* * API to handle the process state. * In case of Down, Zebra keep monitoring the MLAG state. * all the state Notifications will be published to clients */ void zebra_mlag_handle_process_state(enum zebra_mlag_state state) { if (state == MLAG_UP) { zrouter.mlag_info.connected = true; zebra_mlag_publish_process_state(NULL, ZEBRA_MLAG_PROCESS_UP); zebra_mlag_send_register(); } else if (state == MLAG_DOWN) { zrouter.mlag_info.connected = false; zebra_mlag_publish_process_state(NULL, ZEBRA_MLAG_PROCESS_DOWN); zebra_mlag_private_monitor_state(); } } /***********************End of MLAG Thread processing*************************/ /*************************Multi-entratnt Api's********************************/ /* * Provider api to signal that work/events are available * for the Zebra MLAG Write pthread. * This API is called from 2 pthreads.. * 1) by main thread when client posts a MLAG Message * 2) by MLAG Thread, in case of yield * though this api, is called from two threads we don't need any locking * because Thread task enqueue is thread safe means internally it had * necessary protection */ static int zebra_mlag_signal_write_thread(void) { if (zrouter.mlag_info.zebra_pth_mlag) { if (IS_ZEBRA_DEBUG_MLAG) zlog_debug(":%s: Scheduling MLAG write", __func__); thread_add_event(zrouter.mlag_info.th_master, zebra_mlag_client_msg_handler, NULL, 0, &zrouter.mlag_info.t_write); } return 0; } /* * API will be used to publish the MLAG state to interested clients * In case client is passed, state is posted only for that client, * otherwise to all interested clients * this api can be called from two threads. * 1) from main thread: when client is passed * 2) from MLAG Thread: when client is NULL * * In second case, to avoid global data access data will be post to Main * thread, so that actual posting to clients will happen from Main thread. */ static void zebra_mlag_publish_process_state(struct zserv *client, zebra_message_types_t msg_type) { struct stream *s; if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("%s: Publishing MLAG process state:%s to %s Client", __func__, (msg_type == ZEBRA_MLAG_PROCESS_UP) ? "UP" : "DOWN", (client) ? "one" : "all"); if (client) { s = stream_new(ZEBRA_HEADER_SIZE); zclient_create_header(s, msg_type, VRF_DEFAULT); zserv_send_message(client, s); return; } /* * additional four bytes are for mesasge type */ s = stream_new(ZEBRA_HEADER_SIZE + ZEBRA_MLAG_METADATA_LEN); stream_putl(s, ZEBRA_MLAG_MSG_BCAST); zclient_create_header(s, msg_type, VRF_DEFAULT); thread_add_event(zrouter.master, zebra_mlag_post_data_from_main_thread, s, 0, NULL); } /**************************End of Multi-entrant Apis**************************/ /***********************Zebra Main thread processing**************************/ /* * To avoid data corruption, messages will be post to clients only from * main thread, because for that access was needed for clients list. * so instead of forcing the locks, messages will be posted from main thread. */ static int zebra_mlag_post_data_from_main_thread(struct thread *thread) { struct stream *s = THREAD_ARG(thread); struct stream *zebra_s = NULL; struct listnode *node; struct zserv *client; uint32_t msg_type = 0; uint32_t msg_len = 0; if (!s) return -1; STREAM_GETL(s, msg_type); if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "%s: Posting MLAG data for msg_type:0x%x to interested cleints", __func__, msg_type); msg_len = s->endp - ZEBRA_MLAG_METADATA_LEN; for (ALL_LIST_ELEMENTS_RO(zrouter.client_list, node, client)) { if (client->mlag_updates_interested == true) { if (msg_type != ZEBRA_MLAG_MSG_BCAST && !CHECK_FLAG(client->mlag_reg_mask1, (1 << msg_type))) { continue; } if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "%s: Posting MLAG data of length-%d to client:%d ", __func__, msg_len, client->proto); zebra_s = stream_new(msg_len); STREAM_GET(zebra_s->data, s, msg_len); zebra_s->endp = msg_len; stream_putw_at(zebra_s, 0, msg_len); /* * This stream will be enqueued to client_obuf, it will * be freed after posting to client socket. */ zserv_send_message(client, zebra_s); zebra_s = NULL; } } stream_free(s); return 0; stream_failure: stream_free(s); if (zebra_s) stream_free(zebra_s); return 0; } /* * Start the MLAG Thread, this will be used to write client data on to * MLAG Process and to read the data from MLAG and post to cleints. * when all clients are un-registered, this Thread will be * suspended. */ static void zebra_mlag_spawn_pthread(void) { /* Start MLAG write pthread */ struct frr_pthread_attr pattr = {.start = frr_pthread_attr_default.start, .stop = frr_pthread_attr_default.stop}; zrouter.mlag_info.zebra_pth_mlag = frr_pthread_new(&pattr, "Zebra MLAG thread", "Zebra MLAG"); zrouter.mlag_info.th_master = zrouter.mlag_info.zebra_pth_mlag->master; /* Enqueue an initial event for the dataplane pthread */ zebra_mlag_signal_write_thread(); frr_pthread_run(zrouter.mlag_info.zebra_pth_mlag, NULL); } /* * all clients are un-registered for MLAG Updates, terminate the * MLAG write thread */ static int zebra_mlag_terminate_pthread(struct thread *event) { if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Zebra MLAG write thread terminate called"); if (zrouter.mlag_info.clients_interested_cnt) { if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "Zebra MLAG: still some clients are interested"); return 0; } frr_pthread_stop(zrouter.mlag_info.zebra_pth_mlag, NULL); /* Destroy pthread */ frr_pthread_destroy(zrouter.mlag_info.zebra_pth_mlag); zrouter.mlag_info.zebra_pth_mlag = NULL; zrouter.mlag_info.th_master = NULL; zrouter.mlag_info.t_read = NULL; zrouter.mlag_info.t_write = NULL; /* * Send Notification to clean private data */ zebra_mlag_private_cleanup_data(); return 0; } /* * API to register zebra client for MLAG Updates */ void zebra_mlag_client_register(ZAPI_HANDLER_ARGS) { struct stream *s; uint32_t reg_mask = 0; int rc = 0; if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Received MLAG Registration from client-proto:%d", client->proto); /* Get input stream. */ s = msg; /* Get data. */ STREAM_GETL(s, reg_mask); if (client->mlag_updates_interested == true) { if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "Client is registered, existing mask: 0x%x, new mask: 0x%x", client->mlag_reg_mask1, reg_mask); if (client->mlag_reg_mask1 != reg_mask) client->mlag_reg_mask1 = reg_mask; /* * Client might missed MLAG-UP Notification, post-it again */ zebra_mlag_publish_process_state(client, ZEBRA_MLAG_PROCESS_UP); return; } client->mlag_updates_interested = true; client->mlag_reg_mask1 = reg_mask; if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Registering for MLAG Updates with mask: 0x%x, ", client->mlag_reg_mask1); zrouter.mlag_info.clients_interested_cnt++; if (zrouter.mlag_info.clients_interested_cnt == 1) { /* * First-client for MLAG Updates,open the communication channel * with MLAG */ if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "First client, opening the channel with MLAG"); zebra_mlag_spawn_pthread(); rc = zebra_mlag_private_open_channel(); if (rc < 0) { /* * For some reason, zebra not able to open the * comm-channel with MLAG, so post MLAG-DOWN to client. * later when the channel is open, zebra will send * MLAG-UP */ if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "Fail to open channel with MLAG,rc:%d, post Proto-down", rc); zebra_mlag_publish_process_state( client, ZEBRA_MLAG_PROCESS_DOWN); } } if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Client Registered successfully for MLAG Updates"); if (zrouter.mlag_info.connected == true) zebra_mlag_publish_process_state(client, ZEBRA_MLAG_PROCESS_UP); stream_failure: return; } /* * API to un-register for MLAG Updates */ void zebra_mlag_client_unregister(ZAPI_HANDLER_ARGS) { if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Received MLAG De-Registration from client-proto:%d", client->proto); if (client->mlag_updates_interested == false) /* Unexpected */ return; client->mlag_updates_interested = false; client->mlag_reg_mask1 = 0; zrouter.mlag_info.clients_interested_cnt--; if (zrouter.mlag_info.clients_interested_cnt == 0) { /* * No-client is interested for MLAG Updates,close the * communication channel with MLAG */ if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Last client for MLAG, close the channel "); /* * Clean up flow: * ============= * 1) main thread calls socket close which posts De-register * to MLAG write thread * 2) after MLAG write thread posts De-register it sends a * signal back to main thread to do the thread cleanup * this was mainly to make sure De-register is posted to MCLAGD. */ zebra_mlag_private_close_channel(); } if (IS_ZEBRA_DEBUG_MLAG) zlog_debug( "Client De-Registered successfully for MLAG Updates"); } /* * Does following things. * 1) allocated new local stream, and copies the client data and enqueue * to MLAG Thread * 2) MLAG Thread after dequeing, encode the client data using protobuf * and write on to MLAG */ void zebra_mlag_forward_client_msg(ZAPI_HANDLER_ARGS) { struct stream *zebra_s; struct stream *mlag_s; if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Received Client MLAG Data from client-proto:%d", client->proto); /* Get input stream. */ zebra_s = msg; mlag_s = stream_new(zebra_s->endp); /* * Client data is | Zebra Header + MLAG Data | * we need to enqueue only the MLAG data, skipping Zebra Header */ stream_put(mlag_s, zebra_s->data + zebra_s->getp, STREAM_READABLE(zebra_s)); stream_fifo_push_safe(zrouter.mlag_info.mlag_fifo, mlag_s); zebra_mlag_signal_write_thread(); if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("%s: Enqueued Client:%d data to MLAG Thread ", __func__, client->proto); } /***********************End of Zebra Main thread processing*************/ enum mlag_role zebra_mlag_get_role(void) { return zrouter.mlag_info.role; } DEFUN_HIDDEN (show_mlag, show_mlag_cmd, "show zebra mlag", SHOW_STR ZEBRA_STR "The mlag role on this machine\n") { char buf[80]; vty_out(vty, "MLag is configured to: %s\n", mlag_role2str(zrouter.mlag_info.role, buf, sizeof(buf))); return CMD_SUCCESS; } DEFPY(test_mlag, test_mlag_cmd, "test zebra mlag ", "Test code\n" ZEBRA_STR "Modify the Mlag state\n" "Mlag is not setup on the machine\n" "Mlag is setup to be primary\n" "Mlag is setup to be the secondary\n") { enum mlag_role orig = zrouter.mlag_info.role; char buf1[80], buf2[80]; if (none) zrouter.mlag_info.role = MLAG_ROLE_NONE; if (primary) zrouter.mlag_info.role = MLAG_ROLE_PRIMARY; if (secondary) zrouter.mlag_info.role = MLAG_ROLE_SECONDARY; if (IS_ZEBRA_DEBUG_MLAG) zlog_debug("Test: Changing role from %s to %s", mlag_role2str(orig, buf1, sizeof(buf1)), mlag_role2str(orig, buf2, sizeof(buf2))); if (orig != zrouter.mlag_info.role) { zsend_capabilities_all_clients(); if (zrouter.mlag_info.role != MLAG_ROLE_NONE) { if (zrouter.mlag_info.clients_interested_cnt == 0 && test_mlag_in_progress == false) { if (zrouter.mlag_info.zebra_pth_mlag == NULL) zebra_mlag_spawn_pthread(); zrouter.mlag_info.clients_interested_cnt++; test_mlag_in_progress = true; zebra_mlag_private_open_channel(); } } else { if (test_mlag_in_progress == true) { test_mlag_in_progress = false; zrouter.mlag_info.clients_interested_cnt--; zebra_mlag_private_close_channel(); } } } return CMD_SUCCESS; } void zebra_mlag_init(void) { install_element(VIEW_NODE, &show_mlag_cmd); install_element(ENABLE_NODE, &test_mlag_cmd); /* * Intialiaze teh MLAG Global variableis * write thread will be craeted during actual registration with MCLAG */ zrouter.mlag_info.clients_interested_cnt = 0; zrouter.mlag_info.connected = false; zrouter.mlag_info.timer_running = false; zrouter.mlag_info.mlag_fifo = stream_fifo_new(); zrouter.mlag_info.zebra_pth_mlag = NULL; zrouter.mlag_info.th_master = NULL; zrouter.mlag_info.t_read = NULL; zrouter.mlag_info.t_write = NULL; test_mlag_in_progress = false; zebra_mlag_reset_write_buffer(); zebra_mlag_reset_read_buffer(); } void zebra_mlag_terminate(void) { } /* * * ProtoBuf Encoding APIs */ int zebra_mlag_protobuf_encode_client_data(struct stream *s, uint32_t *msg_type) { return 0; } int zebra_mlag_protobuf_decode_message(struct stream **s, uint8_t *data, uint32_t len) { return 0; }