cmservice.c

/* AFS Cache Manager Service
 *
 * Copyright (C) 2002 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 *
 * 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
 * 2 of the License, or (at your option) any later version.
 */

#include <linux/module.h>
#include <linux/init.h>
#include <linux/sched.h>
#include <linux/completion.h>
#include "server.h"
#include "cell.h"
#include "transport.h"
#include <rxrpc/rxrpc.h>
#include <rxrpc/transport.h>
#include <rxrpc/connection.h>
#include <rxrpc/call.h>
#include "cmservice.h"
#include "internal.h"

static unsigned afscm_usage;		/* AFS cache manager usage count */
static struct rw_semaphore afscm_sem;	/* AFS cache manager start/stop semaphore */

static int afscm_new_call(struct rxrpc_call *call);
static void afscm_attention(struct rxrpc_call *call);
static void afscm_error(struct rxrpc_call *call);
static void afscm_aemap(struct rxrpc_call *call);

static void _SRXAFSCM_CallBack(struct rxrpc_call *call);
static void _SRXAFSCM_InitCallBackState(struct rxrpc_call *call);
static void _SRXAFSCM_Probe(struct rxrpc_call *call);

typedef void (*_SRXAFSCM_xxxx_t)(struct rxrpc_call *call);

static const struct rxrpc_operation AFSCM_ops[] = {
	{
		.id	= 204,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "CallBack",
		.user	= _SRXAFSCM_CallBack,
	},
	{
		.id	= 205,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "InitCallBackState",
		.user	= _SRXAFSCM_InitCallBackState,
	},
	{
		.id	= 206,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "Probe",
		.user	= _SRXAFSCM_Probe,
	},
#if 0
	{
		.id	= 207,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "GetLock",
		.user	= _SRXAFSCM_GetLock,
	},
	{
		.id	= 208,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "GetCE",
		.user	= _SRXAFSCM_GetCE,
	},
	{
		.id	= 209,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "GetXStatsVersion",
		.user	= _SRXAFSCM_GetXStatsVersion,
	},
	{
		.id	= 210,
		.asize	= RXRPC_APP_MARK_EOF,
		.name	= "GetXStats",
		.user	= _SRXAFSCM_GetXStats,
	}
#endif
};

static struct rxrpc_service AFSCM_service = {
	.name		= "AFS/CM",
	.owner		= THIS_MODULE,
	.link		= LIST_HEAD_INIT(AFSCM_service.link),
	.new_call	= afscm_new_call,
	.service_id	= 1,
	.attn_func	= afscm_attention,
	.error_func	= afscm_error,
	.aemap_func	= afscm_aemap,
	.ops_begin	= &AFSCM_ops[0],
	.ops_end	= &AFSCM_ops[ARRAY_SIZE(AFSCM_ops)],
};

static DECLARE_COMPLETION(kafscmd_alive);
static DECLARE_COMPLETION(kafscmd_dead);
static DECLARE_WAIT_QUEUE_HEAD(kafscmd_sleepq);
static LIST_HEAD(kafscmd_attention_list);
static LIST_HEAD(afscm_calls);
static DEFINE_SPINLOCK(afscm_calls_lock);
static DEFINE_SPINLOCK(kafscmd_attention_lock);
static int kafscmd_die;

/*
 * AFS Cache Manager kernel thread
 */
static int kafscmd(void *arg)
{
	DECLARE_WAITQUEUE(myself, current);

	struct rxrpc_call *call;
	_SRXAFSCM_xxxx_t func;
	int die;

	printk(KERN_INFO "kAFS: Started kafscmd %d\n", current->pid);

	daemonize("kafscmd");

	complete(&kafscmd_alive);

	/* loop around looking for things to attend to */
	do {
		if (list_empty(&kafscmd_attention_list)) {
			set_current_state(TASK_INTERRUPTIBLE);
			add_wait_queue(&kafscmd_sleepq, &myself);

			for (;;) {
				set_current_state(TASK_INTERRUPTIBLE);
				if (!list_empty(&kafscmd_attention_list) ||
				    signal_pending(current) ||
				    kafscmd_die)
					break;

				schedule();
			}

			remove_wait_queue(&kafscmd_sleepq, &myself);
			set_current_state(TASK_RUNNING);
		}

		die = kafscmd_die;

		/* dequeue the next call requiring attention */
		call = NULL;
		spin_lock(&kafscmd_attention_lock);

		if (!list_empty(&kafscmd_attention_list)) {
			call = list_entry(kafscmd_attention_list.next,
					  struct rxrpc_call,
					  app_attn_link);
			list_del_init(&call->app_attn_link);
			die = 0;
		}

		spin_unlock(&kafscmd_attention_lock);

		if (call) {
			/* act upon it */
			_debug("@@@ Begin Attend Call %p", call);

			func = call->app_user;
			if (func)
				func(call);

			rxrpc_put_call(call);

			_debug("@@@ End Attend Call %p", call);
		}

	} while(!die);

	/* and that's all */
	complete_and_exit(&kafscmd_dead, 0);
}

