Line data Source code
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 0 : static void pim_mlag_zebra_fill_header(enum mlag_msg_type msg_type)
36 : {
37 0 : uint32_t fill_msg_type = msg_type;
38 0 : uint16_t data_len = 0;
39 0 : uint16_t msg_cnt = 1;
40 :
41 0 : if (msg_type == MLAG_MSG_NONE)
42 : return;
43 :
44 0 : switch (msg_type) {
45 0 : case MLAG_REGISTER:
46 : case MLAG_DEREGISTER:
47 0 : data_len = sizeof(struct mlag_msg);
48 0 : break;
49 0 : case MLAG_MROUTE_ADD:
50 0 : data_len = sizeof(struct mlag_mroute_add);
51 0 : fill_msg_type = MLAG_MROUTE_ADD_BULK;
52 0 : break;
53 0 : case MLAG_MROUTE_DEL:
54 0 : data_len = sizeof(struct mlag_mroute_del);
55 0 : fill_msg_type = MLAG_MROUTE_DEL_BULK;
56 0 : 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 0 : stream_reset(router->mlag_stream);
70 : /* ADD Hedaer */
71 0 : 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 0 : stream_putw(router->mlag_stream, data_len);
77 0 : stream_putw(router->mlag_stream, msg_cnt);
78 :
79 0 : if (PIM_DEBUG_MLAG)
80 0 : zlog_debug(":%s: msg_type: %d/%d len %d",
81 : __func__, msg_type, fill_msg_type, data_len);
82 : }
83 :
84 0 : static void pim_mlag_zebra_flush_buffer(void)
85 : {
86 0 : uint32_t msg_type;
87 :
88 : /* Stream had bulk messages update the Hedaer */
89 0 : if (mlag_bulk_cnt > 1) {
90 : /*
91 : * No need to reset the pointer, below api reads from data[0]
92 : */
93 0 : STREAM_GETL(router->mlag_stream, msg_type);
94 0 : if (msg_type == MLAG_MROUTE_ADD_BULK) {
95 0 : stream_putw_at(
96 0 : router->mlag_stream, 4,
97 : (mlag_bulk_cnt * sizeof(struct mlag_mroute_add)));
98 0 : stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
99 0 : } else if (msg_type == MLAG_MROUTE_DEL_BULK) {
100 0 : stream_putw_at(
101 0 : router->mlag_stream, 4,
102 : (mlag_bulk_cnt * sizeof(struct mlag_mroute_del)));
103 0 : stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
104 : } else {
105 0 : flog_err(EC_LIB_ZAPI_ENCODE,
106 : "unknown bulk message type %d bulk_count %d",
107 : msg_type, mlag_bulk_cnt);
108 0 : stream_reset(router->mlag_stream);
109 0 : mlag_bulk_cnt = 0;
110 0 : return;
111 : }
112 : }
113 :
114 0 : zclient_send_mlag_data(zclient, router->mlag_stream);
115 0 : stream_failure:
116 0 : stream_reset(router->mlag_stream);
117 0 : 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 0 : 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 0 : if (prev_msg_type == MLAG_MSG_NONE) {
132 0 : mlag_bulk_cnt = 1;
133 0 : return;
134 : }
135 :
136 : /*msg type is route add & delete, keep bulking */
137 0 : if (curr_msg_type == prev_msg_type
138 0 : && (curr_msg_type == MLAG_MROUTE_ADD
139 0 : || curr_msg_type == MLAG_MROUTE_DEL)) {
140 0 : mlag_bulk_cnt++;
141 0 : return;
142 : }
143 :
144 0 : 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 0 : static void pim_mlag_zthread_handler(struct thread *event)
152 : {
153 0 : struct stream *read_s;
154 0 : uint32_t wr_count = 0;
155 0 : uint32_t prev_msg_type = MLAG_MSG_NONE;
156 0 : uint32_t curr_msg_type = MLAG_MSG_NONE;
157 :
158 0 : router->zpthread_mlag_write = NULL;
159 0 : wr_count = stream_fifo_count_safe(router->mlag_fifo);
160 :
161 0 : if (PIM_DEBUG_MLAG)
162 0 : zlog_debug(":%s: Processing MLAG write, %d messages in queue",
163 : __func__, wr_count);
164 :
165 0 : if (wr_count == 0)
166 : return;
167 :
168 0 : for (wr_count = 0; wr_count < PIM_MLAG_POST_LIMIT; wr_count++) {
169 : /* FIFO is empty,wait for teh message to be add */
170 0 : if (stream_fifo_count_safe(router->mlag_fifo) == 0)
171 : break;
172 :
173 0 : read_s = stream_fifo_pop_safe(router->mlag_fifo);
174 0 : if (!read_s) {
175 0 : zlog_debug(":%s: Got a NULL Messages, some thing wrong",
176 : __func__);
177 0 : break;
178 : }
179 0 : 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 0 : if (router->mlag_stream->endp + read_s->endp + ZEBRA_HEADER_SIZE
185 : > MLAG_BUF_LIMIT)
186 0 : pim_mlag_zebra_flush_buffer();
187 :
188 0 : 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 0 : if (router->mlag_stream->endp == 0)
195 0 : pim_mlag_zebra_fill_header(curr_msg_type);
196 :
197 : /*
198 : * add the data now
199 : */
200 0 : stream_put(router->mlag_stream, read_s->data + read_s->getp,
201 0 : read_s->endp - read_s->getp);
202 :
203 0 : stream_free(read_s);
204 0 : prev_msg_type = curr_msg_type;
205 : }
206 :
207 0 : 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 0 : pim_mlag_zebra_flush_buffer();
215 :
216 0 : if (wr_count >= PIM_MLAG_POST_LIMIT)
217 0 : pim_mlag_signal_zpthread();
218 : }
219 :
220 :
221 0 : int pim_mlag_signal_zpthread(void)
222 : {
223 0 : if (router->master) {
224 0 : if (PIM_DEBUG_MLAG)
225 0 : zlog_debug(":%s: Scheduling PIM MLAG write Thread",
226 : __func__);
227 0 : thread_add_event(router->master, pim_mlag_zthread_handler, NULL,
228 : 0, &router->zpthread_mlag_write);
229 : }
230 0 : return (0);
231 : }
|
