samba-mirror/source3/lib/fncall.c

/*
 * Unix SMB/CIFS implementation.
 * Async fn calls
 * Copyright (C) Volker Lendecke 2009
 *
 * 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 "includes.h"
#include "../lib/util/tevent_unix.h"

#include "../lib/pthreadpool/pthreadpool_pipe.h"

struct fncall_state {
	struct fncall_context *ctx;
	int job_id;
	bool done;

	void *private_parent;
	void *job_private;
};

struct fncall_context {
	struct pthreadpool_pipe *pool;
	int next_job_id;
	int sig_fd;
	struct tevent_req **pending;

	struct fncall_state **orphaned;
	int num_orphaned;

	struct tevent_fd *fde;
};

static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
			   uint16_t flags, void *private_data);

static int fncall_context_destructor(struct fncall_context *ctx)
{
	while (talloc_array_length(ctx->pending) != 0) {
		/* No TALLOC_FREE here */
		talloc_free(ctx->pending[0]);
	}

	while (ctx->num_orphaned != 0) {
		/*
		 * We've got jobs in the queue for which the tevent_req has
		 * been finished already. Wait for all of them to finish.
		 */
		fncall_handler(NULL, NULL, TEVENT_FD_READ, ctx);
	}

	pthreadpool_pipe_destroy(ctx->pool);
	ctx->pool = NULL;

	return 0;
}

struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
					   int max_threads)
{
	struct fncall_context *ctx;
	int ret;

	ctx = talloc_zero(mem_ctx, struct fncall_context);
	if (ctx == NULL) {
		return NULL;
	}

	ret = pthreadpool_pipe_init(max_threads, &ctx->pool);
	if (ret != 0) {
		TALLOC_FREE(ctx);
		return NULL;
	}
	talloc_set_destructor(ctx, fncall_context_destructor);

	ctx->sig_fd = pthreadpool_pipe_signal_fd(ctx->pool);
	if (ctx->sig_fd == -1) {
		TALLOC_FREE(ctx);
		return NULL;
	}

	return ctx;
}

static int fncall_next_job_id(struct fncall_context *ctx)
{
	int num_pending = talloc_array_length(ctx->pending);
	int result;

	while (true) {
		int i;

		result = ctx->next_job_id++;
		if (result == 0) {
			continue;
		}

		for (i=0; i<num_pending; i++) {
			struct fncall_state *state = tevent_req_data(
				ctx->pending[i], struct fncall_state);

			if (result == state->job_id) {
				break;
			}
		}
		if (i == num_pending) {
			return result;
		}
	}
}

static void fncall_unset_pending(struct tevent_req *req);
static void fncall_cleanup(struct tevent_req *req,
			   enum tevent_req_state req_state);

static bool fncall_set_pending(struct tevent_req *req,
			       struct fncall_context *ctx,
			       struct tevent_context *ev)
{
	struct tevent_req **pending;
	int num_pending, orphaned_array_length;

	num_pending = talloc_array_length(ctx->pending);

	pending = talloc_realloc(ctx, ctx->pending, struct tevent_req *,
				 num_pending+1);
	if (pending == NULL) {
		return false;
	}
	pending[num_pending] = req;
	num_pending += 1;
	ctx->pending = pending;
	tevent_req_set_cleanup_fn(req, fncall_cleanup);

	/*
	 * Make sure that the orphaned array of fncall_state structs has
	 * enough space. A job can change from pending to orphaned in
	 * fncall_cleanup, and to fail in a talloc destructor should be
	 * avoided if possible.
	 */

	orphaned_array_length = talloc_array_length(ctx->orphaned);
	if (num_pending > orphaned_array_length) {
		struct fncall_state **orphaned;

		orphaned = talloc_realloc(ctx, ctx->orphaned,
					  struct fncall_state *,
					  orphaned_array_length + 1);
		if (orphaned == NULL) {
			fncall_unset_pending(req);
			return false;
		}
		ctx->orphaned = orphaned;
	}

	if (ctx->fde != NULL) {
		return true;
	}

	ctx->fde = tevent_add_fd(ev, ctx->pending, ctx->sig_fd, TEVENT_FD_READ,
				 fncall_handler, ctx);
	if (ctx->fde == NULL) {
		fncall_unset_pending(req);
		return false;
	}
	return true;
}