/*
 * handle a call coming in to the cache manager
 * - if I want to keep the call, I must increment its usage count
 * - the return value will be negated and passed back in an abort packet if
 *   non-zero
 * - serialised by virtue of there only being one krxiod
 */
static int afscm_new_call(struct rxrpc_call *call)
{
	_enter("%p{cid=%u u=%d}",
	       call, ntohl(call->call_id), atomic_read(&call->usage));

	rxrpc_get_call(call);

	/* add to my current call list */
	spin_lock(&afscm_calls_lock);
	list_add(&call->app_link,&afscm_calls);
	spin_unlock(&afscm_calls_lock);

	_leave(" = 0");
	return 0;
}

/*
 * queue on the kafscmd queue for attention
 */
static void afscm_attention(struct rxrpc_call *call)
{
	_enter("%p{cid=%u u=%d}",
	       call, ntohl(call->call_id), atomic_read(&call->usage));

	spin_lock(&kafscmd_attention_lock);

	if (list_empty(&call->app_attn_link)) {
		list_add_tail(&call->app_attn_link, &kafscmd_attention_list);
		rxrpc_get_call(call);
	}

	spin_unlock(&kafscmd_attention_lock);

	wake_up(&kafscmd_sleepq);

	_leave(" {u=%d}", atomic_read(&call->usage));
}

/*
 * handle my call being aborted
 * - clean up, dequeue and put my ref to the call
 */
static void afscm_error(struct rxrpc_call *call)
{
	int removed;

	_enter("%p{est=%s ac=%u er=%d}",
	       call,
	       rxrpc_call_error_states[call->app_err_state],
	       call->app_abort_code,
	       call->app_errno);

	spin_lock(&kafscmd_attention_lock);

	if (list_empty(&call->app_attn_link)) {
		list_add_tail(&call->app_attn_link, &kafscmd_attention_list);
		rxrpc_get_call(call);
	}

	spin_unlock(&kafscmd_attention_lock);

	removed = 0;
	spin_lock(&afscm_calls_lock);
	if (!list_empty(&call->app_link)) {
		list_del_init(&call->app_link);
		removed = 1;
	}
	spin_unlock(&afscm_calls_lock);

	if (removed)
		rxrpc_put_call(call);

	wake_up(&kafscmd_sleepq);

	_leave("");
}

/*
 * map afs abort codes to/from Linux error codes
 * - called with call->lock held
 */
static void afscm_aemap(struct rxrpc_call *call)
{
	switch (call->app_err_state) {
	case RXRPC_ESTATE_LOCAL_ABORT:
		call->app_abort_code = -call->app_errno;
		break;
	case RXRPC_ESTATE_PEER_ABORT:
		call->app_errno = -ECONNABORTED;
		break;
	default:
		break;
	}
}

/*
 * start the cache manager service if not already started
 */
int afscm_start(void)
{
	int ret;

	down_write(&afscm_sem);
	if (!afscm_usage) {
		ret = kernel_thread(kafscmd, NULL, 0);
		if (ret < 0)
			goto out;

		wait_for_completion(&kafscmd_alive);

		ret = rxrpc_add_service(afs_transport, &AFSCM_service);
		if (ret < 0)
			goto kill;

		afs_kafstimod_add_timer(&afs_mntpt_expiry_timer,
					afs_mntpt_expiry_timeout * HZ);
	}

	afscm_usage++;
	up_write(&afscm_sem);

	return 0;

kill:
	kafscmd_die = 1;
	wake_up(&kafscmd_sleepq);
	wait_for_completion(&kafscmd_dead);

out:
	up_write(&afscm_sem);
	return ret;
}

