From e01c0eed38335e7b421ab4f79410f08ab1d31482 Mon Sep 17 00:00:00 2001 From: Amitay Isaacs Date: Mon, 6 Apr 2015 17:26:29 +1000 Subject: [PATCH] ctdb-common: Add packet write abstraction Signed-off-by: Amitay Isaacs Reviewed-by: Martin Schwenke --- ctdb/common/pkt_write.c | 101 +++++++++++ ctdb/common/pkt_write.h | 79 ++++++++ ctdb/tests/cunit/pkt_write_001.sh | 7 + ctdb/tests/src/pkt_write_test.c | 370 ++++++++++++++++++++++++++++++++++++++ ctdb/wscript | 3 +- 5 files changed, 559 insertions(+), 1 deletion(-) create mode 100644 ctdb/common/pkt_write.c create mode 100644 ctdb/common/pkt_write.h create mode 100755 ctdb/tests/cunit/pkt_write_001.sh create mode 100644 ctdb/tests/src/pkt_write_test.c diff --git a/ctdb/common/pkt_write.c b/ctdb/common/pkt_write.c new file mode 100644 index 00000000000..b1c17305c6d --- /dev/null +++ b/ctdb/common/pkt_write.c @@ -0,0 +1,101 @@ +/* + Write a packet + + Copyright (C) Amitay Isaacs 2015 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, see . +*/ + +#include "replace.h" +#include "system/network.h" + +#include +#include + +#include "lib/util/tevent_unix.h" + +#include "pkt_write.h" + +/* + * Write a packet + */ + +struct pkt_write_state { + int fd; + uint8_t *buf; + size_t buflen, offset; +}; + +struct tevent_req *pkt_write_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + int fd, uint8_t *buf, size_t buflen) +{ + struct tevent_req *req; + struct pkt_write_state *state; + + req = tevent_req_create(mem_ctx, &state, struct pkt_write_state); + if (req == NULL) { + return NULL; + } + + state->fd = fd; + state->buf = buf; + state->buflen = buflen; + state->offset = 0; + + return req; +} + +void pkt_write_handler(struct tevent_context *ev, struct tevent_fd *fde, + uint16_t flags, struct tevent_req *req) +{ + struct pkt_write_state *state = tevent_req_data( + req, struct pkt_write_state); + ssize_t nwritten; + + nwritten = write(state->fd, state->buf + state->offset, + state->buflen - state->offset); + if ((nwritten == -1) && (errno == EINTR)) { + /* retry */ + return; + } + if (nwritten == -1) { + tevent_req_error(req, errno); + return; + } + if (nwritten == 0) { + /* retry */ + return; + } + + state->offset += nwritten; + if (state->offset < state->buflen) { + /* come back later */ + return; + } + + tevent_req_done(req); +} + +ssize_t pkt_write_recv(struct tevent_req *req, int *perrno) +{ + struct pkt_write_state *state = tevent_req_data( + req, struct pkt_write_state); + + if (tevent_req_is_unix_error(req, perrno)) { + return -1; + } + + return state->offset; +} diff --git a/ctdb/common/pkt_write.h b/ctdb/common/pkt_write.h new file mode 100644 index 00000000000..19d8045343e --- /dev/null +++ b/ctdb/common/pkt_write.h @@ -0,0 +1,79 @@ +/* + API for writing a packet + + Copyright (C) Amitay Isaacs 2015 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, see . +*/ + +#ifndef __CTDB_PKT_WRITE_H__ +#define __CTDB_PKT_WRITE_H__ + +#include +#include + +/** + * @file pkt_write.h + * + * @brief Write a packet. + * + * Write a complete packet with possibly multiple system calls. + */ + +/** + * @brief Start async computation to write a packet + * + * This returns a tevent request to write a packet to given fd. The fd + * should be nonblocking. Freeing this request will free all the memory + * associated with the request. + * + * @param[in] mem_ctx Talloc memory context + * @param[in] ev Tevent context + * @param[in] fd The non-blocking file/socket descriptor to write to + * @param[in] buf The data + * @param[in] buflen The size of the data + * @return new tevent request or NULL on failure + */ +struct tevent_req *pkt_write_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + int fd, uint8_t *buf, size_t buflen); + +/** + * @brief Function to actually write data to the socket + * + * This function should be called, when tevent fd event is triggered + * for TEVENT_FD_WRITE event. This function has the syntax of + * tevent_fd_handler_t. The private_data for this function is the tevent + * request created by pkt_write_send function. + * + * @param[in] ev Tevent context + * @param[in] fde Tevent fd context + * @param[in] flags Tevent fd flags + * @param[in] req The active tevent request + */ +void pkt_write_handler(struct tevent_context *ev, struct tevent_fd *fde, + uint16_t flags, struct tevent_req *req); + +/** + * @brief Packet is sent + * + * This function returns the number of bytes written. + * + * @param[in] req Tevent request + * @param[out] perrno errno in case of failure + * @return the number of bytes written, or -1 on failure + */ +ssize_t pkt_write_recv(struct tevent_req *req, int *perrno); + +#endif /* __CTDB_PKT_WRITE_H__ */ diff --git a/ctdb/tests/cunit/pkt_write_001.sh b/ctdb/tests/cunit/pkt_write_001.sh new file mode 100755 index 00000000000..131af05c5b6 --- /dev/null +++ b/ctdb/tests/cunit/pkt_write_001.sh @@ -0,0 +1,7 @@ +#!/bin/sh + +. "${TEST_SCRIPTS_DIR}/unit.sh" + +ok_null + +unit_test pkt_write_test diff --git a/ctdb/tests/src/pkt_write_test.c b/ctdb/tests/src/pkt_write_test.c new file mode 100644 index 00000000000..0a5c6540671 --- /dev/null +++ b/ctdb/tests/src/pkt_write_test.c @@ -0,0 +1,370 @@ +/* + packet write tests + + Copyright (C) Amitay Isaacs 2015 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, see . +*/ + +#include "replace.h" +#include "system/filesys.h" + +#include + +#include "common/pkt_read.c" +#include "common/pkt_write.c" + +struct writer_state { + struct tevent_context *ev; + int fd; + uint8_t *buf; + size_t buflen; + int count; + struct tevent_req *subreq; +}; + +static void writer_next(struct tevent_req *subreq); + +static struct tevent_req *writer_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + int fd, uint8_t *buf, size_t buflen) +{ + struct tevent_req *req, *subreq; + struct writer_state *state; + + req = tevent_req_create(mem_ctx, &state, struct writer_state); + if (req == NULL) { + return NULL; + } + + state->ev = ev; + state->fd = fd; + state->buf = buf; + state->buflen = buflen; + state->count = 0; + + subreq = pkt_write_send(state, state->ev, state->fd, + state->buf, state->buflen); + if (tevent_req_nomem(subreq, req)) { + return tevent_req_post(req, ev); + } + + state->subreq = subreq; + tevent_req_set_callback(subreq, writer_next, req); + return req; +} + +static void writer_next(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct writer_state *state = tevent_req_data( + req, struct writer_state); + ssize_t nwritten; + int err = 0; + + nwritten = pkt_write_recv(subreq, &err); + TALLOC_FREE(subreq); + state->subreq = NULL; + if (nwritten == -1) { + tevent_req_error(req, err); + return; + } + + if (nwritten != state->buflen) { + tevent_req_error(req, EIO); + return; + } + + state->count++; + if (state->count >= 1000) { + tevent_req_done(req); + return; + } + + subreq = pkt_write_send(state, state->ev, state->fd, + state->buf, state->buflen); + if (tevent_req_nomem(subreq, req)) { + return; + } + + state->subreq = subreq; + tevent_req_set_callback(subreq, writer_next, req); +} + +static void writer_recv(struct tevent_req *req, int *perr) +{ + struct writer_state *state = tevent_req_data( + req, struct writer_state); + int err = 0; + + if (state->subreq != NULL) { + *perr = -1; + return; + } + + if (tevent_req_is_unix_error(req, &err)) { + *perr = err; + return; + } + + *perr = 0; +} + +static void writer_handler(struct tevent_context *ev, struct tevent_fd *fde, + uint16_t flags, void *private_data) +{ + struct tevent_req *req = talloc_get_type_abort( + private_data, struct tevent_req); + struct writer_state *state = tevent_req_data( + req, struct writer_state); + + assert(state->subreq != NULL); + pkt_write_handler(ev, fde, flags, state->subreq); +} + +static void writer(int fd) +{ + TALLOC_CTX *mem_ctx; + struct tevent_context *ev; + struct tevent_fd *fde; + struct tevent_req *req; + uint8_t buf[1024*1024]; + size_t buflen; + size_t pkt_size[4] = { 100, 500, 1024, 1024*1024 }; + int i, err; + + mem_ctx = talloc_new(NULL); + assert(mem_ctx != NULL); + + ev = tevent_context_init(mem_ctx); + assert(ev != NULL); + + for (i=0; i<1024*1024; i++) { + buf[i] = i%256; + } + + for (i=0; i<4; i++) { + buflen = pkt_size[i]; + *(uint32_t *)buf = buflen; + + req = writer_send(mem_ctx, ev, fd, buf, buflen); + assert(req != NULL); + + fde = tevent_add_fd(ev, mem_ctx, fd, TEVENT_FD_WRITE, + writer_handler, req); + assert(fde != NULL); + + tevent_req_poll(req, ev); + + writer_recv(req, &err); + assert(err == 0); + + talloc_free(fde); + talloc_free(req); + } + + close(fd); + + talloc_free(mem_ctx); +} + +struct reader_state { + struct tevent_context *ev; + int fd; + uint8_t buf[1024]; + struct tevent_req *subreq; +}; + +static ssize_t reader_more(uint8_t *buf, size_t buflen, void *private_data); +static void reader_done(struct tevent_req *subreq); + +static struct tevent_req *reader_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + int fd) +{ + struct tevent_req *req, *subreq; + struct reader_state *state; + + req = tevent_req_create(mem_ctx, &state, struct reader_state); + if (req == NULL) { + return NULL; + } + + state->ev = ev; + state->fd = fd; + + subreq = pkt_read_send(state, state->ev, state->fd, 4, + state->buf, 1024, reader_more, NULL); + if (tevent_req_nomem(subreq, req)) { + tevent_req_post(req, ev); + } + + state->subreq = subreq; + tevent_req_set_callback(subreq, reader_done, req); + return req; +} + +static ssize_t reader_more(uint8_t *buf, size_t buflen, void *private_data) +{ + uint32_t pkt_len; + + if (buflen < sizeof(pkt_len)) { + return sizeof(pkt_len) - buflen; + } + + pkt_len = *(uint32_t *)buf; + return pkt_len - buflen; +} + +static void reader_done(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct reader_state *state = tevent_req_data( + req, struct reader_state); + ssize_t nread; + uint8_t *buf; + bool free_buf; + int err = 0; + + nread = pkt_read_recv(subreq, state, &buf, &free_buf, &err); + TALLOC_FREE(subreq); + state->subreq = NULL; + if (nread == -1) { + if (err == EPIPE) { + tevent_req_done(req); + } else { + tevent_req_error(req, err); + } + return; + } + + if (free_buf) { + talloc_free(buf); + } + + subreq = pkt_read_send(state, state->ev, state->fd, 4, + state->buf, 1024, reader_more, NULL); + if (tevent_req_nomem(subreq, req)) { + return; + } + + state->subreq = subreq; + tevent_req_set_callback(subreq, reader_done, req); +} + +static void reader_recv(struct tevent_req *req, int *perr) +{ + struct reader_state *state = tevent_req_data( + req, struct reader_state); + int err = 0; + + if (state->subreq != NULL) { + *perr = -1; + } + + if (tevent_req_is_unix_error(req, &err)) { + *perr = err; + return; + } + + *perr = 0; +} + +static void reader_handler(struct tevent_context *ev, struct tevent_fd *fde, + uint16_t flags, void *private_data) +{ + struct tevent_req *req = talloc_get_type_abort( + private_data, struct tevent_req); + struct reader_state *state = tevent_req_data( + req, struct reader_state); + + assert(state->subreq != NULL); + pkt_read_handler(ev, fde, flags, state->subreq); +} + +static void reader(int fd) +{ + TALLOC_CTX *mem_ctx; + struct tevent_context *ev; + struct tevent_fd *fde; + struct tevent_req *req; + int err; + + mem_ctx = talloc_new(NULL); + assert(mem_ctx != NULL); + + ev = tevent_context_init(mem_ctx); + assert(ev != NULL); + + req = reader_send(mem_ctx, ev, fd); + assert(req != NULL); + + fde = tevent_add_fd(ev, mem_ctx, fd, TEVENT_FD_READ, + reader_handler, req); + assert(fde != NULL); + + tevent_req_poll(req, ev); + + reader_recv(req, &err); + assert(err == 0); + + close(fd); + + talloc_free(mem_ctx); +} + +static bool set_nonblocking(int fd) +{ + int v; + + v = fcntl(fd, F_GETFL, 0); + if (v == -1) { + return false; + } + if (fcntl(fd, F_SETFL, v | O_NONBLOCK) == -1) { + return false; + } + return true; +} + +int main(void) +{ + int fd[2]; + int ret; + pid_t pid; + + ret = pipe(fd); + assert(ret == 0); + + pid = fork(); + assert(pid != -1); + + if (pid == 0) { + /* Child process */ + close(fd[0]); + writer(fd[1]); + exit(0); + } + + close(fd[1]); + if (!set_nonblocking(fd[0])) { + exit(1); + } + + reader(fd[0]); + + return 0; +} diff --git a/ctdb/wscript b/ctdb/wscript index a4523d85c1e..454f03f9727 100755 --- a/ctdb/wscript +++ b/ctdb/wscript @@ -336,7 +336,7 @@ def build(bld): bld.SAMBA_SUBSYSTEM('ctdb-util', source=bld.SUBDIR('common', '''db_hash.c srvid.c reqid.c - pkt_read.c'''), + pkt_read.c pkt_write.c'''), deps='replace talloc tevent tdb tevent-unix-util') bld.SAMBA_SUBSYSTEM('ctdb-client', @@ -579,6 +579,7 @@ def build(bld): 'db_hash_test', 'srvid_test', 'pkt_read_test', + 'pkt_write_test', ] for target in ctdb_unit_tests: -- 2.11.4.GIT