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 /* 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, getp: %lu, endp: %lu\n", \
  56              (void *)(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 struct stream *
 158 stream_dupcat (struct stream *s1, struct stream *s2, size_t offset)
 159 {
 160   struct stream *new;
 161
 162   STREAM_VERIFY_SANE (s1);
 163   STREAM_VERIFY_SANE (s2);
 164
 165   if ( (new = stream_new (s1->endp + s2->endp)) == NULL)
 166     return NULL;
 167
 168   memcpy (new->data, s1->data, offset);
 169   memcpy (new->data + offset, s2->data, s2->endp);
 170   memcpy (new->data + offset + s2->endp, s1->data + offset,
 171           (s1->endp - offset));
 172   new->endp = s1->endp + s2->endp;
 173   return new;
 174 }
 175
 176 size_t
 177 stream_resize (struct stream *s, size_t newsize)
 178 {
 179   u_char *newdata;
 180   STREAM_VERIFY_SANE (s);
 181
 182   newdata = XREALLOC (MTYPE_STREAM_DATA, s->data, newsize);
 183
 184   if (newdata == NULL)
 185     return s->size;
 186
 187   s->data = newdata;
 188   s->size = newsize;
 189
 190   if (s->endp > s->size)
 191     s->endp = s->size;
 192   if (s->getp > s->endp)
 193     s->getp = s->endp;
 194
 195   STREAM_VERIFY_SANE (s);
 196
 197   return s->size;
 198 }
 199
 200 size_t
 201 stream_get_getp (struct stream *s)
 202 {
 203   STREAM_VERIFY_SANE(s);
 204   return s->getp;
 205 }
 206
 207 size_t
 208 stream_get_endp (struct stream *s)
 209 {
 210   STREAM_VERIFY_SANE(s);
 211   return s->endp;
 212 }
 213
 214 size_t
 215 stream_get_size (struct stream *s)
 216 {
 217   STREAM_VERIFY_SANE(s);
 218   return s->size;
 219 }
 220
 221 /* Stream structre' stream pointer related functions.  */
 222 void
 223 stream_set_getp (struct stream *s, size_t pos)
 224 {
 225   STREAM_VERIFY_SANE(s);
 226
 227   if (!GETP_VALID (s, pos))
 228     {
 229       STREAM_BOUND_WARN (s, "set getp");
 230       pos = s->endp;
 231     }
 232
 233   s->getp = pos;
 234 }
 235
 236 void
 237 stream_set_endp (struct stream *s, size_t pos)
 238 {
 239   STREAM_VERIFY_SANE(s);
 240
 241   if (!ENDP_VALID(s, pos))
 242     {
 243       STREAM_BOUND_WARN (s, "set endp");
 244       return;
 245     }
 246
 247   /*
 248    * Make sure the current read pointer is not beyond the new endp.
 249    */
 250   if (s->getp > pos)
 251     {
 252       STREAM_BOUND_WARN(s, "set endp");
 253       return;
 254     }
 255
 256   s->endp = pos;
 257   STREAM_VERIFY_SANE(s);
 258 }
 259
 260 /* Forward pointer. */
 261 void
 262 stream_forward_getp (struct stream *s, size_t size)
 263 {
 264   STREAM_VERIFY_SANE(s);
 265
 266   if (!GETP_VALID (s, s->getp + size))
 267     {
 268       STREAM_BOUND_WARN (s, "seek getp");
 269       return;
 270     }
 271
 272   s->getp += size;
 273 }
 274
 275 void
 276 stream_forward_endp (struct stream *s, size_t size)
 277 {
 278   STREAM_VERIFY_SANE(s);
 279
 280   if (!ENDP_VALID (s, s->endp + size))
 281     {
 282       STREAM_BOUND_WARN (s, "seek endp");
 283       return;
 284     }
 285
 286   s->endp += size;
 287 }
 288
 289 /* Copy from stream to destination. */
 290 void
 291 stream_get (void *dst, struct stream *s, size_t size)
 292 {
 293   STREAM_VERIFY_SANE(s);
 294
 295   if (STREAM_READABLE(s) < size)
 296     {
 297       STREAM_BOUND_WARN (s, "get");
 298       return;
 299     }
 300
 301   memcpy (dst, s->data + s->getp, size);
 302   s->getp += size;
 303 }
 304
 305 /* Get next character from the stream. */
 306 u_char
 307 stream_getc (struct stream *s)
 308 {
 309   u_char c;
 310
 311   STREAM_VERIFY_SANE (s);
 312
 313   if (STREAM_READABLE(s) < sizeof (u_char))
 314     {
 315       STREAM_BOUND_WARN (s, "get char");
 316       return 0;
 317     }
 318   c = s->data[s->getp++];
 319
 320   return c;
 321 }
 322
 323 /* Get next character from the stream. */
 324 u_char
 325 stream_getc_from (struct stream *s, size_t from)
 326 {
 327   u_char c;
 328
 329   STREAM_VERIFY_SANE(s);
 330
 331   if (!GETP_VALID (s, from + sizeof (u_char)))
 332     {
 333       STREAM_BOUND_WARN (s, "get char");
 334       return 0;
 335     }
 336
 337   c = s->data[from];
 338
 339   return c;
 340 }
 341
 342 /* Get next word from the stream. */
 343 u_int16_t
 344 stream_getw (struct stream *s)
 345 {
 346   u_int16_t w;
 347
 348   STREAM_VERIFY_SANE (s);
 349
 350   if (STREAM_READABLE (s) < sizeof (u_int16_t))
 351     {
 352       STREAM_BOUND_WARN (s, "get ");
 353       return 0;
 354     }
 355
 356   w = s->data[s->getp++] << 8;
 357   w |= s->data[s->getp++];
 358
 359   return w;
 360 }
 361
 362 /* Get next word from the stream. */
 363 u_int16_t
 364 stream_getw_from (struct stream *s, size_t from)
 365 {
 366   u_int16_t w;
 367
 368   STREAM_VERIFY_SANE(s);
 369
 370   if (!GETP_VALID (s, from + sizeof (u_int16_t)))
 371     {
 372       STREAM_BOUND_WARN (s, "get ");
 373       return 0;
 374     }
 375
 376   w = s->data[from++] << 8;
 377   w |= s->data[from];
 378
 379   return w;
 380 }
 381
 382 /* Get next long word from the stream. */
 383 u_int32_t
 384 stream_getl_from (struct stream *s, size_t from)
 385 {
 386   u_int32_t l;
 387
 388   STREAM_VERIFY_SANE(s);
 389
 390   if (!GETP_VALID (s, from + sizeof (u_int32_t)))
 391     {
 392       STREAM_BOUND_WARN (s, "get long");
 393       return 0;
 394     }
 395
 396   l  = s->data[from++] << 24;
 397   l |= s->data[from++] << 16;
 398   l |= s->data[from++] << 8;
 399   l |= s->data[from];
 400
 401   return l;
 402 }
 403
 404 u_int32_t
 405 stream_getl (struct stream *s)
 406 {
 407   u_int32_t l;
 408
 409   STREAM_VERIFY_SANE(s);
 410
 411   if (STREAM_READABLE (s) < sizeof (u_int32_t))
 412     {
 413       STREAM_BOUND_WARN (s, "get long");
 414       return 0;
 415     }
 416
 417   l  = s->data[s->getp++] << 24;
 418   l |= s->data[s->getp++] << 16;
 419   l |= s->data[s->getp++] << 8;
 420   l |= s->data[s->getp++];
 421
 422   return l;
 423 }
 424
 425 /* Get next quad word from the stream. */
 426 uint64_t
 427 stream_getq_from (struct stream *s, size_t from)
 428 {
 429   uint64_t q;
 430
 431   STREAM_VERIFY_SANE(s);
 432
 433   if (!GETP_VALID (s, from + sizeof (uint64_t)))
 434     {
 435       STREAM_BOUND_WARN (s, "get quad");
 436       return 0;
 437     }
 438
 439   q  = ((uint64_t) s->data[from++]) << 56;
 440   q |= ((uint64_t) s->data[from++]) << 48;
 441   q |= ((uint64_t) s->data[from++]) << 40;
 442   q |= ((uint64_t) s->data[from++]) << 32;
 443   q |= ((uint64_t) s->data[from++]) << 24;
 444   q |= ((uint64_t) s->data[from++]) << 16;
 445   q |= ((uint64_t) s->data[from++]) << 8;
 446   q |= ((uint64_t) s->data[from++]);
 447
 448   return q;
 449 }
 450
 451 uint64_t
 452 stream_getq (struct stream *s)
 453 {
 454   uint64_t q;
 455
 456   STREAM_VERIFY_SANE(s);
 457
 458   if (STREAM_READABLE (s) < sizeof (uint64_t))
 459     {
 460       STREAM_BOUND_WARN (s, "get quad");
 461       return 0;
 462     }
 463
 464   q  = ((uint64_t) s->data[s->getp++]) << 56;
 465   q |= ((uint64_t) s->data[s->getp++]) << 48;
 466   q |= ((uint64_t) s->data[s->getp++]) << 40;
 467   q |= ((uint64_t) s->data[s->getp++]) << 32;
 468   q |= ((uint64_t) s->data[s->getp++]) << 24;
 469   q |= ((uint64_t) s->data[s->getp++]) << 16;
 470   q |= ((uint64_t) s->data[s->getp++]) << 8;
 471   q |= ((uint64_t) s->data[s->getp++]);
 472
 473   return q;
 474 }
 475
 476 /* Get next long word from the stream. */
 477 u_int32_t
 478 stream_get_ipv4 (struct stream *s)
 479 {
 480   u_int32_t l;
 481
 482   STREAM_VERIFY_SANE(s);
 483
 484   if (STREAM_READABLE (s) < sizeof(u_int32_t))
 485     {
 486       STREAM_BOUND_WARN (s, "get ipv4");
 487       return 0;
 488     }
 489
 490   memcpy (&l, s->data + s->getp, sizeof(u_int32_t));
 491   s->getp += sizeof(u_int32_t);
 492
 493   return l;
 494 }
 495
 496 float
 497 stream_getf (struct stream *s)
 498 {
 499 #if !defined(__STDC_IEC_559__) && __GCC_IEC_559 < 1
 500 #warning "Unknown floating-point format, __func__ may be wrong"
 501 #endif
 502 /* we assume 'float' is in the single precision IEC 60559 binary
 503    format, in host byte order */
 504   union {
 505     float r;
 506     uint32_t d;
 507   } u;
 508   u.d = stream_getl (s);
 509   return u.r;
 510 }
 511
 512 double
 513 stream_getd (struct stream *s)
 514 {
 515 #if !defined(__STDC_IEC_559__) && __GCC_IEC_559 < 1
 516 #warning "Unknown floating-point format, __func__ may be wrong"
 517 #endif
 518   union {
 519     double r;
 520     uint64_t d;
 521   } u;
 522   u.d = stream_getq (s);
 523   return u.r;
 524 }
 525
 526 /* Copy to source to stream.
 527  *
 528  * XXX: This uses CHECK_SIZE and hence has funny semantics -> Size will wrap
 529  * around. This should be fixed once the stream updates are working.
 530  *
 531  * stream_write() is saner
 532  */
 533 void
 534 stream_put (struct stream *s, const void *src, size_t size)
 535 {
 536
 537   /* XXX: CHECK_SIZE has strange semantics. It should be deprecated */
 538   CHECK_SIZE(s, size);
 539
 540   STREAM_VERIFY_SANE(s);
 541
 542   if (STREAM_WRITEABLE (s) < size)
 543     {
 544       STREAM_BOUND_WARN (s, "put");
 545       return;
 546     }
 547
 548   if (src)
 549     memcpy (s->data + s->endp, src, size);
 550   else
 551     memset (s->data + s->endp, 0, size);
 552
 553   s->endp += size;
 554 }
 555
 556 /* Put character to the stream. */
 557 int
 558 stream_putc (struct stream *s, u_char c)
 559 {
 560   STREAM_VERIFY_SANE(s);
 561
 562   if (STREAM_WRITEABLE (s) < sizeof(u_char))
 563     {
 564       STREAM_BOUND_WARN (s, "put");
 565       return 0;
 566     }
 567
 568   s->data[s->endp++] = c;
 569   return sizeof (u_char);
 570 }
 571
 572 /* Put word to the stream. */
 573 int
 574 stream_putw (struct stream *s, u_int16_t w)
 575 {
 576   STREAM_VERIFY_SANE (s);
 577
 578   if (STREAM_WRITEABLE (s) < sizeof (u_int16_t))
 579     {
 580       STREAM_BOUND_WARN (s, "put");
 581       return 0;
 582     }
 583
 584   s->data[s->endp++] = (u_char)(w >>  8);
 585   s->data[s->endp++] = (u_char) w;
 586
 587   return 2;
 588 }
 589
 590 /* Put long word to the stream. */
 591 int
 592 stream_putl (struct stream *s, u_int32_t l)
 593 {
 594   STREAM_VERIFY_SANE (s);
 595
 596   if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
 597     {
 598       STREAM_BOUND_WARN (s, "put");
 599       return 0;
 600     }
 601
 602   s->data[s->endp++] = (u_char)(l >> 24);
 603   s->data[s->endp++] = (u_char)(l >> 16);
 604   s->data[s->endp++] = (u_char)(l >>  8);
 605   s->data[s->endp++] = (u_char)l;
 606
 607   return 4;
 608 }
 609
 610 /* Put quad word to the stream. */
 611 int
 612 stream_putq (struct stream *s, uint64_t q)
 613 {
 614   STREAM_VERIFY_SANE (s);
 615
 616   if (STREAM_WRITEABLE (s) < sizeof (uint64_t))
 617     {
 618       STREAM_BOUND_WARN (s, "put quad");
 619       return 0;
 620     }
 621
 622   s->data[s->endp++] = (u_char)(q >> 56);
 623   s->data[s->endp++] = (u_char)(q >> 48);
 624   s->data[s->endp++] = (u_char)(q >> 40);
 625   s->data[s->endp++] = (u_char)(q >> 32);
 626   s->data[s->endp++] = (u_char)(q >> 24);
 627   s->data[s->endp++] = (u_char)(q >> 16);
 628   s->data[s->endp++] = (u_char)(q >>  8);
 629   s->data[s->endp++] = (u_char)q;
 630
 631   return 8;
 632 }
 633
 634 int
 635 stream_putf (struct stream *s, float f)
 636 {
 637 #if !defined(__STDC_IEC_559__) && __GCC_IEC_559 < 1
 638 #warning "Unknown floating-point format, __func__ may be wrong"
 639 #endif
 640
 641 /* we can safely assume 'float' is in the single precision
 642    IEC 60559 binary format in host order */
 643   union {
 644     float i;
 645     uint32_t o;
 646   } u;
 647   u.i = f;
 648   return stream_putl (s, u.o);
 649 }
 650
 651 int
 652 stream_putd (struct stream *s, double d)
 653 {
 654 #if !defined(__STDC_IEC_559__) && __GCC_IEC_559 < 1
 655 #warning "Unknown floating-point format, __func__ may be wrong"
 656 #endif
 657   union {
 658     double i;
 659     uint64_t o;
 660   } u;
 661   u.i = d;
 662   return stream_putq (s, u.o);
 663 }
 664
 665 int
 666 stream_putc_at (struct stream *s, size_t putp, u_char c)
 667 {
 668   STREAM_VERIFY_SANE(s);
 669
 670   if (!PUT_AT_VALID (s, putp + sizeof (u_char)))
 671     {
 672       STREAM_BOUND_WARN (s, "put");
 673       return 0;
 674     }
 675
 676   s->data[putp] = c;
 677
 678   return 1;
 679 }
 680
 681 int
 682 stream_putw_at (struct stream *s, size_t putp, u_int16_t w)
 683 {
 684   STREAM_VERIFY_SANE(s);
 685
 686   if (!PUT_AT_VALID (s, putp + sizeof (u_int16_t)))
 687     {
 688       STREAM_BOUND_WARN (s, "put");
 689       return 0;
 690     }
 691
 692   s->data[putp] = (u_char)(w >>  8);
 693   s->data[putp + 1] = (u_char) w;
 694
 695   return 2;
 696 }
 697
 698 int
 699 stream_putl_at (struct stream *s, size_t putp, u_int32_t l)
 700 {
 701   STREAM_VERIFY_SANE(s);
 702
 703   if (!PUT_AT_VALID (s, putp + sizeof (u_int32_t)))
 704     {
 705       STREAM_BOUND_WARN (s, "put");
 706       return 0;
 707     }
 708   s->data[putp] = (u_char)(l >> 24);
 709   s->data[putp + 1] = (u_char)(l >> 16);
 710   s->data[putp + 2] = (u_char)(l >>  8);
 711   s->data[putp + 3] = (u_char)l;
 712
 713   return 4;
 714 }
 715
 716 int
 717 stream_putq_at (struct stream *s, size_t putp, uint64_t q)
 718 {
 719   STREAM_VERIFY_SANE(s);
 720
 721   if (!PUT_AT_VALID (s, putp + sizeof (uint64_t)))
 722     {
 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
 740 stream_put_ipv4 (struct stream *s, u_int32_t l)
 741 {
 742   STREAM_VERIFY_SANE(s);
 743
 744   if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
 745     {
 746       STREAM_BOUND_WARN (s, "put");
 747       return 0;
 748     }
 749   memcpy (s->data + s->endp, &l, sizeof (u_int32_t));
 750   s->endp += sizeof (u_int32_t);
 751
 752   return sizeof (u_int32_t);
 753 }
 754
 755 /* Put long word to the stream. */
 756 int
 757 stream_put_in_addr (struct stream *s, struct in_addr *addr)
 758 {
 759   STREAM_VERIFY_SANE(s);
 760
 761   if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
 762     {
 763       STREAM_BOUND_WARN (s, "put");
 764       return 0;
 765     }
 766
 767   memcpy (s->data + s->endp, addr, sizeof (u_int32_t));
 768   s->endp += sizeof (u_int32_t);
 769
 770   return sizeof (u_int32_t);
 771 }
 772
 773 /* Put prefix by nlri type format. */
 774 int
 775 stream_put_prefix (struct stream *s, struct prefix *p)
 776 {
 777   size_t psize;
 778
 779   STREAM_VERIFY_SANE(s);
 780
 781   psize = PSIZE (p->prefixlen);
 782
 783   if (STREAM_WRITEABLE (s) < (psize + sizeof (u_char)))
 784     {
 785       STREAM_BOUND_WARN (s, "put");
 786       return 0;
 787     }
 788
 789   s->data[s->endp++] = p->prefixlen;
 790   memcpy (s->data + s->endp, &p->u.prefix, psize);
 791   s->endp += psize;
 792
 793   return psize;
 794 }
 795
 796 /* Read size from fd. */
 797 int
 798 stream_read (struct stream *s, int fd, size_t size)
 799 {
 800   int nbytes;
 801
 802   STREAM_VERIFY_SANE(s);
 803
 804   if (STREAM_WRITEABLE (s) < size)
 805     {
 806       STREAM_BOUND_WARN (s, "put");
 807       return 0;
 808     }
 809
 810   nbytes = readn (fd, s->data + s->endp, size);
 811
 812   if (nbytes > 0)
 813     s->endp += nbytes;
 814
 815   return nbytes;
 816 }
 817
 818 ssize_t
 819 stream_read_try(struct stream *s, int fd, size_t size)
 820 {
 821   ssize_t nbytes;
 822
 823   STREAM_VERIFY_SANE(s);
 824
 825   if (STREAM_WRITEABLE(s) < size)
 826     {
 827       STREAM_BOUND_WARN (s, "put");
 828       /* Fatal (not transient) error, since retrying will not help
 829          (stream is too small to contain the desired data). */
 830       return -1;
 831     }
 832
 833   if ((nbytes = read(fd, s->data + s->endp, size)) >= 0)
 834     {
 835       s->endp += nbytes;
 836       return nbytes;
 837     }
 838   /* Error: was it transient (return -2) or fatal (return -1)? */
 839   if (ERRNO_IO_RETRY(errno))
 840     return -2;
 841   zlog_warn("%s: read failed on fd %d: %s", __func__, fd, safe_strerror(errno));
 842   return -1;
 843 }
 844
 845 /* Read up to size bytes into the stream from the fd, using recvmsgfrom
 846  * whose arguments match the remaining arguments to this function
 847  */
 848 ssize_t
 849 stream_recvfrom (struct stream *s, int fd, size_t size, int flags,
 850                  struct sockaddr *from, socklen_t *fromlen)
 851 {
 852   ssize_t nbytes;
 853
 854   STREAM_VERIFY_SANE(s);
 855
 856   if (STREAM_WRITEABLE(s) < size)
 857     {
 858       STREAM_BOUND_WARN (s, "put");
 859       /* Fatal (not transient) error, since retrying will not help
 860          (stream is too small to contain the desired data). */
 861       return -1;
 862     }
 863
 864   if ((nbytes = recvfrom (fd, s->data + s->endp, size,
 865                           flags, from, fromlen)) >= 0)
 866     {
 867       s->endp += nbytes;
 868       return nbytes;
 869     }
 870   /* Error: was it transient (return -2) or fatal (return -1)? */
 871   if (ERRNO_IO_RETRY(errno))
 872     return -2;
 873   zlog_warn("%s: read failed on fd %d: %s", __func__, fd, safe_strerror(errno));
 874   return -1;
 875 }
 876
 877 /* Read up to smaller of size or SIZE_REMAIN() bytes to the stream, starting
 878  * from endp.
 879  * First iovec will be used to receive the data.
 880  * Stream need not be empty.
 881  */
 882 ssize_t
 883 stream_recvmsg (struct stream *s, int fd, struct msghdr *msgh, int flags,
 884                 size_t size)
 885 {
 886   int nbytes;
 887   struct iovec *iov;
 888
 889   STREAM_VERIFY_SANE(s);
 890   assert (msgh->msg_iovlen > 0);
 891
 892   if (STREAM_WRITEABLE (s) < size)
 893     {
 894       STREAM_BOUND_WARN (s, "put");
 895       /* This is a logic error in the calling code: the stream is too small
 896          to hold the desired data! */
 897       return -1;
 898     }
 899
 900   iov = &(msgh->msg_iov[0]);
 901   iov->iov_base = (s->data + s->endp);
 902   iov->iov_len = size;
 903
 904   nbytes = recvmsg (fd, msgh, flags);
 905
 906   if (nbytes > 0)
 907     s->endp += nbytes;
 908
 909   return nbytes;
 910 }
 911
 912 /* Write data to buffer. */
 913 size_t
 914 stream_write (struct stream *s, const void *ptr, size_t size)
 915 {
 916
 917   CHECK_SIZE(s, size);
 918
 919   STREAM_VERIFY_SANE(s);
 920
 921   if (STREAM_WRITEABLE (s) < size)
 922     {
 923       STREAM_BOUND_WARN (s, "put");
 924       return 0;
 925     }
 926
 927   memcpy (s->data + s->endp, ptr, size);
 928   s->endp += size;
 929
 930   return size;
 931 }
 932
 933 /* Return current read pointer.
 934  * DEPRECATED!
 935  * Use stream_get_pnt_to if you must, but decoding streams properly
 936  * is preferred
 937  */
 938 u_char *
 939 stream_pnt (struct stream *s)
 940 {
 941   STREAM_VERIFY_SANE(s);
 942   return s->data + s->getp;
 943 }
 944
 945 /* Check does this stream empty? */
 946 int
 947 stream_empty (struct stream *s)
 948 {
 949   STREAM_VERIFY_SANE(s);
 950
 951   return (s->endp == 0);
 952 }
 953
 954 /* Reset stream. */
 955 void
 956 stream_reset (struct stream *s)
 957 {
 958   STREAM_VERIFY_SANE (s);
 959
 960   s->getp = s->endp = 0;
 961 }
 962
 963 /* Discard read data (prior to the getp), and move the unread data
 964  * to the beginning of the stream.
 965  *
 966  * See also stream_fifo_* functions, for another approach to managing
 967  * streams.
 968  */
 969 void
 970 stream_discard (struct stream *s)
 971 {
 972   STREAM_VERIFY_SANE (s);
 973
 974   if (s->getp == 0)
 975     return;
 976
 977   if (s->getp == s->endp)
 978     {
 979       stream_reset (s);
 980       return;
 981     }
 982
 983   s->data = memmove (s->data, s->data + s->getp, s->endp - s->getp);
 984   s->endp -= s->getp;
 985   s->getp = 0;
 986 }
 987
 988 /* Write stream contens to the file discriptor. */
 989 int
 990 stream_flush (struct stream *s, int fd)
 991 {
 992   int nbytes;
 993
 994   STREAM_VERIFY_SANE(s);
 995
 996   nbytes = write (fd, s->data + s->getp, s->endp - s->getp);
 997
 998   return nbytes;
 999 }
1000
1001 /* Stream first in first out queue. */
1002
1003 struct stream_fifo *
1004 stream_fifo_new (void)
1005 {
1006   struct stream_fifo *new;
1007
1008   new = XCALLOC (MTYPE_STREAM_FIFO, sizeof (struct stream_fifo));
1009   return new;
1010 }
1011
1012 /* Add new stream to fifo. */
1013 void
1014 stream_fifo_push (struct stream_fifo *fifo, struct stream *s)
1015 {
1016   if (fifo->tail)
1017     fifo->tail->next = s;
1018   else
1019     fifo->head = s;
1020
1021   fifo->tail = s;
1022
1023   fifo->count++;
1024 }
1025
1026 /* Delete first stream from fifo. */
1027 struct stream *
1028 stream_fifo_pop (struct stream_fifo *fifo)
1029 {
1030   struct stream *s;
1031
1032   s = fifo->head;
1033
1034   if (s)
1035     {
1036       fifo->head = s->next;
1037
1038       if (fifo->head == NULL)
1039         fifo->tail = NULL;
1040
1041       fifo->count--;
1042     }
1043
1044   return s;
1045 }
1046
1047 /* Return first fifo entry. */
1048 struct stream *
1049 stream_fifo_head (struct stream_fifo *fifo)
1050 {
1051   return fifo->head;
1052 }
1053
1054 void
1055 stream_fifo_clean (struct stream_fifo *fifo)
1056 {
1057   struct stream *s;
1058   struct stream *next;
1059
1060   for (s = fifo->head; s; s = next)
1061     {
1062       next = s->next;
1063       stream_free (s);
1064     }
1065   fifo->head = fifo->tail = NULL;
1066   fifo->count = 0;
1067 }
1068
1069 void
1070 stream_fifo_free (struct stream_fifo *fifo)
1071 {
1072   stream_fifo_clean (fifo);
1073   XFREE (MTYPE_STREAM_FIFO, fifo);
1074 }