/*
 * stop the cache manager service
 */
void afscm_stop(void)
{
	struct rxrpc_call *call;

	down_write(&afscm_sem);

	BUG_ON(afscm_usage == 0);
	afscm_usage--;

	if (afscm_usage == 0) {
		/* don't want more incoming calls */
		rxrpc_del_service(afs_transport, &AFSCM_service);

		/* abort any calls I've still got open (the afscm_error() will
		 * dequeue them) */
		spin_lock(&afscm_calls_lock);
		while (!list_empty(&afscm_calls)) {
			call = list_entry(afscm_calls.next,
					  struct rxrpc_call,
					  app_link);

			list_del_init(&call->app_link);
			rxrpc_get_call(call);
			spin_unlock(&afscm_calls_lock);

			rxrpc_call_abort(call, -ESRCH); /* abort, dequeue and
							 * put */

			_debug("nuking active call %08x.%d",
			       ntohl(call->conn->conn_id),
			       ntohl(call->call_id));
			rxrpc_put_call(call);
			rxrpc_put_call(call);

			spin_lock(&afscm_calls_lock);
		}
		spin_unlock(&afscm_calls_lock);

		/* get rid of my daemon */
		kafscmd_die = 1;
		wake_up(&kafscmd_sleepq);
		wait_for_completion(&kafscmd_dead);

		/* dispose of any calls waiting for attention */
		spin_lock(&kafscmd_attention_lock);
		while (!list_empty(&kafscmd_attention_list)) {
			call = list_entry(kafscmd_attention_list.next,
					  struct rxrpc_call,
					  app_attn_link);

			list_del_init(&call->app_attn_link);
			spin_unlock(&kafscmd_attention_lock);

			rxrpc_put_call(call);

			spin_lock(&kafscmd_attention_lock);
		}
		spin_unlock(&kafscmd_attention_lock);

		afs_kafstimod_del_timer(&afs_mntpt_expiry_timer);
	}

	up_write(&afscm_sem);
}

/*
 * handle the fileserver breaking a set of callbacks
 */
static void _SRXAFSCM_CallBack(struct rxrpc_call *call)
{
	struct afs_server *server;
	size_t count, qty, tmp;
	int ret = 0, removed;

	_enter("%p{acs=%s}", call, rxrpc_call_states[call->app_call_state]);

	server = afs_server_get_from_peer(call->conn->peer);

	switch (call->app_call_state) {
		/* we've received the last packet
		 * - drain all the data from the call and send the reply
		 */
	case RXRPC_CSTATE_SRVR_GOT_ARGS:
		ret = -EBADMSG;
		qty = call->app_ready_qty;
		if (qty < 8 || qty > 50 * (6 * 4) + 8)
			break;

		{
			struct afs_callback *cb, *pcb;
			int loop;
			__be32 *fp, *bp;

			fp = rxrpc_call_alloc_scratch(call, qty);

			/* drag the entire argument block out to the scratch
			 * space */
			ret = rxrpc_call_read_data(call, fp, qty, 0);
			if (ret < 0)
				break;

			/* and unmarshall the parameter block */
			ret = -EBADMSG;
			count = ntohl(*fp++);
			if (count>AFSCBMAX ||
			    (count * (3 * 4) + 8 != qty &&
			     count * (6 * 4) + 8 != qty))
				break;

			bp = fp + count*3;
			tmp = ntohl(*bp++);
			if (tmp > 0 && tmp != count)
				break;
			if (tmp == 0)
				bp = NULL;

			pcb = cb = rxrpc_call_alloc_scratch_s(
				call, struct afs_callback);

			for (loop = count - 1; loop >= 0; loop--) {
				pcb->fid.vid	= ntohl(*fp++);
				pcb->fid.vnode	= ntohl(*fp++);
				pcb->fid.unique	= ntohl(*fp++);
				if (bp) {
					pcb->version	= ntohl(*bp++);
					pcb->expiry	= ntohl(*bp++);
					pcb->type	= ntohl(*bp++);
				} else {
					pcb->version	= 0;
					pcb->expiry	= 0;
					pcb->type	= AFSCM_CB_UNTYPED;
				}
				pcb++;
			}

			/* invoke the actual service routine */
			ret = SRXAFSCM_CallBack(server, count, cb);
			if (ret < 0)
				break;
		}

		/* send the reply */
		ret = rxrpc_call_write_data(call, 0, NULL, RXRPC_LAST_PACKET,
					    GFP_KERNEL, 0, &count);
		if (ret < 0)
			break;
		break;

		/* operation complete */
	case RXRPC_CSTATE_COMPLETE:
		call->app_user = NULL;
		removed = 0;
		spin_lock(&afscm_calls_lock);
		if (!list_empty(&call->app_link)) {
			list_del_init(&call->app_link);
			removed = 1;
		}
		spin_unlock(&afscm_calls_lock);

		if (removed)
			rxrpc_put_call(call);
		break;

		/* operation terminated on error */
	case RXRPC_CSTATE_ERROR:
		call->app_user = NULL;
		break;

	default:
		break;
	}

	if (ret < 0)
		rxrpc_call_abort(call, ret);

	afs_put_server(server);

	_leave(" = %d", ret);
}

