sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-01-08 21:18:16 +03:00
Code

ctdb-common: Add packet write abstraction

Browse Source 
Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>
This commit is contained in:
Amitay Isaacs 2015-04-06 17:26:29 +10:00 committed by Amitay Isaacs
parent c77d3bb183
commit e01c0eed38
5 changed files with 559 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
ctdb/common/pkt_write.c Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
*/
#include "replace.h"
#include "system/network.h"
#include <talloc.h>
#include <tevent.h>
#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;
}

79
ctdb/common/pkt_write.h Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
*/
#ifndef __CTDB_PKT_WRITE_H__
#define __CTDB_PKT_WRITE_H__
#include <talloc.h>
#include <tevent.h>
/**
* @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__ */

7
ctdb/tests/cunit/pkt_write_001.sh Executable file
View File

@ -0,0 +1,7 @@
#!/bin/sh
. "${TEST_SCRIPTS_DIR}/unit.sh"
ok_null
unit_test pkt_write_test

370
ctdb/tests/src/pkt_write_test.c Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
*/
#include "replace.h"
#include "system/filesys.h"
#include <assert.h>
#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;
}

3
ctdb/wscript
View File

@ -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: