pimd/pim_msdp_packet.c

   1 /*
   2  * IP MSDP packet helper
   3  * Copyright (C) 2016 Cumulus Networks, Inc.
   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 #include <zebra.h>
  20
  21 #include <lib/log.h>
  22 #include <lib/network.h>
  23 #include <lib/stream.h>
  24 #include <lib/thread.h>
  25 #include <lib/vty.h>
  26 #include <lib/lib_errors.h>
  27
  28 #include "pimd.h"
  29 #include "pim_str.h"
  30 #include "pim_errors.h"
  31
  32 #include "pim_msdp.h"
  33 #include "pim_msdp_packet.h"
  34 #include "pim_msdp_socket.h"
  35
  36 static char *pim_msdp_pkt_type_dump(enum pim_msdp_tlv type, char *buf,
  37                                     int buf_size)
  38 {
  39         switch (type) {
  40         case PIM_MSDP_V4_SOURCE_ACTIVE:
  41                 snprintf(buf, buf_size, "%s", "SA");
  42                 break;
  43         case PIM_MSDP_V4_SOURCE_ACTIVE_REQUEST:
  44                 snprintf(buf, buf_size, "%s", "SA_REQ");
  45                 break;
  46         case PIM_MSDP_V4_SOURCE_ACTIVE_RESPONSE:
  47                 snprintf(buf, buf_size, "%s", "SA_RESP");
  48                 break;
  49         case PIM_MSDP_KEEPALIVE:
  50                 snprintf(buf, buf_size, "%s", "KA");
  51                 break;
  52         case PIM_MSDP_RESERVED:
  53                 snprintf(buf, buf_size, "%s", "RSVD");
  54                 break;
  55         case PIM_MSDP_TRACEROUTE_PROGRESS:
  56                 snprintf(buf, buf_size, "%s", "TRACE_PROG");
  57                 break;
  58         case PIM_MSDP_TRACEROUTE_REPLY:
  59                 snprintf(buf, buf_size, "%s", "TRACE_REPLY");
  60                 break;
  61         default:
  62                 snprintf(buf, buf_size, "UNK-%d", type);
  63         }
  64         return buf;
  65 }
  66
  67 static void pim_msdp_pkt_sa_dump_one(struct stream *s)
  68 {
  69         struct prefix_sg sg;
  70
  71         /* just throw away the three reserved bytes */
  72         stream_get3(s);
  73         /* throw away the prefix length also */
  74         stream_getc(s);
  75
  76         memset(&sg, 0, sizeof(struct prefix_sg));
  77         sg.grp.s_addr = stream_get_ipv4(s);
  78         sg.src.s_addr = stream_get_ipv4(s);
  79
  80         zlog_debug("  sg %s", pim_str_sg_dump(&sg));
  81 }
  82
  83 static void pim_msdp_pkt_sa_dump(struct stream *s)
  84 {
  85         int entry_cnt;
  86         int i;
  87         struct in_addr rp; /* Last RP address associated with this SA */
  88
  89         entry_cnt = stream_getc(s);
  90         rp.s_addr = stream_get_ipv4(s);
  91
  92         if (PIM_DEBUG_MSDP_PACKETS) {
  93                 char rp_str[INET_ADDRSTRLEN];
  94                 pim_inet4_dump("<rp?>", rp, rp_str, sizeof(rp_str));
  95                 zlog_debug("  entry_cnt %d rp %s", entry_cnt, rp_str);
  96         }
  97
  98         /* dump SAs */
  99         for (i = 0; i < entry_cnt; ++i) {
 100                 pim_msdp_pkt_sa_dump_one(s);
 101         }
 102 }
 103
 104 static void pim_msdp_pkt_dump(struct pim_msdp_peer *mp, int type, int len,
 105                               bool rx, struct stream *s)
 106 {
 107         char type_str[PIM_MSDP_PKT_TYPE_STRLEN];
 108
 109         pim_msdp_pkt_type_dump(type, type_str, sizeof(type_str));
 110
 111         zlog_debug("MSDP peer %s pkt %s type %s len %d", mp->key_str,
 112                    rx ? "rx" : "tx", type_str, len);
 113
 114         if (!s) {
 115                 return;
 116         }
 117
 118         switch (type) {
 119         case PIM_MSDP_V4_SOURCE_ACTIVE:
 120                 pim_msdp_pkt_sa_dump(s);
 121                 break;
 122         default:;
 123         }
 124 }
 125
 126 /* Check file descriptor whether connect is established. */
 127 static void pim_msdp_connect_check(struct pim_msdp_peer *mp)
 128 {
 129         int status;
 130         socklen_t slen;
 131         int ret;
 132
 133         if (mp->state != PIM_MSDP_CONNECTING) {
 134                 /* if we are here it means we are not in a connecting or
 135                  * established state
 136                  * for now treat this as a fatal error */
 137                 pim_msdp_peer_reset_tcp_conn(mp, "invalid-state");
 138                 return;
 139         }
 140
 141         PIM_MSDP_PEER_READ_OFF(mp);
 142         PIM_MSDP_PEER_WRITE_OFF(mp);
 143
 144         /* Check file descriptor. */
 145         slen = sizeof(status);
 146         ret = getsockopt(mp->fd, SOL_SOCKET, SO_ERROR, (void *)&status, &slen);
 147
 148         /* If getsockopt is fail, this is fatal error. */
 149         if (ret < 0) {
 150                 flog_err_sys(EC_LIB_SOCKET,
 151                              "can't get sockopt for nonblocking connect");
 152                 pim_msdp_peer_reset_tcp_conn(mp, "connect-failed");
 153                 return;
 154         }
 155
 156         /* When status is 0 then TCP connection is established. */
 157         if (PIM_DEBUG_MSDP_INTERNAL) {
 158                 zlog_debug("MSDP peer %s pim_connect_check %s", mp->key_str,
 159                            status ? "fail" : "success");
 160         }
 161         if (status == 0) {
 162                 pim_msdp_peer_established(mp);
 163         } else {
 164                 pim_msdp_peer_reset_tcp_conn(mp, "connect-failed");
 165         }
 166 }
 167
 168 static void pim_msdp_pkt_delete(struct pim_msdp_peer *mp)
 169 {
 170         stream_free(stream_fifo_pop(mp->obuf));
 171 }
 172
 173 static void pim_msdp_pkt_add(struct pim_msdp_peer *mp, struct stream *s)
 174 {
 175         stream_fifo_push(mp->obuf, s);
 176 }
 177
 178 static void pim_msdp_write_proceed_actions(struct pim_msdp_peer *mp)
 179 {
 180         if (stream_fifo_head(mp->obuf)) {
 181                 PIM_MSDP_PEER_WRITE_ON(mp);
 182         }
 183 }
 184
 185 int pim_msdp_write(struct thread *thread)
 186 {
 187         struct pim_msdp_peer *mp;
 188         struct stream *s;
 189         int num;
 190         enum pim_msdp_tlv type;
 191         int len;
 192         int work_cnt = 0;
 193         int work_max_cnt = 100;
 194
 195         mp = THREAD_ARG(thread);
 196         mp->t_write = NULL;
 197
 198         if (PIM_DEBUG_MSDP_INTERNAL) {
 199                 zlog_debug("MSDP peer %s pim_msdp_write", mp->key_str);
 200         }
 201         if (mp->fd < 0) {
 202                 return -1;
 203         }
 204
 205         /* check if TCP connection is established */
 206         if (mp->state != PIM_MSDP_ESTABLISHED) {
 207                 pim_msdp_connect_check(mp);
 208                 return 0;
 209         }
 210
 211         s = stream_fifo_head(mp->obuf);
 212         if (!s) {
 213                 pim_msdp_write_proceed_actions(mp);
 214                 return 0;
 215         }
 216
 217         sockopt_cork(mp->fd, 1);
 218
 219         /* Nonblocking write until TCP output buffer is full  */
 220         do {
 221                 int writenum;
 222
 223                 /* Number of bytes to be sent */
 224                 writenum = stream_get_endp(s) - stream_get_getp(s);
 225
 226                 /* Call write() system call */
 227                 num = write(mp->fd, stream_pnt(s), writenum);
 228                 if (num < 0) {
 229                         /* write failed either retry needed or error */
 230                         if (ERRNO_IO_RETRY(errno)) {
 231                                 if (PIM_DEBUG_MSDP_INTERNAL) {
 232                                         zlog_debug(
 233                                                 "MSDP peer %s pim_msdp_write io retry",
 234                                                 mp->key_str);
 235                                 }
 236                                 break;
 237                         }
 238
 239                         pim_msdp_peer_reset_tcp_conn(mp, "pkt-tx-failed");
 240                         return 0;
 241                 }
 242
 243                 if (num != writenum) {
 244                         /* Partial write */
 245                         stream_forward_getp(s, num);
 246                         if (PIM_DEBUG_MSDP_INTERNAL) {
 247                                 zlog_debug(
 248                                         "MSDP peer %s pim_msdp_partial_write",
 249                                         mp->key_str);
 250                         }
 251                         break;
 252                 }
 253
 254                 /* Retrieve msdp packet type. */
 255                 stream_set_getp(s, 0);
 256                 type = stream_getc(s);
 257                 len = stream_getw(s);
 258                 switch (type) {
 259                 case PIM_MSDP_KEEPALIVE:
 260                         mp->ka_tx_cnt++;
 261                         break;
 262                 case PIM_MSDP_V4_SOURCE_ACTIVE:
 263                         mp->sa_tx_cnt++;
 264                         break;
 265                 default:;
 266                 }
 267                 if (PIM_DEBUG_MSDP_PACKETS) {
 268                         pim_msdp_pkt_dump(mp, type, len, false /*rx*/, s);
 269                 }
 270
 271                 /* packet sent delete it. */
 272                 pim_msdp_pkt_delete(mp);
 273
 274                 ++work_cnt;
 275                 /* may need to pause if we have done too much work in this
 276                  * loop */
 277                 if (work_cnt >= work_max_cnt) {
 278                         break;
 279                 }
 280         } while ((s = stream_fifo_head(mp->obuf)) != NULL);
 281         pim_msdp_write_proceed_actions(mp);
 282
 283         sockopt_cork(mp->fd, 0);
 284
 285         if (PIM_DEBUG_MSDP_INTERNAL) {
 286                 zlog_debug("MSDP peer %s pim_msdp_write wrote %d packets",
 287                            mp->key_str, work_cnt);
 288         }
 289
 290         return 0;
 291 }
 292
 293 static void pim_msdp_pkt_send(struct pim_msdp_peer *mp, struct stream *s)
 294 {
 295         /* Add packet to the end of list. */
 296         pim_msdp_pkt_add(mp, s);
 297
 298         PIM_MSDP_PEER_WRITE_ON(mp);
 299 }
 300
 301 void pim_msdp_pkt_ka_tx(struct pim_msdp_peer *mp)
 302 {
 303         struct stream *s;
 304
 305         if (mp->state != PIM_MSDP_ESTABLISHED) {
 306                 /* don't tx anything unless a session is established */
 307                 return;
 308         }
 309         s = stream_new(PIM_MSDP_KA_TLV_MAX_SIZE);
 310         stream_putc(s, PIM_MSDP_KEEPALIVE);
 311         stream_putw(s, PIM_MSDP_KA_TLV_MAX_SIZE);
 312
 313         pim_msdp_pkt_send(mp, s);
 314 }
 315
 316 static void pim_msdp_pkt_sa_push_to_one_peer(struct pim_instance *pim,
 317                                              struct pim_msdp_peer *mp)
 318 {
 319         struct stream *s;
 320
 321         if (mp->state != PIM_MSDP_ESTABLISHED) {
 322                 /* don't tx anything unless a session is established */
 323                 return;
 324         }
 325         s = stream_dup(pim->msdp.work_obuf);
 326         if (s) {
 327                 pim_msdp_pkt_send(mp, s);
 328                 mp->flags |= PIM_MSDP_PEERF_SA_JUST_SENT;
 329         }
 330 }
 331
 332 /* push the stream into the obuf fifo of all the peers */
 333 static void pim_msdp_pkt_sa_push(struct pim_instance *pim,
 334                                  struct pim_msdp_peer *mp)
 335 {
 336         struct listnode *mpnode;
 337
 338         if (mp) {
 339                 pim_msdp_pkt_sa_push_to_one_peer(pim, mp);
 340         } else {
 341                 for (ALL_LIST_ELEMENTS_RO(pim->msdp.peer_list, mpnode, mp)) {
 342                         if (PIM_DEBUG_MSDP_INTERNAL) {
 343                                 zlog_debug("MSDP peer %s pim_msdp_pkt_sa_push",
 344                                            mp->key_str);
 345                         }
 346                         pim_msdp_pkt_sa_push_to_one_peer(pim, mp);
 347                 }
 348         }
 349 }
 350
 351 static int pim_msdp_pkt_sa_fill_hdr(struct pim_instance *pim, int local_cnt)
 352 {
 353         int curr_tlv_ecnt;
 354
 355         stream_reset(pim->msdp.work_obuf);
 356         curr_tlv_ecnt = local_cnt > PIM_MSDP_SA_MAX_ENTRY_CNT
 357                                 ? PIM_MSDP_SA_MAX_ENTRY_CNT
 358                                 : local_cnt;
 359         local_cnt -= curr_tlv_ecnt;
 360         stream_putc(pim->msdp.work_obuf, PIM_MSDP_V4_SOURCE_ACTIVE);
 361         stream_putw(pim->msdp.work_obuf,
 362                     PIM_MSDP_SA_ENTRY_CNT2SIZE(curr_tlv_ecnt));
 363         stream_putc(pim->msdp.work_obuf, curr_tlv_ecnt);
 364         stream_put_ipv4(pim->msdp.work_obuf, pim->msdp.originator_id.s_addr);
 365
 366         return local_cnt;
 367 }
 368
 369 static void pim_msdp_pkt_sa_fill_one(struct pim_msdp_sa *sa)
 370 {
 371         stream_put3(sa->pim->msdp.work_obuf, 0 /* reserved */);
 372         stream_putc(sa->pim->msdp.work_obuf, 32 /* sprefix len */);
 373         stream_put_ipv4(sa->pim->msdp.work_obuf, sa->sg.grp.s_addr);
 374         stream_put_ipv4(sa->pim->msdp.work_obuf, sa->sg.src.s_addr);
 375 }
 376
 377 static void pim_msdp_pkt_sa_gen(struct pim_instance *pim,
 378                                 struct pim_msdp_peer *mp)
 379 {
 380         struct listnode *sanode;
 381         struct pim_msdp_sa *sa;
 382         int sa_count;
 383         int local_cnt = pim->msdp.local_cnt;
 384
 385         sa_count = 0;
 386         if (PIM_DEBUG_MSDP_INTERNAL) {
 387                 zlog_debug("  sa gen  %d", local_cnt);
 388         }
 389
 390         local_cnt = pim_msdp_pkt_sa_fill_hdr(pim, local_cnt);
 391
 392         for (ALL_LIST_ELEMENTS_RO(pim->msdp.sa_list, sanode, sa)) {
 393                 if (!(sa->flags & PIM_MSDP_SAF_LOCAL)) {
 394                         /* current implementation of MSDP is for anycast i.e.
 395                          * full mesh. so
 396                          * no re-forwarding of SAs that we learnt from other
 397                          * peers */
 398                         continue;
 399                 }
 400                 /* add sa into scratch pad */
 401                 pim_msdp_pkt_sa_fill_one(sa);
 402                 ++sa_count;
 403                 if (sa_count >= PIM_MSDP_SA_MAX_ENTRY_CNT) {
 404                         pim_msdp_pkt_sa_push(pim, mp);
 405                         /* reset headers */
 406                         sa_count = 0;
 407                         if (PIM_DEBUG_MSDP_INTERNAL) {
 408                                 zlog_debug("  sa gen for remainder %d",
 409                                            local_cnt);
 410                         }
 411                         local_cnt = pim_msdp_pkt_sa_fill_hdr(pim, local_cnt);
 412                 }
 413         }
 414
 415         if (sa_count) {
 416                 pim_msdp_pkt_sa_push(pim, mp);
 417         }
 418         return;
 419 }
 420
 421 static void pim_msdp_pkt_sa_tx_done(struct pim_instance *pim)
 422 {
 423         struct listnode *mpnode;
 424         struct pim_msdp_peer *mp;
 425
 426         /* if SA were sent to the peers we restart ka timer and avoid
 427          * unnecessary ka noise */
 428         for (ALL_LIST_ELEMENTS_RO(pim->msdp.peer_list, mpnode, mp)) {
 429                 if (mp->flags & PIM_MSDP_PEERF_SA_JUST_SENT) {
 430                         mp->flags &= ~PIM_MSDP_PEERF_SA_JUST_SENT;
 431                         pim_msdp_peer_pkt_txed(mp);
 432                 }
 433         }
 434 }
 435
 436 void pim_msdp_pkt_sa_tx(struct pim_instance *pim)
 437 {
 438         pim_msdp_pkt_sa_gen(pim, NULL /* mp */);
 439         pim_msdp_pkt_sa_tx_done(pim);
 440 }
 441
 442 void pim_msdp_pkt_sa_tx_one(struct pim_msdp_sa *sa)
 443 {
 444         pim_msdp_pkt_sa_fill_hdr(sa->pim, 1 /* cnt */);
 445         pim_msdp_pkt_sa_fill_one(sa);
 446         pim_msdp_pkt_sa_push(sa->pim, NULL);
 447         pim_msdp_pkt_sa_tx_done(sa->pim);
 448 }
 449
 450 /* when a connection is first established we push all SAs immediately */
 451 void pim_msdp_pkt_sa_tx_to_one_peer(struct pim_msdp_peer *mp)
 452 {
 453         pim_msdp_pkt_sa_gen(mp->pim, mp);
 454         pim_msdp_pkt_sa_tx_done(mp->pim);
 455 }
 456
 457 static void pim_msdp_pkt_rxed_with_fatal_error(struct pim_msdp_peer *mp)
 458 {
 459         pim_msdp_peer_reset_tcp_conn(mp, "invalid-pkt-rx");
 460 }
 461
 462 static void pim_msdp_pkt_ka_rx(struct pim_msdp_peer *mp, int len)
 463 {
 464         mp->ka_rx_cnt++;
 465         if (len != PIM_MSDP_KA_TLV_MAX_SIZE) {
 466                 pim_msdp_pkt_rxed_with_fatal_error(mp);
 467                 return;
 468         }
 469         pim_msdp_peer_pkt_rxed(mp);
 470 }
 471
 472 static void pim_msdp_pkt_sa_rx_one(struct pim_msdp_peer *mp, struct in_addr rp)
 473 {
 474         int prefix_len;
 475         struct prefix_sg sg;
 476
 477         /* just throw away the three reserved bytes */
 478         stream_get3(mp->ibuf);
 479         prefix_len = stream_getc(mp->ibuf);
 480
 481         memset(&sg, 0, sizeof(struct prefix_sg));
 482         sg.grp.s_addr = stream_get_ipv4(mp->ibuf);
 483         sg.src.s_addr = stream_get_ipv4(mp->ibuf);
 484
 485         if (prefix_len != 32) {
 486                 /* ignore SA update if the prefix length is not 32 */
 487                 flog_err(EC_PIM_MSDP_PACKET,
 488                          "rxed sa update with invalid prefix length %d",
 489                          prefix_len);
 490                 return;
 491         }
 492         if (PIM_DEBUG_MSDP_PACKETS) {
 493                 zlog_debug("  sg %s", pim_str_sg_dump(&sg));
 494         }
 495         pim_msdp_sa_ref(mp->pim, mp, &sg, rp);
 496 }
 497
 498 static void pim_msdp_pkt_sa_rx(struct pim_msdp_peer *mp, int len)
 499 {
 500         int entry_cnt;
 501         int i;
 502         struct in_addr rp; /* Last RP address associated with this SA */
 503
 504         mp->sa_rx_cnt++;
 505
 506         if (len < PIM_MSDP_SA_TLV_MIN_SIZE) {
 507                 pim_msdp_pkt_rxed_with_fatal_error(mp);
 508                 return;
 509         }
 510
 511         entry_cnt = stream_getc(mp->ibuf);
 512         /* some vendors include the actual multicast data in the tlv (at the
 513          * end).
 514          * we will ignore such data. in the future we may consider pushing it
 515          * down
 516          * the RPT */
 517         if (len < PIM_MSDP_SA_ENTRY_CNT2SIZE(entry_cnt)) {
 518                 pim_msdp_pkt_rxed_with_fatal_error(mp);
 519                 return;
 520         }
 521         rp.s_addr = stream_get_ipv4(mp->ibuf);
 522
 523         if (PIM_DEBUG_MSDP_PACKETS) {
 524                 char rp_str[INET_ADDRSTRLEN];
 525                 pim_inet4_dump("<rp?>", rp, rp_str, sizeof(rp_str));
 526                 zlog_debug("  entry_cnt %d rp %s", entry_cnt, rp_str);
 527         }
 528
 529         if (!pim_msdp_peer_rpf_check(mp, rp)) {
 530                 /* if peer-RPF check fails don't process the packet any further
 531                  */
 532                 if (PIM_DEBUG_MSDP_PACKETS) {
 533                         zlog_debug("  peer RPF check failed");
 534                 }
 535                 return;
 536         }
 537
 538         pim_msdp_peer_pkt_rxed(mp);
 539
 540         /* update SA cache */
 541         for (i = 0; i < entry_cnt; ++i) {
 542                 pim_msdp_pkt_sa_rx_one(mp, rp);
 543         }
 544 }
 545
 546 static void pim_msdp_pkt_rx(struct pim_msdp_peer *mp)
 547 {
 548         enum pim_msdp_tlv type;
 549         int len;
 550
 551         /* re-read type and len */
 552         type = stream_getc_from(mp->ibuf, 0);
 553         len = stream_getw_from(mp->ibuf, 1);
 554         if (len < PIM_MSDP_HEADER_SIZE) {
 555                 pim_msdp_pkt_rxed_with_fatal_error(mp);
 556                 return;
 557         }
 558
 559         if (len > PIM_MSDP_SA_TLV_MAX_SIZE) {
 560                 /* if tlv size if greater than max just ignore the tlv */
 561                 return;
 562         }
 563
 564         if (PIM_DEBUG_MSDP_PACKETS) {
 565                 pim_msdp_pkt_dump(mp, type, len, true /*rx*/, NULL /*s*/);
 566         }
 567
 568         switch (type) {
 569         case PIM_MSDP_KEEPALIVE:
 570                 pim_msdp_pkt_ka_rx(mp, len);
 571                 break;
 572         case PIM_MSDP_V4_SOURCE_ACTIVE:
 573                 mp->sa_rx_cnt++;
 574                 pim_msdp_pkt_sa_rx(mp, len);
 575                 break;
 576         default:
 577                 mp->unk_rx_cnt++;
 578         }
 579 }
 580
 581 /* pim msdp read utility function. */
 582 static int pim_msdp_read_packet(struct pim_msdp_peer *mp)
 583 {
 584         int nbytes;
 585         int readsize;
 586         int old_endp;
 587         int new_endp;
 588
 589         old_endp = stream_get_endp(mp->ibuf);
 590         readsize = mp->packet_size - old_endp;
 591         if (!readsize) {
 592                 return 0;
 593         }
 594
 595         /* Read packet from fd */
 596         nbytes = stream_read_try(mp->ibuf, mp->fd, readsize);
 597         new_endp = stream_get_endp(mp->ibuf);
 598         if (nbytes < 0) {
 599                 if (PIM_DEBUG_MSDP_INTERNAL) {
 600                         zlog_debug("MSDP peer %s read failed %d", mp->key_str,
 601                                    nbytes);
 602                 }
 603                 if (nbytes == -2) {
 604                         if (PIM_DEBUG_MSDP_INTERNAL) {
 605                                 zlog_debug(
 606                                         "MSDP peer %s pim_msdp_read io retry old_end: %d new_end: %d",
 607                                         mp->key_str, old_endp, new_endp);
 608                         }
 609                         /* transient error retry */
 610                         return -1;
 611                 }
 612                 pim_msdp_pkt_rxed_with_fatal_error(mp);
 613                 return -1;
 614         }
 615
 616         if (!nbytes) {
 617                 if (PIM_DEBUG_MSDP_INTERNAL) {
 618                         zlog_debug("MSDP peer %s read failed %d", mp->key_str,
 619                                    nbytes);
 620                 }
 621                 pim_msdp_peer_reset_tcp_conn(mp, "peer-down");
 622                 return -1;
 623         }
 624
 625         /* We read partial packet. */
 626         if (stream_get_endp(mp->ibuf) != mp->packet_size) {
 627                 if (PIM_DEBUG_MSDP_INTERNAL) {
 628                         zlog_debug(
 629                                 "MSDP peer %s read partial len %d old_endp %d new_endp %d",
 630                                 mp->key_str, mp->packet_size, old_endp,
 631                                 new_endp);
 632                 }
 633                 return -1;
 634         }
 635
 636         return 0;
 637 }
 638
 639 int pim_msdp_read(struct thread *thread)
 640 {
 641         struct pim_msdp_peer *mp;
 642         int rc;
 643         uint32_t len;
 644
 645         mp = THREAD_ARG(thread);
 646         mp->t_read = NULL;
 647
 648         if (PIM_DEBUG_MSDP_INTERNAL) {
 649                 zlog_debug("MSDP peer %s pim_msdp_read", mp->key_str);
 650         }
 651
 652         if (mp->fd < 0) {
 653                 return -1;
 654         }
 655
 656         /* check if TCP connection is established */
 657         if (mp->state != PIM_MSDP_ESTABLISHED) {
 658                 pim_msdp_connect_check(mp);
 659                 return 0;
 660         }
 661
 662         PIM_MSDP_PEER_READ_ON(mp);
 663
 664         if (!mp->packet_size) {
 665                 mp->packet_size = PIM_MSDP_HEADER_SIZE;
 666         }
 667
 668         if (stream_get_endp(mp->ibuf) < PIM_MSDP_HEADER_SIZE) {
 669                 /* start by reading the TLV header */
 670                 rc = pim_msdp_read_packet(mp);
 671                 if (rc < 0) {
 672                         goto pim_msdp_read_end;
 673                 }
 674
 675                 /* Find TLV type and len  */
 676                 stream_getc(mp->ibuf);
 677                 len = stream_getw(mp->ibuf);
 678                 if (len < PIM_MSDP_HEADER_SIZE) {
 679                         pim_msdp_pkt_rxed_with_fatal_error(mp);
 680                         goto pim_msdp_read_end;
 681                 }
 682                 /* read complete TLV */
 683                 mp->packet_size = len;
 684         }
 685
 686         rc = pim_msdp_read_packet(mp);
 687         if (rc < 0) {
 688                 goto pim_msdp_read_end;
 689         }
 690
 691         pim_msdp_pkt_rx(mp);
 692
 693         /* reset input buffers and get ready for the next packet */
 694         mp->packet_size = 0;
 695         stream_reset(mp->ibuf);
 696
 697 pim_msdp_read_end:
 698         return 0;
 699 }