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 read 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-09-02 22:32:50 +10:00 committed by Amitay Isaacs
parent facd3c864e
commit c77d3bb183
5 changed files with 542 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

190
ctdb/common/pkt_read.c Normal file
View File

@ -0,0 +1,190 @@
/*
Reading packets using fixed and dynamic buffer
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/>.
*/
/* This is similar to read_packet abstraction. The main different is that
* tevent fd event is created only once.
*/
#include "replace.h"
#include "system/network.h"
#include <talloc.h>
#include <tevent.h>
#include "lib/util/tevent_unix.h"
#include "pkt_read.h"
/*
* Read a packet using fixed buffer
*/
struct pkt_read_state {
int fd;
uint8_t *buf;
size_t buflen;
size_t nread, total;
bool use_fixed;
ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
void *private_data;
};
struct tevent_req *pkt_read_send(TALLOC_CTX *mem_ctx,
struct tevent_context *ev,
int fd, size_t initial,
uint8_t *buf, size_t buflen,
ssize_t (*more)(uint8_t *buf,
size_t buflen,
void *private_data),
void *private_data)
{
struct tevent_req *req;
struct pkt_read_state *state;
req = tevent_req_create(mem_ctx, &state, struct pkt_read_state);
if (req == NULL) {
return NULL;
}
state->fd = fd;
if (buf == NULL || buflen == 0) {
state->use_fixed = false;
state->buf = talloc_array(state, uint8_t, initial);
if (state->buf == NULL) {
talloc_free(req);
return NULL;
}
state->buflen = initial;
} else {
state->use_fixed = true;
state->buf = buf;
state->buflen = buflen;
}
state->nread = 0;
state->total = initial;
state->more = more;
state->private_data = private_data;
return req;
}
void pkt_read_handler(struct tevent_context *ev, struct tevent_fd *fde,
uint16_t flags, struct tevent_req *req)
{
struct pkt_read_state *state = tevent_req_data(
req, struct pkt_read_state);
ssize_t nread, more;
uint8_t *tmp;
nread = read(state->fd, state->buf + state->nread,
state->total - state->nread);
if ((nread == -1) && (errno == EINTR)) {
/* retry */
return;
}
if (nread == -1) {
tevent_req_error(req, errno);
return;
}
if (nread == 0) {
/* fd closed */
tevent_req_error(req, EPIPE);
return;
}
state->nread += nread;
if (state->nread < state->total) {
/* come back later */
return;
}
/* Check if "more" asks for more data */
if (state->more == NULL) {
tevent_req_done(req);
return;
}
more = state->more(state->buf, state->nread, state->private_data);
if (more == -1) {
/* invalid packet */
tevent_req_error(req, EIO);
return;
}
if (more == 0) {
tevent_req_done(req);
return;
}
if (state->total + more < state->total) {
/* int wrapped */
tevent_req_error(req, EMSGSIZE);
return;
}
if (state->total + more < state->buflen) {
/* continue using fixed buffer */
state->total += more;
return;
}
if (state->use_fixed) {
/* switch to dynamic buffer */
tmp = talloc_array(state, uint8_t, state->total + more);
if (tevent_req_nomem(tmp, req)) {
return;
}
memcpy(tmp, state->buf, state->total);
state->use_fixed = false;
} else {
tmp = talloc_realloc(state, state->buf, uint8_t,
state->total + more);
if (tevent_req_nomem(tmp, req)) {
return;
}
}
state->buf = tmp;
state->buflen = state->total + more;
state->total += more;
}
ssize_t pkt_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
uint8_t **pbuf, bool *free_buf, int *perrno)
{
struct pkt_read_state *state = tevent_req_data(
req, struct pkt_read_state);
if (tevent_req_is_unix_error(req, perrno)) {
return -1;
}
if (state->use_fixed) {
*pbuf = state->buf;
*free_buf = false;
} else {
*pbuf = talloc_steal(mem_ctx, state->buf);
*free_buf = true;
}
return state->total;
}

98
ctdb/common/pkt_read.h Normal file
View File

