[linux-2.6-block.git] / net / rxrpc / ar-local.c

/* AF_RXRPC local endpoint management
 *
 * Copyright (C) 2007 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/net.h>
#include <linux/skbuff.h>
#include <linux/slab.h>
#include <linux/udp.h>
#include <linux/ip.h>
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include <generated/utsrelease.h>
#include "ar-internal.h"

static const char rxrpc_version_string[65] = "linux-" UTS_RELEASE " AF_RXRPC";

static LIST_HEAD(rxrpc_locals);
DEFINE_RWLOCK(rxrpc_local_lock);
static DECLARE_RWSEM(rxrpc_local_sem);
static DECLARE_WAIT_QUEUE_HEAD(rxrpc_local_wq);

static void rxrpc_destroy_local(struct work_struct *work);
static void rxrpc_process_local_events(struct work_struct *work);

/*
 * allocate a new local
 */
static
struct rxrpc_local *rxrpc_alloc_local(struct sockaddr_rxrpc *srx)
{
	struct rxrpc_local *local;

	local = kzalloc(sizeof(struct rxrpc_local), GFP_KERNEL);
	if (local) {
		INIT_WORK(&local->destroyer, &rxrpc_destroy_local);
		INIT_WORK(&local->acceptor, &rxrpc_accept_incoming_calls);
		INIT_WORK(&local->rejecter, &rxrpc_reject_packets);
		INIT_WORK(&local->event_processor, &rxrpc_process_local_events);
		INIT_LIST_HEAD(&local->services);
		INIT_LIST_HEAD(&local->link);
		init_rwsem(&local->defrag_sem);
		skb_queue_head_init(&local->accept_queue);
		skb_queue_head_init(&local->reject_queue);
		skb_queue_head_init(&local->event_queue);
		spin_lock_init(&local->lock);
		rwlock_init(&local->services_lock);
		atomic_set(&local->usage, 1);
		local->debug_id = atomic_inc_return(&rxrpc_debug_id);
		memcpy(&local->srx, srx, sizeof(*srx));
	}

	_leave(" = %p", local);
	return local;
}

/*
 * create the local socket
 * - must be called with rxrpc_local_sem writelocked
 */
