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