/*

 *   teamd_workq.c - Teamd work queue

 *   Copyright (C) 2013-2015 Jiri Pirko <jiri@resnulli.us>

 *

 *   This library is free software; you can redistribute it and/or

 *   modify it under the terms of the GNU Lesser General Public

 *   License as published by the Free Software Foundation; either

 *   version 2.1 of the License, or (at your option) any later version.

 *

 *   This library 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

 *   Lesser General Public License for more details.

 *

 *   You should have received a copy of the GNU Lesser General Public

 *   License along with this library; if not, write to the Free Software

 *   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

 */

#include <stdlib.h>

#include <unistd.h>

#include <fcntl.h>

#include <errno.h>

#include <private/misc.h>

#include "teamd_workq.h"

#define WORKQ_CB_NAME "workq"

static int teamd_workq_callback_socket(struct teamd_context *ctx, int events,

				       void *priv)

{

	struct teamd_workq *workq;

	struct teamd_workq *tmp;

	char bytes[16];

	int ret;

	int err;

again:

	ret = read(ctx->workq.pipe_r, bytes, sizeof(bytes));

	if (ret == -1) {

		if (errno == EINTR)

			goto again;

		else if (errno != EAGAIN)

			return -errno;

	}

	teamd_loop_callback_disable(ctx, WORKQ_CB_NAME, ctx);

	list_for_each_node_entry_safe(workq, tmp, &ctx->workq.work_list, list) {

		list_del(&workq->list);

		list_init(&workq->list);

		err = workq->func(ctx, workq);

		if (err)

			return err;

	}

	return 0;

}

int teamd_workq_init(struct teamd_context *ctx)

{

	int fds[2];

	int err;

	list_init(&ctx->workq.work_list);

	err = pipe2(fds, O_NONBLOCK);

	if (err)

		return -errno;

	ctx->workq.pipe_r = fds[0];

	ctx->workq.pipe_w = fds[1];

	err = teamd_loop_callback_fd_add_tail(ctx, WORKQ_CB_NAME, ctx,

					      teamd_workq_callback_socket,

					      ctx->workq.pipe_r,

					      TEAMD_LOOP_FD_EVENT_READ);

	if (err) {

		teamd_log_err("Failed add workq callback.");

		goto close_pipe;

	}

	return 0;

close_pipe:

	close(ctx->workq.pipe_r);

	close(ctx->workq.pipe_w);

	return 0;

}

void teamd_workq_fini(struct teamd_context *ctx)

{

	struct teamd_workq *workq;

	struct teamd_workq *tmp;

	teamd_loop_callback_del(ctx, WORKQ_CB_NAME, ctx);

	close(ctx->workq.pipe_r);

	close(ctx->workq.pipe_w);

	list_for_each_node_entry_safe(workq, tmp, &ctx->workq.work_list, list) {

		list_del(&workq->list);

		list_init(&workq->list);

	}

}

static void teamd_workq_set_for_process(struct teamd_context *ctx)

{

	int err;

	const char byte = 0;

retry:

	err = write(ctx->workq.pipe_w, &byte, 1);

	if (err == -1 && errno == EINTR)

		goto retry;

	teamd_loop_callback_enable(ctx, WORKQ_CB_NAME, ctx);

}

void teamd_workq_schedule_work(struct teamd_context *ctx,

			       struct teamd_workq *workq)

{

	if (!list_empty(&workq->list))

		return;

	list_add_tail(&ctx->workq.work_list, &workq->list);

	teamd_workq_set_for_process(ctx);

}

void teamd_workq_init_work(struct teamd_workq *workq, teamd_workq_func_t func)

{

	workq->func = func;

	list_init(&workq->list);

}