ctdb/tests/src/sock_io_test.c

   1 /*
   2    sock I/O tests
   3
   4    Copyright (C) Amitay Isaacs  2017
   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 #include "system/network.h"
  23 #include "system/wait.h"
  24
  25 #include <assert.h>
  26
  27 #include "common/sock_io.c"
  28
  29 static int socket_init(const char *sockpath)
  30 {
  31         struct sockaddr_un addr;
  32         int fd, ret;
  33         size_t len;
  34
  35         memset(&addr, 0, sizeof(addr));
  36         addr.sun_family = AF_UNIX;
  37
  38         len = strlcpy(addr.sun_path, sockpath, sizeof(addr.sun_path));
  39         assert(len < sizeof(addr.sun_path));
  40
  41         fd = socket(AF_UNIX, SOCK_STREAM, 0);
  42         assert(fd != -1);
  43
  44         ret = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
  45         assert(ret != -1);
  46
  47         ret = listen(fd, 10);
  48         assert(ret != -1);
  49
  50         return fd;
  51 }
  52
  53 static void test1_writer(int fd)
  54 {
  55         uint8_t buf[1024];
  56         ssize_t nwritten;
  57         uint32_t len;
  58
  59         for (len = 10; len < 1000; len += 10) {
  60                 int value = len / 10;
  61                 uint32_t buflen = len + sizeof(uint32_t);
  62
  63                 memset(buf,  value, buflen);
  64                 memcpy(buf, &buflen, sizeof(uint32_t));
  65
  66                 nwritten = sys_write(fd, buf, buflen);
  67                 assert(nwritten == buflen);
  68         }
  69 }
  70
  71 struct test1_reader_state {
  72         size_t pkt_len;
  73         bool done;
  74 };
  75
  76 static void test1_reader(uint8_t *buf, size_t buflen, void *private_data)
  77 {
  78         struct test1_reader_state *state =
  79                 (struct test1_reader_state *)private_data;
  80
  81         if (buflen == 0) {
  82                 state->done = true;
  83                 return;
  84         }
  85
  86         assert(buflen == state->pkt_len);
  87
  88         state->pkt_len += 10;
  89 }
  90
  91 static void test1(TALLOC_CTX *mem_ctx, const char *sockpath)
  92 {
  93         struct test1_reader_state state;
  94         struct tevent_context *ev;
  95         struct sock_queue *queue;
  96         pid_t pid;
  97         int pfd[2], fd, ret;
  98         ssize_t n;
  99
 100         ret = pipe(pfd);
 101         assert(ret == 0);
 102
 103         pid = fork();
 104         assert(pid != -1);
 105
 106         if (pid == 0) {
 107                 int newfd;
 108
 109                 close(pfd[0]);
 110
 111                 fd = socket_init(sockpath);
 112                 assert(fd != -1);
 113
 114                 ret = 1;
 115                 n = sys_write(pfd[1], &ret, sizeof(int));
 116                 assert(n == sizeof(int));
 117
 118                 newfd = accept(fd, NULL, NULL);
 119                 assert(newfd != -1);
 120
 121                 test1_writer(newfd);
 122                 close(newfd);
 123                 unlink(sockpath);
 124
 125                 exit(0);
 126         }
 127
 128         close(pfd[1]);
 129
 130         n = sys_read(pfd[0], &ret, sizeof(int));
 131         assert(n == sizeof(int));
 132         assert(ret == 1);
 133
 134         close(pfd[0]);
 135
 136         fd = sock_connect(sockpath);
 137         assert(fd != -1);
 138
 139         ev = tevent_context_init(mem_ctx);
 140         assert(ev != NULL);
 141
 142         state.pkt_len = 10 + sizeof(uint32_t);
 143         state.done = false;
 144
 145         queue = sock_queue_setup(mem_ctx, ev, fd, test1_reader, &state);
 146         assert(queue != NULL);
 147
 148         while (! state.done) {
 149                 tevent_loop_once(ev);
 150         }
 151
 152         talloc_free(queue);
 153         talloc_free(ev);
 154
 155         pid = wait(&ret);
 156         assert(pid != -1);
 157 }
 158
 159 static void test2_reader(int fd)
 160 {
 161         uint8_t buf[1024];
 162         size_t pkt_len = 10 + sizeof(uint32_t);
 163         ssize_t n;
 164
 165         while (1) {
 166                 n = sys_read(fd, buf, 1024);
 167                 assert(n != -1);
 168
 169                 if (n == 0) {
 170                         return;
 171                 }
 172
 173                 assert((size_t)n == pkt_len);
 174                 pkt_len += 10;
 175         }
 176 }
 177
 178 static void test2_dummy_reader(uint8_t *buf, size_t buflen,
 179                                void *private_data)
 180 {
 181         abort();
 182 }
 183
 184 static void test2_writer(struct sock_queue *queue)
 185 {
 186         uint8_t buf[1024];
 187         uint32_t len;
 188         int ret;
 189
 190         for (len = 10; len < 1000; len += 10) {
 191                 int value = len / 10;
 192                 uint32_t buflen = len + sizeof(uint32_t);
 193
 194                 memset(buf,  value, buflen);
 195                 memcpy(buf, &buflen, sizeof(uint32_t));
 196
 197                 ret = sock_queue_write(queue, buf, buflen);
 198                 assert(ret == 0);
 199         }
 200 }
 201
 202 static void test2(TALLOC_CTX *mem_ctx, const char *sockpath)
 203 {
 204         struct tevent_context *ev;
 205         struct sock_queue *queue;
 206         pid_t pid;
 207         int pfd[2], fd, ret;
 208         ssize_t n;
 209
 210         ret = pipe(pfd);
 211         assert(ret == 0);
 212
 213         pid = fork();
 214         assert(pid != -1);
 215
 216         if (pid == 0) {
 217                 int newfd;
 218
 219                 close(pfd[0]);
 220
 221                 fd = socket_init(sockpath);
 222                 assert(fd != -1);
 223
 224                 ret = 1;
 225                 n = sys_write(pfd[1], &ret, sizeof(int));
 226                 assert(n == sizeof(int));
 227
 228                 newfd = accept(fd, NULL, NULL);
 229                 assert(newfd != -1);
 230
 231                 test2_reader(newfd);
 232                 close(newfd);
 233                 unlink(sockpath);
 234
 235                 exit(0);
 236         }
 237
 238         close(pfd[1]);
 239
 240         n = sys_read(pfd[0], &ret, sizeof(int));
 241         assert(n == sizeof(int));
 242         assert(ret == 1);
 243
 244         close(pfd[0]);
 245
 246         fd = sock_connect(sockpath);
 247         assert(fd != -1);
 248
 249         ev = tevent_context_init(mem_ctx);
 250         assert(ev != NULL);
 251
 252         queue = sock_queue_setup(mem_ctx, ev, fd, test2_dummy_reader, NULL);
 253         assert(queue != NULL);
 254
 255         test2_writer(queue);
 256
 257         talloc_free(queue);
 258         talloc_free(ev);
 259
 260         pid = wait(&ret);
 261         assert(pid != -1);
 262 }
 263
 264 int main(int argc, const char **argv)
 265 {
 266         TALLOC_CTX *mem_ctx;
 267         const char *sockpath;
 268
 269         if (argc != 2) {
 270                 fprintf(stderr, "%s <sockpath>\n", argv[0]);
 271                 exit(1);
 272         }
 273
 274         sockpath = argv[1];
 275
 276         mem_ctx = talloc_new(NULL);
 277         assert(mem_ctx != NULL);
 278
 279         test1(mem_ctx, sockpath);
 280         test2(mem_ctx, sockpath);
 281
 282         return 0;
 283 }