lib/stream.c

   1 /*
   2 *Packet interface
   3 *Copyright (C) 1999 Kunihiro Ishiguro
   4 *
   5 *This file is part of GNU Zebra.
   6 *
   7 *GNU Zebra is free software; you can redistribute it and/or modify it
   8 *under the terms of the GNU General Public License as published by the
   9 *Free Software Foundation; either version 2, or (at your option) any
  10 *later version.
  11 *
  12 *GNU Zebra is distributed in the hope that it will be useful, but
  13 *WITHOUT ANY WARRANTY; without even the implied warranty of
  14 *MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15 *General Public License for more details.
  16 *
  17 *You should have received a copy of the GNU General Public License
  18 *along with GNU Zebra; see the file COPYING.  If not, write to the Free
  19 *Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
  20 *02111-1307, USA.
  21 */
  22
  23 #include <zebra.h>
  24 #include <stddef.h>
  25
  26 #include "stream.h"
  27 #include "memory.h"
  28 #include "network.h"
  29 #include "prefix.h"
  30 #include "log.h"
  31
  32 DEFINE_MTYPE_STATIC(LIB, STREAM, "Stream")
  33 DEFINE_MTYPE_STATIC(LIB, STREAM_DATA, "Stream data")
  34 DEFINE_MTYPE_STATIC(LIB, STREAM_FIFO, "Stream FIFO")
  35
  36 /* Tests whether a position is valid */
  37 #define GETP_VALID(S, G) ((G) <= (S)->endp)
  38 #define PUT_AT_VALID(S,G) GETP_VALID(S,G)
  39 #define ENDP_VALID(S, E) ((E) <= (S)->size)
  40
  41 /* asserting sanity checks. Following must be true before
  42  * stream functions are called:
  43  *
  44  * Following must always be true of stream elements
  45  * before and after calls to stream functions:
  46  *
  47  * getp <= endp <= size
  48  *
  49  * Note that after a stream function is called following may be true:
  50  * if (getp == endp) then stream is no longer readable
  51  * if (endp == size) then stream is no longer writeable
  52  *
  53  * It is valid to put to anywhere within the size of the stream, but only
  54  * using stream_put..._at() functions.
  55  */
  56 #define STREAM_WARN_OFFSETS(S)                                                 \
  57         zlog_warn("&(struct stream): %p, size: %lu, getp: %lu, endp: %lu\n",   \
  58                   (void *)(S), (unsigned long)(S)->size,                       \
  59                   (unsigned long)(S)->getp, (unsigned long)(S)->endp)
  60
  61 #define STREAM_VERIFY_SANE(S)                                                  \
  62         do {                                                                   \
  63                 if (!(GETP_VALID(S, (S)->getp) && ENDP_VALID(S, (S)->endp)))   \
  64                         STREAM_WARN_OFFSETS(S);                                \
  65                 assert(GETP_VALID(S, (S)->getp));                              \
  66                 assert(ENDP_VALID(S, (S)->endp));                              \
  67         } while (0)
  68
  69 #define STREAM_BOUND_WARN(S, WHAT)                                             \
  70         do {                                                                   \
  71                 zlog_warn("%s: Attempt to %s out of bounds", __func__,         \
  72                           (WHAT));                                             \
  73                 STREAM_WARN_OFFSETS(S);                                        \
  74                 assert(0);                                                     \
  75         } while (0)
  76
  77 /* XXX: Deprecated macro: do not use */
  78 #define CHECK_SIZE(S, Z)                                                       \
  79         do {                                                                   \
  80                 if (((S)->endp + (Z)) > (S)->size) {                           \
  81                         zlog_warn(                                             \
  82                                 "CHECK_SIZE: truncating requested size %lu\n", \
  83                                 (unsigned long)(Z));                           \
  84                         STREAM_WARN_OFFSETS(S);                                \
  85                         (Z) = (S)->size - (S)->endp;                           \
  86                 }                                                              \
  87         } while (0);
  88
  89 /* Make stream buffer. */
  90 struct stream *stream_new(size_t size)
  91 {
  92         struct stream *s;
  93
  94         assert(size > 0);
  95
  96         s = XCALLOC(MTYPE_STREAM, sizeof(struct stream));
  97
  98         if (s == NULL)
  99                 return s;
 100
 101         if ((s->data = XMALLOC(MTYPE_STREAM_DATA, size)) == NULL) {
 102                 XFREE(MTYPE_STREAM, s);
 103                 return NULL;
 104         }
 105
 106         s->size = size;
 107         return s;
 108 }
 109
 110 /* Free it now. */
 111 void stream_free(struct stream *s)
 112 {
 113         if (!s)
 114                 return;
 115
 116         XFREE(MTYPE_STREAM_DATA, s->data);
 117         XFREE(MTYPE_STREAM, s);
 118 }
 119
 120 struct stream *stream_copy(struct stream *new, struct stream *src)
 121 {
 122         STREAM_VERIFY_SANE(src);
 123
 124         assert(new != NULL);
 125         assert(STREAM_SIZE(new) >= src->endp);
 126
 127         new->endp = src->endp;
 128         new->getp = src->getp;
 129
 130         memcpy(new->data, src->data, src->endp);
 131
 132         return new;
 133 }
 134
 135 struct stream *stream_dup(struct stream *s)
 136 {
 137         struct stream *new;
 138
 139         STREAM_VERIFY_SANE(s);
 140
 141         if ((new = stream_new(s->endp)) == NULL)
 142                 return NULL;
 143
 144         return (stream_copy(new, s));
 145 }
 146
 147 struct stream *stream_dupcat(struct stream *s1, struct stream *s2,
 148                              size_t offset)
 149 {
 150         struct stream *new;
 151
 152         STREAM_VERIFY_SANE(s1);
 153         STREAM_VERIFY_SANE(s2);
 154
 155         if ((new = stream_new(s1->endp + s2->endp)) == NULL)
 156                 return NULL;
 157
 158         memcpy(new->data, s1->data, offset);
 159         memcpy(new->data + offset, s2->data, s2->endp);
 160         memcpy(new->data + offset + s2->endp, s1->data + offset,
 161                (s1->endp - offset));
 162         new->endp = s1->endp + s2->endp;
 163         return new;
 164 }
 165
 166 size_t stream_resize(struct stream *s, size_t newsize)
 167 {
 168         u_char *newdata;
 169         STREAM_VERIFY_SANE(s);
 170
 171         newdata = XREALLOC(MTYPE_STREAM_DATA, s->data, newsize);
 172
 173         if (newdata == NULL)
 174                 return s->size;
 175
 176         s->data = newdata;
 177         s->size = newsize;
 178
 179         if (s->endp > s->size)
 180                 s->endp = s->size;
 181         if (s->getp > s->endp)
 182                 s->getp = s->endp;
 183
 184         STREAM_VERIFY_SANE(s);
 185
 186         return s->size;
 187 }
 188
 189 size_t stream_get_getp(struct stream *s)
 190 {
 191         STREAM_VERIFY_SANE(s);
 192         return s->getp;
 193 }
 194
 195 size_t stream_get_endp(struct stream *s)
 196 {
 197         STREAM_VERIFY_SANE(s);
 198         return s->endp;
 199 }
 200
 201 size_t stream_get_size(struct stream *s)
 202 {
 203         STREAM_VERIFY_SANE(s);
 204         return s->size;
 205 }
 206
 207 /* Stream structre' stream pointer related functions.  */
 208 void stream_set_getp(struct stream *s, size_t pos)
 209 {
 210         STREAM_VERIFY_SANE(s);
 211
 212         if (!GETP_VALID(s, pos)) {
 213                 STREAM_BOUND_WARN(s, "set getp");
 214                 pos = s->endp;
 215         }
 216
 217         s->getp = pos;
 218 }
 219
 220 void stream_set_endp(struct stream *s, size_t pos)
 221 {
 222         STREAM_VERIFY_SANE(s);
 223
 224         if (!ENDP_VALID(s, pos)) {
 225                 STREAM_BOUND_WARN(s, "set endp");
 226                 return;
 227         }
 228
 229         /*
 230          * Make sure the current read pointer is not beyond the new endp.
 231          */
 232         if (s->getp > pos) {
 233                 STREAM_BOUND_WARN(s, "set endp");
 234                 return;
 235         }
 236
 237         s->endp = pos;
 238         STREAM_VERIFY_SANE(s);
 239 }
 240
 241 /* Forward pointer. */
 242 void stream_forward_getp(struct stream *s, size_t size)
 243 {
 244         STREAM_VERIFY_SANE(s);
 245
 246         if (!GETP_VALID(s, s->getp + size)) {
 247                 STREAM_BOUND_WARN(s, "seek getp");
 248                 return;
 249         }
 250
 251         s->getp += size;
 252 }
 253
 254 void stream_forward_endp(struct stream *s, size_t size)
 255 {
 256         STREAM_VERIFY_SANE(s);
 257
 258         if (!ENDP_VALID(s, s->endp + size)) {
 259                 STREAM_BOUND_WARN(s, "seek endp");
 260                 return;
 261         }
 262
 263         s->endp += size;
 264 }
 265
 266 /* Copy from stream to destination. */
 267 void stream_get(void *dst, struct stream *s, size_t size)
 268 {
 269         STREAM_VERIFY_SANE(s);
 270
 271         if (STREAM_READABLE(s) < size) {
 272                 STREAM_BOUND_WARN(s, "get");
 273                 return;
 274         }
 275
 276         memcpy(dst, s->data + s->getp, size);
 277         s->getp += size;
 278 }
 279
 280 /* Get next character from the stream. */
 281 u_char stream_getc(struct stream *s)
 282 {
 283         u_char c;
 284
 285         STREAM_VERIFY_SANE(s);
 286
 287         if (STREAM_READABLE(s) < sizeof(u_char)) {
 288                 STREAM_BOUND_WARN(s, "get char");
 289                 return 0;
 290         }
 291         c = s->data[s->getp++];
 292
 293         return c;
 294 }
 295
 296 /* Get next character from the stream. */
 297 u_char stream_getc_from(struct stream *s, size_t from)
 298 {
 299         u_char c;
 300
 301         STREAM_VERIFY_SANE(s);
 302
 303         if (!GETP_VALID(s, from + sizeof(u_char))) {
 304                 STREAM_BOUND_WARN(s, "get char");
 305                 return 0;
 306         }
 307
 308         c = s->data[from];
 309
 310         return c;
 311 }
 312
 313 /* Get next word from the stream. */
 314 u_int16_t stream_getw(struct stream *s)
 315 {
 316         u_int16_t w;
 317
 318         STREAM_VERIFY_SANE(s);
 319
 320         if (STREAM_READABLE(s) < sizeof(u_int16_t)) {
 321                 STREAM_BOUND_WARN(s, "get ");
 322                 return 0;
 323         }
 324
 325         w = s->data[s->getp++] << 8;
 326         w |= s->data[s->getp++];
 327
 328         return w;
 329 }
 330
 331 /* Get next word from the stream. */
 332 u_int16_t stream_getw_from(struct stream *s, size_t from)
 333 {
 334         u_int16_t w;
 335
 336         STREAM_VERIFY_SANE(s);
 337
 338         if (!GETP_VALID(s, from + sizeof(u_int16_t))) {
 339                 STREAM_BOUND_WARN(s, "get ");
 340                 return 0;
 341         }
 342
 343         w = s->data[from++] << 8;
 344         w |= s->data[from];
 345
 346         return w;
 347 }
 348
 349 /* Get next 3-byte from the stream. */
 350 u_int32_t stream_get3_from(struct stream *s, size_t from)
 351 {
 352         u_int32_t l;
 353
 354         STREAM_VERIFY_SANE(s);
 355
 356         if (!GETP_VALID(s, from + 3)) {
 357                 STREAM_BOUND_WARN(s, "get 3byte");
 358                 return 0;
 359         }
 360
 361         l = s->data[from++] << 16;
 362         l |= s->data[from++] << 8;
 363         l |= s->data[from];
 364
 365         return l;
 366 }
 367
 368 u_int32_t stream_get3(struct stream *s)
 369 {
 370         u_int32_t l;
 371
 372         STREAM_VERIFY_SANE(s);
 373
 374         if (STREAM_READABLE(s) < 3) {
 375                 STREAM_BOUND_WARN(s, "get 3byte");
 376                 return 0;
 377         }
 378
 379         l = s->data[s->getp++] << 16;
 380         l |= s->data[s->getp++] << 8;
 381         l |= s->data[s->getp++];
 382
 383         return l;
 384 }
 385
 386 /* Get next long word from the stream. */
 387 u_int32_t stream_getl_from(struct stream *s, size_t from)
 388 {
 389         u_int32_t l;
 390
 391         STREAM_VERIFY_SANE(s);
 392
 393         if (!GETP_VALID(s, from + sizeof(u_int32_t))) {
 394                 STREAM_BOUND_WARN(s, "get long");
 395                 return 0;
 396         }
 397
 398         l = s->data[from++] << 24;
 399         l |= s->data[from++] << 16;
 400         l |= s->data[from++] << 8;
 401         l |= s->data[from];
 402
 403         return l;
 404 }
 405
 406 /* Copy from stream at specific location to destination. */
 407 void stream_get_from(void *dst, struct stream *s, size_t from, size_t size)
 408 {
 409         STREAM_VERIFY_SANE(s);
 410
 411         if (!GETP_VALID(s, from + size)) {
 412                 STREAM_BOUND_WARN(s, "get from");
 413                 return;
 414         }
 415
 416         memcpy(dst, s->data + from, size);
 417 }
 418
 419 u_int32_t stream_getl(struct stream *s)
 420 {
 421         u_int32_t l;
 422
 423         STREAM_VERIFY_SANE(s);
 424
 425         if (STREAM_READABLE(s) < sizeof(u_int32_t)) {
 426                 STREAM_BOUND_WARN(s, "get long");
 427                 return 0;
 428         }
 429
 430         l = s->data[s->getp++] << 24;
 431         l |= s->data[s->getp++] << 16;
 432         l |= s->data[s->getp++] << 8;
 433         l |= s->data[s->getp++];
 434
 435         return l;
 436 }
 437
 438 /* Get next quad word from the stream. */
 439 uint64_t stream_getq_from(struct stream *s, size_t from)
 440 {
 441         uint64_t q;
 442
 443         STREAM_VERIFY_SANE(s);
 444
 445         if (!GETP_VALID(s, from + sizeof(uint64_t))) {
 446                 STREAM_BOUND_WARN(s, "get quad");
 447                 return 0;
 448         }
 449
 450         q = ((uint64_t)s->data[from++]) << 56;
 451         q |= ((uint64_t)s->data[from++]) << 48;
 452         q |= ((uint64_t)s->data[from++]) << 40;
 453         q |= ((uint64_t)s->data[from++]) << 32;
 454         q |= ((uint64_t)s->data[from++]) << 24;
 455         q |= ((uint64_t)s->data[from++]) << 16;
 456         q |= ((uint64_t)s->data[from++]) << 8;
 457         q |= ((uint64_t)s->data[from++]);
 458
 459         return q;
 460 }
 461
 462 uint64_t stream_getq(struct stream *s)
 463 {
 464         uint64_t q;
 465
 466         STREAM_VERIFY_SANE(s);
 467
 468         if (STREAM_READABLE(s) < sizeof(uint64_t)) {
 469                 STREAM_BOUND_WARN(s, "get quad");
 470                 return 0;
 471         }
 472
 473         q = ((uint64_t)s->data[s->getp++]) << 56;
 474         q |= ((uint64_t)s->data[s->getp++]) << 48;
 475         q |= ((uint64_t)s->data[s->getp++]) << 40;
 476         q |= ((uint64_t)s->data[s->getp++]) << 32;
 477         q |= ((uint64_t)s->data[s->getp++]) << 24;
 478         q |= ((uint64_t)s->data[s->getp++]) << 16;
 479         q |= ((uint64_t)s->data[s->getp++]) << 8;
 480         q |= ((uint64_t)s->data[s->getp++]);
 481
 482         return q;
 483 }
 484
 485 /* Get next long word from the stream. */
 486 u_int32_t stream_get_ipv4(struct stream *s)
 487 {
 488         u_int32_t l;
 489
 490         STREAM_VERIFY_SANE(s);
 491
 492         if (STREAM_READABLE(s) < sizeof(u_int32_t)) {
 493                 STREAM_BOUND_WARN(s, "get ipv4");
 494                 return 0;
 495         }
 496
 497         memcpy(&l, s->data + s->getp, sizeof(u_int32_t));
 498         s->getp += sizeof(u_int32_t);
 499
 500         return l;
 501 }
 502
 503 float stream_getf(struct stream *s)
 504 {
 505         union {
 506                 float r;
 507                 uint32_t d;
 508         } u;
 509         u.d = stream_getl(s);
 510         return u.r;
 511 }
 512
 513 double stream_getd(struct stream *s)
 514 {
 515         union {
 516                 double r;
 517                 uint64_t d;
 518         } u;
 519         u.d = stream_getq(s);
 520         return u.r;
 521 }
 522
 523 /* Copy to source to stream.
 524  *
 525  * XXX: This uses CHECK_SIZE and hence has funny semantics -> Size will wrap
 526  * around. This should be fixed once the stream updates are working.
 527  *
 528  * stream_write() is saner
 529  */
 530 void stream_put(struct stream *s, const void *src, size_t size)
 531 {
 532
 533         /* XXX: CHECK_SIZE has strange semantics. It should be deprecated */
 534         CHECK_SIZE(s, size);
 535
 536         STREAM_VERIFY_SANE(s);
 537
 538         if (STREAM_WRITEABLE(s) < size) {
 539                 STREAM_BOUND_WARN(s, "put");
 540                 return;
 541         }
 542
 543         if (src)
 544                 memcpy(s->data + s->endp, src, size);
 545         else
 546                 memset(s->data + s->endp, 0, size);
 547
 548         s->endp += size;
 549 }
 550
 551 /* Put character to the stream. */
 552 int stream_putc(struct stream *s, u_char c)
 553 {
 554         STREAM_VERIFY_SANE(s);
 555
 556         if (STREAM_WRITEABLE(s) < sizeof(u_char)) {
 557                 STREAM_BOUND_WARN(s, "put");
 558                 return 0;
 559         }
 560
 561         s->data[s->endp++] = c;
 562         return sizeof(u_char);
 563 }
 564
 565 /* Put word to the stream. */
 566 int stream_putw(struct stream *s, u_int16_t w)
 567 {
 568         STREAM_VERIFY_SANE(s);
 569
 570         if (STREAM_WRITEABLE(s) < sizeof(u_int16_t)) {
 571                 STREAM_BOUND_WARN(s, "put");
 572                 return 0;
 573         }
 574
 575         s->data[s->endp++] = (u_char)(w >> 8);
 576         s->data[s->endp++] = (u_char)w;
 577
 578         return 2;
 579 }
 580
 581 /* Put long word to the stream. */
 582 int stream_put3(struct stream *s, u_int32_t l)
 583 {
 584         STREAM_VERIFY_SANE(s);
 585
 586         if (STREAM_WRITEABLE(s) < 3) {
 587                 STREAM_BOUND_WARN(s, "put");
 588                 return 0;
 589         }
 590
 591         s->data[s->endp++] = (u_char)(l >> 16);
 592         s->data[s->endp++] = (u_char)(l >> 8);
 593         s->data[s->endp++] = (u_char)l;
 594
 595         return 3;
 596 }
 597
 598 /* Put long word to the stream. */
 599 int stream_putl(struct stream *s, u_int32_t l)
 600 {
 601         STREAM_VERIFY_SANE(s);
 602
 603         if (STREAM_WRITEABLE(s) < sizeof(u_int32_t)) {
 604                 STREAM_BOUND_WARN(s, "put");
 605                 return 0;
 606         }
 607
 608         s->data[s->endp++] = (u_char)(l >> 24);
 609         s->data[s->endp++] = (u_char)(l >> 16);
 610         s->data[s->endp++] = (u_char)(l >> 8);
 611         s->data[s->endp++] = (u_char)l;
 612
 613         return 4;
 614 }
 615
 616 /* Put quad word to the stream. */
 617 int stream_putq(struct stream *s, uint64_t q)
 618 {
 619         STREAM_VERIFY_SANE(s);
 620
 621         if (STREAM_WRITEABLE(s) < sizeof(uint64_t)) {
 622                 STREAM_BOUND_WARN(s, "put quad");
 623                 return 0;
 624         }
 625
 626         s->data[s->endp++] = (u_char)(q >> 56);
 627         s->data[s->endp++] = (u_char)(q >> 48);
 628         s->data[s->endp++] = (u_char)(q >> 40);
 629         s->data[s->endp++] = (u_char)(q >> 32);
 630         s->data[s->endp++] = (u_char)(q >> 24);
 631         s->data[s->endp++] = (u_char)(q >> 16);
 632         s->data[s->endp++] = (u_char)(q >> 8);
 633         s->data[s->endp++] = (u_char)q;
 634
 635         return 8;
 636 }
 637
 638 int stream_putf(struct stream *s, float f)
 639 {
 640         union {
 641                 float i;
 642                 uint32_t o;
 643         } u;
 644         u.i = f;
 645         return stream_putl(s, u.o);
 646 }
 647
 648 int stream_putd(struct stream *s, double d)
 649 {
 650         union {
 651                 double i;
 652                 uint64_t o;
 653         } u;
 654         u.i = d;
 655         return stream_putq(s, u.o);
 656 }
 657
 658 int stream_putc_at(struct stream *s, size_t putp, u_char c)
 659 {
 660         STREAM_VERIFY_SANE(s);
 661
 662         if (!PUT_AT_VALID(s, putp + sizeof(u_char))) {
 663                 STREAM_BOUND_WARN(s, "put");
 664                 return 0;
 665         }
 666
 667         s->data[putp] = c;
 668
 669         return 1;
 670 }
 671
 672 int stream_putw_at(struct stream *s, size_t putp, u_int16_t w)
 673 {
 674         STREAM_VERIFY_SANE(s);
 675
 676         if (!PUT_AT_VALID(s, putp + sizeof(u_int16_t))) {
 677                 STREAM_BOUND_WARN(s, "put");
 678                 return 0;
 679         }
 680
 681         s->data[putp] = (u_char)(w >> 8);
 682         s->data[putp + 1] = (u_char)w;
 683
 684         return 2;
 685 }
 686
 687 int stream_put3_at(struct stream *s, size_t putp, u_int32_t l)
 688 {
 689         STREAM_VERIFY_SANE(s);
 690
 691         if (!PUT_AT_VALID(s, putp + 3)) {
 692                 STREAM_BOUND_WARN(s, "put");
 693                 return 0;
 694         }
 695         s->data[putp] = (u_char)(l >> 16);
 696         s->data[putp + 1] = (u_char)(l >> 8);
 697         s->data[putp + 2] = (u_char)l;
 698
 699         return 3;
 700 }
 701
 702 int stream_putl_at(struct stream *s, size_t putp, u_int32_t l)
 703 {
 704         STREAM_VERIFY_SANE(s);
 705
 706         if (!PUT_AT_VALID(s, putp + sizeof(u_int32_t))) {
 707                 STREAM_BOUND_WARN(s, "put");
 708                 return 0;
 709         }
 710         s->data[putp] = (u_char)(l >> 24);
 711         s->data[putp + 1] = (u_char)(l >> 16);
 712         s->data[putp + 2] = (u_char)(l >> 8);
 713         s->data[putp + 3] = (u_char)l;
 714
 715         return 4;
 716 }
 717
 718 int stream_putq_at(struct stream *s, size_t putp, uint64_t q)
 719 {
 720         STREAM_VERIFY_SANE(s);
 721
 722         if (!PUT_AT_VALID(s, putp + sizeof(uint64_t))) {
 723                 STREAM_BOUND_WARN(s, "put");
 724                 return 0;
 725         }
 726         s->data[putp] = (u_char)(q >> 56);
 727         s->data[putp + 1] = (u_char)(q >> 48);
 728         s->data[putp + 2] = (u_char)(q >> 40);
 729         s->data[putp + 3] = (u_char)(q >> 32);
 730         s->data[putp + 4] = (u_char)(q >> 24);
 731         s->data[putp + 5] = (u_char)(q >> 16);
 732         s->data[putp + 6] = (u_char)(q >> 8);
 733         s->data[putp + 7] = (u_char)q;
 734
 735         return 8;
 736 }
 737
 738 /* Put long word to the stream. */
 739 int stream_put_ipv4(struct stream *s, u_int32_t l)
 740 {
 741         STREAM_VERIFY_SANE(s);
 742
 743         if (STREAM_WRITEABLE(s) < sizeof(u_int32_t)) {
 744                 STREAM_BOUND_WARN(s, "put");
 745                 return 0;
 746         }
 747         memcpy(s->data + s->endp, &l, sizeof(u_int32_t));
 748         s->endp += sizeof(u_int32_t);
 749
 750         return sizeof(u_int32_t);
 751 }
 752
 753 /* Put long word to the stream. */
 754 int stream_put_in_addr(struct stream *s, struct in_addr *addr)
 755 {
 756         STREAM_VERIFY_SANE(s);
 757
 758         if (STREAM_WRITEABLE(s) < sizeof(u_int32_t)) {
 759                 STREAM_BOUND_WARN(s, "put");
 760                 return 0;
 761         }
 762
 763         memcpy(s->data + s->endp, addr, sizeof(u_int32_t));
 764         s->endp += sizeof(u_int32_t);
 765
 766         return sizeof(u_int32_t);
 767 }
 768
 769 /* Put in_addr at location in the stream. */
 770 int stream_put_in_addr_at(struct stream *s, size_t putp, struct in_addr *addr)
 771 {
 772         STREAM_VERIFY_SANE(s);
 773
 774         if (!PUT_AT_VALID(s, putp + 4)) {
 775                 STREAM_BOUND_WARN(s, "put");
 776                 return 0;
 777         }
 778
 779         memcpy(&s->data[putp], addr, 4);
 780         return 4;
 781 }
 782
 783 /* Put in6_addr at location in the stream. */
 784 int stream_put_in6_addr_at(struct stream *s, size_t putp, struct in6_addr *addr)
 785 {
 786         STREAM_VERIFY_SANE(s);
 787
 788         if (!PUT_AT_VALID(s, putp + 16)) {
 789                 STREAM_BOUND_WARN(s, "put");
 790                 return 0;
 791         }
 792
 793         memcpy(&s->data[putp], addr, 16);
 794         return 16;
 795 }
 796
 797 /* Put prefix by nlri type format. */
 798 int stream_put_prefix_addpath(struct stream *s, struct prefix *p,
 799                               int addpath_encode, u_int32_t addpath_tx_id)
 800 {
 801         size_t psize;
 802         size_t psize_with_addpath;
 803
 804         STREAM_VERIFY_SANE(s);
 805
 806         psize = PSIZE(p->prefixlen);
 807
 808         if (addpath_encode)
 809                 psize_with_addpath = psize + 4;
 810         else
 811                 psize_with_addpath = psize;
 812
 813         if (STREAM_WRITEABLE(s) < (psize_with_addpath + sizeof(u_char))) {
 814                 STREAM_BOUND_WARN(s, "put");
 815                 return 0;
 816         }
 817
 818         if (addpath_encode) {
 819                 s->data[s->endp++] = (u_char)(addpath_tx_id >> 24);
 820                 s->data[s->endp++] = (u_char)(addpath_tx_id >> 16);
 821                 s->data[s->endp++] = (u_char)(addpath_tx_id >> 8);
 822                 s->data[s->endp++] = (u_char)addpath_tx_id;
 823         }
 824
 825         s->data[s->endp++] = p->prefixlen;
 826         memcpy(s->data + s->endp, &p->u.prefix, psize);
 827         s->endp += psize;
 828
 829         return psize;
 830 }
 831
 832 int stream_put_prefix(struct stream *s, struct prefix *p)
 833 {
 834         return stream_put_prefix_addpath(s, p, 0, 0);
 835 }
 836
 837
 838 /* Read size from fd. */
 839 int stream_read(struct stream *s, int fd, size_t size)
 840 {
 841         int nbytes;
 842
 843         STREAM_VERIFY_SANE(s);
 844
 845         if (STREAM_WRITEABLE(s) < size) {
 846                 STREAM_BOUND_WARN(s, "put");
 847                 return 0;
 848         }
 849
 850         nbytes = readn(fd, s->data + s->endp, size);
 851
 852         if (nbytes > 0)
 853                 s->endp += nbytes;
 854
 855         return nbytes;
 856 }
 857
 858 ssize_t stream_read_try(struct stream *s, int fd, size_t size)
 859 {
 860         ssize_t nbytes;
 861
 862         STREAM_VERIFY_SANE(s);
 863
 864         if (STREAM_WRITEABLE(s) < size) {
 865                 STREAM_BOUND_WARN(s, "put");
 866                 /* Fatal (not transient) error, since retrying will not help
 867                    (stream is too small to contain the desired data). */
 868                 return -1;
 869         }
 870
 871         if ((nbytes = read(fd, s->data + s->endp, size)) >= 0) {
 872                 s->endp += nbytes;
 873                 return nbytes;
 874         }
 875         /* Error: was it transient (return -2) or fatal (return -1)? */
 876         if (ERRNO_IO_RETRY(errno))
 877                 return -2;
 878         zlog_warn("%s: read failed on fd %d: %s", __func__, fd,
 879                   safe_strerror(errno));
 880         return -1;
 881 }
 882
 883 /* Read up to size bytes into the stream from the fd, using recvmsgfrom
 884  * whose arguments match the remaining arguments to this function
 885  */
 886 ssize_t stream_recvfrom(struct stream *s, int fd, size_t size, int flags,
 887                         struct sockaddr *from, socklen_t *fromlen)
 888 {
 889         ssize_t nbytes;
 890
 891         STREAM_VERIFY_SANE(s);
 892
 893         if (STREAM_WRITEABLE(s) < size) {
 894                 STREAM_BOUND_WARN(s, "put");
 895                 /* Fatal (not transient) error, since retrying will not help
 896                    (stream is too small to contain the desired data). */
 897                 return -1;
 898         }
 899
 900         if ((nbytes = recvfrom(fd, s->data + s->endp, size, flags, from,
 901                                fromlen))
 902             >= 0) {
 903                 s->endp += nbytes;
 904                 return nbytes;
 905         }
 906         /* Error: was it transient (return -2) or fatal (return -1)? */
 907         if (ERRNO_IO_RETRY(errno))
 908                 return -2;
 909         zlog_warn("%s: read failed on fd %d: %s", __func__, fd,
 910                   safe_strerror(errno));
 911         return -1;
 912 }
 913
 914 /* Read up to smaller of size or SIZE_REMAIN() bytes to the stream, starting
 915  * from endp.
 916  * First iovec will be used to receive the data.
 917  * Stream need not be empty.
 918  */
 919 ssize_t stream_recvmsg(struct stream *s, int fd, struct msghdr *msgh, int flags,
 920                        size_t size)
 921 {
 922         int nbytes;
 923         struct iovec *iov;
 924
 925         STREAM_VERIFY_SANE(s);
 926         assert(msgh->msg_iovlen > 0);
 927
 928         if (STREAM_WRITEABLE(s) < size) {
 929                 STREAM_BOUND_WARN(s, "put");
 930                 /* This is a logic error in the calling code: the stream is too
 931                    small
 932                    to hold the desired data! */
 933                 return -1;
 934         }
 935
 936         iov = &(msgh->msg_iov[0]);
 937         iov->iov_base = (s->data + s->endp);
 938         iov->iov_len = size;
 939
 940         nbytes = recvmsg(fd, msgh, flags);
 941
 942         if (nbytes > 0)
 943                 s->endp += nbytes;
 944
 945         return nbytes;
 946 }
 947
 948 /* Write data to buffer. */
 949 size_t stream_write(struct stream *s, const void *ptr, size_t size)
 950 {
 951
 952         CHECK_SIZE(s, size);
 953
 954         STREAM_VERIFY_SANE(s);
 955
 956         if (STREAM_WRITEABLE(s) < size) {
 957                 STREAM_BOUND_WARN(s, "put");
 958                 return 0;
 959         }
 960
 961         memcpy(s->data + s->endp, ptr, size);
 962         s->endp += size;
 963
 964         return size;
 965 }
 966
 967 /* Return current read pointer.
 968  * DEPRECATED!
 969  * Use stream_get_pnt_to if you must, but decoding streams properly
 970  * is preferred
 971  */
 972 u_char *stream_pnt(struct stream *s)
 973 {
 974         STREAM_VERIFY_SANE(s);
 975         return s->data + s->getp;
 976 }
 977
 978 /* Check does this stream empty? */
 979 int stream_empty(struct stream *s)
 980 {
 981         STREAM_VERIFY_SANE(s);
 982
 983         return (s->endp == 0);
 984 }
 985
 986 /* Reset stream. */
 987 void stream_reset(struct stream *s)
 988 {
 989         STREAM_VERIFY_SANE(s);
 990
 991         s->getp = s->endp = 0;
 992 }
 993
 994 /* Write stream contens to the file discriptor. */
 995 int stream_flush(struct stream *s, int fd)
 996 {
 997         int nbytes;
 998
 999         STREAM_VERIFY_SANE(s);
1000
1001         nbytes = write(fd, s->data + s->getp, s->endp - s->getp);
1002
1003         return nbytes;
1004 }
1005
1006 /* Stream first in first out queue. */
1007
1008 struct stream_fifo *stream_fifo_new(void)
1009 {
1010         struct stream_fifo *new;
1011
1012         new = XCALLOC(MTYPE_STREAM_FIFO, sizeof(struct stream_fifo));
1013         return new;
1014 }
1015
1016 /* Add new stream to fifo. */
1017 void stream_fifo_push(struct stream_fifo *fifo, struct stream *s)
1018 {
1019         if (fifo->tail)
1020                 fifo->tail->next = s;
1021         else
1022                 fifo->head = s;
1023
1024         fifo->tail = s;
1025
1026         fifo->count++;
1027 }
1028
1029 /* Delete first stream from fifo. */
1030 struct stream *stream_fifo_pop(struct stream_fifo *fifo)
1031 {
1032         struct stream *s;
1033
1034         s = fifo->head;
1035
1036         if (s) {
1037                 fifo->head = s->next;
1038
1039                 if (fifo->head == NULL)
1040                         fifo->tail = NULL;
1041
1042                 fifo->count--;
1043         }
1044
1045         return s;
1046 }
1047
1048 /* Return first fifo entry. */
1049 struct stream *stream_fifo_head(struct stream_fifo *fifo)
1050 {
1051         return fifo->head;
1052 }
1053
1054 void stream_fifo_clean(struct stream_fifo *fifo)
1055 {
1056         struct stream *s;
1057         struct stream *next;
1058
1059         for (s = fifo->head; s; s = next) {
1060                 next = s->next;
1061                 stream_free(s);
1062         }
1063         fifo->head = fifo->tail = NULL;
1064         fifo->count = 0;
1065 }
1066
1067 void stream_fifo_free(struct stream_fifo *fifo)
1068 {
1069         stream_fifo_clean(fifo);
1070         XFREE(MTYPE_STREAM_FIFO, fifo);
1071 }