include/iobuf.h

   1 /*
   2  * PgBouncer - Lightweight connection pooler for PostgreSQL.
   3  *
   4  * Copyright (c) 2007-2009  Marko Kreen, Skype Technologies OÜ
   5  *
   6  * Permission to use, copy, modify, and/or distribute this software for any
   7  * purpose with or without fee is hereby granted, provided that the above
   8  * copyright notice and this permission notice appear in all copies.
   9  *
  10  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
  11  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  12  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  13  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  14  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  15  * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
  16  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  17  */
  18
  19 /*
  20  * Temporary buffer for single i/o.
  21  *
  22  * Pattern:
  23  *
  24  *      iobuf_get_and_reset()
  25  * start:
  26  *      iobuf_recv()
  27  * loop:
  28  *      if (new_pkt)
  29  *              iobuf_parse()
  30  *
  31  *      if (send) {
  32  *              iobuf_tag_send()
  33  *      } else {
  34  *              send_pending()
  35  *              iobuf_tag_skip()
  36  *      }
  37  *      if (more-unparsed)
  38  *              goto loop;
  39  *      send_pending();
  40  */
  41
  42 /*
  43  * 0 .. done_pos         -- sent
  44  * done_pos .. parse_pos -- parsed, to send
  45  * parse_pos .. recv_pos -- received, to parse
  46  */
  47 struct iobuf {
  48         unsigned done_pos;
  49         unsigned parse_pos;
  50         unsigned recv_pos;
  51         uint8_t buf[FLEX_ARRAY];
  52 };
  53 typedef struct iobuf IOBuf;
  54
  55 static inline bool iobuf_sane(const IOBuf *io)
  56 {
  57         return (io == NULL) ||
  58                 (  io->parse_pos >= io->done_pos
  59                 && io->recv_pos >= io->parse_pos
  60                 && (unsigned)cf_sbuf_len >= io->recv_pos);
  61 }
  62
  63 static inline bool iobuf_empty(const IOBuf *io)
  64 {
  65         return io == NULL || io->done_pos == io->recv_pos;
  66 }
  67
  68 /* unsent amount */
  69 static inline unsigned iobuf_amount_pending(const IOBuf *buf)
  70 {
  71         return buf->parse_pos - buf->done_pos;
  72 }
  73
  74 /* max possible to parse (tag_send/tag_skip) */
  75 static inline unsigned iobuf_amount_parse(const IOBuf *buf)
  76 {
  77         return buf->recv_pos - buf->parse_pos;
  78 }
  79
  80 /* max possible to recv */
  81 static inline unsigned iobuf_amount_recv(const IOBuf *buf)
  82 {
  83         return cf_sbuf_len - buf->recv_pos;
  84 }
  85
  86 /* put all unparsed to mbuf */
  87 static inline unsigned iobuf_parse_all(const IOBuf *buf, MBuf *mbuf)
  88 {
  89         unsigned avail = iobuf_amount_parse(buf);
  90         const uint8_t *pos = buf->buf + buf->parse_pos;
  91         mbuf_init(mbuf, pos, avail);
  92         return avail;
  93 }
  94
  95 /* put all unparsed to mbuf, with size limit */
  96 static inline unsigned iobuf_parse_limit(const IOBuf *buf, MBuf *mbuf, unsigned limit)
  97 {
  98         unsigned avail = iobuf_amount_parse(buf);
  99         const uint8_t *pos = buf->buf + buf->parse_pos;
 100         if (avail > limit)
 101                 avail = limit;
 102         mbuf_init(mbuf, pos, avail);
 103         return avail;
 104 }
 105
 106 /* recv */
 107 static inline int _MUSTCHECK iobuf_recv_limit(IOBuf *io, int fd, unsigned len)
 108 {
 109         uint8_t *pos = io->buf + io->recv_pos;
 110         int got;
 111         unsigned avail = iobuf_amount_recv(io);
 112
 113         if (len > avail)
 114                 len = avail;
 115
 116         Assert(len > 0);
 117
 118         got = safe_recv(fd, pos, len, 0);
 119         if (got > 0)
 120                 io->recv_pos += got;
 121         return got;
 122 }
 123
 124 static inline int _MUSTCHECK iobuf_recv_max(IOBuf *io, int fd)
 125 {
 126         return iobuf_recv_limit(io, fd, iobuf_amount_recv(io));
 127 }
 128
 129 /* send tagged data */
 130 static inline int _MUSTCHECK iobuf_send_pending(IOBuf *io, int fd)
 131 {
 132         uint8_t *pos = io->buf + io->done_pos;
 133         int len, res;
 134
 135         len = io->parse_pos - io->done_pos;
 136         Assert(len > 0);
 137
 138         res = safe_send(fd, pos, len, 0);
 139         if (res > 0)
 140                 io->done_pos += res;
 141         return res;
 142 }
 143
 144 static inline void iobuf_tag_send(IOBuf *io, unsigned len)
 145 {
 146         Assert(len > 0 && len <= iobuf_amount_parse(io));
 147
 148         io->parse_pos += len;
 149 }
 150
 151 static inline void iobuf_tag_skip(IOBuf *io, unsigned len)
 152 {
 153         Assert(io->parse_pos == io->done_pos); /* no send pending */
 154         Assert(len > 0 && len <= iobuf_amount_parse(io));
 155
 156         io->parse_pos += len;
 157         io->done_pos = io->parse_pos;
 158 }
 159
 160 static inline void iobuf_try_resync(IOBuf *io, unsigned small_pkt)
 161 {
 162         unsigned avail = io->recv_pos - io->done_pos;
 163         if (avail == 0) {
 164                 if (io->recv_pos > 0)
 165                         io->recv_pos = io->parse_pos = io->done_pos = 0;
 166         } else if (avail <= small_pkt && io->done_pos > 0) {
 167                 memmove(io->buf, io->buf + io->done_pos, avail);
 168                 io->parse_pos -= io->done_pos;
 169                 io->recv_pos = avail;
 170                 io->done_pos = 0;
 171         }
 172 }
 173
 174 static inline void iobuf_reset(IOBuf *io)
 175 {
 176         io->recv_pos = io->parse_pos = io->done_pos = 0;
 177 }
 178