static int rxrpc_create_local(struct rxrpc_local *local)
{
	struct sock *sock;
	int ret, opt;

	_enter("%p{%d}", local, local->srx.transport_type);

	/* create a socket to represent the local endpoint */
	ret = sock_create_kern(&init_net, PF_INET, local->srx.transport_type,
			       IPPROTO_UDP, &local->socket);
	if (ret < 0) {
		_leave(" = %d [socket]", ret);
		return ret;
	}

	/* if a local address was supplied then bind it */
	if (local->srx.transport_len > sizeof(sa_family_t)) {
		_debug("bind");
		ret = kernel_bind(local->socket,
				  (struct sockaddr *) &local->srx.transport,
				  local->srx.transport_len);
		if (ret < 0) {
			_debug("bind failed");
			goto error;
		}
	}

	/* we want to receive ICMP errors */
	opt = 1;
	ret = kernel_setsockopt(local->socket, SOL_IP, IP_RECVERR,
				(char *) &opt, sizeof(opt));
	if (ret < 0) {
		_debug("setsockopt failed");
		goto error;
	}

	/* we want to set the don't fragment bit */
	opt = IP_PMTUDISC_DO;
	ret = kernel_setsockopt(local->socket, SOL_IP, IP_MTU_DISCOVER,
				(char *) &opt, sizeof(opt));
	if (ret < 0) {
		_debug("setsockopt failed");
		goto error;
	}

	write_lock_bh(&rxrpc_local_lock);
	list_add(&local->link, &rxrpc_locals);
	write_unlock_bh(&rxrpc_local_lock);

	/* set the socket up */
	sock = local->socket->sk;
	sock->sk_user_data	= local;
	sock->sk_data_ready	= rxrpc_data_ready;
	sock->sk_error_report	= rxrpc_UDP_error_report;
	_leave(" = 0");
	return 0;

error:
	kernel_sock_shutdown(local->socket, SHUT_RDWR);
	local->socket->sk->sk_user_data = NULL;
	sock_release(local->socket);
	local->socket = NULL;

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

/*
 * create a new local endpoint using the specified UDP address
 */
struct rxrpc_local *rxrpc_lookup_local(struct sockaddr_rxrpc *srx)
{
	struct rxrpc_local *local;
	int ret;

	_enter("{%d,%u,%pI4+%hu}",
	       srx->transport_type,
	       srx->transport.family,
	       &srx->transport.sin.sin_addr,
	       ntohs(srx->transport.sin.sin_port));

	down_write(&rxrpc_local_sem);

	/* see if we have a suitable local local endpoint already */
	read_lock_bh(&rxrpc_local_lock);

	list_for_each_entry(local, &rxrpc_locals, link) {
		_debug("CMP {%d,%u,%pI4+%hu}",
		       local->srx.transport_type,
		       local->srx.transport.family,
		       &local->srx.transport.sin.sin_addr,
		       ntohs(local->srx.transport.sin.sin_port));

		if (local->srx.transport_type != srx->transport_type ||
		    local->srx.transport.family != srx->transport.family)
			continue;

		switch (srx->transport.family) {
		case AF_INET:
			if (local->srx.transport.sin.sin_port !=
			    srx->transport.sin.sin_port)
				continue;
			if (memcmp(&local->srx.transport.sin.sin_addr,
				   &srx->transport.sin.sin_addr,
				   sizeof(struct in_addr)) != 0)
				continue;
			goto found_local;

		default:
			BUG();
		}
	}

	read_unlock_bh(&rxrpc_local_lock);

	/* we didn't find one, so we need to create one */
	local = rxrpc_alloc_local(srx);
	if (!local) {
		up_write(&rxrpc_local_sem);
		return ERR_PTR(-ENOMEM);
	}

	ret = rxrpc_create_local(local);
	if (ret < 0) {
		up_write(&rxrpc_local_sem);
		kfree(local);
		_leave(" = %d", ret);
		return ERR_PTR(ret);
	}

	up_write(&rxrpc_local_sem);

	_net("LOCAL new %d {%d,%u,%pI4+%hu}",
	     local->debug_id,
	     local->srx.transport_type,
	     local->srx.transport.family,
	     &local->srx.transport.sin.sin_addr,
	     ntohs(local->srx.transport.sin.sin_port));

	_leave(" = %p [new]", local);
	return local;

found_local:
	rxrpc_get_local(local);
	read_unlock_bh(&rxrpc_local_lock);
	up_write(&rxrpc_local_sem);

	_net("LOCAL old %d {%d,%u,%pI4+%hu}",
	     local->debug_id,
	     local->srx.transport_type,
	     local->srx.transport.family,
	     &local->srx.transport.sin.sin_addr,
	     ntohs(local->srx.transport.sin.sin_port));

	_leave(" = %p [reuse]", local);
	return local;
}

/*
 * release a local endpoint
 */
void rxrpc_put_local(struct rxrpc_local *local)
{
	_enter("%p{u=%d}", local, atomic_read(&local->usage));

	ASSERTCMP(atomic_read(&local->usage), >, 0);

	/* to prevent a race, the decrement and the dequeue must be effectively
	 * atomic */
	write_lock_bh(&rxrpc_local_lock);
	if (unlikely(atomic_dec_and_test(&local->usage))) {
		_debug("destroy local");
		rxrpc_queue_work(&local->destroyer);
	}
	write_unlock_bh(&rxrpc_local_lock);
	_leave("");
}

/*
 * destroy a local endpoint
 */
static void rxrpc_destroy_local(struct work_struct *work)
{
	struct rxrpc_local *local =
		container_of(work, struct rxrpc_local, destroyer);

	_enter("%p{%d}", local, atomic_read(&local->usage));

	down_write(&rxrpc_local_sem);

	write_lock_bh(&rxrpc_local_lock);
	if (atomic_read(&local->usage) > 0) {
		write_unlock_bh(&rxrpc_local_lock);
		up_read(&rxrpc_local_sem);
		_leave(" [resurrected]");
		return;
	}

	list_del(&local->link);
	local->socket->sk->sk_user_data = NULL;
	write_unlock_bh(&rxrpc_local_lock);

	downgrade_write(&rxrpc_local_sem);

	ASSERT(list_empty(&local->services));
	ASSERT(!work_pending(&local->acceptor));
	ASSERT(!work_pending(&local->rejecter));
	ASSERT(!work_pending(&local->event_processor));

	/* finish cleaning up the local descriptor */
	rxrpc_purge_queue(&local->accept_queue);
	rxrpc_purge_queue(&local->reject_queue);
	rxrpc_purge_queue(&local->event_queue);
	kernel_sock_shutdown(local->socket, SHUT_RDWR);
	sock_release(local->socket);

	up_read(&rxrpc_local_sem);

	_net("DESTROY LOCAL %d", local->debug_id);
	kfree(local);

	if (list_empty(&rxrpc_locals))
		wake_up_all(&rxrpc_local_wq);

	_leave("");
}

/*
 * preemptively destroy all local local endpoint rather than waiting for
 * them to be destroyed
 */
void __exit rxrpc_destroy_all_locals(void)
{
	DECLARE_WAITQUEUE(myself,current);

	_enter("");

	/* we simply have to wait for them to go away */
	if (!list_empty(&rxrpc_locals)) {
		set_current_state(TASK_UNINTERRUPTIBLE);
		add_wait_queue(&rxrpc_local_wq, &myself);

		while (!list_empty(&rxrpc_locals)) {
			schedule();
			set_current_state(TASK_UNINTERRUPTIBLE);
		}

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

	_leave("");
}

/*
 * Reply to a version request
 */
static void rxrpc_send_version_request(struct rxrpc_local *local,
				       struct rxrpc_host_header *hdr,
				       struct sk_buff *skb)
{
	struct rxrpc_wire_header whdr;
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	struct sockaddr_in sin;
	struct msghdr msg;
	struct kvec iov[2];
	size_t len;
	int ret;

	_enter("");

	sin.sin_family = AF_INET;
	sin.sin_port = udp_hdr(skb)->source;
	sin.sin_addr.s_addr = ip_hdr(skb)->saddr;

	msg.msg_name	= &sin;
	msg.msg_namelen	= sizeof(sin);
	msg.msg_control	= NULL;
	msg.msg_controllen = 0;
	msg.msg_flags	= 0;

	whdr.epoch	= htonl(sp->hdr.epoch);
	whdr.cid	= htonl(sp->hdr.cid);
	whdr.callNumber	= htonl(sp->hdr.callNumber);
	whdr.seq	= 0;
	whdr.serial	= 0;
	whdr.type	= RXRPC_PACKET_TYPE_VERSION;
	whdr.flags	= RXRPC_LAST_PACKET | (~hdr->flags & RXRPC_CLIENT_INITIATED);
	whdr.userStatus	= 0;
	whdr.securityIndex = 0;
	whdr._rsvd	= 0;
	whdr.serviceId	= htons(sp->hdr.serviceId);

	iov[0].iov_base	= &whdr;
	iov[0].iov_len	= sizeof(whdr);
	iov[1].iov_base	= (char *)rxrpc_version_string;
	iov[1].iov_len	= sizeof(rxrpc_version_string);

	len = iov[0].iov_len + iov[1].iov_len;

	_proto("Tx VERSION (reply)");

	ret = kernel_sendmsg(local->socket, &msg, iov, 2, len);
	if (ret < 0)
		_debug("sendmsg failed: %d", ret);

	_leave("");
}

/*
 * Process event packets targetted at a local endpoint.
 */
static void rxrpc_process_local_events(struct work_struct *work)
{
	struct rxrpc_local *local = container_of(work, struct rxrpc_local, event_processor);
	struct sk_buff *skb;
	char v;

	_enter("");

	atomic_inc(&local->usage);
	
	while ((skb = skb_dequeue(&local->event_queue))) {
		struct rxrpc_skb_priv *sp = rxrpc_skb(skb);

		_debug("{%d},{%u}", local->debug_id, sp->hdr.type);

		switch (sp->hdr.type) {
		case RXRPC_PACKET_TYPE_VERSION:
			if (skb_copy_bits(skb, 0, &v, 1) < 0)
				return;
			_proto("Rx VERSION { %02x }", v);
			if (v == 0)
				rxrpc_send_version_request(local, &sp->hdr, skb);
			break;

		default:
			/* Just ignore anything we don't understand */
			break;
		}

		rxrpc_put_local(local);
		rxrpc_free_skb(skb);
	}

	rxrpc_put_local(local);
	_leave("");
}
Commit	Line	Data
17926a79 DH	1	/* AF_RXRPC local endpoint management
	2	*
	3	* Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
	4	* Written by David Howells (dhowells@redhat.com)
	5	*
	6	* This program is free software; you can redistribute it and/or
	7	* modify it under the terms of the GNU General Public License
	8	* as published by the Free Software Foundation; either version
	9	* 2 of the License, or (at your option) any later version.
	10	*/
	11
	12	#include <linux/module.h>
	13	#include <linux/net.h>
	14	#include <linux/skbuff.h>
5a0e3ad6	15	#include <linux/slab.h>
44ba0698 DH	16	#include <linux/udp.h>
44ba0698 DH	17	#include <linux/ip.h>
17926a79 DH	18	#include <net/sock.h>
17926a79 DH	19	#include <net/af_rxrpc.h>
44ba0698	20	#include <generated/utsrelease.h>
17926a79 DH	21	#include "ar-internal.h"
17926a79 DH	22
44ba0698 DH	23	static const char rxrpc_version_string[65] = "linux-" UTS_RELEASE " AF_RXRPC";
44ba0698 DH	24
17926a79 DH	25	static LIST_HEAD(rxrpc_locals);
	26	DEFINE_RWLOCK(rxrpc_local_lock);
	27	static DECLARE_RWSEM(rxrpc_local_sem);
	28	static DECLARE_WAIT_QUEUE_HEAD(rxrpc_local_wq);
	29
	30	static void rxrpc_destroy_local(struct work_struct *work);
44ba0698	31	static void rxrpc_process_local_events(struct work_struct *work);
17926a79 DH	32
	33	/*
	34	* allocate a new local
	35	*/
	36	static
	37	struct rxrpc_local rxrpc_alloc_local(struct sockaddr_rxrpc srx)
	38	{
	39	struct rxrpc_local *local;
	40
	41	local = kzalloc(sizeof(struct rxrpc_local), GFP_KERNEL);
	42	if (local) {
	43	INIT_WORK(&local->destroyer, &rxrpc_destroy_local);
	44	INIT_WORK(&local->acceptor, &rxrpc_accept_incoming_calls);
	45	INIT_WORK(&local->rejecter, &rxrpc_reject_packets);
44ba0698	46	INIT_WORK(&local->event_processor, &rxrpc_process_local_events);
17926a79 DH	47	INIT_LIST_HEAD(&local->services);
	48	INIT_LIST_HEAD(&local->link);
	49	init_rwsem(&local->defrag_sem);
	50	skb_queue_head_init(&local->accept_queue);
	51	skb_queue_head_init(&local->reject_queue);
44ba0698	52	skb_queue_head_init(&local->event_queue);
17926a79 DH	53	spin_lock_init(&local->lock);
	54	rwlock_init(&local->services_lock);
	55	atomic_set(&local->usage, 1);
	56	local->debug_id = atomic_inc_return(&rxrpc_debug_id);
	57	memcpy(&local->srx, srx, sizeof(*srx));
	58	}
	59
	60	_leave(" = %p", local);
	61	return local;
	62	}
	63
	64	/*
	65	* create the local socket
	66	* - must be called with rxrpc_local_sem writelocked
	67	*/
	68	static int rxrpc_create_local(struct rxrpc_local *local)
	69	{
	70	struct sock *sock;
	71	int ret, opt;
	72
	73	_enter("%p{%d}", local, local->srx.transport_type);
	74
	75	/* create a socket to represent the local endpoint */
eeb1bd5c EB	76	ret = sock_create_kern(&init_net, PF_INET, local->srx.transport_type,
eeb1bd5c EB	77	IPPROTO_UDP, &local->socket);
17926a79 DH	78	if (ret < 0) {
	79	_leave(" = %d [socket]", ret);
	80	return ret;
	81	}
	82
	83	/* if a local address was supplied then bind it */
	84	if (local->srx.transport_len > sizeof(sa_family_t)) {
	85	_debug("bind");
	86	ret = kernel_bind(local->socket,
	87	(struct sockaddr *) &local->srx.transport,
	88	local->srx.transport_len);
	89	if (ret < 0) {
	90	_debug("bind failed");
	91	goto error;
	92	}
	93	}
	94
	95	/* we want to receive ICMP errors */
	96	opt = 1;
	97	ret = kernel_setsockopt(local->socket, SOL_IP, IP_RECVERR,
	98	(char *) &opt, sizeof(opt));
	99	if (ret < 0) {
	100	_debug("setsockopt failed");
	101	goto error;
	102	}
	103
	104	/* we want to set the don't fragment bit */
	105	opt = IP_PMTUDISC_DO;
	106	ret = kernel_setsockopt(local->socket, SOL_IP, IP_MTU_DISCOVER,
	107	(char *) &opt, sizeof(opt));
	108	if (ret < 0) {
	109	_debug("setsockopt failed");
	110	goto error;
	111	}
	112
	113	write_lock_bh(&rxrpc_local_lock);
	114	list_add(&local->link, &rxrpc_locals);
	115	write_unlock_bh(&rxrpc_local_lock);
	116
	117	/* set the socket up */
	118	sock = local->socket->sk;
	119	sock->sk_user_data = local;
	120	sock->sk_data_ready = rxrpc_data_ready;
	121	sock->sk_error_report = rxrpc_UDP_error_report;
	122	_leave(" = 0");
	123	return 0;
	124
	125	error:
91cf45f0	126	kernel_sock_shutdown(local->socket, SHUT_RDWR);
17926a79 DH	127	local->socket->sk->sk_user_data = NULL;
	128	sock_release(local->socket);
	129	local->socket = NULL;
	130
	131	_leave(" = %d", ret);
	132	return ret;
	133	}
	134
	135	/*
	136	* create a new local endpoint using the specified UDP address
	137	*/
	138	struct rxrpc_local rxrpc_lookup_local(struct sockaddr_rxrpc srx)
	139	{
	140	struct rxrpc_local *local;
	141	int ret;
	142
21454aaa	143	_enter("{%d,%u,%pI4+%hu}",
17926a79 DH	144	srx->transport_type,
17926a79 DH	145	srx->transport.family,
21454aaa	146	&srx->transport.sin.sin_addr,
17926a79 DH	147	ntohs(srx->transport.sin.sin_port));
	148
	149	down_write(&rxrpc_local_sem);
	150
	151	/* see if we have a suitable local local endpoint already */
	152	read_lock_bh(&rxrpc_local_lock);
	153
	154	list_for_each_entry(local, &rxrpc_locals, link) {
21454aaa	155	_debug("CMP {%d,%u,%pI4+%hu}",
17926a79 DH	156	local->srx.transport_type,
17926a79 DH	157	local->srx.transport.family,
21454aaa	158	&local->srx.transport.sin.sin_addr,
17926a79 DH	159	ntohs(local->srx.transport.sin.sin_port));
	160
	161	if (local->srx.transport_type != srx->transport_type \|\|
	162	local->srx.transport.family != srx->transport.family)
	163	continue;
	164
	165	switch (srx->transport.family) {
	166	case AF_INET:
	167	if (local->srx.transport.sin.sin_port !=
	168	srx->transport.sin.sin_port)
	169	continue;
	170	if (memcmp(&local->srx.transport.sin.sin_addr,
	171	&srx->transport.sin.sin_addr,
	172	sizeof(struct in_addr)) != 0)
	173	continue;
	174	goto found_local;
	175
	176	default:
	177	BUG();
	178	}
	179	}
	180
	181	read_unlock_bh(&rxrpc_local_lock);
	182
	183	/* we didn't find one, so we need to create one */
	184	local = rxrpc_alloc_local(srx);
	185	if (!local) {
	186	up_write(&rxrpc_local_sem);
	187	return ERR_PTR(-ENOMEM);
	188	}
	189
	190	ret = rxrpc_create_local(local);
	191	if (ret < 0) {
	192	up_write(&rxrpc_local_sem);
	193	kfree(local);
	194	_leave(" = %d", ret);
	195	return ERR_PTR(ret);
	196	}
	197
	198	up_write(&rxrpc_local_sem);
	199
21454aaa	200	_net("LOCAL new %d {%d,%u,%pI4+%hu}",
17926a79 DH	201	local->debug_id,
	202	local->srx.transport_type,
	203	local->srx.transport.family,
21454aaa	204	&local->srx.transport.sin.sin_addr,
17926a79 DH	205	ntohs(local->srx.transport.sin.sin_port));
	206
	207	_leave(" = %p [new]", local);
	208	return local;
	209
	210	found_local:
	211	rxrpc_get_local(local);
	212	read_unlock_bh(&rxrpc_local_lock);
	213	up_write(&rxrpc_local_sem);
	214
21454aaa	215	_net("LOCAL old %d {%d,%u,%pI4+%hu}",
17926a79 DH	216	local->debug_id,
	217	local->srx.transport_type,
	218	local->srx.transport.family,
21454aaa	219	&local->srx.transport.sin.sin_addr,
17926a79 DH	220	ntohs(local->srx.transport.sin.sin_port));
	221
	222	_leave(" = %p [reuse]", local);
	223	return local;
	224	}
	225
	226	/*
	227	* release a local endpoint
	228	*/
	229	void rxrpc_put_local(struct rxrpc_local *local)
	230	{
	231	_enter("%p{u=%d}", local, atomic_read(&local->usage));
	232
	233	ASSERTCMP(atomic_read(&local->usage), >, 0);
	234
	235	/* to prevent a race, the decrement and the dequeue must be effectively
	236	* atomic */
	237	write_lock_bh(&rxrpc_local_lock);
	238	if (unlikely(atomic_dec_and_test(&local->usage))) {
	239	_debug("destroy local");
651350d1	240	rxrpc_queue_work(&local->destroyer);
17926a79 DH	241	}
	242	write_unlock_bh(&rxrpc_local_lock);
	243	_leave("");
	244	}
	245
	246	/*
	247	* destroy a local endpoint
	248	*/
	249	static void rxrpc_destroy_local(struct work_struct *work)
	250	{
	251	struct rxrpc_local *local =
	252	container_of(work, struct rxrpc_local, destroyer);
	253
	254	_enter("%p{%d}", local, atomic_read(&local->usage));
	255
	256	down_write(&rxrpc_local_sem);
	257
	258	write_lock_bh(&rxrpc_local_lock);
	259	if (atomic_read(&local->usage) > 0) {
	260	write_unlock_bh(&rxrpc_local_lock);
	261	up_read(&rxrpc_local_sem);
	262	_leave(" [resurrected]");
	263	return;
	264	}
	265
	266	list_del(&local->link);
	267	local->socket->sk->sk_user_data = NULL;
	268	write_unlock_bh(&rxrpc_local_lock);
	269
	270	downgrade_write(&rxrpc_local_sem);
	271
	272	ASSERT(list_empty(&local->services));
	273	ASSERT(!work_pending(&local->acceptor));
	274	ASSERT(!work_pending(&local->rejecter));
44ba0698	275	ASSERT(!work_pending(&local->event_processor));
17926a79 DH	276
	277	/* finish cleaning up the local descriptor */
	278	rxrpc_purge_queue(&local->accept_queue);
	279	rxrpc_purge_queue(&local->reject_queue);
44ba0698	280	rxrpc_purge_queue(&local->event_queue);
91cf45f0	281	kernel_sock_shutdown(local->socket, SHUT_RDWR);
17926a79 DH	282	sock_release(local->socket);
	283
	284	up_read(&rxrpc_local_sem);
	285
	286	_net("DESTROY LOCAL %d", local->debug_id);
	287	kfree(local);
	288
	289	if (list_empty(&rxrpc_locals))
	290	wake_up_all(&rxrpc_local_wq);
	291
	292	_leave("");
	293	}
	294
	295	/*
	296	* preemptively destroy all local local endpoint rather than waiting for
	297	* them to be destroyed
	298	*/
	299	void __exit rxrpc_destroy_all_locals(void)
	300	{
	301	DECLARE_WAITQUEUE(myself,current);
	302
	303	_enter("");
	304
	305	/* we simply have to wait for them to go away */
	306	if (!list_empty(&rxrpc_locals)) {
	307	set_current_state(TASK_UNINTERRUPTIBLE);
	308	add_wait_queue(&rxrpc_local_wq, &myself);
	309
	310	while (!list_empty(&rxrpc_locals)) {
	311	schedule();
	312	set_current_state(TASK_UNINTERRUPTIBLE);
	313	}
	314
	315	remove_wait_queue(&rxrpc_local_wq, &myself);
	316	set_current_state(TASK_RUNNING);
	317	}
	318
	319	_leave("");
	320	}
44ba0698 DH	321
	322	/*
	323	* Reply to a version request
	324	*/
	325	static void rxrpc_send_version_request(struct rxrpc_local *local,
0d12f8a4	326	struct rxrpc_host_header *hdr,
44ba0698 DH	327	struct sk_buff *skb)
44ba0698 DH	328	{
0d12f8a4 DH	329	struct rxrpc_wire_header whdr;
0d12f8a4 DH	330	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
44ba0698 DH	331	struct sockaddr_in sin;
	332	struct msghdr msg;
	333	struct kvec iov[2];
	334	size_t len;
	335	int ret;
	336
	337	_enter("");
	338
	339	sin.sin_family = AF_INET;
	340	sin.sin_port = udp_hdr(skb)->source;
	341	sin.sin_addr.s_addr = ip_hdr(skb)->saddr;
	342
	343	msg.msg_name = &sin;
	344	msg.msg_namelen = sizeof(sin);
	345	msg.msg_control = NULL;
	346	msg.msg_controllen = 0;
	347	msg.msg_flags = 0;
	348
0d12f8a4 DH	349	whdr.epoch = htonl(sp->hdr.epoch);
	350	whdr.cid = htonl(sp->hdr.cid);
	351	whdr.callNumber = htonl(sp->hdr.callNumber);
	352	whdr.seq = 0;
	353	whdr.serial = 0;
	354	whdr.type = RXRPC_PACKET_TYPE_VERSION;
	355	whdr.flags = RXRPC_LAST_PACKET \| (~hdr->flags & RXRPC_CLIENT_INITIATED);
	356	whdr.userStatus = 0;
	357	whdr.securityIndex = 0;
	358	whdr._rsvd = 0;
	359	whdr.serviceId = htons(sp->hdr.serviceId);
	360
	361	iov[0].iov_base = &whdr;
	362	iov[0].iov_len = sizeof(whdr);
44ba0698 DH	363	iov[1].iov_base = (char *)rxrpc_version_string;
	364	iov[1].iov_len = sizeof(rxrpc_version_string);
	365
	366	len = iov[0].iov_len + iov[1].iov_len;
	367
	368	_proto("Tx VERSION (reply)");
	369
	370	ret = kernel_sendmsg(local->socket, &msg, iov, 2, len);
	371	if (ret < 0)
	372	_debug("sendmsg failed: %d", ret);
	373
	374	_leave("");
	375	}
	376
	377	/*
	378	* Process event packets targetted at a local endpoint.
	379	*/
	380	static void rxrpc_process_local_events(struct work_struct *work)
	381	{
	382	struct rxrpc_local *local = container_of(work, struct rxrpc_local, event_processor);
	383	struct sk_buff *skb;
	384	char v;
	385
	386	_enter("");
	387
	388	atomic_inc(&local->usage);
	389
	390	while ((skb = skb_dequeue(&local->event_queue))) {
	391	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	392
0d12f8a4	393	_debug("{%d},{%u}", local->debug_id, sp->hdr.type);
44ba0698 DH	394
	395	switch (sp->hdr.type) {
	396	case RXRPC_PACKET_TYPE_VERSION:
	397	if (skb_copy_bits(skb, 0, &v, 1) < 0)
	398	return;
	399	_proto("Rx VERSION { %02x }", v);
	400	if (v == 0)
	401	rxrpc_send_version_request(local, &sp->hdr, skb);
	402	break;
	403
	404	default:
	405	/* Just ignore anything we don't understand */
	406	break;
	407	}
	408
	409	rxrpc_put_local(local);
	410	rxrpc_free_skb(skb);
	411	}
	412
	413	rxrpc_put_local(local);
	414	_leave("");
	415	}