static void fncall_unset_pending(struct tevent_req *req)
{
	struct fncall_state *state = tevent_req_data(req, struct fncall_state);
	struct fncall_context *ctx = state->ctx;
	int num_pending = talloc_array_length(ctx->pending);
	int i;

	tevent_req_set_cleanup_fn(req, NULL);

	if (num_pending == 1) {
		TALLOC_FREE(ctx->fde);
		TALLOC_FREE(ctx->pending);
		return;
	}

	for (i=0; i<num_pending; i++) {
		if (req == ctx->pending[i]) {
			break;
		}
	}
	if (i == num_pending) {
		return;
	}
	if (num_pending > 1) {
		ctx->pending[i] = ctx->pending[num_pending-1];
	}
	ctx->pending = talloc_realloc(NULL, ctx->pending, struct tevent_req *,
				      num_pending - 1);
}

static void fncall_cleanup(struct tevent_req *req,
			   enum tevent_req_state req_state)
{
	struct fncall_state *state = tevent_req_data(
		req, struct fncall_state);
	struct fncall_context *ctx = state->ctx;

	switch (req_state) {
	case TEVENT_REQ_RECEIVED:
		break;
	default:
		return;
	}

	fncall_unset_pending(req);

	if (state->done) {
		return;
	}

	/*
	 * Keep around the state of the deleted request until the request has
	 * finished in the helper thread. fncall_handler will destroy it.
	 */
	ctx->orphaned[ctx->num_orphaned] = talloc_move(ctx->orphaned, &state);
	ctx->num_orphaned += 1;
}

struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
			       struct fncall_context *ctx,
			       void (*fn)(void *private_data),
			       void *private_data)
{
	struct tevent_req *req;
	struct fncall_state *state;
	int ret;

	req = tevent_req_create(mem_ctx, &state, struct fncall_state);
	if (req == NULL) {
		return NULL;
	}
	state->ctx = ctx;
	state->job_id = fncall_next_job_id(state->ctx);
	state->done = false;

	/*
	 * We need to keep the private data we handed out to the thread around
	 * as long as the job is not finished. This is a bit of an abstraction
	 * violation, because the "req->state1->subreq->state2" (we're
	 * "subreq", "req" is the request our caller creates) is broken to
	 * "ctx->state2->state1", but we are right now in the destructor for
	 * "subreq2", so what can we do. We need to keep state1 around,
	 * otherwise the helper thread will have no place to put its results.
	 */

	state->private_parent = talloc_parent(private_data);
	state->job_private = talloc_move(state, &private_data);

	ret = pthreadpool_pipe_add_job(state->ctx->pool, state->job_id, fn,
				       state->job_private);
	if (ret == -1) {
		tevent_req_error(req, errno);
		return tevent_req_post(req, ev);
	}
	if (!fncall_set_pending(req, state->ctx, ev)) {
		tevent_req_oom(req);
		return tevent_req_post(req, ev);
	}
	return req;
}

static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
			   uint16_t flags, void *private_data)
{
	struct fncall_context *ctx = talloc_get_type_abort(
		private_data, struct fncall_context);
	int i, num_pending;
	int job_id;

	if (pthreadpool_pipe_finished_jobs(ctx->pool, &job_id, 1) < 0) {
		return;
	}

	num_pending = talloc_array_length(ctx->pending);

	for (i=0; i<num_pending; i++) {
		struct fncall_state *state = tevent_req_data(
			ctx->pending[i], struct fncall_state);

		if (job_id == state->job_id) {
			state->done = true;
			talloc_move(state->private_parent,
				    &state->job_private);
			tevent_req_done(ctx->pending[i]);
			return;
		}
	}

	for (i=0; i<ctx->num_orphaned; i++) {
		if (job_id == ctx->orphaned[i]->job_id) {
			break;
		}
	}
	if (i == ctx->num_orphaned) {
		return;
	}

	TALLOC_FREE(ctx->orphaned[i]);

	if (i < ctx->num_orphaned-1) {
		ctx->orphaned[i] = ctx->orphaned[ctx->num_orphaned-1];
	}
	ctx->num_orphaned -= 1;
}

int fncall_recv(struct tevent_req *req, int *perr)
{
	if (tevent_req_is_unix_error(req, perr)) {
		return -1;
	}
	return 0;
}