@ -0,0 +1,98 @@
/*
API for reading packets using fixed and dynamic buffer
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_READ_H__
#define __CTDB_PKT_READ_H__
#include <talloc.h>
#include <tevent.h>
/**
* @file pkt_read.h
*
* @brief Read a packet using fixed size buffer or allocated memory.
*
* CTDB communication uses lots of small packets. This abstraction avoids the
* need to allocate memory for small packets. Only if the received packet is
* larger than the fixed memory buffer, use talloc to allocate memory.
*/
/**
* @brief Start async computation to read a packet
*
* This returns a tevent request to read a packet from 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 read from
* @param[in] initial Initial amount of data to read
* @param[in] buf The static buffer to read data in
* @param[in] buflen The size of the static buffer
* @param[in] more The function to check if the bytes read forms a packet
* @param[in] private_data Private data to pass to more function
* @return new tevent request or NULL on failure
*/
struct tevent_req *pkt_read_send(TALLOC_CTX *mem_ctx,
struct tevent_context *ev,
int fd, size_t initial,
uint8_t *buf, size_t buflen,
ssize_t (*more)(uint8_t *buf,
size_t buflen,
void *private_data),
void *private_data);
/**
* @brief Function to actually read data from the socket
*
* This function should be called, when tevent fd event is triggered. This
* function has the syntax of tevent_fd_handler_t. The private_data for this
* function is the tevent request created by pkt_read_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_read_handler(struct tevent_context *ev, struct tevent_fd *fde,
uint16_t flags, struct tevent_req *req);
/**
* @brief Retrieve a packet
*
* This function returns the pkt read from fd.
*
* @param[in] req Tevent request
* @param[in] mem_ctx Talloc memory context
* @param[out] pbuf The pointer to the buffer
* @param[out] free_buf Boolean to indicate that caller should free buffer
* @param[out] perrno errno in case of failure
* @return the size of the pkt, or -1 on failure
*
* If the pkt data is dynamically allocated, then it is moved under the
* specified talloc memory context and free_buf is set to true. It is the
* responsibility of the caller to the free the memory returned.
*
* If the pkt data is stored in the fixed buffer, then free_buf is set to false.
*/
ssize_t pkt_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
uint8_t **pbuf, bool *free_buf, int *perrno);
#endif /* __CTDB_PKT_READ_H__ */

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

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

242
ctdb/tests/src/pkt_read_test.c Normal file
View File

@ -0,0 +1,242 @@
/*
packet read 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"
static void writer(int fd)
{
uint8_t buf[1024*1024];
size_t buflen;
size_t pkt_size[4] = { 100, 500, 1024, 1024*1024 };
int i, j;
int ret;
for (i=0; i<1024*1024; i++) {
buf[i] = i%256;
}
for (i=0; i<1000; i++) {
for (j=0; j<4; j++) {
buflen = pkt_size[j];
*(uint32_t *)buf = buflen;
ret = write(fd, buf, buflen);
if (ret < 0) {
printf("write error: %s\n", strerror(errno));
assert(ret > 0);
}
}
}
close(fd);
}
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;
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;
}

8
ctdb/wscript
View File

@ -335,8 +335,9 @@ def build(bld):
bld.SAMBA_SUBSYSTEM('ctdb-util',
source=bld.SUBDIR('common',
'''db_hash.c srvid.c reqid.c'''),
deps='replace talloc tevent tdb')
'''db_hash.c srvid.c reqid.c
pkt_read.c'''),
deps='replace talloc tevent tdb tevent-unix-util')
bld.SAMBA_SUBSYSTEM('ctdb-client',
source=bld.SUBDIR('client', 'ctdb_client.c'),
@ -577,6 +578,7 @@ def build(bld):
ctdb_unit_tests = [
'db_hash_test',
'srvid_test',
'pkt_read_test',
]
for target in ctdb_unit_tests:
@ -584,7 +586,7 @@ def build(bld):
bld.SAMBA_BINARY(target,
source=src,
deps='talloc tevent tdb',
deps='talloc tevent tdb tevent-unix-util',
install_path='${CTDB_TEST_LIBDIR}')
bld.SAMBA_BINARY('reqid_test',