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