stream.c 19 KB

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