/*
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 "lib/util/blocking.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 ((size_t)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];
memcpy(buf, &buflen, sizeof(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);
}
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]);
ret = set_blocking(fd[0], false);
if (ret == -1) {
exit(1);
}
reader(fd[0]);
return 0;
}