ctdb/tests/src/pkt_read_test.c

   1 /*
   2    packet read tests
   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 #include "replace.h"
  21 #include "system/filesys.h"
  22
  23 #include <assert.h>
  24
  25 #include "common/pkt_read.c"
  26
  27 static void writer(int fd)
  28 {
  29         uint8_t buf[1024*1024];
  30         size_t buflen;
  31         size_t pkt_size[4] = { 100, 500, 1024, 1024*1024 };
  32         int i, j;
  33         int ret;
  34
  35         for (i=0; i<1024*1024; i++) {
  36                 buf[i] = i%256;
  37         }
  38
  39         for (i=0; i<1000; i++) {
  40                 for (j=0; j<4; j++) {
  41                         buflen = pkt_size[j];
  42                         memcpy(buf, &buflen, sizeof(buflen));
  43
  44                         ret = write(fd, buf, buflen);
  45                         if (ret < 0) {
  46                                 printf("write error: %s\n", strerror(errno));
  47                                 assert(ret > 0);
  48                         }
  49                 }
  50         }
  51
  52         close(fd);
  53 }
  54
  55 struct reader_state {
  56         struct tevent_context *ev;
  57         int fd;
  58         uint8_t buf[1024];
  59         struct tevent_req *subreq;
  60 };
  61
  62 static ssize_t reader_more(uint8_t *buf, size_t buflen, void *private_data);
  63 static void reader_done(struct tevent_req *subreq);
  64
  65 static struct tevent_req *reader_send(TALLOC_CTX *mem_ctx,
  66                                       struct tevent_context *ev,
  67                                       int fd)
  68 {
  69         struct tevent_req *req, *subreq;
  70         struct reader_state *state;
  71
  72         req = tevent_req_create(mem_ctx, &state, struct reader_state);
  73         if (req == NULL) {
  74                 return NULL;
  75         }
  76
  77         state->ev = ev;
  78         state->fd = fd;
  79
  80         subreq = pkt_read_send(state, state->ev, state->fd, 4,
  81                                state->buf, 1024, reader_more, NULL);
  82         if (tevent_req_nomem(subreq, req)) {
  83                 tevent_req_post(req, ev);
  84         }
  85
  86         state->subreq = subreq;
  87         tevent_req_set_callback(subreq, reader_done, req);
  88         return req;
  89 }
  90
  91 static ssize_t reader_more(uint8_t *buf, size_t buflen, void *private_data)
  92 {
  93         uint32_t pkt_len;
  94
  95         if (buflen < sizeof(pkt_len)) {
  96                 return sizeof(pkt_len) - buflen;
  97         }
  98
  99         pkt_len = *(uint32_t *)buf;
 100         return pkt_len - buflen;
 101 }
 102
 103 static void reader_done(struct tevent_req *subreq)
 104 {
 105         struct tevent_req *req = tevent_req_callback_data(
 106                 subreq, struct tevent_req);
 107         struct reader_state *state = tevent_req_data(
 108                 req, struct reader_state);
 109         ssize_t nread;
 110         uint8_t *buf;
 111         bool free_buf;
 112         int err;
 113
 114         nread = pkt_read_recv(subreq, state, &buf, &free_buf, &err);
 115         TALLOC_FREE(subreq);
 116         state->subreq = NULL;
 117         if (nread == -1) {
 118                 if (err == EPIPE) {
 119                         tevent_req_done(req);
 120                 } else {
 121                         tevent_req_error(req, err);
 122                 }
 123                 return;
 124         }
 125
 126         if (free_buf) {
 127                 talloc_free(buf);
 128         }
 129
 130         subreq = pkt_read_send(state, state->ev, state->fd, 4,
 131                                state->buf, 1024, reader_more, NULL);
 132         if (tevent_req_nomem(subreq, req)) {
 133                 return;
 134         }
 135
 136         state->subreq = subreq;
 137         tevent_req_set_callback(subreq, reader_done, req);
 138 }
 139
 140 static void reader_recv(struct tevent_req *req, int *perr)
 141 {
 142         struct reader_state *state = tevent_req_data(
 143                 req, struct reader_state);
 144         int err = 0;
 145
 146         if (state->subreq != NULL) {
 147                 *perr = -1;
 148         }
 149
 150         if (tevent_req_is_unix_error(req, &err)) {
 151                 *perr = err;
 152                 return;
 153         }
 154
 155         *perr = 0;
 156 }
 157
 158 static void reader_handler(struct tevent_context *ev, struct tevent_fd *fde,
 159                            uint16_t flags, void *private_data)
 160 {
 161         struct tevent_req *req = talloc_get_type_abort(
 162                 private_data, struct tevent_req);
 163         struct reader_state *state = tevent_req_data(
 164                 req, struct reader_state);
 165
 166         assert(state->subreq != NULL);
 167         pkt_read_handler(ev, fde, flags, state->subreq);
 168 }
 169
 170 static void reader(int fd)
 171 {
 172         TALLOC_CTX *mem_ctx;
 173         struct tevent_context *ev;
 174         struct tevent_fd *fde;
 175         struct tevent_req *req;
 176         int err;
 177
 178         mem_ctx = talloc_new(NULL);
 179         assert(mem_ctx != NULL);
 180
 181         ev = tevent_context_init(mem_ctx);
 182         assert(ev != NULL);
 183
 184         req = reader_send(mem_ctx, ev, fd);
 185         assert(req != NULL);
 186
 187         fde = tevent_add_fd(ev, mem_ctx, fd, TEVENT_FD_READ,
 188                             reader_handler, req);
 189         assert(fde != NULL);
 190
 191         tevent_req_poll(req, ev);
 192
 193         reader_recv(req, &err);
 194         assert(err == 0);
 195
 196         close(fd);
 197
 198         talloc_free(mem_ctx);
 199 }
 200
 201 static bool set_nonblocking(int fd)
 202 {
 203         int v;
 204
 205         v = fcntl(fd, F_GETFL, 0);
 206         if (v == -1) {
 207                 return false;
 208         }
 209         if (fcntl(fd, F_SETFL, v | O_NONBLOCK) == -1) {
 210                 return false;
 211         }
 212         return true;
 213 }
 214
 215 int main(void)
 216 {
 217         int fd[2];
 218         int ret;
 219         pid_t pid;
 220
 221         ret = pipe(fd);
 222         assert(ret == 0);
 223
 224         pid = fork();
 225         assert(pid != -1);
 226
 227         if (pid == 0) {
 228                 /* Child process */
 229                 close(fd[0]);
 230                 writer(fd[1]);
 231                 exit(0);
 232         }
 233
 234         close(fd[1]);
 235         if (!set_nonblocking(fd[0])) {
 236                 exit(1);
 237         }
 238
 239         reader(fd[0]);
 240
 241         return 0;
 242 }