]>
git.proxmox.com Git - mirror_frr.git/blob - lib/stream.c
0eb790b7530b9c5b1b1d56831a8292e963556a18
3 * Copyright (C) 1999 Kunihiro Ishiguro
5 * This file is part of GNU Zebra.
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
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.
17 * You should have received a copy of the GNU General Public License along
18 * with this program; see the file COPYING; if not, write to the Free Software
19 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
31 DEFINE_MTYPE_STATIC(LIB
, STREAM
, "Stream")
32 DEFINE_MTYPE_STATIC(LIB
, STREAM_DATA
, "Stream data")
33 DEFINE_MTYPE_STATIC(LIB
, STREAM_FIFO
, "Stream FIFO")
35 /* Tests whether a position is valid */
36 #define GETP_VALID(S, G) ((G) <= (S)->endp)
37 #define PUT_AT_VALID(S,G) GETP_VALID(S,G)
38 #define ENDP_VALID(S, E) ((E) <= (S)->size)
40 /* asserting sanity checks. Following must be true before
41 * stream functions are called:
43 * Following must always be true of stream elements
44 * before and after calls to stream functions:
46 * getp <= endp <= size
48 * Note that after a stream function is called following may be true:
49 * if (getp == endp) then stream is no longer readable
50 * if (endp == size) then stream is no longer writeable
52 * It is valid to put to anywhere within the size of the stream, but only
53 * using stream_put..._at() functions.
55 #define STREAM_WARN_OFFSETS(S) \
56 zlog_warn("&(struct stream): %p, size: %lu, getp: %lu, endp: %lu\n", \
57 (void *)(S), (unsigned long)(S)->size, \
58 (unsigned long)(S)->getp, (unsigned long)(S)->endp)
60 #define STREAM_VERIFY_SANE(S) \
62 if (!(GETP_VALID(S, (S)->getp) && ENDP_VALID(S, (S)->endp))) \
63 STREAM_WARN_OFFSETS(S); \
64 assert(GETP_VALID(S, (S)->getp)); \
65 assert(ENDP_VALID(S, (S)->endp)); \
68 #define STREAM_BOUND_WARN(S, WHAT) \
70 zlog_warn("%s: Attempt to %s out of bounds", __func__, \
72 STREAM_WARN_OFFSETS(S); \
76 #define STREAM_BOUND_WARN2(S, WHAT) \
78 zlog_warn("%s: Attempt to %s out of bounds", __func__, \
80 STREAM_WARN_OFFSETS(S); \
83 /* XXX: Deprecated macro: do not use */
84 #define CHECK_SIZE(S, Z) \
86 if (((S)->endp + (Z)) > (S)->size) { \
88 "CHECK_SIZE: truncating requested size %lu\n", \
89 (unsigned long)(Z)); \
90 STREAM_WARN_OFFSETS(S); \
91 (Z) = (S)->size - (S)->endp; \
95 /* Make stream buffer. */
96 struct stream
*stream_new(size_t size
)
102 s
= XCALLOC(MTYPE_STREAM
, sizeof(struct stream
));
107 if ((s
->data
= XMALLOC(MTYPE_STREAM_DATA
, size
)) == NULL
) {
108 XFREE(MTYPE_STREAM
, s
);
117 void stream_free(struct stream
*s
)
122 XFREE(MTYPE_STREAM_DATA
, s
->data
);
123 XFREE(MTYPE_STREAM
, s
);
126 struct stream
*stream_copy(struct stream
*new, struct stream
*src
)
128 STREAM_VERIFY_SANE(src
);
131 assert(STREAM_SIZE(new) >= src
->endp
);
133 new->endp
= src
->endp
;
134 new->getp
= src
->getp
;
136 memcpy(new->data
, src
->data
, src
->endp
);
141 struct stream
*stream_dup(struct stream
*s
)
145 STREAM_VERIFY_SANE(s
);
147 if ((new = stream_new(s
->endp
)) == NULL
)
150 return (stream_copy(new, s
));
153 struct stream
*stream_dupcat(struct stream
*s1
, struct stream
*s2
,
158 STREAM_VERIFY_SANE(s1
);
159 STREAM_VERIFY_SANE(s2
);
161 if ((new = stream_new(s1
->endp
+ s2
->endp
)) == NULL
)
164 memcpy(new->data
, s1
->data
, offset
);
165 memcpy(new->data
+ offset
, s2
->data
, s2
->endp
);
166 memcpy(new->data
+ offset
+ s2
->endp
, s1
->data
+ offset
,
167 (s1
->endp
- offset
));
168 new->endp
= s1
->endp
+ s2
->endp
;
172 size_t stream_resize(struct stream
*s
, size_t newsize
)
175 STREAM_VERIFY_SANE(s
);
177 newdata
= XREALLOC(MTYPE_STREAM_DATA
, s
->data
, newsize
);
185 if (s
->endp
> s
->size
)
187 if (s
->getp
> s
->endp
)
190 STREAM_VERIFY_SANE(s
);
195 size_t stream_get_getp(struct stream
*s
)
197 STREAM_VERIFY_SANE(s
);
201 size_t stream_get_endp(struct stream
*s
)
203 STREAM_VERIFY_SANE(s
);
207 size_t stream_get_size(struct stream
*s
)
209 STREAM_VERIFY_SANE(s
);
213 /* Stream structre' stream pointer related functions. */
214 void stream_set_getp(struct stream
*s
, size_t pos
)
216 STREAM_VERIFY_SANE(s
);
218 if (!GETP_VALID(s
, pos
)) {
219 STREAM_BOUND_WARN(s
, "set getp");
226 void stream_set_endp(struct stream
*s
, size_t pos
)
228 STREAM_VERIFY_SANE(s
);
230 if (!ENDP_VALID(s
, pos
)) {
231 STREAM_BOUND_WARN(s
, "set endp");
236 * Make sure the current read pointer is not beyond the new endp.
239 STREAM_BOUND_WARN(s
, "set endp");
244 STREAM_VERIFY_SANE(s
);
247 /* Forward pointer. */
248 void stream_forward_getp(struct stream
*s
, size_t size
)
250 STREAM_VERIFY_SANE(s
);
252 if (!GETP_VALID(s
, s
->getp
+ size
)) {
253 STREAM_BOUND_WARN(s
, "seek getp");
260 void stream_forward_endp(struct stream
*s
, size_t size
)
262 STREAM_VERIFY_SANE(s
);
264 if (!ENDP_VALID(s
, s
->endp
+ size
)) {
265 STREAM_BOUND_WARN(s
, "seek endp");
272 /* Copy from stream to destination. */
273 inline bool stream_get2(void *dst
, struct stream
*s
, size_t size
)
275 STREAM_VERIFY_SANE(s
);
277 if (STREAM_READABLE(s
) < size
) {
278 STREAM_BOUND_WARN2(s
, "get");
282 memcpy(dst
, s
->data
+ s
->getp
, size
);
288 void stream_get(void *dst
, struct stream
*s
, size_t size
)
290 STREAM_VERIFY_SANE(s
);
292 if (STREAM_READABLE(s
) < size
) {
293 STREAM_BOUND_WARN(s
, "get");
297 memcpy(dst
, s
->data
+ s
->getp
, size
);
301 /* Get next character from the stream. */
302 inline bool stream_getc2(struct stream
*s
, u_char
*byte
)
304 STREAM_VERIFY_SANE(s
);
306 if (STREAM_READABLE(s
) < sizeof(u_char
)) {
307 STREAM_BOUND_WARN2(s
, "get char");
310 *byte
= s
->data
[s
->getp
++];
315 u_char
stream_getc(struct stream
*s
)
319 STREAM_VERIFY_SANE(s
);
321 if (STREAM_READABLE(s
) < sizeof(u_char
)) {
322 STREAM_BOUND_WARN(s
, "get char");
325 c
= s
->data
[s
->getp
++];
330 /* Get next character from the stream. */
331 u_char
stream_getc_from(struct stream
*s
, size_t from
)
335 STREAM_VERIFY_SANE(s
);
337 if (!GETP_VALID(s
, from
+ sizeof(u_char
))) {
338 STREAM_BOUND_WARN(s
, "get char");
347 inline bool stream_getw2(struct stream
*s
, uint16_t *word
)
349 STREAM_VERIFY_SANE(s
);
351 if (STREAM_READABLE(s
) < sizeof(uint16_t)) {
352 STREAM_BOUND_WARN2(s
, "get ");
356 *word
= s
->data
[s
->getp
++] << 8;
357 *word
|= s
->data
[s
->getp
++];
362 /* Get next word from the stream. */
363 u_int16_t
stream_getw(struct stream
*s
)
367 STREAM_VERIFY_SANE(s
);
369 if (STREAM_READABLE(s
) < sizeof(u_int16_t
)) {
370 STREAM_BOUND_WARN(s
, "get ");
374 w
= s
->data
[s
->getp
++] << 8;
375 w
|= s
->data
[s
->getp
++];
380 /* Get next word from the stream. */
381 u_int16_t
stream_getw_from(struct stream
*s
, size_t from
)
385 STREAM_VERIFY_SANE(s
);
387 if (!GETP_VALID(s
, from
+ sizeof(u_int16_t
))) {
388 STREAM_BOUND_WARN(s
, "get ");
392 w
= s
->data
[from
++] << 8;
398 /* Get next 3-byte from the stream. */
399 u_int32_t
stream_get3_from(struct stream
*s
, size_t from
)
403 STREAM_VERIFY_SANE(s
);
405 if (!GETP_VALID(s
, from
+ 3)) {
406 STREAM_BOUND_WARN(s
, "get 3byte");
410 l
= s
->data
[from
++] << 16;
411 l
|= s
->data
[from
++] << 8;
417 u_int32_t
stream_get3(struct stream
*s
)
421 STREAM_VERIFY_SANE(s
);
423 if (STREAM_READABLE(s
) < 3) {
424 STREAM_BOUND_WARN(s
, "get 3byte");
428 l
= s
->data
[s
->getp
++] << 16;
429 l
|= s
->data
[s
->getp
++] << 8;
430 l
|= s
->data
[s
->getp
++];
435 /* Get next long word from the stream. */
436 u_int32_t
stream_getl_from(struct stream
*s
, size_t from
)
440 STREAM_VERIFY_SANE(s
);
442 if (!GETP_VALID(s
, from
+ sizeof(u_int32_t
))) {
443 STREAM_BOUND_WARN(s
, "get long");
447 l
= (unsigned)(s
->data
[from
++]) << 24;
448 l
|= s
->data
[from
++] << 16;
449 l
|= s
->data
[from
++] << 8;
455 /* Copy from stream at specific location to destination. */
456 void stream_get_from(void *dst
, struct stream
*s
, size_t from
, size_t size
)
458 STREAM_VERIFY_SANE(s
);
460 if (!GETP_VALID(s
, from
+ size
)) {
461 STREAM_BOUND_WARN(s
, "get from");
465 memcpy(dst
, s
->data
+ from
, size
);
468 inline bool stream_getl2(struct stream
*s
, uint32_t *l
)
470 STREAM_VERIFY_SANE(s
);
472 if (STREAM_READABLE(s
) < sizeof(uint32_t)) {
473 STREAM_BOUND_WARN2(s
, "get long");
477 *l
= (unsigned int)(s
->data
[s
->getp
++]) << 24;
478 *l
|= s
->data
[s
->getp
++] << 16;
479 *l
|= s
->data
[s
->getp
++] << 8;
480 *l
|= s
->data
[s
->getp
++];
486 u_int32_t
stream_getl(struct stream
*s
)
490 STREAM_VERIFY_SANE(s
);
492 if (STREAM_READABLE(s
) < sizeof(u_int32_t
)) {
493 STREAM_BOUND_WARN(s
, "get long");
497 l
= (unsigned)(s
->data
[s
->getp
++]) << 24;
498 l
|= s
->data
[s
->getp
++] << 16;
499 l
|= s
->data
[s
->getp
++] << 8;
500 l
|= s
->data
[s
->getp
++];
505 /* Get next quad word from the stream. */
506 uint64_t stream_getq_from(struct stream
*s
, size_t from
)
510 STREAM_VERIFY_SANE(s
);
512 if (!GETP_VALID(s
, from
+ sizeof(uint64_t))) {
513 STREAM_BOUND_WARN(s
, "get quad");
517 q
= ((uint64_t)s
->data
[from
++]) << 56;
518 q
|= ((uint64_t)s
->data
[from
++]) << 48;
519 q
|= ((uint64_t)s
->data
[from
++]) << 40;
520 q
|= ((uint64_t)s
->data
[from
++]) << 32;
521 q
|= ((uint64_t)s
->data
[from
++]) << 24;
522 q
|= ((uint64_t)s
->data
[from
++]) << 16;
523 q
|= ((uint64_t)s
->data
[from
++]) << 8;
524 q
|= ((uint64_t)s
->data
[from
++]);
529 uint64_t stream_getq(struct stream
*s
)
533 STREAM_VERIFY_SANE(s
);
535 if (STREAM_READABLE(s
) < sizeof(uint64_t)) {
536 STREAM_BOUND_WARN(s
, "get quad");
540 q
= ((uint64_t)s
->data
[s
->getp
++]) << 56;
541 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 48;
542 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 40;
543 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 32;
544 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 24;
545 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 16;
546 q
|= ((uint64_t)s
->data
[s
->getp
++]) << 8;
547 q
|= ((uint64_t)s
->data
[s
->getp
++]);
552 /* Get next long word from the stream. */
553 u_int32_t
stream_get_ipv4(struct stream
*s
)
557 STREAM_VERIFY_SANE(s
);
559 if (STREAM_READABLE(s
) < sizeof(u_int32_t
)) {
560 STREAM_BOUND_WARN(s
, "get ipv4");
564 memcpy(&l
, s
->data
+ s
->getp
, sizeof(u_int32_t
));
565 s
->getp
+= sizeof(u_int32_t
);
570 float stream_getf(struct stream
*s
)
576 u
.d
= stream_getl(s
);
580 double stream_getd(struct stream
*s
)
586 u
.d
= stream_getq(s
);
590 /* Copy to source to stream.
592 * XXX: This uses CHECK_SIZE and hence has funny semantics -> Size will wrap
593 * around. This should be fixed once the stream updates are working.
595 * stream_write() is saner
597 void stream_put(struct stream
*s
, const void *src
, size_t size
)
600 /* XXX: CHECK_SIZE has strange semantics. It should be deprecated */
603 STREAM_VERIFY_SANE(s
);
605 if (STREAM_WRITEABLE(s
) < size
) {
606 STREAM_BOUND_WARN(s
, "put");
611 memcpy(s
->data
+ s
->endp
, src
, size
);
613 memset(s
->data
+ s
->endp
, 0, size
);
618 /* Put character to the stream. */
619 int stream_putc(struct stream
*s
, u_char c
)
621 STREAM_VERIFY_SANE(s
);
623 if (STREAM_WRITEABLE(s
) < sizeof(u_char
)) {
624 STREAM_BOUND_WARN(s
, "put");
628 s
->data
[s
->endp
++] = c
;
629 return sizeof(u_char
);
632 /* Put word to the stream. */
633 int stream_putw(struct stream
*s
, u_int16_t w
)
635 STREAM_VERIFY_SANE(s
);
637 if (STREAM_WRITEABLE(s
) < sizeof(u_int16_t
)) {
638 STREAM_BOUND_WARN(s
, "put");
642 s
->data
[s
->endp
++] = (u_char
)(w
>> 8);
643 s
->data
[s
->endp
++] = (u_char
)w
;
648 /* Put long word to the stream. */
649 int stream_put3(struct stream
*s
, u_int32_t l
)
651 STREAM_VERIFY_SANE(s
);
653 if (STREAM_WRITEABLE(s
) < 3) {
654 STREAM_BOUND_WARN(s
, "put");
658 s
->data
[s
->endp
++] = (u_char
)(l
>> 16);
659 s
->data
[s
->endp
++] = (u_char
)(l
>> 8);
660 s
->data
[s
->endp
++] = (u_char
)l
;
665 /* Put long word to the stream. */
666 int stream_putl(struct stream
*s
, u_int32_t l
)
668 STREAM_VERIFY_SANE(s
);
670 if (STREAM_WRITEABLE(s
) < sizeof(u_int32_t
)) {
671 STREAM_BOUND_WARN(s
, "put");
675 s
->data
[s
->endp
++] = (u_char
)(l
>> 24);
676 s
->data
[s
->endp
++] = (u_char
)(l
>> 16);
677 s
->data
[s
->endp
++] = (u_char
)(l
>> 8);
678 s
->data
[s
->endp
++] = (u_char
)l
;
683 /* Put quad word to the stream. */
684 int stream_putq(struct stream
*s
, uint64_t q
)
686 STREAM_VERIFY_SANE(s
);
688 if (STREAM_WRITEABLE(s
) < sizeof(uint64_t)) {
689 STREAM_BOUND_WARN(s
, "put quad");
693 s
->data
[s
->endp
++] = (u_char
)(q
>> 56);
694 s
->data
[s
->endp
++] = (u_char
)(q
>> 48);
695 s
->data
[s
->endp
++] = (u_char
)(q
>> 40);
696 s
->data
[s
->endp
++] = (u_char
)(q
>> 32);
697 s
->data
[s
->endp
++] = (u_char
)(q
>> 24);
698 s
->data
[s
->endp
++] = (u_char
)(q
>> 16);
699 s
->data
[s
->endp
++] = (u_char
)(q
>> 8);
700 s
->data
[s
->endp
++] = (u_char
)q
;
705 int stream_putf(struct stream
*s
, float f
)
712 return stream_putl(s
, u
.o
);
715 int stream_putd(struct stream
*s
, double d
)
722 return stream_putq(s
, u
.o
);
725 int stream_putc_at(struct stream
*s
, size_t putp
, u_char c
)
727 STREAM_VERIFY_SANE(s
);
729 if (!PUT_AT_VALID(s
, putp
+ sizeof(u_char
))) {
730 STREAM_BOUND_WARN(s
, "put");
739 int stream_putw_at(struct stream
*s
, size_t putp
, u_int16_t w
)
741 STREAM_VERIFY_SANE(s
);
743 if (!PUT_AT_VALID(s
, putp
+ sizeof(u_int16_t
))) {
744 STREAM_BOUND_WARN(s
, "put");
748 s
->data
[putp
] = (u_char
)(w
>> 8);
749 s
->data
[putp
+ 1] = (u_char
)w
;
754 int stream_put3_at(struct stream
*s
, size_t putp
, u_int32_t l
)
756 STREAM_VERIFY_SANE(s
);
758 if (!PUT_AT_VALID(s
, putp
+ 3)) {
759 STREAM_BOUND_WARN(s
, "put");
762 s
->data
[putp
] = (u_char
)(l
>> 16);
763 s
->data
[putp
+ 1] = (u_char
)(l
>> 8);
764 s
->data
[putp
+ 2] = (u_char
)l
;
769 int stream_putl_at(struct stream
*s
, size_t putp
, u_int32_t l
)
771 STREAM_VERIFY_SANE(s
);
773 if (!PUT_AT_VALID(s
, putp
+ sizeof(u_int32_t
))) {
774 STREAM_BOUND_WARN(s
, "put");
777 s
->data
[putp
] = (u_char
)(l
>> 24);
778 s
->data
[putp
+ 1] = (u_char
)(l
>> 16);
779 s
->data
[putp
+ 2] = (u_char
)(l
>> 8);
780 s
->data
[putp
+ 3] = (u_char
)l
;
785 int stream_putq_at(struct stream
*s
, size_t putp
, uint64_t q
)
787 STREAM_VERIFY_SANE(s
);
789 if (!PUT_AT_VALID(s
, putp
+ sizeof(uint64_t))) {
790 STREAM_BOUND_WARN(s
, "put");
793 s
->data
[putp
] = (u_char
)(q
>> 56);
794 s
->data
[putp
+ 1] = (u_char
)(q
>> 48);
795 s
->data
[putp
+ 2] = (u_char
)(q
>> 40);
796 s
->data
[putp
+ 3] = (u_char
)(q
>> 32);
797 s
->data
[putp
+ 4] = (u_char
)(q
>> 24);
798 s
->data
[putp
+ 5] = (u_char
)(q
>> 16);
799 s
->data
[putp
+ 6] = (u_char
)(q
>> 8);
800 s
->data
[putp
+ 7] = (u_char
)q
;
805 /* Put long word to the stream. */
806 int stream_put_ipv4(struct stream
*s
, u_int32_t l
)
808 STREAM_VERIFY_SANE(s
);
810 if (STREAM_WRITEABLE(s
) < sizeof(u_int32_t
)) {
811 STREAM_BOUND_WARN(s
, "put");
814 memcpy(s
->data
+ s
->endp
, &l
, sizeof(u_int32_t
));
815 s
->endp
+= sizeof(u_int32_t
);
817 return sizeof(u_int32_t
);
820 /* Put long word to the stream. */
821 int stream_put_in_addr(struct stream
*s
, struct in_addr
*addr
)
823 STREAM_VERIFY_SANE(s
);
825 if (STREAM_WRITEABLE(s
) < sizeof(u_int32_t
)) {
826 STREAM_BOUND_WARN(s
, "put");
830 memcpy(s
->data
+ s
->endp
, addr
, sizeof(u_int32_t
));
831 s
->endp
+= sizeof(u_int32_t
);
833 return sizeof(u_int32_t
);
836 /* Put in_addr at location in the stream. */
837 int stream_put_in_addr_at(struct stream
*s
, size_t putp
, struct in_addr
*addr
)
839 STREAM_VERIFY_SANE(s
);
841 if (!PUT_AT_VALID(s
, putp
+ 4)) {
842 STREAM_BOUND_WARN(s
, "put");
846 memcpy(&s
->data
[putp
], addr
, 4);
850 /* Put in6_addr at location in the stream. */
851 int stream_put_in6_addr_at(struct stream
*s
, size_t putp
, struct in6_addr
*addr
)
853 STREAM_VERIFY_SANE(s
);
855 if (!PUT_AT_VALID(s
, putp
+ 16)) {
856 STREAM_BOUND_WARN(s
, "put");
860 memcpy(&s
->data
[putp
], addr
, 16);
864 /* Put prefix by nlri type format. */
865 int stream_put_prefix_addpath(struct stream
*s
, struct prefix
*p
,
866 int addpath_encode
, u_int32_t addpath_tx_id
)
869 size_t psize_with_addpath
;
871 STREAM_VERIFY_SANE(s
);
873 psize
= PSIZE(p
->prefixlen
);
876 psize_with_addpath
= psize
+ 4;
878 psize_with_addpath
= psize
;
880 if (STREAM_WRITEABLE(s
) < (psize_with_addpath
+ sizeof(u_char
))) {
881 STREAM_BOUND_WARN(s
, "put");
885 if (addpath_encode
) {
886 s
->data
[s
->endp
++] = (u_char
)(addpath_tx_id
>> 24);
887 s
->data
[s
->endp
++] = (u_char
)(addpath_tx_id
>> 16);
888 s
->data
[s
->endp
++] = (u_char
)(addpath_tx_id
>> 8);
889 s
->data
[s
->endp
++] = (u_char
)addpath_tx_id
;
892 s
->data
[s
->endp
++] = p
->prefixlen
;
893 memcpy(s
->data
+ s
->endp
, &p
->u
.prefix
, psize
);
899 int stream_put_prefix(struct stream
*s
, struct prefix
*p
)
901 return stream_put_prefix_addpath(s
, p
, 0, 0);
904 /* Put NLRI with label */
905 int stream_put_labeled_prefix(struct stream
*s
, struct prefix
*p
,
909 u_char
*label_pnt
= (u_char
*)label
;
911 STREAM_VERIFY_SANE(s
);
913 psize
= PSIZE(p
->prefixlen
);
915 if (STREAM_WRITEABLE(s
) < (psize
+ 3)) {
916 STREAM_BOUND_WARN(s
, "put");
920 stream_putc(s
, (p
->prefixlen
+ 24));
921 stream_putc(s
, label_pnt
[0]);
922 stream_putc(s
, label_pnt
[1]);
923 stream_putc(s
, label_pnt
[2]);
924 memcpy(s
->data
+ s
->endp
, &p
->u
.prefix
, psize
);
930 /* Read size from fd. */
931 int stream_read(struct stream
*s
, int fd
, size_t size
)
935 STREAM_VERIFY_SANE(s
);
937 if (STREAM_WRITEABLE(s
) < size
) {
938 STREAM_BOUND_WARN(s
, "put");
942 nbytes
= readn(fd
, s
->data
+ s
->endp
, size
);
950 ssize_t
stream_read_try(struct stream
*s
, int fd
, size_t size
)
954 STREAM_VERIFY_SANE(s
);
956 if (STREAM_WRITEABLE(s
) < size
) {
957 STREAM_BOUND_WARN(s
, "put");
958 /* Fatal (not transient) error, since retrying will not help
959 (stream is too small to contain the desired data). */
963 if ((nbytes
= read(fd
, s
->data
+ s
->endp
, size
)) >= 0) {
967 /* Error: was it transient (return -2) or fatal (return -1)? */
968 if (ERRNO_IO_RETRY(errno
))
970 zlog_warn("%s: read failed on fd %d: %s", __func__
, fd
,
971 safe_strerror(errno
));
975 /* Read up to size bytes into the stream from the fd, using recvmsgfrom
976 * whose arguments match the remaining arguments to this function
978 ssize_t
stream_recvfrom(struct stream
*s
, int fd
, size_t size
, int flags
,
979 struct sockaddr
*from
, socklen_t
*fromlen
)
983 STREAM_VERIFY_SANE(s
);
985 if (STREAM_WRITEABLE(s
) < size
) {
986 STREAM_BOUND_WARN(s
, "put");
987 /* Fatal (not transient) error, since retrying will not help
988 (stream is too small to contain the desired data). */
992 if ((nbytes
= recvfrom(fd
, s
->data
+ s
->endp
, size
, flags
, from
,
998 /* Error: was it transient (return -2) or fatal (return -1)? */
999 if (ERRNO_IO_RETRY(errno
))
1001 zlog_warn("%s: read failed on fd %d: %s", __func__
, fd
,
1002 safe_strerror(errno
));
1006 /* Read up to smaller of size or SIZE_REMAIN() bytes to the stream, starting
1008 * First iovec will be used to receive the data.
1009 * Stream need not be empty.
1011 ssize_t
stream_recvmsg(struct stream
*s
, int fd
, struct msghdr
*msgh
, int flags
,
1017 STREAM_VERIFY_SANE(s
);
1018 assert(msgh
->msg_iovlen
> 0);
1020 if (STREAM_WRITEABLE(s
) < size
) {
1021 STREAM_BOUND_WARN(s
, "put");
1022 /* This is a logic error in the calling code: the stream is too
1024 to hold the desired data! */
1028 iov
= &(msgh
->msg_iov
[0]);
1029 iov
->iov_base
= (s
->data
+ s
->endp
);
1030 iov
->iov_len
= size
;
1032 nbytes
= recvmsg(fd
, msgh
, flags
);
1040 /* Write data to buffer. */
1041 size_t stream_write(struct stream
*s
, const void *ptr
, size_t size
)
1044 CHECK_SIZE(s
, size
);
1046 STREAM_VERIFY_SANE(s
);
1048 if (STREAM_WRITEABLE(s
) < size
) {
1049 STREAM_BOUND_WARN(s
, "put");
1053 memcpy(s
->data
+ s
->endp
, ptr
, size
);
1059 /* Return current read pointer.
1061 * Use stream_get_pnt_to if you must, but decoding streams properly
1064 u_char
*stream_pnt(struct stream
*s
)
1066 STREAM_VERIFY_SANE(s
);
1067 return s
->data
+ s
->getp
;
1070 /* Check does this stream empty? */
1071 int stream_empty(struct stream
*s
)
1073 STREAM_VERIFY_SANE(s
);
1075 return (s
->endp
== 0);
1079 void stream_reset(struct stream
*s
)
1081 STREAM_VERIFY_SANE(s
);
1083 s
->getp
= s
->endp
= 0;
1086 /* Write stream contens to the file discriptor. */
1087 int stream_flush(struct stream
*s
, int fd
)
1091 STREAM_VERIFY_SANE(s
);
1093 nbytes
= write(fd
, s
->data
+ s
->getp
, s
->endp
- s
->getp
);
1098 /* Stream first in first out queue. */
1100 struct stream_fifo
*stream_fifo_new(void)
1102 struct stream_fifo
*new;
1104 new = XCALLOC(MTYPE_STREAM_FIFO
, sizeof(struct stream_fifo
));
1108 /* Add new stream to fifo. */
1109 void stream_fifo_push(struct stream_fifo
*fifo
, struct stream
*s
)
1112 fifo
->tail
->next
= s
;
1121 /* Delete first stream from fifo. */
1122 struct stream
*stream_fifo_pop(struct stream_fifo
*fifo
)
1129 fifo
->head
= s
->next
;
1131 if (fifo
->head
== NULL
)
1140 /* Return first fifo entry. */
1141 struct stream
*stream_fifo_head(struct stream_fifo
*fifo
)
1146 void stream_fifo_clean(struct stream_fifo
*fifo
)
1149 struct stream
*next
;
1151 for (s
= fifo
->head
; s
; s
= next
) {
1155 fifo
->head
= fifo
->tail
= NULL
;
1159 void stream_fifo_free(struct stream_fifo
*fifo
)
1161 stream_fifo_clean(fifo
);
1162 XFREE(MTYPE_STREAM_FIFO
, fifo
);