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_mlag.h"
  26 #include "pim_zebra.h"
  27
  28 extern struct zclient *zclient;
  29
  30 #define PIM_MLAG_POST_LIMIT 100
  31
  32 int32_t mlag_bulk_cnt;
  33
  34 static void pim_mlag_zebra_fill_header(enum mlag_msg_type msg_type)
  35 {
  36         uint32_t fill_msg_type = msg_type;
  37         uint16_t data_len;
  38         uint16_t msg_cnt = 1;
  39
  40         if (msg_type == MLAG_MSG_NONE)
  41                 return;
  42
  43         switch (msg_type) {
  44         case MLAG_REGISTER:
  45         case MLAG_DEREGISTER:
  46                 data_len = sizeof(struct mlag_msg);
  47                 break;
  48         case MLAG_MROUTE_ADD:
  49                 data_len = sizeof(struct mlag_mroute_add);
  50                 fill_msg_type = MLAG_MROUTE_ADD_BULK;
  51                 break;
  52         case MLAG_MROUTE_DEL:
  53                 data_len = sizeof(struct mlag_mroute_del);
  54                 fill_msg_type = MLAG_MROUTE_DEL_BULK;
  55                 break;
  56         default:
  57                 data_len = 0;
  58                 break;
  59         }
  60
  61         stream_reset(router->mlag_stream);
  62         /* ADD Hedaer */
  63         stream_putl(router->mlag_stream, fill_msg_type);
  64         /*
  65          * In case of Bulk actual size & msg_cnt will be updated
  66          * just before writing onto zebra
  67          */
  68         stream_putw(router->mlag_stream, data_len);
  69         stream_putw(router->mlag_stream, msg_cnt);
  70
  71         if (PIM_DEBUG_MLAG)
  72                 zlog_debug(":%s: msg_type: %d/%d len %d",
  73                            __func__, msg_type, fill_msg_type, data_len);
  74 }
  75
  76 static void pim_mlag_zebra_flush_buffer(void)
  77 {
  78         uint32_t msg_type;
  79
  80         /* Stream had bulk messages update the Hedaer */
  81         if (mlag_bulk_cnt > 1) {
  82                 /*
  83                  * No need to reset the pointer, below api reads from data[0]
  84                  */
  85                 STREAM_GETL(router->mlag_stream, msg_type);
  86                 if (msg_type == MLAG_MROUTE_ADD_BULK) {
  87                         stream_putw_at(
  88                                 router->mlag_stream, 4,
  89                                 (mlag_bulk_cnt * sizeof(struct mlag_mroute_add)));
  90                         stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
  91                 } else if (msg_type == MLAG_MROUTE_DEL_BULK) {
  92                         stream_putw_at(
  93                                 router->mlag_stream, 4,
  94                                 (mlag_bulk_cnt * sizeof(struct mlag_mroute_del)));
  95                         stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
  96                 } else {
  97                         flog_err(EC_LIB_ZAPI_ENCODE,
  98                                 "unknown bulk message type %d bulk_count %d",
  99                                 msg_type, mlag_bulk_cnt);
 100                         stream_reset(router->mlag_stream);
 101                         mlag_bulk_cnt = 0;
 102                         return;
 103                 }
 104         }
 105
 106         zclient_send_mlag_data(zclient, router->mlag_stream);
 107 stream_failure:
 108         stream_reset(router->mlag_stream);
 109         mlag_bulk_cnt = 0;
 110 }
 111
 112 /*
 113  * Only ROUTE add & Delete will be bulked.
 114  * Buffer will be flushed, when
 115  * 1) there were no messages in the queue
 116  * 2) Curr_msg_type != prev_msg_type
 117  */
 118
 119 static void pim_mlag_zebra_check_for_buffer_flush(uint32_t curr_msg_type,
 120                                                   uint32_t prev_msg_type)
 121 {
 122         /* First Message, keep bulking */
 123         if (prev_msg_type == MLAG_MSG_NONE) {
 124                 mlag_bulk_cnt = 1;
 125                 return;
 126         }
 127
 128         /*msg type is route add & delete, keep bulking */
 129         if (curr_msg_type == prev_msg_type
 130             && (curr_msg_type == MLAG_MROUTE_ADD
 131                 || curr_msg_type == MLAG_MROUTE_DEL)) {
 132                 mlag_bulk_cnt++;
 133                 return;
 134         }
 135
 136         pim_mlag_zebra_flush_buffer();
 137 }
 138
 139 /*
 140  * Thsi thread reads the clients data from the Gloabl queue and encodes with
 141  * protobuf and pass on to the MLAG socket.
 142  */
 143 static int pim_mlag_zthread_handler(struct thread *event)
 144 {
 145         struct stream *read_s;
 146         uint32_t wr_count = 0;
 147         uint32_t prev_msg_type = MLAG_MSG_NONE;
 148         uint32_t curr_msg_type = MLAG_MSG_NONE;
 149
 150         router->zpthread_mlag_write = NULL;
 151         wr_count = stream_fifo_count_safe(router->mlag_fifo);
 152
 153         if (PIM_DEBUG_MLAG)
 154                 zlog_debug(":%s: Processing MLAG write, %d messages in queue",
 155                            __func__, wr_count);
 156
 157         if (wr_count == 0)
 158                 return 0;
 159
 160         for (wr_count = 0; wr_count < PIM_MLAG_POST_LIMIT; wr_count++) {
 161                 /* FIFO is empty,wait for teh message to be add */
 162                 if (stream_fifo_count_safe(router->mlag_fifo) == 0)
 163                         break;
 164
 165                 read_s = stream_fifo_pop_safe(router->mlag_fifo);
 166                 if (!read_s) {
 167                         zlog_debug(":%s: Got a NULL Messages, some thing wrong",
 168                                    __func__);
 169                         break;
 170                 }
 171                 STREAM_GETL(read_s, curr_msg_type);
 172                 /*
 173                  * Check for Buffer Overflow,
 174                  * MLAG Can't process more than 'PIM_MLAG_BUF_LIMIT' bytes
 175                  */
 176                 if (router->mlag_stream->endp + read_s->endp + ZEBRA_HEADER_SIZE
 177                     > MLAG_BUF_LIMIT)
 178                         pim_mlag_zebra_flush_buffer();
 179
 180                 pim_mlag_zebra_check_for_buffer_flush(curr_msg_type,
 181                                                       prev_msg_type);
 182
 183                 /*
 184                  * First message to Buffer, fill the Header
 185                  */
 186                 if (router->mlag_stream->endp == 0)
 187                         pim_mlag_zebra_fill_header(curr_msg_type);
 188
 189                 /*
 190                  * add the data now
 191                  */
 192                 stream_put(router->mlag_stream, read_s->data + read_s->getp,
 193                            read_s->endp - read_s->getp);
 194
 195                 stream_free(read_s);
 196                 prev_msg_type = curr_msg_type;
 197         }
 198
 199 stream_failure:
 200         /*
 201          * we are here , because
 202          * 1. Queue might be empty
 203          * 2. we crossed the max Q Read limit
 204          * In any acse flush the buffer towards zebra
 205          */
 206         pim_mlag_zebra_flush_buffer();
 207
 208         if (wr_count >= PIM_MLAG_POST_LIMIT)
 209                 pim_mlag_signal_zpthread();
 210
 211         return 0;
 212 }
 213
 214
 215 int pim_mlag_signal_zpthread(void)
 216 {
 217         if (router->master) {
 218                 if (PIM_DEBUG_MLAG)
 219                         zlog_debug(":%s: Scheduling PIM MLAG write Thread",
 220                                    __func__);
 221                 thread_add_event(router->master, pim_mlag_zthread_handler, NULL,
 222                                  0, &router->zpthread_mlag_write);
 223         }
 224         return (0);
 225 }