bgpd/bgp_io.c

   1 /* BGP I/O.
   2  * Implements packet I/O in a pthread.
   3  * Copyright (C) 2017  Cumulus Networks
   4  * Quentin Young
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 2 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; see the file COPYING; if not, write to the
  18  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston,
  19  * MA 02110-1301 USA
  20  */
  21
  22 /* clang-format off */
  23 #include <zebra.h>
  24 #include <pthread.h>            // for pthread_mutex_unlock, pthread_mutex_lock
  25 #include <sys/uio.h>            // for writev
  26
  27 #include "frr_pthread.h"
  28 #include "linklist.h"           // for list_delete, list_delete_all_node, lis...
  29 #include "log.h"                // for zlog_debug, safe_strerror, zlog_err
  30 #include "memory.h"             // for MTYPE_TMP, XCALLOC, XFREE
  31 #include "network.h"            // for ERRNO_IO_RETRY
  32 #include "stream.h"             // for stream_get_endp, stream_getw_from, str...
  33 #include "ringbuf.h"            // for ringbuf_remain, ringbuf_peek, ringbuf_...
  34 #include "thread.h"             // for THREAD_OFF, THREAD_ARG, thread...
  35 #include "zassert.h"            // for assert
  36
  37 #include "bgpd/bgp_io.h"
  38 #include "bgpd/bgp_debug.h"     // for bgp_debug_neighbor_events, bgp_type_str
  39 #include "bgpd/bgp_errors.h"    // for expanded error reference information
  40 #include "bgpd/bgp_fsm.h"       // for BGP_EVENT_ADD, bgp_event
  41 #include "bgpd/bgp_packet.h"    // for bgp_notify_send_with_data, bgp_notify...
  42 #include "bgpd/bgp_trace.h"     // for frrtraces
  43 #include "bgpd/bgpd.h"          // for peer, BGP_MARKER_SIZE, bgp_master, bm
  44 /* clang-format on */
  45
  46 /* forward declarations */
  47 static uint16_t bgp_write(struct peer *);
  48 static uint16_t bgp_read(struct peer *peer, int *code_p);
  49 static int bgp_process_writes(struct thread *);
  50 static int bgp_process_reads(struct thread *);
  51 static bool validate_header(struct peer *);
  52
  53 /* generic i/o status codes */
  54 #define BGP_IO_TRANS_ERR (1 << 0) // EAGAIN or similar occurred
  55 #define BGP_IO_FATAL_ERR (1 << 1) // some kind of fatal TCP error
  56
  57 /* Thread external API ----------------------------------------------------- */
  58
  59 void bgp_writes_on(struct peer *peer)
  60 {
  61         struct frr_pthread *fpt = bgp_pth_io;
  62         assert(fpt->running);
  63
  64         assert(peer->status != Deleted);
  65         assert(peer->obuf);
  66         assert(peer->ibuf);
  67         assert(peer->ibuf_work);
  68         assert(!peer->t_connect_check_r);
  69         assert(!peer->t_connect_check_w);
  70         assert(peer->fd);
  71
  72         thread_add_write(fpt->master, bgp_process_writes, peer, peer->fd,
  73                          &peer->t_write);
  74         SET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
  75 }
  76
  77 void bgp_writes_off(struct peer *peer)
  78 {
  79         struct frr_pthread *fpt = bgp_pth_io;
  80         assert(fpt->running);
  81
  82         thread_cancel_async(fpt->master, &peer->t_write, NULL);
  83         THREAD_OFF(peer->t_generate_updgrp_packets);
  84
  85         UNSET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
  86 }
  87
  88 void bgp_reads_on(struct peer *peer)
  89 {
  90         struct frr_pthread *fpt = bgp_pth_io;
  91         assert(fpt->running);
  92
  93         assert(peer->status != Deleted);
  94         assert(peer->ibuf);
  95         assert(peer->fd);
  96         assert(peer->ibuf_work);
  97         assert(peer->obuf);
  98         assert(!peer->t_connect_check_r);
  99         assert(!peer->t_connect_check_w);
 100         assert(peer->fd);
 101
 102         thread_add_read(fpt->master, bgp_process_reads, peer, peer->fd,
 103                         &peer->t_read);
 104
 105         SET_FLAG(peer->thread_flags, PEER_THREAD_READS_ON);
 106 }
 107
 108 void bgp_reads_off(struct peer *peer)
 109 {
 110         struct frr_pthread *fpt = bgp_pth_io;
 111         assert(fpt->running);
 112
 113         thread_cancel_async(fpt->master, &peer->t_read, NULL);
 114         THREAD_OFF(peer->t_process_packet);
 115
 116         UNSET_FLAG(peer->thread_flags, PEER_THREAD_READS_ON);
 117 }
 118
 119 /* Thread internal functions ----------------------------------------------- */
 120
 121 /*
 122  * Called from I/O pthread when a file descriptor has become ready for writing.
 123  */
 124 static int bgp_process_writes(struct thread *thread)
 125 {
 126         static struct peer *peer;
 127         peer = THREAD_ARG(thread);
 128         uint16_t status;
 129         bool reschedule;
 130         bool fatal = false;
 131
 132         if (peer->fd < 0)
 133                 return -1;
 134
 135         struct frr_pthread *fpt = bgp_pth_io;
 136
 137         frr_with_mutex(&peer->io_mtx) {
 138                 status = bgp_write(peer);
 139                 reschedule = (stream_fifo_head(peer->obuf) != NULL);
 140         }
 141
 142         /* no problem */
 143         if (CHECK_FLAG(status, BGP_IO_TRANS_ERR)) {
 144         }
 145
 146         /* problem */
 147         if (CHECK_FLAG(status, BGP_IO_FATAL_ERR)) {
 148                 reschedule = false;
 149                 fatal = true;
 150         }
 151
 152         /* If suppress fib pending is enabled, route is advertised to peers when
 153          * the status is received from the FIB. The delay is added
 154          * to update group packet generate which will allow more routes to be
 155          * sent in the update message
 156          */
 157         if (reschedule) {
 158                 thread_add_write(fpt->master, bgp_process_writes, peer,
 159                                  peer->fd, &peer->t_write);
 160         } else if (!fatal) {
 161                 BGP_UPDATE_GROUP_TIMER_ON(&peer->t_generate_updgrp_packets,
 162                                           bgp_generate_updgrp_packets);
 163         }
 164
 165         return 0;
 166 }
 167
 168 /*
 169  * Called from I/O pthread when a file descriptor has become ready for reading,
 170  * or has hung up.
 171  *
 172  * We read as much data as possible, process as many packets as we can and
 173  * place them on peer->ibuf for secondary processing by the main thread.
 174  */
 175 static int bgp_process_reads(struct thread *thread)
 176 {
 177         /* clang-format off */
 178         static struct peer *peer;       // peer to read from
 179         uint16_t status;                // bgp_read status code
 180         bool more = true;               // whether we got more data
 181         bool fatal = false;             // whether fatal error occurred
 182         bool added_pkt = false;         // whether we pushed onto ->ibuf
 183         int code = 0;                   // FSM code if error occurred
 184         /* clang-format on */
 185
 186         peer = THREAD_ARG(thread);
 187
 188         if (peer->fd < 0 || bm->terminating)
 189                 return -1;
 190
 191         struct frr_pthread *fpt = bgp_pth_io;
 192
 193         frr_with_mutex(&peer->io_mtx) {
 194                 status = bgp_read(peer, &code);
 195         }
 196
 197         /* error checking phase */
 198         if (CHECK_FLAG(status, BGP_IO_TRANS_ERR)) {
 199                 /* no problem; just don't process packets */
 200                 more = false;
 201         }
 202
 203         if (CHECK_FLAG(status, BGP_IO_FATAL_ERR)) {
 204                 /* problem; tear down session */
 205                 more = false;
 206                 fatal = true;
 207
 208                 /* Handle the error in the main pthread, include the
 209                  * specific state change from 'bgp_read'.
 210                  */
 211                 thread_add_event(bm->master, bgp_packet_process_error,
 212                                  peer, code, NULL);
 213         }
 214
 215         while (more) {
 216                 /* static buffer for transferring packets */
 217                 /* shorter alias to peer's input buffer */
 218                 struct ringbuf *ibw = peer->ibuf_work;
 219                 /* packet size as given by header */
 220                 uint16_t pktsize = 0;
 221
 222                 /* check that we have enough data for a header */
 223                 if (ringbuf_remain(ibw) < BGP_HEADER_SIZE)
 224                         break;
 225
 226                 /* check that header is valid */
 227                 if (!validate_header(peer)) {
 228                         fatal = true;
 229                         break;
 230                 }
 231
 232                 /* header is valid; retrieve packet size */
 233                 ringbuf_peek(ibw, BGP_MARKER_SIZE, &pktsize, sizeof(pktsize));
 234
 235                 pktsize = ntohs(pktsize);
 236
 237                 /* if this fails we are seriously screwed */
 238                 assert(pktsize <= peer->max_packet_size);
 239
 240                 /*
 241                  * If we have that much data, chuck it into its own
 242                  * stream and append to input queue for processing.
 243                  */
 244                 if (ringbuf_remain(ibw) >= pktsize) {
 245                         struct stream *pkt = stream_new(pktsize);
 246
 247                         assert(STREAM_WRITEABLE(pkt) == pktsize);
 248                         assert(ringbuf_get(ibw, pkt->data, pktsize) == pktsize);
 249                         stream_set_endp(pkt, pktsize);
 250
 251                         frrtrace(2, frr_bgp, packet_read, peer, pkt);
 252                         frr_with_mutex(&peer->io_mtx) {
 253                                 stream_fifo_push(peer->ibuf, pkt);
 254                         }
 255
 256                         added_pkt = true;
 257                 } else
 258                         break;
 259         }
 260
 261         /* handle invalid header */
 262         if (fatal) {
 263                 /* wipe buffer just in case someone screwed up */
 264                 ringbuf_wipe(peer->ibuf_work);
 265         } else {
 266                 assert(ringbuf_space(peer->ibuf_work) >= peer->max_packet_size);
 267
 268                 thread_add_read(fpt->master, bgp_process_reads, peer, peer->fd,
 269                                 &peer->t_read);
 270                 if (added_pkt)
 271                         thread_add_event(bm->master, bgp_process_packet,
 272                                          peer, 0, &peer->t_process_packet);
 273         }
 274
 275         return 0;
 276 }
 277
 278 /*
 279  * Flush peer output buffer.
 280  *
 281  * This function pops packets off of peer->obuf and writes them to peer->fd.
 282  * The amount of packets written is equal to the minimum of peer->wpkt_quanta
 283  * and the number of packets on the output buffer, unless an error occurs.
 284  *
 285  * If write() returns an error, the appropriate FSM event is generated.
 286  *
 287  * The return value is equal to the number of packets written
 288  * (which may be zero).
 289  */
 290 static uint16_t bgp_write(struct peer *peer)
 291 {
 292         uint8_t type;
 293         struct stream *s;
 294         int update_last_write = 0;
 295         unsigned int count;
 296         uint32_t uo = 0;
 297         uint16_t status = 0;
 298         uint32_t wpkt_quanta_old;
 299
 300         int writenum = 0;
 301         int num;
 302         unsigned int iovsz;
 303         unsigned int strmsz;
 304         unsigned int total_written;
 305
 306         wpkt_quanta_old = atomic_load_explicit(&peer->bgp->wpkt_quanta,
 307                                                memory_order_relaxed);
 308         struct stream *ostreams[wpkt_quanta_old];
 309         struct stream **streams = ostreams;
 310         struct iovec iov[wpkt_quanta_old];
 311
 312         s = stream_fifo_head(peer->obuf);
 313
 314         if (!s)
 315                 goto done;
 316
 317         count = iovsz = 0;
 318         while (count < wpkt_quanta_old && iovsz < array_size(iov) && s) {
 319                 ostreams[iovsz] = s;
 320                 iov[iovsz].iov_base = stream_pnt(s);
 321                 iov[iovsz].iov_len = STREAM_READABLE(s);
 322                 writenum += STREAM_READABLE(s);
 323                 s = s->next;
 324                 ++iovsz;
 325                 ++count;
 326         }
 327
 328         strmsz = iovsz;
 329         total_written = 0;
 330
 331         do {
 332                 num = writev(peer->fd, iov, iovsz);
 333
 334                 if (num < 0) {
 335                         if (!ERRNO_IO_RETRY(errno)) {
 336                                 BGP_EVENT_ADD(peer, TCP_fatal_error);
 337                                 SET_FLAG(status, BGP_IO_FATAL_ERR);
 338                         } else {
 339                                 SET_FLAG(status, BGP_IO_TRANS_ERR);
 340                         }
 341
 342                         break;
 343                 } else if (num != writenum) {
 344                         unsigned int msg_written = 0;
 345                         unsigned int ic = iovsz;
 346
 347                         for (unsigned int i = 0; i < ic; i++) {
 348                                 size_t ss = iov[i].iov_len;
 349
 350                                 if (ss > (unsigned int) num)
 351                                         break;
 352
 353                                 msg_written++;
 354                                 iovsz--;
 355                                 writenum -= ss;
 356                                 num -= ss;
 357                         }
 358
 359                         total_written += msg_written;
 360
 361                         assert(total_written < count);
 362
 363                         memmove(&iov, &iov[msg_written],
 364                                 sizeof(iov[0]) * iovsz);
 365                         streams = &streams[msg_written];
 366                         stream_forward_getp(streams[0], num);
 367                         iov[0].iov_base = stream_pnt(streams[0]);
 368                         iov[0].iov_len = STREAM_READABLE(streams[0]);
 369
 370                         writenum -= num;
 371                         num = 0;
 372                         assert(writenum > 0);
 373                 } else {
 374                         total_written = strmsz;
 375                 }
 376
 377         } while (num != writenum);
 378
 379         /* Handle statistics */
 380         for (unsigned int i = 0; i < total_written; i++) {
 381                 s = stream_fifo_pop(peer->obuf);
 382
 383                 assert(s == ostreams[i]);
 384
 385                 /* Retrieve BGP packet type. */
 386                 stream_set_getp(s, BGP_MARKER_SIZE + 2);
 387                 type = stream_getc(s);
 388
 389                 switch (type) {
 390                 case BGP_MSG_OPEN:
 391                         atomic_fetch_add_explicit(&peer->open_out, 1,
 392                                                   memory_order_relaxed);
 393                         break;
 394                 case BGP_MSG_UPDATE:
 395                         atomic_fetch_add_explicit(&peer->update_out, 1,
 396                                                   memory_order_relaxed);
 397                         uo++;
 398                         break;
 399                 case BGP_MSG_NOTIFY:
 400                         atomic_fetch_add_explicit(&peer->notify_out, 1,
 401                                                   memory_order_relaxed);
 402                         /* Double start timer. */
 403                         peer->v_start *= 2;
 404
 405                         /* Overflow check. */
 406                         if (peer->v_start >= (60 * 2))
 407                                 peer->v_start = (60 * 2);
 408
 409                         /*
 410                          * Handle Graceful Restart case where the state changes
 411                          * to Connect instead of Idle.
 412                          */
 413                         BGP_EVENT_ADD(peer, BGP_Stop);
 414                         goto done;
 415
 416                 case BGP_MSG_KEEPALIVE:
 417                         atomic_fetch_add_explicit(&peer->keepalive_out, 1,
 418                                                   memory_order_relaxed);
 419                         break;
 420                 case BGP_MSG_ROUTE_REFRESH_NEW:
 421                 case BGP_MSG_ROUTE_REFRESH_OLD:
 422                         atomic_fetch_add_explicit(&peer->refresh_out, 1,
 423                                                   memory_order_relaxed);
 424                         break;
 425                 case BGP_MSG_CAPABILITY:
 426                         atomic_fetch_add_explicit(&peer->dynamic_cap_out, 1,
 427                                                   memory_order_relaxed);
 428                         break;
 429                 }
 430
 431                 stream_free(s);
 432                 ostreams[i] = NULL;
 433                 update_last_write = 1;
 434         }
 435
 436 done : {
 437         /*
 438          * Update last_update if UPDATEs were written.
 439          * Note: that these are only updated at end,
 440          *       not per message (i.e., per loop)
 441          */
 442         if (uo)
 443                 atomic_store_explicit(&peer->last_update, bgp_clock(),
 444                                       memory_order_relaxed);
 445
 446         /* If we TXed any flavor of packet */
 447         if (update_last_write)
 448                 atomic_store_explicit(&peer->last_write, bgp_clock(),
 449                                       memory_order_relaxed);
 450 }
 451
 452         return status;
 453 }
 454
 455 /*
 456  * Reads a chunk of data from peer->fd into peer->ibuf_work.
 457  *
 458  * code_p
 459  *    Pointer to location to store FSM event code in case of fatal error.
 460  *
 461  * @return status flag (see top-of-file)
 462  */
 463 static uint16_t bgp_read(struct peer *peer, int *code_p)
 464 {
 465         ssize_t nbytes;  // how many bytes we actually read
 466         uint16_t status = 0;
 467
 468         nbytes = ringbuf_read(peer->ibuf_work, peer->fd);
 469
 470         /* EAGAIN or EWOULDBLOCK; come back later */
 471         if (nbytes < 0 && ERRNO_IO_RETRY(errno)) {
 472                 SET_FLAG(status, BGP_IO_TRANS_ERR);
 473         } else if (nbytes < 0) {
 474                 /* Fatal error; tear down session */
 475                 flog_err(EC_BGP_UPDATE_RCV,
 476                          "%s [Error] bgp_read_packet error: %s", peer->host,
 477                          safe_strerror(errno));
 478
 479                 /* Handle the error in the main pthread. */
 480                 if (code_p)
 481                         *code_p = TCP_fatal_error;
 482
 483                 SET_FLAG(status, BGP_IO_FATAL_ERR);
 484
 485         } else if (nbytes == 0) {
 486                 /* Received EOF / TCP session closed */
 487                 if (bgp_debug_neighbor_events(peer))
 488                         zlog_debug("%s [Event] BGP connection closed fd %d",
 489                                    peer->host, peer->fd);
 490
 491                 /* Handle the error in the main pthread. */
 492                 if (code_p)
 493                         *code_p = TCP_connection_closed;
 494
 495                 SET_FLAG(status, BGP_IO_FATAL_ERR);
 496         }
 497
 498         return status;
 499 }
 500
 501 /*
 502  * Called after we have read a BGP packet header. Validates marker, message
 503  * type and packet length. If any of these aren't correct, sends a notify.
 504  *
 505  * Assumes that there are at least BGP_HEADER_SIZE readable bytes in the input
 506  * buffer.
 507  */
 508 static bool validate_header(struct peer *peer)
 509 {
 510         uint16_t size;
 511         uint8_t type;
 512         struct ringbuf *pkt = peer->ibuf_work;
 513
 514         static const uint8_t m_correct[BGP_MARKER_SIZE] = {
 515                 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
 516                 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
 517         uint8_t m_rx[BGP_MARKER_SIZE] = {0x00};
 518
 519         if (ringbuf_peek(pkt, 0, m_rx, BGP_MARKER_SIZE) != BGP_MARKER_SIZE)
 520                 return false;
 521
 522         if (memcmp(m_correct, m_rx, BGP_MARKER_SIZE) != 0) {
 523                 bgp_notify_send(peer, BGP_NOTIFY_HEADER_ERR,
 524                                 BGP_NOTIFY_HEADER_NOT_SYNC);
 525                 return false;
 526         }
 527
 528         /* Get size and type in network byte order. */
 529         ringbuf_peek(pkt, BGP_MARKER_SIZE, &size, sizeof(size));
 530         ringbuf_peek(pkt, BGP_MARKER_SIZE + 2, &type, sizeof(type));
 531
 532         size = ntohs(size);
 533
 534         /* BGP type check. */
 535         if (type != BGP_MSG_OPEN && type != BGP_MSG_UPDATE
 536             && type != BGP_MSG_NOTIFY && type != BGP_MSG_KEEPALIVE
 537             && type != BGP_MSG_ROUTE_REFRESH_NEW
 538             && type != BGP_MSG_ROUTE_REFRESH_OLD
 539             && type != BGP_MSG_CAPABILITY) {
 540                 if (bgp_debug_neighbor_events(peer))
 541                         zlog_debug("%s unknown message type 0x%02x", peer->host,
 542                                    type);
 543
 544                 bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR,
 545                                           BGP_NOTIFY_HEADER_BAD_MESTYPE, &type,
 546                                           1);
 547                 return false;
 548         }
 549
 550         /* Minimum packet length check. */
 551         if ((size < BGP_HEADER_SIZE) || (size > peer->max_packet_size)
 552             || (type == BGP_MSG_OPEN && size < BGP_MSG_OPEN_MIN_SIZE)
 553             || (type == BGP_MSG_UPDATE && size < BGP_MSG_UPDATE_MIN_SIZE)
 554             || (type == BGP_MSG_NOTIFY && size < BGP_MSG_NOTIFY_MIN_SIZE)
 555             || (type == BGP_MSG_KEEPALIVE && size != BGP_MSG_KEEPALIVE_MIN_SIZE)
 556             || (type == BGP_MSG_ROUTE_REFRESH_NEW
 557                 && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE)
 558             || (type == BGP_MSG_ROUTE_REFRESH_OLD
 559                 && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE)
 560             || (type == BGP_MSG_CAPABILITY
 561                 && size < BGP_MSG_CAPABILITY_MIN_SIZE)) {
 562                 if (bgp_debug_neighbor_events(peer)) {
 563                         zlog_debug("%s bad message length - %d for %s",
 564                                    peer->host, size,
 565                                    type == 128 ? "ROUTE-REFRESH"
 566                                                : bgp_type_str[(int)type]);
 567                 }
 568
 569                 uint16_t nsize = htons(size);
 570
 571                 bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR,
 572                                           BGP_NOTIFY_HEADER_BAD_MESLEN,
 573                                           (unsigned char *)&nsize, 2);
 574                 return false;
 575         }
 576
 577         return true;
 578 }