1 /* Maintain an RxRPC server socket to do AFS communications through
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/slab.h>
13 #include <net/sock.h>
14 #include <net/af_rxrpc.h>
15 #include <rxrpc/packet.h>
16 #include "internal.h"
17 #include "afs_cm.h"
18 
19 static struct socket *afs_socket; /* my RxRPC socket */
20 static struct workqueue_struct *afs_async_calls;
21 static atomic_t afs_outstanding_calls;
22 static atomic_t afs_outstanding_skbs;
23 
24 static void afs_wake_up_call_waiter(struct afs_call *);
25 static int afs_wait_for_call_to_complete(struct afs_call *);
26 static void afs_wake_up_async_call(struct afs_call *);
27 static int afs_dont_wait_for_call_to_complete(struct afs_call *);
28 static void afs_process_async_call(struct work_struct *);
29 static void afs_rx_interceptor(struct sock *, unsigned long, struct sk_buff *);
30 static int afs_deliver_cm_op_id(struct afs_call *, struct sk_buff *, bool);
31 
32 /* synchronous call management */
33 const struct afs_wait_mode afs_sync_call = {
34 	.rx_wakeup	= afs_wake_up_call_waiter,
35 	.wait		= afs_wait_for_call_to_complete,
36 };
37 
38 /* asynchronous call management */
39 const struct afs_wait_mode afs_async_call = {
40 	.rx_wakeup	= afs_wake_up_async_call,
41 	.wait		= afs_dont_wait_for_call_to_complete,
42 };
43 
44 /* asynchronous incoming call management */
45 static const struct afs_wait_mode afs_async_incoming_call = {
46 	.rx_wakeup	= afs_wake_up_async_call,
47 };
48 
49 /* asynchronous incoming call initial processing */
50 static const struct afs_call_type afs_RXCMxxxx = {
51 	.name		= "CB.xxxx",
52 	.deliver	= afs_deliver_cm_op_id,
53 	.abort_to_error	= afs_abort_to_error,
54 };
55 
56 static void afs_collect_incoming_call(struct work_struct *);
57 
58 static struct sk_buff_head afs_incoming_calls;
59 static DECLARE_WORK(afs_collect_incoming_call_work, afs_collect_incoming_call);
60 
61 /*
62  * open an RxRPC socket and bind it to be a server for callback notifications
63  * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT
64  */
afs_open_socket(void)65 int afs_open_socket(void)
66 {
67 	struct sockaddr_rxrpc srx;
68 	struct socket *socket;
69 	int ret;
70 
71 	_enter("");
72 
73 	skb_queue_head_init(&afs_incoming_calls);
74 
75 	afs_async_calls = create_singlethread_workqueue("kafsd");
76 	if (!afs_async_calls) {
77 		_leave(" = -ENOMEM [wq]");
78 		return -ENOMEM;
79 	}
80 
81 	ret = sock_create_kern(AF_RXRPC, SOCK_DGRAM, PF_INET, &socket);
82 	if (ret < 0) {
83 		destroy_workqueue(afs_async_calls);
84 		_leave(" = %d [socket]", ret);
85 		return ret;
86 	}
87 
88 	socket->sk->sk_allocation = GFP_NOFS;
89 
90 	/* bind the callback manager's address to make this a server socket */
91 	srx.srx_family			= AF_RXRPC;
92 	srx.srx_service			= CM_SERVICE;
93 	srx.transport_type		= SOCK_DGRAM;
94 	srx.transport_len		= sizeof(srx.transport.sin);
95 	srx.transport.sin.sin_family	= AF_INET;
96 	srx.transport.sin.sin_port	= htons(AFS_CM_PORT);
97 	memset(&srx.transport.sin.sin_addr, 0,
98 	       sizeof(srx.transport.sin.sin_addr));
99 
100 	ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx));
101 	if (ret < 0) {
102 		sock_release(socket);
103 		destroy_workqueue(afs_async_calls);
104 		_leave(" = %d [bind]", ret);
105 		return ret;
106 	}
107 
108 	rxrpc_kernel_intercept_rx_messages(socket, afs_rx_interceptor);
109 
110 	afs_socket = socket;
111 	_leave(" = 0");
112 	return 0;
113 }
114 
115 /*
116  * close the RxRPC socket AFS was using
117  */
afs_close_socket(void)118 void afs_close_socket(void)
119 {
120 	_enter("");
121 
122 	sock_release(afs_socket);
123 
124 	_debug("dework");
125 	destroy_workqueue(afs_async_calls);
126 
127 	ASSERTCMP(atomic_read(&afs_outstanding_skbs), ==, 0);
128 	ASSERTCMP(atomic_read(&afs_outstanding_calls), ==, 0);
129 	_leave("");
130 }
131 
132 /*
133  * note that the data in a socket buffer is now delivered and that the buffer
134  * should be freed
135  */
afs_data_delivered(struct sk_buff * skb)136 static void afs_data_delivered(struct sk_buff *skb)
137 {
138 	if (!skb) {
139 		_debug("DLVR NULL [%d]", atomic_read(&afs_outstanding_skbs));
140 		dump_stack();
141 	} else {
142 		_debug("DLVR %p{%u} [%d]",
143 		       skb, skb->mark, atomic_read(&afs_outstanding_skbs));
144 		if (atomic_dec_return(&afs_outstanding_skbs) == -1)
145 			BUG();
146 		rxrpc_kernel_data_delivered(skb);
147 	}
148 }
149 
150 /*
151  * free a socket buffer
152  */
afs_free_skb(struct sk_buff * skb)153 static void afs_free_skb(struct sk_buff *skb)
154 {
155 	if (!skb) {
156 		_debug("FREE NULL [%d]", atomic_read(&afs_outstanding_skbs));
157 		dump_stack();
158 	} else {
159 		_debug("FREE %p{%u} [%d]",
160 		       skb, skb->mark, atomic_read(&afs_outstanding_skbs));
161 		if (atomic_dec_return(&afs_outstanding_skbs) == -1)
162 			BUG();
163 		rxrpc_kernel_free_skb(skb);
164 	}
165 }
166 
167 /*
168  * free a call
169  */
afs_free_call(struct afs_call * call)170 static void afs_free_call(struct afs_call *call)
171 {
172 	_debug("DONE %p{%s} [%d]",
173 	       call, call->type->name, atomic_read(&afs_outstanding_calls));
174 	if (atomic_dec_return(&afs_outstanding_calls) == -1)
175 		BUG();
176 
177 	ASSERTCMP(call->rxcall, ==, NULL);
178 	ASSERT(!work_pending(&call->async_work));
179 	ASSERT(skb_queue_empty(&call->rx_queue));
180 	ASSERT(call->type->name != NULL);
181 
182 	kfree(call->request);
183 	kfree(call);
184 }
185 
186 /*
187  * allocate a call with flat request and reply buffers
188  */
afs_alloc_flat_call(const struct afs_call_type * type,size_t request_size,size_t reply_size)189 struct afs_call *afs_alloc_flat_call(const struct afs_call_type *type,
190 				     size_t request_size, size_t reply_size)
191 {
192 	struct afs_call *call;
193 
194 	call = kzalloc(sizeof(*call), GFP_NOFS);
195 	if (!call)
196 		goto nomem_call;
197 
198 	_debug("CALL %p{%s} [%d]",
199 	       call, type->name, atomic_read(&afs_outstanding_calls));
200 	atomic_inc(&afs_outstanding_calls);
201 
202 	call->type = type;
203 	call->request_size = request_size;
204 	call->reply_max = reply_size;
205 
206 	if (request_size) {
207 		call->request = kmalloc(request_size, GFP_NOFS);
208 		if (!call->request)
209 			goto nomem_free;
210 	}
211 
212 	if (reply_size) {
213 		call->buffer = kmalloc(reply_size, GFP_NOFS);
214 		if (!call->buffer)
215 			goto nomem_free;
216 	}
217 
218 	init_waitqueue_head(&call->waitq);
219 	skb_queue_head_init(&call->rx_queue);
220 	return call;
221 
222 nomem_free:
223 	afs_free_call(call);
224 nomem_call:
225 	return NULL;
226 }
227 
228 /*
229  * clean up a call with flat buffer
230  */
afs_flat_call_destructor(struct afs_call * call)231 void afs_flat_call_destructor(struct afs_call *call)
232 {
233 	_enter("");
234 
235 	kfree(call->request);
236 	call->request = NULL;
237 	kfree(call->buffer);
238 	call->buffer = NULL;
239 }
240 
241 /*
242  * attach the data from a bunch of pages on an inode to a call
243  */
afs_send_pages(struct afs_call * call,struct msghdr * msg,struct kvec * iov)244 static int afs_send_pages(struct afs_call *call, struct msghdr *msg,
245 			  struct kvec *iov)
246 {
247 	struct page *pages[8];
248 	unsigned count, n, loop, offset, to;
249 	pgoff_t first = call->first, last = call->last;
250 	int ret;
251 
252 	_enter("");
253 
254 	offset = call->first_offset;
255 	call->first_offset = 0;
256 
257 	do {
258 		_debug("attach %lx-%lx", first, last);
259 
260 		count = last - first + 1;
261 		if (count > ARRAY_SIZE(pages))
262 			count = ARRAY_SIZE(pages);
263 		n = find_get_pages_contig(call->mapping, first, count, pages);
264 		ASSERTCMP(n, ==, count);
265 
266 		loop = 0;
267 		do {
268 			msg->msg_flags = 0;
269 			to = PAGE_SIZE;
270 			if (first + loop >= last)
271 				to = call->last_to;
272 			else
273 				msg->msg_flags = MSG_MORE;
274 			iov->iov_base = kmap(pages[loop]) + offset;
275 			iov->iov_len = to - offset;
276 			offset = 0;
277 
278 			_debug("- range %u-%u%s",
279 			       offset, to, msg->msg_flags ? " [more]" : "");
280 			msg->msg_iov = (struct iovec *) iov;
281 			msg->msg_iovlen = 1;
282 
283 			/* have to change the state *before* sending the last
284 			 * packet as RxRPC might give us the reply before it
285 			 * returns from sending the request */
286 			if (first + loop >= last)
287 				call->state = AFS_CALL_AWAIT_REPLY;
288 			ret = rxrpc_kernel_send_data(call->rxcall, msg,
289 						     to - offset);
290 			kunmap(pages[loop]);
291 			if (ret < 0)
292 				break;
293 		} while (++loop < count);
294 		first += count;
295 
296 		for (loop = 0; loop < count; loop++)
297 			put_page(pages[loop]);
298 		if (ret < 0)
299 			break;
300 	} while (first <= last);
301 
302 	_leave(" = %d", ret);
303 	return ret;
304 }
305 
306 /*
307  * initiate a call
308  */
afs_make_call(struct in_addr * addr,struct afs_call * call,gfp_t gfp,const struct afs_wait_mode * wait_mode)309 int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
310 		  const struct afs_wait_mode *wait_mode)
311 {
312 	struct sockaddr_rxrpc srx;
313 	struct rxrpc_call *rxcall;
314 	struct msghdr msg;
315 	struct kvec iov[1];
316 	int ret;
317 	struct sk_buff *skb;
318 
319 	_enter("%x,{%d},", addr->s_addr, ntohs(call->port));
320 
321 	ASSERT(call->type != NULL);
322 	ASSERT(call->type->name != NULL);
323 
324 	_debug("____MAKE %p{%s,%x} [%d]____",
325 	       call, call->type->name, key_serial(call->key),
326 	       atomic_read(&afs_outstanding_calls));
327 
328 	call->wait_mode = wait_mode;
329 	INIT_WORK(&call->async_work, afs_process_async_call);
330 
331 	memset(&srx, 0, sizeof(srx));
332 	srx.srx_family = AF_RXRPC;
333 	srx.srx_service = call->service_id;
334 	srx.transport_type = SOCK_DGRAM;
335 	srx.transport_len = sizeof(srx.transport.sin);
336 	srx.transport.sin.sin_family = AF_INET;
337 	srx.transport.sin.sin_port = call->port;
338 	memcpy(&srx.transport.sin.sin_addr, addr, 4);
339 
340 	/* create a call */
341 	rxcall = rxrpc_kernel_begin_call(afs_socket, &srx, call->key,
342 					 (unsigned long) call, gfp);
343 	call->key = NULL;
344 	if (IS_ERR(rxcall)) {
345 		ret = PTR_ERR(rxcall);
346 		goto error_kill_call;
347 	}
348 
349 	call->rxcall = rxcall;
350 
351 	/* send the request */
352 	iov[0].iov_base	= call->request;
353 	iov[0].iov_len	= call->request_size;
354 
355 	msg.msg_name		= NULL;
356 	msg.msg_namelen		= 0;
357 	msg.msg_iov		= (struct iovec *) iov;
358 	msg.msg_iovlen		= 1;
359 	msg.msg_control		= NULL;
360 	msg.msg_controllen	= 0;
361 	msg.msg_flags		= (call->send_pages ? MSG_MORE : 0);
362 
363 	/* have to change the state *before* sending the last packet as RxRPC
364 	 * might give us the reply before it returns from sending the
365 	 * request */
366 	if (!call->send_pages)
367 		call->state = AFS_CALL_AWAIT_REPLY;
368 	ret = rxrpc_kernel_send_data(rxcall, &msg, call->request_size);
369 	if (ret < 0)
370 		goto error_do_abort;
371 
372 	if (call->send_pages) {
373 		ret = afs_send_pages(call, &msg, iov);
374 		if (ret < 0)
375 			goto error_do_abort;
376 	}
377 
378 	/* at this point, an async call may no longer exist as it may have
379 	 * already completed */
380 	return wait_mode->wait(call);
381 
382 error_do_abort:
383 	rxrpc_kernel_abort_call(rxcall, RX_USER_ABORT);
384 	while ((skb = skb_dequeue(&call->rx_queue)))
385 		afs_free_skb(skb);
386 	rxrpc_kernel_end_call(rxcall);
387 	call->rxcall = NULL;
388 error_kill_call:
389 	call->type->destructor(call);
390 	afs_free_call(call);
391 	_leave(" = %d", ret);
392 	return ret;
393 }
394 
395 /*
396  * handles intercepted messages that were arriving in the socket's Rx queue
397  * - called with the socket receive queue lock held to ensure message ordering
398  * - called with softirqs disabled
399  */
afs_rx_interceptor(struct sock * sk,unsigned long user_call_ID,struct sk_buff * skb)400 static void afs_rx_interceptor(struct sock *sk, unsigned long user_call_ID,
401 			       struct sk_buff *skb)
402 {
403 	struct afs_call *call = (struct afs_call *) user_call_ID;
404 
405 	_enter("%p,,%u", call, skb->mark);
406 
407 	_debug("ICPT %p{%u} [%d]",
408 	       skb, skb->mark, atomic_read(&afs_outstanding_skbs));
409 
410 	ASSERTCMP(sk, ==, afs_socket->sk);
411 	atomic_inc(&afs_outstanding_skbs);
412 
413 	if (!call) {
414 		/* its an incoming call for our callback service */
415 		skb_queue_tail(&afs_incoming_calls, skb);
416 		queue_work(afs_wq, &afs_collect_incoming_call_work);
417 	} else {
418 		/* route the messages directly to the appropriate call */
419 		skb_queue_tail(&call->rx_queue, skb);
420 		call->wait_mode->rx_wakeup(call);
421 	}
422 
423 	_leave("");
424 }
425 
426 /*
427  * deliver messages to a call
428  */
afs_deliver_to_call(struct afs_call * call)429 static void afs_deliver_to_call(struct afs_call *call)
430 {
431 	struct sk_buff *skb;
432 	bool last;
433 	u32 abort_code;
434 	int ret;
435 
436 	_enter("");
437 
438 	while ((call->state == AFS_CALL_AWAIT_REPLY ||
439 		call->state == AFS_CALL_AWAIT_OP_ID ||
440 		call->state == AFS_CALL_AWAIT_REQUEST ||
441 		call->state == AFS_CALL_AWAIT_ACK) &&
442 	       (skb = skb_dequeue(&call->rx_queue))) {
443 		switch (skb->mark) {
444 		case RXRPC_SKB_MARK_DATA:
445 			_debug("Rcv DATA");
446 			last = rxrpc_kernel_is_data_last(skb);
447 			ret = call->type->deliver(call, skb, last);
448 			switch (ret) {
449 			case 0:
450 				if (last &&
451 				    call->state == AFS_CALL_AWAIT_REPLY)
452 					call->state = AFS_CALL_COMPLETE;
453 				break;
454 			case -ENOTCONN:
455 				abort_code = RX_CALL_DEAD;
456 				goto do_abort;
457 			case -ENOTSUPP:
458 				abort_code = RX_INVALID_OPERATION;
459 				goto do_abort;
460 			default:
461 				abort_code = RXGEN_CC_UNMARSHAL;
462 				if (call->state != AFS_CALL_AWAIT_REPLY)
463 					abort_code = RXGEN_SS_UNMARSHAL;
464 			do_abort:
465 				rxrpc_kernel_abort_call(call->rxcall,
466 							abort_code);
467 				call->error = ret;
468 				call->state = AFS_CALL_ERROR;
469 				break;
470 			}
471 			afs_data_delivered(skb);
472 			skb = NULL;
473 			continue;
474 		case RXRPC_SKB_MARK_FINAL_ACK:
475 			_debug("Rcv ACK");
476 			call->state = AFS_CALL_COMPLETE;
477 			break;
478 		case RXRPC_SKB_MARK_BUSY:
479 			_debug("Rcv BUSY");
480 			call->error = -EBUSY;
481 			call->state = AFS_CALL_BUSY;
482 			break;
483 		case RXRPC_SKB_MARK_REMOTE_ABORT:
484 			abort_code = rxrpc_kernel_get_abort_code(skb);
485 			call->error = call->type->abort_to_error(abort_code);
486 			call->state = AFS_CALL_ABORTED;
487 			_debug("Rcv ABORT %u -> %d", abort_code, call->error);
488 			break;
489 		case RXRPC_SKB_MARK_NET_ERROR:
490 			call->error = -rxrpc_kernel_get_error_number(skb);
491 			call->state = AFS_CALL_ERROR;
492 			_debug("Rcv NET ERROR %d", call->error);
493 			break;
494 		case RXRPC_SKB_MARK_LOCAL_ERROR:
495 			call->error = -rxrpc_kernel_get_error_number(skb);
496 			call->state = AFS_CALL_ERROR;
497 			_debug("Rcv LOCAL ERROR %d", call->error);
498 			break;
499 		default:
500 			BUG();
501 			break;
502 		}
503 
504 		afs_free_skb(skb);
505 	}
506 
507 	/* make sure the queue is empty if the call is done with (we might have
508 	 * aborted the call early because of an unmarshalling error) */
509 	if (call->state >= AFS_CALL_COMPLETE) {
510 		while ((skb = skb_dequeue(&call->rx_queue)))
511 			afs_free_skb(skb);
512 		if (call->incoming) {
513 			rxrpc_kernel_end_call(call->rxcall);
514 			call->rxcall = NULL;
515 			call->type->destructor(call);
516 			afs_free_call(call);
517 		}
518 	}
519 
520 	_leave("");
521 }
522 
523 /*
524  * wait synchronously for a call to complete
525  */
afs_wait_for_call_to_complete(struct afs_call * call)526 static int afs_wait_for_call_to_complete(struct afs_call *call)
527 {
528 	struct sk_buff *skb;
529 	int ret;
530 
531 	DECLARE_WAITQUEUE(myself, current);
532 
533 	_enter("");
534 
535 	add_wait_queue(&call->waitq, &myself);
536 	for (;;) {
537 		set_current_state(TASK_INTERRUPTIBLE);
538 
539 		/* deliver any messages that are in the queue */
540 		if (!skb_queue_empty(&call->rx_queue)) {
541 			__set_current_state(TASK_RUNNING);
542 			afs_deliver_to_call(call);
543 			continue;
544 		}
545 
546 		ret = call->error;
547 		if (call->state >= AFS_CALL_COMPLETE)
548 			break;
549 		ret = -EINTR;
550 		if (signal_pending(current))
551 			break;
552 		schedule();
553 	}
554 
555 	remove_wait_queue(&call->waitq, &myself);
556 	__set_current_state(TASK_RUNNING);
557 
558 	/* kill the call */
559 	if (call->state < AFS_CALL_COMPLETE) {
560 		_debug("call incomplete");
561 		rxrpc_kernel_abort_call(call->rxcall, RX_CALL_DEAD);
562 		while ((skb = skb_dequeue(&call->rx_queue)))
563 			afs_free_skb(skb);
564 	}
565 
566 	_debug("call complete");
567 	rxrpc_kernel_end_call(call->rxcall);
568 	call->rxcall = NULL;
569 	call->type->destructor(call);
570 	afs_free_call(call);
571 	_leave(" = %d", ret);
572 	return ret;
573 }
574 
575 /*
576  * wake up a waiting call
577  */
afs_wake_up_call_waiter(struct afs_call * call)578 static void afs_wake_up_call_waiter(struct afs_call *call)
579 {
580 	wake_up(&call->waitq);
581 }
582 
583 /*
584  * wake up an asynchronous call
585  */
afs_wake_up_async_call(struct afs_call * call)586 static void afs_wake_up_async_call(struct afs_call *call)
587 {
588 	_enter("");
589 	queue_work(afs_async_calls, &call->async_work);
590 }
591 
592 /*
593  * put a call into asynchronous mode
594  * - mustn't touch the call descriptor as the call my have completed by the
595  *   time we get here
596  */
afs_dont_wait_for_call_to_complete(struct afs_call * call)597 static int afs_dont_wait_for_call_to_complete(struct afs_call *call)
598 {
599 	_enter("");
600 	return -EINPROGRESS;
601 }
602 
603 /*
604  * delete an asynchronous call
605  */
afs_delete_async_call(struct work_struct * work)606 static void afs_delete_async_call(struct work_struct *work)
607 {
608 	struct afs_call *call =
609 		container_of(work, struct afs_call, async_work);
610 
611 	_enter("");
612 
613 	afs_free_call(call);
614 
615 	_leave("");
616 }
617 
618 /*
619  * perform processing on an asynchronous call
620  * - on a multiple-thread workqueue this work item may try to run on several
621  *   CPUs at the same time
622  */
afs_process_async_call(struct work_struct * work)623 static void afs_process_async_call(struct work_struct *work)
624 {
625 	struct afs_call *call =
626 		container_of(work, struct afs_call, async_work);
627 
628 	_enter("");
629 
630 	if (!skb_queue_empty(&call->rx_queue))
631 		afs_deliver_to_call(call);
632 
633 	if (call->state >= AFS_CALL_COMPLETE && call->wait_mode) {
634 		if (call->wait_mode->async_complete)
635 			call->wait_mode->async_complete(call->reply,
636 							call->error);
637 		call->reply = NULL;
638 
639 		/* kill the call */
640 		rxrpc_kernel_end_call(call->rxcall);
641 		call->rxcall = NULL;
642 		if (call->type->destructor)
643 			call->type->destructor(call);
644 
645 		/* we can't just delete the call because the work item may be
646 		 * queued */
647 		PREPARE_WORK(&call->async_work, afs_delete_async_call);
648 		queue_work(afs_async_calls, &call->async_work);
649 	}
650 
651 	_leave("");
652 }
653 
654 /*
655  * empty a socket buffer into a flat reply buffer
656  */
afs_transfer_reply(struct afs_call * call,struct sk_buff * skb)657 void afs_transfer_reply(struct afs_call *call, struct sk_buff *skb)
658 {
659 	size_t len = skb->len;
660 
661 	if (skb_copy_bits(skb, 0, call->buffer + call->reply_size, len) < 0)
662 		BUG();
663 	call->reply_size += len;
664 }
665 
666 /*
667  * accept the backlog of incoming calls
668  */
afs_collect_incoming_call(struct work_struct * work)669 static void afs_collect_incoming_call(struct work_struct *work)
670 {
671 	struct rxrpc_call *rxcall;
672 	struct afs_call *call = NULL;
673 	struct sk_buff *skb;
674 
675 	while ((skb = skb_dequeue(&afs_incoming_calls))) {
676 		_debug("new call");
677 
678 		/* don't need the notification */
679 		afs_free_skb(skb);
680 
681 		if (!call) {
682 			call = kzalloc(sizeof(struct afs_call), GFP_KERNEL);
683 			if (!call) {
684 				rxrpc_kernel_reject_call(afs_socket);
685 				return;
686 			}
687 
688 			INIT_WORK(&call->async_work, afs_process_async_call);
689 			call->wait_mode = &afs_async_incoming_call;
690 			call->type = &afs_RXCMxxxx;
691 			init_waitqueue_head(&call->waitq);
692 			skb_queue_head_init(&call->rx_queue);
693 			call->state = AFS_CALL_AWAIT_OP_ID;
694 
695 			_debug("CALL %p{%s} [%d]",
696 			       call, call->type->name,
697 			       atomic_read(&afs_outstanding_calls));
698 			atomic_inc(&afs_outstanding_calls);
699 		}
700 
701 		rxcall = rxrpc_kernel_accept_call(afs_socket,
702 						  (unsigned long) call);
703 		if (!IS_ERR(rxcall)) {
704 			call->rxcall = rxcall;
705 			call = NULL;
706 		}
707 	}
708 
709 	if (call)
710 		afs_free_call(call);
711 }
712 
713 /*
714  * grab the operation ID from an incoming cache manager call
715  */
afs_deliver_cm_op_id(struct afs_call * call,struct sk_buff * skb,bool last)716 static int afs_deliver_cm_op_id(struct afs_call *call, struct sk_buff *skb,
717 				bool last)
718 {
719 	size_t len = skb->len;
720 	void *oibuf = (void *) &call->operation_ID;
721 
722 	_enter("{%u},{%zu},%d", call->offset, len, last);
723 
724 	ASSERTCMP(call->offset, <, 4);
725 
726 	/* the operation ID forms the first four bytes of the request data */
727 	len = min_t(size_t, len, 4 - call->offset);
728 	if (skb_copy_bits(skb, 0, oibuf + call->offset, len) < 0)
729 		BUG();
730 	if (!pskb_pull(skb, len))
731 		BUG();
732 	call->offset += len;
733 
734 	if (call->offset < 4) {
735 		if (last) {
736 			_leave(" = -EBADMSG [op ID short]");
737 			return -EBADMSG;
738 		}
739 		_leave(" = 0 [incomplete]");
740 		return 0;
741 	}
742 
743 	call->state = AFS_CALL_AWAIT_REQUEST;
744 
745 	/* ask the cache manager to route the call (it'll change the call type
746 	 * if successful) */
747 	if (!afs_cm_incoming_call(call))
748 		return -ENOTSUPP;
749 
750 	/* pass responsibility for the remainer of this message off to the
751 	 * cache manager op */
752 	return call->type->deliver(call, skb, last);
753 }
754 
755 /*
756  * send an empty reply
757  */
afs_send_empty_reply(struct afs_call * call)758 void afs_send_empty_reply(struct afs_call *call)
759 {
760 	struct msghdr msg;
761 	struct iovec iov[1];
762 
763 	_enter("");
764 
765 	iov[0].iov_base		= NULL;
766 	iov[0].iov_len		= 0;
767 	msg.msg_name		= NULL;
768 	msg.msg_namelen		= 0;
769 	msg.msg_iov		= iov;
770 	msg.msg_iovlen		= 0;
771 	msg.msg_control		= NULL;
772 	msg.msg_controllen	= 0;
773 	msg.msg_flags		= 0;
774 
775 	call->state = AFS_CALL_AWAIT_ACK;
776 	switch (rxrpc_kernel_send_data(call->rxcall, &msg, 0)) {
777 	case 0:
778 		_leave(" [replied]");
779 		return;
780 
781 	case -ENOMEM:
782 		_debug("oom");
783 		rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
784 	default:
785 		rxrpc_kernel_end_call(call->rxcall);
786 		call->rxcall = NULL;
787 		call->type->destructor(call);
788 		afs_free_call(call);
789 		_leave(" [error]");
790 		return;
791 	}
792 }
793 
794 /*
795  * send a simple reply
796  */
afs_send_simple_reply(struct afs_call * call,const void * buf,size_t len)797 void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len)
798 {
799 	struct msghdr msg;
800 	struct iovec iov[1];
801 	int n;
802 
803 	_enter("");
804 
805 	iov[0].iov_base		= (void *) buf;
806 	iov[0].iov_len		= len;
807 	msg.msg_name		= NULL;
808 	msg.msg_namelen		= 0;
809 	msg.msg_iov		= iov;
810 	msg.msg_iovlen		= 1;
811 	msg.msg_control		= NULL;
812 	msg.msg_controllen	= 0;
813 	msg.msg_flags		= 0;
814 
815 	call->state = AFS_CALL_AWAIT_ACK;
816 	n = rxrpc_kernel_send_data(call->rxcall, &msg, len);
817 	if (n >= 0) {
818 		_leave(" [replied]");
819 		return;
820 	}
821 	if (n == -ENOMEM) {
822 		_debug("oom");
823 		rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
824 	}
825 	rxrpc_kernel_end_call(call->rxcall);
826 	call->rxcall = NULL;
827 	call->type->destructor(call);
828 	afs_free_call(call);
829 	_leave(" [error]");
830 }
831 
832 /*
833  * extract a piece of data from the received data socket buffers
834  */
afs_extract_data(struct afs_call * call,struct sk_buff * skb,bool last,void * buf,size_t count)835 int afs_extract_data(struct afs_call *call, struct sk_buff *skb,
836 		     bool last, void *buf, size_t count)
837 {
838 	size_t len = skb->len;
839 
840 	_enter("{%u},{%zu},%d,,%zu", call->offset, len, last, count);
841 
842 	ASSERTCMP(call->offset, <, count);
843 
844 	len = min_t(size_t, len, count - call->offset);
845 	if (skb_copy_bits(skb, 0, buf + call->offset, len) < 0 ||
846 	    !pskb_pull(skb, len))
847 		BUG();
848 	call->offset += len;
849 
850 	if (call->offset < count) {
851 		if (last) {
852 			_leave(" = -EBADMSG [%d < %zu]", call->offset, count);
853 			return -EBADMSG;
854 		}
855 		_leave(" = -EAGAIN");
856 		return -EAGAIN;
857 	}
858 	return 0;
859 }
860