xref: /openbmc/linux/fs/afs/rxrpc.c (revision 547840bd)
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* Maintain an RxRPC server socket to do AFS communications through
3  *
4  * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
5  * Written by David Howells (dhowells@redhat.com)
6  */
7 
8 #include <linux/slab.h>
9 #include <linux/sched/signal.h>
10 
11 #include <net/sock.h>
12 #include <net/af_rxrpc.h>
13 #include "internal.h"
14 #include "afs_cm.h"
15 #include "protocol_yfs.h"
16 
17 struct workqueue_struct *afs_async_calls;
18 
19 static void afs_wake_up_call_waiter(struct sock *, struct rxrpc_call *, unsigned long);
20 static void afs_wake_up_async_call(struct sock *, struct rxrpc_call *, unsigned long);
21 static void afs_process_async_call(struct work_struct *);
22 static void afs_rx_new_call(struct sock *, struct rxrpc_call *, unsigned long);
23 static void afs_rx_discard_new_call(struct rxrpc_call *, unsigned long);
24 static int afs_deliver_cm_op_id(struct afs_call *);
25 
26 /* asynchronous incoming call initial processing */
27 static const struct afs_call_type afs_RXCMxxxx = {
28 	.name		= "CB.xxxx",
29 	.deliver	= afs_deliver_cm_op_id,
30 };
31 
32 /*
33  * open an RxRPC socket and bind it to be a server for callback notifications
34  * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT
35  */
36 int afs_open_socket(struct afs_net *net)
37 {
38 	struct sockaddr_rxrpc srx;
39 	struct socket *socket;
40 	int ret;
41 
42 	_enter("");
43 
44 	ret = sock_create_kern(net->net, AF_RXRPC, SOCK_DGRAM, PF_INET6, &socket);
45 	if (ret < 0)
46 		goto error_1;
47 
48 	socket->sk->sk_allocation = GFP_NOFS;
49 
50 	/* bind the callback manager's address to make this a server socket */
51 	memset(&srx, 0, sizeof(srx));
52 	srx.srx_family			= AF_RXRPC;
53 	srx.srx_service			= CM_SERVICE;
54 	srx.transport_type		= SOCK_DGRAM;
55 	srx.transport_len		= sizeof(srx.transport.sin6);
56 	srx.transport.sin6.sin6_family	= AF_INET6;
57 	srx.transport.sin6.sin6_port	= htons(AFS_CM_PORT);
58 
59 	ret = rxrpc_sock_set_min_security_level(socket->sk,
60 						RXRPC_SECURITY_ENCRYPT);
61 	if (ret < 0)
62 		goto error_2;
63 
64 	ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx));
65 	if (ret == -EADDRINUSE) {
66 		srx.transport.sin6.sin6_port = 0;
67 		ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx));
68 	}
69 	if (ret < 0)
70 		goto error_2;
71 
72 	srx.srx_service = YFS_CM_SERVICE;
73 	ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx));
74 	if (ret < 0)
75 		goto error_2;
76 
77 	/* Ideally, we'd turn on service upgrade here, but we can't because
78 	 * OpenAFS is buggy and leaks the userStatus field from packet to
79 	 * packet and between FS packets and CB packets - so if we try to do an
80 	 * upgrade on an FS packet, OpenAFS will leak that into the CB packet
81 	 * it sends back to us.
82 	 */
83 
84 	rxrpc_kernel_new_call_notification(socket, afs_rx_new_call,
85 					   afs_rx_discard_new_call);
86 
87 	ret = kernel_listen(socket, INT_MAX);
88 	if (ret < 0)
89 		goto error_2;
90 
91 	net->socket = socket;
92 	afs_charge_preallocation(&net->charge_preallocation_work);
93 	_leave(" = 0");
94 	return 0;
95 
96 error_2:
97 	sock_release(socket);
98 error_1:
99 	_leave(" = %d", ret);
100 	return ret;
101 }
102 
103 /*
104  * close the RxRPC socket AFS was using
105  */
106 void afs_close_socket(struct afs_net *net)
107 {
108 	_enter("");
109 
110 	kernel_listen(net->socket, 0);
111 	flush_workqueue(afs_async_calls);
112 
113 	if (net->spare_incoming_call) {
114 		afs_put_call(net->spare_incoming_call);
115 		net->spare_incoming_call = NULL;
116 	}
117 
118 	_debug("outstanding %u", atomic_read(&net->nr_outstanding_calls));
119 	wait_var_event(&net->nr_outstanding_calls,
120 		       !atomic_read(&net->nr_outstanding_calls));
121 	_debug("no outstanding calls");
122 
123 	kernel_sock_shutdown(net->socket, SHUT_RDWR);
124 	flush_workqueue(afs_async_calls);
125 	sock_release(net->socket);
126 
127 	_debug("dework");
128 	_leave("");
129 }
130 
131 /*
132  * Allocate a call.
133  */
134 static struct afs_call *afs_alloc_call(struct afs_net *net,
135 				       const struct afs_call_type *type,
136 				       gfp_t gfp)
137 {
138 	struct afs_call *call;
139 	int o;
140 
141 	call = kzalloc(sizeof(*call), gfp);
142 	if (!call)
143 		return NULL;
144 
145 	call->type = type;
146 	call->net = net;
147 	call->debug_id = atomic_inc_return(&rxrpc_debug_id);
148 	atomic_set(&call->usage, 1);
149 	INIT_WORK(&call->async_work, afs_process_async_call);
150 	init_waitqueue_head(&call->waitq);
151 	spin_lock_init(&call->state_lock);
152 	call->iter = &call->def_iter;
153 
154 	o = atomic_inc_return(&net->nr_outstanding_calls);
155 	trace_afs_call(call, afs_call_trace_alloc, 1, o,
156 		       __builtin_return_address(0));
157 	return call;
158 }
159 
160 /*
161  * Dispose of a reference on a call.
162  */
163 void afs_put_call(struct afs_call *call)
164 {
165 	struct afs_net *net = call->net;
166 	int n = atomic_dec_return(&call->usage);
167 	int o = atomic_read(&net->nr_outstanding_calls);
168 
169 	trace_afs_call(call, afs_call_trace_put, n, o,
170 		       __builtin_return_address(0));
171 
172 	ASSERTCMP(n, >=, 0);
173 	if (n == 0) {
174 		ASSERT(!work_pending(&call->async_work));
175 		ASSERT(call->type->name != NULL);
176 
177 		if (call->rxcall) {
178 			rxrpc_kernel_end_call(net->socket, call->rxcall);
179 			call->rxcall = NULL;
180 		}
181 		if (call->type->destructor)
182 			call->type->destructor(call);
183 
184 		afs_put_server(call->net, call->server, afs_server_trace_put_call);
185 		afs_put_cb_interest(call->net, call->cbi);
186 		afs_put_addrlist(call->alist);
187 		kfree(call->request);
188 
189 		trace_afs_call(call, afs_call_trace_free, 0, o,
190 			       __builtin_return_address(0));
191 		kfree(call);
192 
193 		o = atomic_dec_return(&net->nr_outstanding_calls);
194 		if (o == 0)
195 			wake_up_var(&net->nr_outstanding_calls);
196 	}
197 }
198 
199 static struct afs_call *afs_get_call(struct afs_call *call,
200 				     enum afs_call_trace why)
201 {
202 	int u = atomic_inc_return(&call->usage);
203 
204 	trace_afs_call(call, why, u,
205 		       atomic_read(&call->net->nr_outstanding_calls),
206 		       __builtin_return_address(0));
207 	return call;
208 }
209 
210 /*
211  * Queue the call for actual work.
212  */
213 static void afs_queue_call_work(struct afs_call *call)
214 {
215 	if (call->type->work) {
216 		INIT_WORK(&call->work, call->type->work);
217 
218 		afs_get_call(call, afs_call_trace_work);
219 		if (!queue_work(afs_wq, &call->work))
220 			afs_put_call(call);
221 	}
222 }
223 
224 /*
225  * allocate a call with flat request and reply buffers
226  */
227 struct afs_call *afs_alloc_flat_call(struct afs_net *net,
228 				     const struct afs_call_type *type,
229 				     size_t request_size, size_t reply_max)
230 {
231 	struct afs_call *call;
232 
233 	call = afs_alloc_call(net, type, GFP_NOFS);
234 	if (!call)
235 		goto nomem_call;
236 
237 	if (request_size) {
238 		call->request_size = request_size;
239 		call->request = kmalloc(request_size, GFP_NOFS);
240 		if (!call->request)
241 			goto nomem_free;
242 	}
243 
244 	if (reply_max) {
245 		call->reply_max = reply_max;
246 		call->buffer = kmalloc(reply_max, GFP_NOFS);
247 		if (!call->buffer)
248 			goto nomem_free;
249 	}
250 
251 	afs_extract_to_buf(call, call->reply_max);
252 	call->operation_ID = type->op;
253 	init_waitqueue_head(&call->waitq);
254 	return call;
255 
256 nomem_free:
257 	afs_put_call(call);
258 nomem_call:
259 	return NULL;
260 }
261 
262 /*
263  * clean up a call with flat buffer
264  */
265 void afs_flat_call_destructor(struct afs_call *call)
266 {
267 	_enter("");
268 
269 	kfree(call->request);
270 	call->request = NULL;
271 	kfree(call->buffer);
272 	call->buffer = NULL;
273 }
274 
275 #define AFS_BVEC_MAX 8
276 
277 /*
278  * Load the given bvec with the next few pages.
279  */
280 static void afs_load_bvec(struct afs_call *call, struct msghdr *msg,
281 			  struct bio_vec *bv, pgoff_t first, pgoff_t last,
282 			  unsigned offset)
283 {
284 	struct page *pages[AFS_BVEC_MAX];
285 	unsigned int nr, n, i, to, bytes = 0;
286 
287 	nr = min_t(pgoff_t, last - first + 1, AFS_BVEC_MAX);
288 	n = find_get_pages_contig(call->mapping, first, nr, pages);
289 	ASSERTCMP(n, ==, nr);
290 
291 	msg->msg_flags |= MSG_MORE;
292 	for (i = 0; i < nr; i++) {
293 		to = PAGE_SIZE;
294 		if (first + i >= last) {
295 			to = call->last_to;
296 			msg->msg_flags &= ~MSG_MORE;
297 		}
298 		bv[i].bv_page = pages[i];
299 		bv[i].bv_len = to - offset;
300 		bv[i].bv_offset = offset;
301 		bytes += to - offset;
302 		offset = 0;
303 	}
304 
305 	iov_iter_bvec(&msg->msg_iter, WRITE, bv, nr, bytes);
306 }
307 
308 /*
309  * Advance the AFS call state when the RxRPC call ends the transmit phase.
310  */
311 static void afs_notify_end_request_tx(struct sock *sock,
312 				      struct rxrpc_call *rxcall,
313 				      unsigned long call_user_ID)
314 {
315 	struct afs_call *call = (struct afs_call *)call_user_ID;
316 
317 	afs_set_call_state(call, AFS_CALL_CL_REQUESTING, AFS_CALL_CL_AWAIT_REPLY);
318 }
319 
320 /*
321  * attach the data from a bunch of pages on an inode to a call
322  */
323 static int afs_send_pages(struct afs_call *call, struct msghdr *msg)
324 {
325 	struct bio_vec bv[AFS_BVEC_MAX];
326 	unsigned int bytes, nr, loop, offset;
327 	pgoff_t first = call->first, last = call->last;
328 	int ret;
329 
330 	offset = call->first_offset;
331 	call->first_offset = 0;
332 
333 	do {
334 		afs_load_bvec(call, msg, bv, first, last, offset);
335 		trace_afs_send_pages(call, msg, first, last, offset);
336 
337 		offset = 0;
338 		bytes = msg->msg_iter.count;
339 		nr = msg->msg_iter.nr_segs;
340 
341 		ret = rxrpc_kernel_send_data(call->net->socket, call->rxcall, msg,
342 					     bytes, afs_notify_end_request_tx);
343 		for (loop = 0; loop < nr; loop++)
344 			put_page(bv[loop].bv_page);
345 		if (ret < 0)
346 			break;
347 
348 		first += nr;
349 	} while (first <= last);
350 
351 	trace_afs_sent_pages(call, call->first, last, first, ret);
352 	return ret;
353 }
354 
355 /*
356  * Initiate a call and synchronously queue up the parameters for dispatch.  Any
357  * error is stored into the call struct, which the caller must check for.
358  */
359 void afs_make_call(struct afs_addr_cursor *ac, struct afs_call *call, gfp_t gfp)
360 {
361 	struct sockaddr_rxrpc *srx = &ac->alist->addrs[ac->index];
362 	struct rxrpc_call *rxcall;
363 	struct msghdr msg;
364 	struct kvec iov[1];
365 	s64 tx_total_len;
366 	int ret;
367 
368 	_enter(",{%pISp},", &srx->transport);
369 
370 	ASSERT(call->type != NULL);
371 	ASSERT(call->type->name != NULL);
372 
373 	_debug("____MAKE %p{%s,%x} [%d]____",
374 	       call, call->type->name, key_serial(call->key),
375 	       atomic_read(&call->net->nr_outstanding_calls));
376 
377 	call->addr_ix = ac->index;
378 	call->alist = afs_get_addrlist(ac->alist);
379 
380 	/* Work out the length we're going to transmit.  This is awkward for
381 	 * calls such as FS.StoreData where there's an extra injection of data
382 	 * after the initial fixed part.
383 	 */
384 	tx_total_len = call->request_size;
385 	if (call->send_pages) {
386 		if (call->last == call->first) {
387 			tx_total_len += call->last_to - call->first_offset;
388 		} else {
389 			/* It looks mathematically like you should be able to
390 			 * combine the following lines with the ones above, but
391 			 * unsigned arithmetic is fun when it wraps...
392 			 */
393 			tx_total_len += PAGE_SIZE - call->first_offset;
394 			tx_total_len += call->last_to;
395 			tx_total_len += (call->last - call->first - 1) * PAGE_SIZE;
396 		}
397 	}
398 
399 	/* If the call is going to be asynchronous, we need an extra ref for
400 	 * the call to hold itself so the caller need not hang on to its ref.
401 	 */
402 	if (call->async) {
403 		afs_get_call(call, afs_call_trace_get);
404 		call->drop_ref = true;
405 	}
406 
407 	/* create a call */
408 	rxcall = rxrpc_kernel_begin_call(call->net->socket, srx, call->key,
409 					 (unsigned long)call,
410 					 tx_total_len, gfp,
411 					 (call->async ?
412 					  afs_wake_up_async_call :
413 					  afs_wake_up_call_waiter),
414 					 call->upgrade,
415 					 (call->intr ? RXRPC_PREINTERRUPTIBLE :
416 					  RXRPC_UNINTERRUPTIBLE),
417 					 call->debug_id);
418 	if (IS_ERR(rxcall)) {
419 		ret = PTR_ERR(rxcall);
420 		call->error = ret;
421 		goto error_kill_call;
422 	}
423 
424 	call->rxcall = rxcall;
425 
426 	if (call->max_lifespan)
427 		rxrpc_kernel_set_max_life(call->net->socket, rxcall,
428 					  call->max_lifespan);
429 
430 	/* send the request */
431 	iov[0].iov_base	= call->request;
432 	iov[0].iov_len	= call->request_size;
433 
434 	msg.msg_name		= NULL;
435 	msg.msg_namelen		= 0;
436 	iov_iter_kvec(&msg.msg_iter, WRITE, iov, 1, call->request_size);
437 	msg.msg_control		= NULL;
438 	msg.msg_controllen	= 0;
439 	msg.msg_flags		= MSG_WAITALL | (call->send_pages ? MSG_MORE : 0);
440 
441 	ret = rxrpc_kernel_send_data(call->net->socket, rxcall,
442 				     &msg, call->request_size,
443 				     afs_notify_end_request_tx);
444 	if (ret < 0)
445 		goto error_do_abort;
446 
447 	if (call->send_pages) {
448 		ret = afs_send_pages(call, &msg);
449 		if (ret < 0)
450 			goto error_do_abort;
451 	}
452 
453 	/* Note that at this point, we may have received the reply or an abort
454 	 * - and an asynchronous call may already have completed.
455 	 *
456 	 * afs_wait_for_call_to_complete(call, ac)
457 	 * must be called to synchronously clean up.
458 	 */
459 	return;
460 
461 error_do_abort:
462 	if (ret != -ECONNABORTED) {
463 		rxrpc_kernel_abort_call(call->net->socket, rxcall,
464 					RX_USER_ABORT, ret, "KSD");
465 	} else {
466 		iov_iter_kvec(&msg.msg_iter, READ, NULL, 0, 0);
467 		rxrpc_kernel_recv_data(call->net->socket, rxcall,
468 				       &msg.msg_iter, false,
469 				       &call->abort_code, &call->service_id);
470 		ac->abort_code = call->abort_code;
471 		ac->responded = true;
472 	}
473 	call->error = ret;
474 	trace_afs_call_done(call);
475 error_kill_call:
476 	if (call->type->done)
477 		call->type->done(call);
478 
479 	/* We need to dispose of the extra ref we grabbed for an async call.
480 	 * The call, however, might be queued on afs_async_calls and we need to
481 	 * make sure we don't get any more notifications that might requeue it.
482 	 */
483 	if (call->rxcall) {
484 		rxrpc_kernel_end_call(call->net->socket, call->rxcall);
485 		call->rxcall = NULL;
486 	}
487 	if (call->async) {
488 		if (cancel_work_sync(&call->async_work))
489 			afs_put_call(call);
490 		afs_put_call(call);
491 	}
492 
493 	ac->error = ret;
494 	call->state = AFS_CALL_COMPLETE;
495 	_leave(" = %d", ret);
496 }
497 
498 /*
499  * deliver messages to a call
500  */
501 static void afs_deliver_to_call(struct afs_call *call)
502 {
503 	enum afs_call_state state;
504 	u32 abort_code, remote_abort = 0;
505 	int ret;
506 
507 	_enter("%s", call->type->name);
508 
509 	while (state = READ_ONCE(call->state),
510 	       state == AFS_CALL_CL_AWAIT_REPLY ||
511 	       state == AFS_CALL_SV_AWAIT_OP_ID ||
512 	       state == AFS_CALL_SV_AWAIT_REQUEST ||
513 	       state == AFS_CALL_SV_AWAIT_ACK
514 	       ) {
515 		if (state == AFS_CALL_SV_AWAIT_ACK) {
516 			iov_iter_kvec(&call->def_iter, READ, NULL, 0, 0);
517 			ret = rxrpc_kernel_recv_data(call->net->socket,
518 						     call->rxcall, &call->def_iter,
519 						     false, &remote_abort,
520 						     &call->service_id);
521 			trace_afs_receive_data(call, &call->def_iter, false, ret);
522 
523 			if (ret == -EINPROGRESS || ret == -EAGAIN)
524 				return;
525 			if (ret < 0 || ret == 1) {
526 				if (ret == 1)
527 					ret = 0;
528 				goto call_complete;
529 			}
530 			return;
531 		}
532 
533 		if (!call->have_reply_time &&
534 		    rxrpc_kernel_get_reply_time(call->net->socket,
535 						call->rxcall,
536 						&call->reply_time))
537 			call->have_reply_time = true;
538 
539 		ret = call->type->deliver(call);
540 		state = READ_ONCE(call->state);
541 		switch (ret) {
542 		case 0:
543 			afs_queue_call_work(call);
544 			if (state == AFS_CALL_CL_PROC_REPLY) {
545 				if (call->cbi)
546 					set_bit(AFS_SERVER_FL_MAY_HAVE_CB,
547 						&call->cbi->server->flags);
548 				goto call_complete;
549 			}
550 			ASSERTCMP(state, >, AFS_CALL_CL_PROC_REPLY);
551 			goto done;
552 		case -EINPROGRESS:
553 		case -EAGAIN:
554 			goto out;
555 		case -ECONNABORTED:
556 			ASSERTCMP(state, ==, AFS_CALL_COMPLETE);
557 			goto done;
558 		case -ENOTSUPP:
559 			abort_code = RXGEN_OPCODE;
560 			rxrpc_kernel_abort_call(call->net->socket, call->rxcall,
561 						abort_code, ret, "KIV");
562 			goto local_abort;
563 		case -EIO:
564 			pr_err("kAFS: Call %u in bad state %u\n",
565 			       call->debug_id, state);
566 			/* Fall through */
567 		case -ENODATA:
568 		case -EBADMSG:
569 		case -EMSGSIZE:
570 			abort_code = RXGEN_CC_UNMARSHAL;
571 			if (state != AFS_CALL_CL_AWAIT_REPLY)
572 				abort_code = RXGEN_SS_UNMARSHAL;
573 			rxrpc_kernel_abort_call(call->net->socket, call->rxcall,
574 						abort_code, ret, "KUM");
575 			goto local_abort;
576 		default:
577 			abort_code = RX_USER_ABORT;
578 			rxrpc_kernel_abort_call(call->net->socket, call->rxcall,
579 						abort_code, ret, "KER");
580 			goto local_abort;
581 		}
582 	}
583 
584 done:
585 	if (call->type->done)
586 		call->type->done(call);
587 out:
588 	_leave("");
589 	return;
590 
591 local_abort:
592 	abort_code = 0;
593 call_complete:
594 	afs_set_call_complete(call, ret, remote_abort);
595 	state = AFS_CALL_COMPLETE;
596 	goto done;
597 }
598 
599 /*
600  * Wait synchronously for a call to complete and clean up the call struct.
601  */
602 long afs_wait_for_call_to_complete(struct afs_call *call,
603 				   struct afs_addr_cursor *ac)
604 {
605 	long ret;
606 	bool rxrpc_complete = false;
607 
608 	DECLARE_WAITQUEUE(myself, current);
609 
610 	_enter("");
611 
612 	ret = call->error;
613 	if (ret < 0)
614 		goto out;
615 
616 	add_wait_queue(&call->waitq, &myself);
617 	for (;;) {
618 		set_current_state(TASK_UNINTERRUPTIBLE);
619 
620 		/* deliver any messages that are in the queue */
621 		if (!afs_check_call_state(call, AFS_CALL_COMPLETE) &&
622 		    call->need_attention) {
623 			call->need_attention = false;
624 			__set_current_state(TASK_RUNNING);
625 			afs_deliver_to_call(call);
626 			continue;
627 		}
628 
629 		if (afs_check_call_state(call, AFS_CALL_COMPLETE))
630 			break;
631 
632 		if (!rxrpc_kernel_check_life(call->net->socket, call->rxcall)) {
633 			/* rxrpc terminated the call. */
634 			rxrpc_complete = true;
635 			break;
636 		}
637 
638 		schedule();
639 	}
640 
641 	remove_wait_queue(&call->waitq, &myself);
642 	__set_current_state(TASK_RUNNING);
643 
644 	if (!afs_check_call_state(call, AFS_CALL_COMPLETE)) {
645 		if (rxrpc_complete) {
646 			afs_set_call_complete(call, call->error, call->abort_code);
647 		} else {
648 			/* Kill off the call if it's still live. */
649 			_debug("call interrupted");
650 			if (rxrpc_kernel_abort_call(call->net->socket, call->rxcall,
651 						    RX_USER_ABORT, -EINTR, "KWI"))
652 				afs_set_call_complete(call, -EINTR, 0);
653 		}
654 	}
655 
656 	spin_lock_bh(&call->state_lock);
657 	ac->abort_code = call->abort_code;
658 	ac->error = call->error;
659 	spin_unlock_bh(&call->state_lock);
660 
661 	ret = ac->error;
662 	switch (ret) {
663 	case 0:
664 		ret = call->ret0;
665 		call->ret0 = 0;
666 
667 		/* Fall through */
668 	case -ECONNABORTED:
669 		ac->responded = true;
670 		break;
671 	}
672 
673 out:
674 	_debug("call complete");
675 	afs_put_call(call);
676 	_leave(" = %p", (void *)ret);
677 	return ret;
678 }
679 
680 /*
681  * wake up a waiting call
682  */
683 static void afs_wake_up_call_waiter(struct sock *sk, struct rxrpc_call *rxcall,
684 				    unsigned long call_user_ID)
685 {
686 	struct afs_call *call = (struct afs_call *)call_user_ID;
687 
688 	call->need_attention = true;
689 	wake_up(&call->waitq);
690 }
691 
692 /*
693  * wake up an asynchronous call
694  */
695 static void afs_wake_up_async_call(struct sock *sk, struct rxrpc_call *rxcall,
696 				   unsigned long call_user_ID)
697 {
698 	struct afs_call *call = (struct afs_call *)call_user_ID;
699 	int u;
700 
701 	trace_afs_notify_call(rxcall, call);
702 	call->need_attention = true;
703 
704 	u = atomic_fetch_add_unless(&call->usage, 1, 0);
705 	if (u != 0) {
706 		trace_afs_call(call, afs_call_trace_wake, u + 1,
707 			       atomic_read(&call->net->nr_outstanding_calls),
708 			       __builtin_return_address(0));
709 
710 		if (!queue_work(afs_async_calls, &call->async_work))
711 			afs_put_call(call);
712 	}
713 }
714 
715 /*
716  * Perform I/O processing on an asynchronous call.  The work item carries a ref
717  * to the call struct that we either need to release or to pass on.
718  */
719 static void afs_process_async_call(struct work_struct *work)
720 {
721 	struct afs_call *call = container_of(work, struct afs_call, async_work);
722 
723 	_enter("");
724 
725 	if (call->state < AFS_CALL_COMPLETE && call->need_attention) {
726 		call->need_attention = false;
727 		afs_deliver_to_call(call);
728 	}
729 
730 	afs_put_call(call);
731 	_leave("");
732 }
733 
734 static void afs_rx_attach(struct rxrpc_call *rxcall, unsigned long user_call_ID)
735 {
736 	struct afs_call *call = (struct afs_call *)user_call_ID;
737 
738 	call->rxcall = rxcall;
739 }
740 
741 /*
742  * Charge the incoming call preallocation.
743  */
744 void afs_charge_preallocation(struct work_struct *work)
745 {
746 	struct afs_net *net =
747 		container_of(work, struct afs_net, charge_preallocation_work);
748 	struct afs_call *call = net->spare_incoming_call;
749 
750 	for (;;) {
751 		if (!call) {
752 			call = afs_alloc_call(net, &afs_RXCMxxxx, GFP_KERNEL);
753 			if (!call)
754 				break;
755 
756 			call->drop_ref = true;
757 			call->async = true;
758 			call->state = AFS_CALL_SV_AWAIT_OP_ID;
759 			init_waitqueue_head(&call->waitq);
760 			afs_extract_to_tmp(call);
761 		}
762 
763 		if (rxrpc_kernel_charge_accept(net->socket,
764 					       afs_wake_up_async_call,
765 					       afs_rx_attach,
766 					       (unsigned long)call,
767 					       GFP_KERNEL,
768 					       call->debug_id) < 0)
769 			break;
770 		call = NULL;
771 	}
772 	net->spare_incoming_call = call;
773 }
774 
775 /*
776  * Discard a preallocated call when a socket is shut down.
777  */
778 static void afs_rx_discard_new_call(struct rxrpc_call *rxcall,
779 				    unsigned long user_call_ID)
780 {
781 	struct afs_call *call = (struct afs_call *)user_call_ID;
782 
783 	call->rxcall = NULL;
784 	afs_put_call(call);
785 }
786 
787 /*
788  * Notification of an incoming call.
789  */
790 static void afs_rx_new_call(struct sock *sk, struct rxrpc_call *rxcall,
791 			    unsigned long user_call_ID)
792 {
793 	struct afs_net *net = afs_sock2net(sk);
794 
795 	queue_work(afs_wq, &net->charge_preallocation_work);
796 }
797 
798 /*
799  * Grab the operation ID from an incoming cache manager call.  The socket
800  * buffer is discarded on error or if we don't yet have sufficient data.
801  */
802 static int afs_deliver_cm_op_id(struct afs_call *call)
803 {
804 	int ret;
805 
806 	_enter("{%zu}", iov_iter_count(call->iter));
807 
808 	/* the operation ID forms the first four bytes of the request data */
809 	ret = afs_extract_data(call, true);
810 	if (ret < 0)
811 		return ret;
812 
813 	call->operation_ID = ntohl(call->tmp);
814 	afs_set_call_state(call, AFS_CALL_SV_AWAIT_OP_ID, AFS_CALL_SV_AWAIT_REQUEST);
815 
816 	/* ask the cache manager to route the call (it'll change the call type
817 	 * if successful) */
818 	if (!afs_cm_incoming_call(call))
819 		return -ENOTSUPP;
820 
821 	trace_afs_cb_call(call);
822 
823 	/* pass responsibility for the remainer of this message off to the
824 	 * cache manager op */
825 	return call->type->deliver(call);
826 }
827 
828 /*
829  * Advance the AFS call state when an RxRPC service call ends the transmit
830  * phase.
831  */
832 static void afs_notify_end_reply_tx(struct sock *sock,
833 				    struct rxrpc_call *rxcall,
834 				    unsigned long call_user_ID)
835 {
836 	struct afs_call *call = (struct afs_call *)call_user_ID;
837 
838 	afs_set_call_state(call, AFS_CALL_SV_REPLYING, AFS_CALL_SV_AWAIT_ACK);
839 }
840 
841 /*
842  * send an empty reply
843  */
844 void afs_send_empty_reply(struct afs_call *call)
845 {
846 	struct afs_net *net = call->net;
847 	struct msghdr msg;
848 
849 	_enter("");
850 
851 	rxrpc_kernel_set_tx_length(net->socket, call->rxcall, 0);
852 
853 	msg.msg_name		= NULL;
854 	msg.msg_namelen		= 0;
855 	iov_iter_kvec(&msg.msg_iter, WRITE, NULL, 0, 0);
856 	msg.msg_control		= NULL;
857 	msg.msg_controllen	= 0;
858 	msg.msg_flags		= 0;
859 
860 	switch (rxrpc_kernel_send_data(net->socket, call->rxcall, &msg, 0,
861 				       afs_notify_end_reply_tx)) {
862 	case 0:
863 		_leave(" [replied]");
864 		return;
865 
866 	case -ENOMEM:
867 		_debug("oom");
868 		rxrpc_kernel_abort_call(net->socket, call->rxcall,
869 					RX_USER_ABORT, -ENOMEM, "KOO");
870 		/* Fall through */
871 	default:
872 		_leave(" [error]");
873 		return;
874 	}
875 }
876 
877 /*
878  * send a simple reply
879  */
880 void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len)
881 {
882 	struct afs_net *net = call->net;
883 	struct msghdr msg;
884 	struct kvec iov[1];
885 	int n;
886 
887 	_enter("");
888 
889 	rxrpc_kernel_set_tx_length(net->socket, call->rxcall, len);
890 
891 	iov[0].iov_base		= (void *) buf;
892 	iov[0].iov_len		= len;
893 	msg.msg_name		= NULL;
894 	msg.msg_namelen		= 0;
895 	iov_iter_kvec(&msg.msg_iter, WRITE, iov, 1, len);
896 	msg.msg_control		= NULL;
897 	msg.msg_controllen	= 0;
898 	msg.msg_flags		= 0;
899 
900 	n = rxrpc_kernel_send_data(net->socket, call->rxcall, &msg, len,
901 				   afs_notify_end_reply_tx);
902 	if (n >= 0) {
903 		/* Success */
904 		_leave(" [replied]");
905 		return;
906 	}
907 
908 	if (n == -ENOMEM) {
909 		_debug("oom");
910 		rxrpc_kernel_abort_call(net->socket, call->rxcall,
911 					RX_USER_ABORT, -ENOMEM, "KOO");
912 	}
913 	_leave(" [error]");
914 }
915 
916 /*
917  * Extract a piece of data from the received data socket buffers.
918  */
919 int afs_extract_data(struct afs_call *call, bool want_more)
920 {
921 	struct afs_net *net = call->net;
922 	struct iov_iter *iter = call->iter;
923 	enum afs_call_state state;
924 	u32 remote_abort = 0;
925 	int ret;
926 
927 	_enter("{%s,%zu},%d", call->type->name, iov_iter_count(iter), want_more);
928 
929 	ret = rxrpc_kernel_recv_data(net->socket, call->rxcall, iter,
930 				     want_more, &remote_abort,
931 				     &call->service_id);
932 	if (ret == 0 || ret == -EAGAIN)
933 		return ret;
934 
935 	state = READ_ONCE(call->state);
936 	if (ret == 1) {
937 		switch (state) {
938 		case AFS_CALL_CL_AWAIT_REPLY:
939 			afs_set_call_state(call, state, AFS_CALL_CL_PROC_REPLY);
940 			break;
941 		case AFS_CALL_SV_AWAIT_REQUEST:
942 			afs_set_call_state(call, state, AFS_CALL_SV_REPLYING);
943 			break;
944 		case AFS_CALL_COMPLETE:
945 			kdebug("prem complete %d", call->error);
946 			return afs_io_error(call, afs_io_error_extract);
947 		default:
948 			break;
949 		}
950 		return 0;
951 	}
952 
953 	afs_set_call_complete(call, ret, remote_abort);
954 	return ret;
955 }
956 
957 /*
958  * Log protocol error production.
959  */
960 noinline int afs_protocol_error(struct afs_call *call, int error,
961 				enum afs_eproto_cause cause)
962 {
963 	trace_afs_protocol_error(call, error, cause);
964 	return error;
965 }
966