pimd/pim_zpthread.c

   1 /*
   2  * PIM for Quagga
   3  * Copyright (C) 2008  Everton da Silva Marques
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; see the file COPYING; if not, write to the Free Software
  17  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  18  */
  19
  20 #include <zebra.h>
  21 #include <lib/log.h>
  22 #include <lib/lib_errors.h>
  23
  24 #include "pimd.h"
  25 #include "pim_instance.h"
  26 #include "pim_mlag.h"
  27 #include "pim_zebra.h"
  28
  29 extern struct zclient *zclient;
  30
  31 #define PIM_MLAG_POST_LIMIT 100
  32
  33 int32_t mlag_bulk_cnt;
  34
  35 static void pim_mlag_zebra_fill_header(enum mlag_msg_type msg_type)
  36 {
  37         uint32_t fill_msg_type = msg_type;
  38         uint16_t data_len = 0;
  39         uint16_t msg_cnt = 1;
  40
  41         if (msg_type == MLAG_MSG_NONE)
  42                 return;
  43
  44         switch (msg_type) {
  45         case MLAG_REGISTER:
  46         case MLAG_DEREGISTER:
  47                 data_len = sizeof(struct mlag_msg);
  48                 break;
  49         case MLAG_MROUTE_ADD:
  50                 data_len = sizeof(struct mlag_mroute_add);
  51                 fill_msg_type = MLAG_MROUTE_ADD_BULK;
  52                 break;
  53         case MLAG_MROUTE_DEL:
  54                 data_len = sizeof(struct mlag_mroute_del);
  55                 fill_msg_type = MLAG_MROUTE_DEL_BULK;
  56                 break;
  57         case MLAG_MSG_NONE:
  58         case MLAG_STATUS_UPDATE:
  59         case MLAG_DUMP:
  60         case MLAG_MROUTE_ADD_BULK:
  61         case MLAG_MROUTE_DEL_BULK:
  62         case MLAG_PIM_CFG_DUMP:
  63         case MLAG_VXLAN_UPDATE:
  64         case MLAG_PEER_FRR_STATUS:
  65                 data_len = 0;
  66                 break;
  67         }
  68
  69         stream_reset(router->mlag_stream);
  70         /* ADD Hedaer */
  71         stream_putl(router->mlag_stream, fill_msg_type);
  72         /*
  73          * In case of Bulk actual size & msg_cnt will be updated
  74          * just before writing onto zebra
  75          */
  76         stream_putw(router->mlag_stream, data_len);
  77         stream_putw(router->mlag_stream, msg_cnt);
  78
  79         if (PIM_DEBUG_MLAG)
  80                 zlog_debug(":%s: msg_type: %d/%d len %d",
  81                            __func__, msg_type, fill_msg_type, data_len);
  82 }
  83
  84 static void pim_mlag_zebra_flush_buffer(void)
  85 {
  86         uint32_t msg_type;
  87
  88         /* Stream had bulk messages update the Hedaer */
  89         if (mlag_bulk_cnt > 1) {
  90                 /*
  91                  * No need to reset the pointer, below api reads from data[0]
  92                  */
  93                 STREAM_GETL(router->mlag_stream, msg_type);
  94                 if (msg_type == MLAG_MROUTE_ADD_BULK) {
  95                         stream_putw_at(
  96                                 router->mlag_stream, 4,
  97                                 (mlag_bulk_cnt * sizeof(struct mlag_mroute_add)));
  98                         stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
  99                 } else if (msg_type == MLAG_MROUTE_DEL_BULK) {
 100                         stream_putw_at(
 101                                 router->mlag_stream, 4,
 102                                 (mlag_bulk_cnt * sizeof(struct mlag_mroute_del)));
 103                         stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
 104                 } else {
 105                         flog_err(EC_LIB_ZAPI_ENCODE,
 106                                 "unknown bulk message type %d bulk_count %d",
 107                                 msg_type, mlag_bulk_cnt);
 108                         stream_reset(router->mlag_stream);
 109                         mlag_bulk_cnt = 0;
 110                         return;
 111                 }
 112         }
 113
 114         zclient_send_mlag_data(zclient, router->mlag_stream);
 115 stream_failure:
 116         stream_reset(router->mlag_stream);
 117         mlag_bulk_cnt = 0;
 118 }
 119
 120 /*
 121  * Only ROUTE add & Delete will be bulked.
 122  * Buffer will be flushed, when
 123  * 1) there were no messages in the queue
 124  * 2) Curr_msg_type != prev_msg_type
 125  */
 126
 127 static void pim_mlag_zebra_check_for_buffer_flush(uint32_t curr_msg_type,
 128                                                   uint32_t prev_msg_type)
 129 {
 130         /* First Message, keep bulking */
 131         if (prev_msg_type == MLAG_MSG_NONE) {
 132                 mlag_bulk_cnt = 1;
 133                 return;
 134         }
 135
 136         /*msg type is route add & delete, keep bulking */
 137         if (curr_msg_type == prev_msg_type
 138             && (curr_msg_type == MLAG_MROUTE_ADD
 139                 || curr_msg_type == MLAG_MROUTE_DEL)) {
 140                 mlag_bulk_cnt++;
 141                 return;
 142         }
 143
 144         pim_mlag_zebra_flush_buffer();
 145 }
 146
 147 /*
 148  * Thsi thread reads the clients data from the Gloabl queue and encodes with
 149  * protobuf and pass on to the MLAG socket.
 150  */
 151 static void pim_mlag_zthread_handler(struct thread *event)
 152 {
 153         struct stream *read_s;
 154         uint32_t wr_count = 0;
 155         uint32_t prev_msg_type = MLAG_MSG_NONE;
 156         uint32_t curr_msg_type = MLAG_MSG_NONE;
 157
 158         router->zpthread_mlag_write = NULL;
 159         wr_count = stream_fifo_count_safe(router->mlag_fifo);
 160
 161         if (PIM_DEBUG_MLAG)
 162                 zlog_debug(":%s: Processing MLAG write, %d messages in queue",
 163                            __func__, wr_count);
 164
 165         if (wr_count == 0)
 166                 return;
 167
 168         for (wr_count = 0; wr_count < PIM_MLAG_POST_LIMIT; wr_count++) {
 169                 /* FIFO is empty,wait for teh message to be add */
 170                 if (stream_fifo_count_safe(router->mlag_fifo) == 0)
 171                         break;
 172
 173                 read_s = stream_fifo_pop_safe(router->mlag_fifo);
 174                 if (!read_s) {
 175                         zlog_debug(":%s: Got a NULL Messages, some thing wrong",
 176                                    __func__);
 177                         break;
 178                 }
 179                 STREAM_GETL(read_s, curr_msg_type);
 180                 /*
 181                  * Check for Buffer Overflow,
 182                  * MLAG Can't process more than 'PIM_MLAG_BUF_LIMIT' bytes
 183                  */
 184                 if (router->mlag_stream->endp + read_s->endp + ZEBRA_HEADER_SIZE
 185                     > MLAG_BUF_LIMIT)
 186                         pim_mlag_zebra_flush_buffer();
 187
 188                 pim_mlag_zebra_check_for_buffer_flush(curr_msg_type,
 189                                                       prev_msg_type);
 190
 191                 /*
 192                  * First message to Buffer, fill the Header
 193                  */
 194                 if (router->mlag_stream->endp == 0)
 195                         pim_mlag_zebra_fill_header(curr_msg_type);
 196
 197                 /*
 198                  * add the data now
 199                  */
 200                 stream_put(router->mlag_stream, read_s->data + read_s->getp,
 201                            read_s->endp - read_s->getp);
 202
 203                 stream_free(read_s);
 204                 prev_msg_type = curr_msg_type;
 205         }
 206
 207 stream_failure:
 208         /*
 209          * we are here , because
 210          * 1. Queue might be empty
 211          * 2. we crossed the max Q Read limit
 212          * In any acse flush the buffer towards zebra
 213          */
 214         pim_mlag_zebra_flush_buffer();
 215
 216         if (wr_count >= PIM_MLAG_POST_LIMIT)
 217                 pim_mlag_signal_zpthread();
 218 }
 219
 220
 221 int pim_mlag_signal_zpthread(void)
 222 {
 223         if (router->master) {
 224                 if (PIM_DEBUG_MLAG)
 225                         zlog_debug(":%s: Scheduling PIM MLAG write Thread",
 226                                    __func__);
 227                 thread_add_event(router->master, pim_mlag_zthread_handler, NULL,
 228                                  0, &router->zpthread_mlag_write);
 229         }
 230         return (0);
 231 }