lib/frr_zmq.c

   1 /*
   2  * libzebra ZeroMQ bindings
   3  * Copyright (C) 2015  David Lamparter
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the Free
   7  * Software Foundation; either version 2 of the License, or (at your option)
   8  * any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13  * 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 /*
  21  * IF YOU MODIFY THIS FILE PLEASE RUN `make check` and ensure that
  22  * the test_zmq.c unit test is still working.  There are dependancies
  23  * between the two that are extremely fragile.  My understanding
  24  * is that there is specialized ownership of the cb pointer based
  25  * upon what is happening.  Those assumptions are supposed to be
  26  * tested in the test_zmq.c
  27  */
  28 #include <zebra.h>
  29 #include <zmq.h>
  30
  31 #include "thread.h"
  32 #include "memory.h"
  33 #include "frr_zmq.h"
  34 #include "log.h"
  35 #include "lib_errors.h"
  36
  37 DEFINE_MTYPE_STATIC(LIB, ZEROMQ_CB, "ZeroMQ callback");
  38
  39 /* libzmq's context */
  40 void *frrzmq_context = NULL;
  41 static unsigned frrzmq_initcount = 0;
  42
  43 void frrzmq_init(void)
  44 {
  45         if (frrzmq_initcount++ == 0) {
  46                 frrzmq_context = zmq_ctx_new();
  47                 zmq_ctx_set(frrzmq_context, ZMQ_IPV6, 1);
  48         }
  49 }
  50
  51 void frrzmq_finish(void)
  52 {
  53         if (--frrzmq_initcount == 0) {
  54                 zmq_ctx_term(frrzmq_context);
  55                 frrzmq_context = NULL;
  56         }
  57 }
  58
  59 static int frrzmq_read_msg(struct thread *t)
  60 {
  61         struct frrzmq_cb **cbp = THREAD_ARG(t);
  62         struct frrzmq_cb *cb;
  63         zmq_msg_t msg;
  64         unsigned partno;
  65         unsigned char read = 0;
  66         int ret, more;
  67         size_t moresz;
  68
  69         if (!cbp)
  70                 return 1;
  71         cb = (*cbp);
  72         if (!cb || !cb->zmqsock)
  73                 return 1;
  74
  75         while (1) {
  76                 zmq_pollitem_t polli = {.socket = cb->zmqsock,
  77                                         .events = ZMQ_POLLIN};
  78                 ret = zmq_poll(&polli, 1, 0);
  79
  80                 if (ret < 0)
  81                         goto out_err;
  82
  83                 if (!(polli.revents & ZMQ_POLLIN))
  84                         break;
  85
  86                 if (cb->read.cb_msg) {
  87                         cb->read.cb_msg(cb->read.arg, cb->zmqsock);
  88                         read = 1;
  89
  90                         if (cb->read.cancelled) {
  91                                 frrzmq_check_events(cbp, &cb->write,
  92                                                     ZMQ_POLLOUT);
  93                                 cb->read.thread = NULL;
  94                                 if (cb->write.cancelled && !cb->write.thread)
  95                                         XFREE(MTYPE_ZEROMQ_CB, *cbp);
  96
  97                                 return 0;
  98                         }
  99                         continue;
 100                 }
 101
 102                 partno = 0;
 103                 if (zmq_msg_init(&msg))
 104                         goto out_err;
 105                 do {
 106                         ret = zmq_msg_recv(&msg, cb->zmqsock, ZMQ_NOBLOCK);
 107                         if (ret < 0) {
 108                                 if (errno == EAGAIN)
 109                                         break;
 110
 111                                 zmq_msg_close(&msg);
 112                                 goto out_err;
 113                         }
 114                         read = 1;
 115
 116                         cb->read.cb_part(cb->read.arg, cb->zmqsock, &msg,
 117                                          partno);
 118                         if (cb->read.cancelled) {
 119                                 zmq_msg_close(&msg);
 120                                 frrzmq_check_events(cbp, &cb->write,
 121                                                     ZMQ_POLLOUT);
 122                                 cb->read.thread = NULL;
 123                                 if (cb->write.cancelled && !cb->write.thread)
 124                                         XFREE(MTYPE_ZEROMQ_CB, *cbp);
 125
 126                                 return 0;
 127                         }
 128
 129                         /* cb_part may have read additional parts of the
 130                          * message; don't use zmq_msg_more here */
 131                         moresz = sizeof(more);
 132                         more = 0;
 133                         ret = zmq_getsockopt(cb->zmqsock, ZMQ_RCVMORE, &more,
 134                                              &moresz);
 135                         if (ret < 0) {
 136                                 zmq_msg_close(&msg);
 137                                 goto out_err;
 138                         }
 139
 140                         partno++;
 141                 } while (more);
 142                 zmq_msg_close(&msg);
 143         }
 144
 145         if (read)
 146                 frrzmq_check_events(cbp, &cb->write, ZMQ_POLLOUT);
 147
 148         thread_add_read(t->master, frrzmq_read_msg, cbp,
 149                         cb->fd, &cb->read.thread);
 150         return 0;
 151
 152 out_err:
 153         flog_err(EC_LIB_ZMQ, "ZeroMQ read error: %s(%d)", strerror(errno),
 154                  errno);
 155         if (cb->read.cb_error)
 156                 cb->read.cb_error(cb->read.arg, cb->zmqsock);
 157         return 1;
 158 }
 159
 160 int _frrzmq_thread_add_read(const struct xref_threadsched *xref,
 161                             struct thread_master *master,
 162                             void (*msgfunc)(void *arg, void *zmqsock),
 163                             void (*partfunc)(void *arg, void *zmqsock,
 164                                              zmq_msg_t *msg, unsigned partnum),
 165                             void (*errfunc)(void *arg, void *zmqsock),
 166                             void *arg, void *zmqsock,
 167                             struct frrzmq_cb **cbp)
 168 {
 169         int fd, events;
 170         size_t len;
 171         struct frrzmq_cb *cb;
 172
 173         if (!cbp)
 174                 return -1;
 175         if (!(msgfunc || partfunc) || (msgfunc && partfunc))
 176                 return -1;
 177         len = sizeof(fd);
 178         if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
 179                 return -1;
 180         len = sizeof(events);
 181         if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
 182                 return -1;
 183
 184         if (*cbp)
 185                 cb = *cbp;
 186         else {
 187                 cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
 188
 189                 cb->write.cancelled = true;
 190                 *cbp = cb;
 191         }
 192
 193         cb->zmqsock = zmqsock;
 194         cb->fd = fd;
 195         cb->read.arg = arg;
 196         cb->read.cb_msg = msgfunc;
 197         cb->read.cb_part = partfunc;
 198         cb->read.cb_error = errfunc;
 199         cb->read.cancelled = false;
 200
 201         if (events & ZMQ_POLLIN) {
 202                 thread_cancel(&cb->read.thread);
 203
 204                 thread_add_event(master, frrzmq_read_msg, cbp, fd,
 205                                   &cb->read.thread);
 206         } else
 207                 thread_add_read(master, frrzmq_read_msg, cbp, fd,
 208                                 &cb->read.thread);
 209         return 0;
 210 }
 211
 212 static int frrzmq_write_msg(struct thread *t)
 213 {
 214         struct frrzmq_cb **cbp = THREAD_ARG(t);
 215         struct frrzmq_cb *cb;
 216         unsigned char written = 0;
 217         int ret;
 218
 219         if (!cbp)
 220                 return 1;
 221         cb = (*cbp);
 222         if (!cb || !cb->zmqsock)
 223                 return 1;
 224
 225         while (1) {
 226                 zmq_pollitem_t polli = {.socket = cb->zmqsock,
 227                                         .events = ZMQ_POLLOUT};
 228                 ret = zmq_poll(&polli, 1, 0);
 229
 230                 if (ret < 0)
 231                         goto out_err;
 232
 233                 if (!(polli.revents & ZMQ_POLLOUT))
 234                         break;
 235
 236                 if (cb->write.cb_msg) {
 237                         cb->write.cb_msg(cb->write.arg, cb->zmqsock);
 238                         written = 1;
 239
 240                         if (cb->write.cancelled) {
 241                                 frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
 242                                 cb->write.thread = NULL;
 243                                 if (cb->read.cancelled && !cb->read.thread)
 244                                         XFREE(MTYPE_ZEROMQ_CB, *cbp);
 245
 246                                 return 0;
 247                         }
 248                         continue;
 249                 }
 250         }
 251
 252         if (written)
 253                 frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
 254
 255         thread_add_write(t->master, frrzmq_write_msg, cbp,
 256                          cb->fd, &cb->write.thread);
 257         return 0;
 258
 259 out_err:
 260         flog_err(EC_LIB_ZMQ, "ZeroMQ write error: %s(%d)", strerror(errno),
 261                  errno);
 262         if (cb->write.cb_error)
 263                 cb->write.cb_error(cb->write.arg, cb->zmqsock);
 264         return 1;
 265 }
 266
 267 int _frrzmq_thread_add_write(const struct xref_threadsched *xref,
 268                              struct thread_master *master,
 269                              void (*msgfunc)(void *arg, void *zmqsock),
 270                              void (*errfunc)(void *arg, void *zmqsock),
 271                              void *arg, void *zmqsock, struct frrzmq_cb **cbp)
 272 {
 273         int fd, events;
 274         size_t len;
 275         struct frrzmq_cb *cb;
 276
 277         if (!cbp)
 278                 return -1;
 279         if (!msgfunc)
 280                 return -1;
 281         len = sizeof(fd);
 282         if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
 283                 return -1;
 284         len = sizeof(events);
 285         if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
 286                 return -1;
 287
 288         if (*cbp)
 289                 cb = *cbp;
 290         else {
 291                 cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
 292
 293                 cb->read.cancelled = true;
 294                 *cbp = cb;
 295         }
 296
 297         cb->zmqsock = zmqsock;
 298         cb->fd = fd;
 299         cb->write.arg = arg;
 300         cb->write.cb_msg = msgfunc;
 301         cb->write.cb_part = NULL;
 302         cb->write.cb_error = errfunc;
 303         cb->write.cancelled = false;
 304
 305         if (events & ZMQ_POLLOUT) {
 306                 thread_cancel(&cb->write.thread);
 307
 308                 _thread_add_event(xref, master, frrzmq_write_msg, cbp, fd,
 309                                   &cb->write.thread);
 310         } else
 311                 thread_add_write(master, frrzmq_write_msg, cbp, fd,
 312                                  &cb->write.thread);
 313         return 0;
 314 }
 315
 316 void frrzmq_thread_cancel(struct frrzmq_cb **cb, struct cb_core *core)
 317 {
 318         if (!cb || !*cb)
 319                 return;
 320         core->cancelled = true;
 321         thread_cancel(&core->thread);
 322
 323         /*
 324          * Looking at this code one would assume that FRR
 325          * would want a `!(*cb)->write.thread.  This was
 326          * attempted in e08165def1c62beee0e87385 but this
 327          * change caused `make check` to stop working
 328          * which was not noticed because our CI system
 329          * does not build with zeromq.  Put this back
 330          * to the code as written in 2017.  e08165de..
 331          * was introduced in 2021.  So someone was ok
 332          * with frrzmq_thread_cancel for 4 years.  This will
 333          * allow those people doing `make check` to continue
 334          * working.  In the meantime if the people using
 335          * this code see an issue they can fix it
 336          */
 337         if ((*cb)->read.cancelled && !(*cb)->read.thread
 338             && (*cb)->write.cancelled && (*cb)->write.thread)
 339                 XFREE(MTYPE_ZEROMQ_CB, *cb);
 340 }
 341
 342 void frrzmq_check_events(struct frrzmq_cb **cbp, struct cb_core *core,
 343                          int event)
 344 {
 345         struct frrzmq_cb *cb;
 346         int events;
 347         size_t len;
 348
 349         if (!cbp)
 350                 return;
 351         cb = (*cbp);
 352         if (!cb || !cb->zmqsock)
 353                 return;
 354
 355         len = sizeof(events);
 356         if (zmq_getsockopt(cb->zmqsock, ZMQ_EVENTS, &events, &len))
 357                 return;
 358         if ((events & event) && core->thread && !core->cancelled) {
 359                 struct thread_master *tm = core->thread->master;
 360
 361                 thread_cancel(&core->thread);
 362
 363                 if (event == ZMQ_POLLIN)
 364                         thread_add_event(tm, frrzmq_read_msg,
 365                                          cbp, cb->fd, &core->thread);
 366                 else
 367                         thread_add_event(tm, frrzmq_write_msg,
 368                                          cbp, cb->fd, &core->thread);
 369         }
 370 }