ctdb/common/pkt_read.c

   1 /*
   2    Reading packets using fixed and dynamic buffer
   3
   4    Copyright (C) Amitay Isaacs 2015
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 /* This is similar to read_packet abstraction.  The main different is that
  21  * tevent fd event is created only once.
  22  */
  23
  24 #include "replace.h"
  25 #include "system/network.h"
  26
  27 #include <talloc.h>
  28 #include <tevent.h>
  29
  30 #include "lib/util/tevent_unix.h"
  31
  32 #include "pkt_read.h"
  33
  34 /*
  35  * Read a packet using fixed buffer
  36  */
  37
  38 struct pkt_read_state {
  39         int fd;
  40         uint8_t *buf;
  41         size_t buflen;
  42         size_t nread, total;
  43         bool use_fixed;
  44         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
  45         void *private_data;
  46 };
  47
  48 struct tevent_req *pkt_read_send(TALLOC_CTX *mem_ctx,
  49                                  struct tevent_context *ev,
  50                                  int fd, size_t initial,
  51                                  uint8_t *buf, size_t buflen,
  52                                  ssize_t (*more)(uint8_t *buf,
  53                                                  size_t buflen,
  54                                                  void *private_data),
  55                                  void *private_data)
  56 {
  57         struct tevent_req *req;
  58         struct pkt_read_state *state;
  59
  60         req = tevent_req_create(mem_ctx, &state, struct pkt_read_state);
  61         if (req == NULL) {
  62                 return NULL;
  63         }
  64
  65         state->fd = fd;
  66
  67         if (buf == NULL || buflen == 0) {
  68                 state->use_fixed = false;
  69                 state->buf = talloc_array(state, uint8_t, initial);
  70                 if (state->buf == NULL) {
  71                         talloc_free(req);
  72                         return NULL;
  73                 }
  74                 state->buflen = initial;
  75         } else {
  76                 state->use_fixed = true;
  77                 state->buf = buf;
  78                 state->buflen = buflen;
  79         }
  80
  81         state->nread = 0;
  82         state->total = initial;
  83
  84         state->more = more;
  85         state->private_data = private_data;
  86
  87         return req;
  88 }
  89
  90 void pkt_read_handler(struct tevent_context *ev, struct tevent_fd *fde,
  91                       uint16_t flags, struct tevent_req *req)
  92 {
  93         struct pkt_read_state *state = tevent_req_data(
  94                 req, struct pkt_read_state);
  95         ssize_t nread, more;
  96         uint8_t *tmp;
  97
  98         nread = read(state->fd, state->buf + state->nread,
  99                      state->total - state->nread);
 100         if ((nread == -1) && (errno == EINTR)) {
 101                 /* retry */
 102                 return;
 103         }
 104         if (nread == -1) {
 105                 tevent_req_error(req, errno);
 106                 return;
 107         }
 108         if (nread == 0) {
 109                 /* fd closed */
 110                 tevent_req_error(req, EPIPE);
 111                 return;
 112         }
 113
 114         state->nread += nread;
 115         if (state->nread < state->total) {
 116                 /* come back later */
 117                 return;
 118         }
 119
 120         /* Check if "more" asks for more data */
 121         if (state->more == NULL) {
 122                 tevent_req_done(req);
 123                 return;
 124         }
 125
 126         more = state->more(state->buf, state->nread, state->private_data);
 127         if (more == -1) {
 128                 /* invalid packet */
 129                 tevent_req_error(req, EIO);
 130                 return;
 131         }
 132         if (more == 0) {
 133                 tevent_req_done(req);
 134                 return;
 135         }
 136
 137         if (state->total + more < state->total) {
 138                 /* int wrapped */
 139                 tevent_req_error(req, EMSGSIZE);
 140                 return;
 141         }
 142
 143         if (state->total + more < state->buflen) {
 144                 /* continue using fixed buffer */
 145                 state->total += more;
 146                 return;
 147         }
 148
 149         if (state->use_fixed) {
 150                 /* switch to dynamic buffer */
 151                 tmp = talloc_array(state, uint8_t, state->total + more);
 152                 if (tevent_req_nomem(tmp, req)) {
 153                         return;
 154                 }
 155
 156                 memcpy(tmp, state->buf, state->total);
 157                 state->use_fixed = false;
 158         } else {
 159                 tmp = talloc_realloc(state, state->buf, uint8_t,
 160                                      state->total + more);
 161                 if (tevent_req_nomem(tmp, req)) {
 162                         return;
 163                 }
 164         }
 165
 166         state->buf = tmp;
 167         state->buflen = state->total + more;
 168         state->total += more;
 169 }
 170
 171 ssize_t pkt_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 172                       uint8_t **pbuf, bool *free_buf, int *perrno)
 173 {
 174         struct pkt_read_state *state = tevent_req_data(
 175                 req, struct pkt_read_state);
 176
 177         if (tevent_req_is_unix_error(req, perrno)) {
 178                 return -1;
 179         }
 180
 181         if (state->use_fixed) {
 182                 *pbuf = state->buf;
 183                 *free_buf = false;
 184         } else {
 185                 *pbuf = talloc_steal(mem_ctx, state->buf);
 186                 *free_buf = true;
 187         }
 188
 189         return state->total;
 190 }