stream.c 19 KB

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