/*
 * handle the fileserver asking us to initialise our callback state
 */
static void _SRXAFSCM_InitCallBackState(struct rxrpc_call *call)
{
	struct afs_server *server;
	size_t count;
	int ret = 0, removed;

	_enter("%p{acs=%s}", call, rxrpc_call_states[call->app_call_state]);

	server = afs_server_get_from_peer(call->conn->peer);

	switch (call->app_call_state) {
		/* we've received the last packet - drain all the data from the
		 * call */
	case RXRPC_CSTATE_SRVR_GOT_ARGS:
		/* shouldn't be any args */
		ret = -EBADMSG;
		break;

		/* send the reply when asked for it */
	case RXRPC_CSTATE_SRVR_SND_REPLY:
		/* invoke the actual service routine */
		ret = SRXAFSCM_InitCallBackState(server);
		if (ret < 0)
			break;

		ret = rxrpc_call_write_data(call, 0, NULL, RXRPC_LAST_PACKET,
					    GFP_KERNEL, 0, &count);
		if (ret < 0)
			break;
		break;

		/* operation complete */
	case RXRPC_CSTATE_COMPLETE:
		call->app_user = NULL;
		removed = 0;
		spin_lock(&afscm_calls_lock);
		if (!list_empty(&call->app_link)) {
			list_del_init(&call->app_link);
			removed = 1;
		}
		spin_unlock(&afscm_calls_lock);

		if (removed)
			rxrpc_put_call(call);
		break;

		/* operation terminated on error */
	case RXRPC_CSTATE_ERROR:
		call->app_user = NULL;
		break;

	default:
		break;
	}

	if (ret < 0)
		rxrpc_call_abort(call, ret);

	afs_put_server(server);

	_leave(" = %d", ret);
}

/*
 * handle a probe from a fileserver
 */
static void _SRXAFSCM_Probe(struct rxrpc_call *call)
{
	struct afs_server *server;
	size_t count;
	int ret = 0, removed;

	_enter("%p{acs=%s}", call, rxrpc_call_states[call->app_call_state]);

	server = afs_server_get_from_peer(call->conn->peer);

	switch (call->app_call_state) {
		/* we've received the last packet - drain all the data from the
		 * call */
	case RXRPC_CSTATE_SRVR_GOT_ARGS:
		/* shouldn't be any args */
		ret = -EBADMSG;
		break;

		/* send the reply when asked for it */
	case RXRPC_CSTATE_SRVR_SND_REPLY:
		/* invoke the actual service routine */
		ret = SRXAFSCM_Probe(server);
		if (ret < 0)
			break;

		ret = rxrpc_call_write_data(call, 0, NULL, RXRPC_LAST_PACKET,
					    GFP_KERNEL, 0, &count);
		if (ret < 0)
			break;
		break;

		/* operation complete */
	case RXRPC_CSTATE_COMPLETE:
		call->app_user = NULL;
		removed = 0;
		spin_lock(&afscm_calls_lock);
		if (!list_empty(&call->app_link)) {
			list_del_init(&call->app_link);
			removed = 1;
		}
		spin_unlock(&afscm_calls_lock);

		if (removed)
			rxrpc_put_call(call);
		break;

		/* operation terminated on error */
	case RXRPC_CSTATE_ERROR:
		call->app_user = NULL;
		break;

	default:
		break;
	}

	if (ret < 0)
		rxrpc_call_abort(call, ret);

	afs_put_server(server);

	_leave(" = %d", ret);
}