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