xref: /linux/net/rxrpc/rxperf.c (revision 4b132aacb0768ac1e652cf517097ea6f237214b9)
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* In-kernel rxperf server for testing purposes.
3  *
4  * Copyright (C) 2022 Red Hat, Inc. All Rights Reserved.
5  * Written by David Howells (dhowells@redhat.com)
6  */
7 
8 #define pr_fmt(fmt) "rxperf: " fmt
9 #include <linux/module.h>
10 #include <linux/slab.h>
11 #include <net/sock.h>
12 #include <net/af_rxrpc.h>
13 #define RXRPC_TRACE_ONLY_DEFINE_ENUMS
14 #include <trace/events/rxrpc.h>
15 
16 MODULE_DESCRIPTION("rxperf test server (afs)");
17 MODULE_AUTHOR("Red Hat, Inc.");
18 MODULE_LICENSE("GPL");
19 
20 #define RXPERF_PORT		7009
21 #define RX_PERF_SERVICE		147
22 #define RX_PERF_VERSION		3
23 #define RX_PERF_SEND		0
24 #define RX_PERF_RECV		1
25 #define RX_PERF_RPC		3
26 #define RX_PERF_FILE		4
27 #define RX_PERF_MAGIC_COOKIE	0x4711
28 
29 struct rxperf_proto_params {
30 	__be32		version;
31 	__be32		type;
32 	__be32		rsize;
33 	__be32		wsize;
34 } __packed;
35 
36 static const u8 rxperf_magic_cookie[] = { 0x00, 0x00, 0x47, 0x11 };
37 static const u8 secret[8] = { 0xa7, 0x83, 0x8a, 0xcb, 0xc7, 0x83, 0xec, 0x94 };
38 
39 enum rxperf_call_state {
40 	RXPERF_CALL_SV_AWAIT_PARAMS,	/* Server: Awaiting parameter block */
41 	RXPERF_CALL_SV_AWAIT_REQUEST,	/* Server: Awaiting request data */
42 	RXPERF_CALL_SV_REPLYING,	/* Server: Replying */
43 	RXPERF_CALL_SV_AWAIT_ACK,	/* Server: Awaiting final ACK */
44 	RXPERF_CALL_COMPLETE,		/* Completed or failed */
45 };
46 
47 struct rxperf_call {
48 	struct rxrpc_call	*rxcall;
49 	struct iov_iter		iter;
50 	struct kvec		kvec[1];
51 	struct work_struct	work;
52 	const char		*type;
53 	size_t			iov_len;
54 	size_t			req_len;	/* Size of request blob */
55 	size_t			reply_len;	/* Size of reply blob */
56 	unsigned int		debug_id;
57 	unsigned int		operation_id;
58 	struct rxperf_proto_params params;
59 	__be32			tmp[2];
60 	s32			abort_code;
61 	enum rxperf_call_state	state;
62 	short			error;
63 	unsigned short		unmarshal;
64 	u16			service_id;
65 	int (*deliver)(struct rxperf_call *call);
66 	void (*processor)(struct work_struct *work);
67 };
68 
69 static struct socket *rxperf_socket;
70 static struct key *rxperf_sec_keyring;	/* Ring of security/crypto keys */
71 static struct workqueue_struct *rxperf_workqueue;
72 
73 static void rxperf_deliver_to_call(struct work_struct *work);
74 static int rxperf_deliver_param_block(struct rxperf_call *call);
75 static int rxperf_deliver_request(struct rxperf_call *call);
76 static int rxperf_process_call(struct rxperf_call *call);
77 static void rxperf_charge_preallocation(struct work_struct *work);
78 
79 static DECLARE_WORK(rxperf_charge_preallocation_work,
80 		    rxperf_charge_preallocation);
81 
82 static inline void rxperf_set_call_state(struct rxperf_call *call,
83 					 enum rxperf_call_state to)
84 {
85 	call->state = to;
86 }
87 
88 static inline void rxperf_set_call_complete(struct rxperf_call *call,
89 					    int error, s32 remote_abort)
90 {
91 	if (call->state != RXPERF_CALL_COMPLETE) {
92 		call->abort_code = remote_abort;
93 		call->error = error;
94 		call->state = RXPERF_CALL_COMPLETE;
95 	}
96 }
97 
98 static void rxperf_rx_discard_new_call(struct rxrpc_call *rxcall,
99 				       unsigned long user_call_ID)
100 {
101 	kfree((struct rxperf_call *)user_call_ID);
102 }
103 
104 static void rxperf_rx_new_call(struct sock *sk, struct rxrpc_call *rxcall,
105 			       unsigned long user_call_ID)
106 {
107 	queue_work(rxperf_workqueue, &rxperf_charge_preallocation_work);
108 }
109 
110 static void rxperf_queue_call_work(struct rxperf_call *call)
111 {
112 	queue_work(rxperf_workqueue, &call->work);
113 }
114 
115 static void rxperf_notify_rx(struct sock *sk, struct rxrpc_call *rxcall,
116 			     unsigned long call_user_ID)
117 {
118 	struct rxperf_call *call = (struct rxperf_call *)call_user_ID;
119 
120 	if (call->state != RXPERF_CALL_COMPLETE)
121 		rxperf_queue_call_work(call);
122 }
123 
124 static void rxperf_rx_attach(struct rxrpc_call *rxcall, unsigned long user_call_ID)
125 {
126 	struct rxperf_call *call = (struct rxperf_call *)user_call_ID;
127 
128 	call->rxcall = rxcall;
129 }
130 
131 static void rxperf_notify_end_reply_tx(struct sock *sock,
132 				       struct rxrpc_call *rxcall,
133 				       unsigned long call_user_ID)
134 {
135 	rxperf_set_call_state((struct rxperf_call *)call_user_ID,
136 			      RXPERF_CALL_SV_AWAIT_ACK);
137 }
138 
139 /*
140  * Charge the incoming call preallocation.
141  */
142 static void rxperf_charge_preallocation(struct work_struct *work)
143 {
144 	struct rxperf_call *call;
145 
146 	for (;;) {
147 		call = kzalloc(sizeof(*call), GFP_KERNEL);
148 		if (!call)
149 			break;
150 
151 		call->type		= "unset";
152 		call->debug_id		= atomic_inc_return(&rxrpc_debug_id);
153 		call->deliver		= rxperf_deliver_param_block;
154 		call->state		= RXPERF_CALL_SV_AWAIT_PARAMS;
155 		call->service_id	= RX_PERF_SERVICE;
156 		call->iov_len		= sizeof(call->params);
157 		call->kvec[0].iov_len	= sizeof(call->params);
158 		call->kvec[0].iov_base	= &call->params;
159 		iov_iter_kvec(&call->iter, READ, call->kvec, 1, call->iov_len);
160 		INIT_WORK(&call->work, rxperf_deliver_to_call);
161 
162 		if (rxrpc_kernel_charge_accept(rxperf_socket,
163 					       rxperf_notify_rx,
164 					       rxperf_rx_attach,
165 					       (unsigned long)call,
166 					       GFP_KERNEL,
167 					       call->debug_id) < 0)
168 			break;
169 		call = NULL;
170 	}
171 
172 	kfree(call);
173 }
174 
175 /*
176  * Open an rxrpc socket and bind it to be a server for callback notifications
177  * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT
178  */
179 static int rxperf_open_socket(void)
180 {
181 	struct sockaddr_rxrpc srx;
182 	struct socket *socket;
183 	int ret;
184 
185 	ret = sock_create_kern(&init_net, AF_RXRPC, SOCK_DGRAM, PF_INET6,
186 			       &socket);
187 	if (ret < 0)
188 		goto error_1;
189 
190 	socket->sk->sk_allocation = GFP_NOFS;
191 
192 	/* bind the callback manager's address to make this a server socket */
193 	memset(&srx, 0, sizeof(srx));
194 	srx.srx_family			= AF_RXRPC;
195 	srx.srx_service			= RX_PERF_SERVICE;
196 	srx.transport_type		= SOCK_DGRAM;
197 	srx.transport_len		= sizeof(srx.transport.sin6);
198 	srx.transport.sin6.sin6_family	= AF_INET6;
199 	srx.transport.sin6.sin6_port	= htons(RXPERF_PORT);
200 
201 	ret = rxrpc_sock_set_min_security_level(socket->sk,
202 						RXRPC_SECURITY_ENCRYPT);
203 	if (ret < 0)
204 		goto error_2;
205 
206 	ret = rxrpc_sock_set_security_keyring(socket->sk, rxperf_sec_keyring);
207 
208 	ret = kernel_bind(socket, (struct sockaddr *)&srx, sizeof(srx));
209 	if (ret < 0)
210 		goto error_2;
211 
212 	rxrpc_kernel_new_call_notification(socket, rxperf_rx_new_call,
213 					   rxperf_rx_discard_new_call);
214 
215 	ret = kernel_listen(socket, INT_MAX);
216 	if (ret < 0)
217 		goto error_2;
218 
219 	rxperf_socket = socket;
220 	rxperf_charge_preallocation(&rxperf_charge_preallocation_work);
221 	return 0;
222 
223 error_2:
224 	sock_release(socket);
225 error_1:
226 	pr_err("Can't set up rxperf socket: %d\n", ret);
227 	return ret;
228 }
229 
230 /*
231  * close the rxrpc socket rxperf was using
232  */
233 static void rxperf_close_socket(void)
234 {
235 	kernel_listen(rxperf_socket, 0);
236 	kernel_sock_shutdown(rxperf_socket, SHUT_RDWR);
237 	flush_workqueue(rxperf_workqueue);
238 	sock_release(rxperf_socket);
239 }
240 
241 /*
242  * Log remote abort codes that indicate that we have a protocol disagreement
243  * with the server.
244  */
245 static void rxperf_log_error(struct rxperf_call *call, s32 remote_abort)
246 {
247 	static int max = 0;
248 	const char *msg;
249 	int m;
250 
251 	switch (remote_abort) {
252 	case RX_EOF:		 msg = "unexpected EOF";	break;
253 	case RXGEN_CC_MARSHAL:	 msg = "client marshalling";	break;
254 	case RXGEN_CC_UNMARSHAL: msg = "client unmarshalling";	break;
255 	case RXGEN_SS_MARSHAL:	 msg = "server marshalling";	break;
256 	case RXGEN_SS_UNMARSHAL: msg = "server unmarshalling";	break;
257 	case RXGEN_DECODE:	 msg = "opcode decode";		break;
258 	case RXGEN_SS_XDRFREE:	 msg = "server XDR cleanup";	break;
259 	case RXGEN_CC_XDRFREE:	 msg = "client XDR cleanup";	break;
260 	case -32:		 msg = "insufficient data";	break;
261 	default:
262 		return;
263 	}
264 
265 	m = max;
266 	if (m < 3) {
267 		max = m + 1;
268 		pr_info("Peer reported %s failure on %s\n", msg, call->type);
269 	}
270 }
271 
272 /*
273  * deliver messages to a call
274  */
275 static void rxperf_deliver_to_call(struct work_struct *work)
276 {
277 	struct rxperf_call *call = container_of(work, struct rxperf_call, work);
278 	enum rxperf_call_state state;
279 	u32 abort_code, remote_abort = 0;
280 	int ret = 0;
281 
282 	if (call->state == RXPERF_CALL_COMPLETE)
283 		return;
284 
285 	while (state = call->state,
286 	       state == RXPERF_CALL_SV_AWAIT_PARAMS ||
287 	       state == RXPERF_CALL_SV_AWAIT_REQUEST ||
288 	       state == RXPERF_CALL_SV_AWAIT_ACK
289 	       ) {
290 		if (state == RXPERF_CALL_SV_AWAIT_ACK) {
291 			if (!rxrpc_kernel_check_life(rxperf_socket, call->rxcall))
292 				goto call_complete;
293 			return;
294 		}
295 
296 		ret = call->deliver(call);
297 		if (ret == 0)
298 			ret = rxperf_process_call(call);
299 
300 		switch (ret) {
301 		case 0:
302 			continue;
303 		case -EINPROGRESS:
304 		case -EAGAIN:
305 			return;
306 		case -ECONNABORTED:
307 			rxperf_log_error(call, call->abort_code);
308 			goto call_complete;
309 		case -EOPNOTSUPP:
310 			abort_code = RXGEN_OPCODE;
311 			rxrpc_kernel_abort_call(rxperf_socket, call->rxcall,
312 						abort_code, ret,
313 						rxperf_abort_op_not_supported);
314 			goto call_complete;
315 		case -ENOTSUPP:
316 			abort_code = RX_USER_ABORT;
317 			rxrpc_kernel_abort_call(rxperf_socket, call->rxcall,
318 						abort_code, ret,
319 						rxperf_abort_op_not_supported);
320 			goto call_complete;
321 		case -EIO:
322 			pr_err("Call %u in bad state %u\n",
323 			       call->debug_id, call->state);
324 			fallthrough;
325 		case -ENODATA:
326 		case -EBADMSG:
327 		case -EMSGSIZE:
328 		case -ENOMEM:
329 		case -EFAULT:
330 			rxrpc_kernel_abort_call(rxperf_socket, call->rxcall,
331 						RXGEN_SS_UNMARSHAL, ret,
332 						rxperf_abort_unmarshal_error);
333 			goto call_complete;
334 		default:
335 			rxrpc_kernel_abort_call(rxperf_socket, call->rxcall,
336 						RX_CALL_DEAD, ret,
337 						rxperf_abort_general_error);
338 			goto call_complete;
339 		}
340 	}
341 
342 call_complete:
343 	rxperf_set_call_complete(call, ret, remote_abort);
344 	/* The call may have been requeued */
345 	rxrpc_kernel_shutdown_call(rxperf_socket, call->rxcall);
346 	rxrpc_kernel_put_call(rxperf_socket, call->rxcall);
347 	cancel_work(&call->work);
348 	kfree(call);
349 }
350 
351 /*
352  * Extract a piece of data from the received data socket buffers.
353  */
354 static int rxperf_extract_data(struct rxperf_call *call, bool want_more)
355 {
356 	u32 remote_abort = 0;
357 	int ret;
358 
359 	ret = rxrpc_kernel_recv_data(rxperf_socket, call->rxcall, &call->iter,
360 				     &call->iov_len, want_more, &remote_abort,
361 				     &call->service_id);
362 	pr_debug("Extract i=%zu l=%zu m=%u ret=%d\n",
363 		 iov_iter_count(&call->iter), call->iov_len, want_more, ret);
364 	if (ret == 0 || ret == -EAGAIN)
365 		return ret;
366 
367 	if (ret == 1) {
368 		switch (call->state) {
369 		case RXPERF_CALL_SV_AWAIT_REQUEST:
370 			rxperf_set_call_state(call, RXPERF_CALL_SV_REPLYING);
371 			break;
372 		case RXPERF_CALL_COMPLETE:
373 			pr_debug("premature completion %d", call->error);
374 			return call->error;
375 		default:
376 			break;
377 		}
378 		return 0;
379 	}
380 
381 	rxperf_set_call_complete(call, ret, remote_abort);
382 	return ret;
383 }
384 
385 /*
386  * Grab the operation ID from an incoming manager call.
387  */
388 static int rxperf_deliver_param_block(struct rxperf_call *call)
389 {
390 	u32 version;
391 	int ret;
392 
393 	/* Extract the parameter block */
394 	ret = rxperf_extract_data(call, true);
395 	if (ret < 0)
396 		return ret;
397 
398 	version			= ntohl(call->params.version);
399 	call->operation_id	= ntohl(call->params.type);
400 	call->deliver		= rxperf_deliver_request;
401 
402 	if (version != RX_PERF_VERSION) {
403 		pr_info("Version mismatch %x\n", version);
404 		return -ENOTSUPP;
405 	}
406 
407 	switch (call->operation_id) {
408 	case RX_PERF_SEND:
409 		call->type = "send";
410 		call->reply_len = 0;
411 		call->iov_len = 4;	/* Expect req size */
412 		break;
413 	case RX_PERF_RECV:
414 		call->type = "recv";
415 		call->req_len = 0;
416 		call->iov_len = 4;	/* Expect reply size */
417 		break;
418 	case RX_PERF_RPC:
419 		call->type = "rpc";
420 		call->iov_len = 8;	/* Expect req size and reply size */
421 		break;
422 	case RX_PERF_FILE:
423 		call->type = "file";
424 		fallthrough;
425 	default:
426 		return -EOPNOTSUPP;
427 	}
428 
429 	rxperf_set_call_state(call, RXPERF_CALL_SV_AWAIT_REQUEST);
430 	return call->deliver(call);
431 }
432 
433 /*
434  * Deliver the request data.
435  */
436 static int rxperf_deliver_request(struct rxperf_call *call)
437 {
438 	int ret;
439 
440 	switch (call->unmarshal) {
441 	case 0:
442 		call->kvec[0].iov_len	= call->iov_len;
443 		call->kvec[0].iov_base	= call->tmp;
444 		iov_iter_kvec(&call->iter, READ, call->kvec, 1, call->iov_len);
445 		call->unmarshal++;
446 		fallthrough;
447 	case 1:
448 		ret = rxperf_extract_data(call, true);
449 		if (ret < 0)
450 			return ret;
451 
452 		switch (call->operation_id) {
453 		case RX_PERF_SEND:
454 			call->type = "send";
455 			call->req_len	= ntohl(call->tmp[0]);
456 			call->reply_len	= 0;
457 			break;
458 		case RX_PERF_RECV:
459 			call->type = "recv";
460 			call->req_len = 0;
461 			call->reply_len	= ntohl(call->tmp[0]);
462 			break;
463 		case RX_PERF_RPC:
464 			call->type = "rpc";
465 			call->req_len	= ntohl(call->tmp[0]);
466 			call->reply_len	= ntohl(call->tmp[1]);
467 			break;
468 		default:
469 			pr_info("Can't parse extra params\n");
470 			return -EIO;
471 		}
472 
473 		pr_debug("CALL op=%s rq=%zx rp=%zx\n",
474 			 call->type, call->req_len, call->reply_len);
475 
476 		call->iov_len = call->req_len;
477 		iov_iter_discard(&call->iter, READ, call->req_len);
478 		call->unmarshal++;
479 		fallthrough;
480 	case 2:
481 		ret = rxperf_extract_data(call, false);
482 		if (ret < 0)
483 			return ret;
484 		call->unmarshal++;
485 		fallthrough;
486 	default:
487 		return 0;
488 	}
489 }
490 
491 /*
492  * Process a call for which we've received the request.
493  */
494 static int rxperf_process_call(struct rxperf_call *call)
495 {
496 	struct msghdr msg = {};
497 	struct bio_vec bv;
498 	struct kvec iov[1];
499 	ssize_t n;
500 	size_t reply_len = call->reply_len, len;
501 
502 	rxrpc_kernel_set_tx_length(rxperf_socket, call->rxcall,
503 				   reply_len + sizeof(rxperf_magic_cookie));
504 
505 	while (reply_len > 0) {
506 		len = min_t(size_t, reply_len, PAGE_SIZE);
507 		bvec_set_page(&bv, ZERO_PAGE(0), len, 0);
508 		iov_iter_bvec(&msg.msg_iter, WRITE, &bv, 1, len);
509 		msg.msg_flags = MSG_MORE;
510 		n = rxrpc_kernel_send_data(rxperf_socket, call->rxcall, &msg,
511 					   len, rxperf_notify_end_reply_tx);
512 		if (n < 0)
513 			return n;
514 		if (n == 0)
515 			return -EIO;
516 		reply_len -= n;
517 	}
518 
519 	len = sizeof(rxperf_magic_cookie);
520 	iov[0].iov_base	= (void *)rxperf_magic_cookie;
521 	iov[0].iov_len	= len;
522 	iov_iter_kvec(&msg.msg_iter, WRITE, iov, 1, len);
523 	msg.msg_flags = 0;
524 	n = rxrpc_kernel_send_data(rxperf_socket, call->rxcall, &msg, len,
525 				   rxperf_notify_end_reply_tx);
526 	if (n >= 0)
527 		return 0; /* Success */
528 
529 	if (n == -ENOMEM)
530 		rxrpc_kernel_abort_call(rxperf_socket, call->rxcall,
531 					RXGEN_SS_MARSHAL, -ENOMEM,
532 					rxperf_abort_oom);
533 	return n;
534 }
535 
536 /*
537  * Add a key to the security keyring.
538  */
539 static int rxperf_add_key(struct key *keyring)
540 {
541 	key_ref_t kref;
542 	int ret;
543 
544 	kref = key_create_or_update(make_key_ref(keyring, true),
545 				    "rxrpc_s",
546 				    __stringify(RX_PERF_SERVICE) ":2",
547 				    secret,
548 				    sizeof(secret),
549 				    KEY_POS_VIEW | KEY_POS_READ | KEY_POS_SEARCH
550 				    | KEY_USR_VIEW,
551 				    KEY_ALLOC_NOT_IN_QUOTA);
552 
553 	if (IS_ERR(kref)) {
554 		pr_err("Can't allocate rxperf server key: %ld\n", PTR_ERR(kref));
555 		return PTR_ERR(kref);
556 	}
557 
558 	ret = key_link(keyring, key_ref_to_ptr(kref));
559 	if (ret < 0)
560 		pr_err("Can't link rxperf server key: %d\n", ret);
561 	key_ref_put(kref);
562 	return ret;
563 }
564 
565 /*
566  * Initialise the rxperf server.
567  */
568 static int __init rxperf_init(void)
569 {
570 	struct key *keyring;
571 	int ret = -ENOMEM;
572 
573 	pr_info("Server registering\n");
574 
575 	rxperf_workqueue = alloc_workqueue("rxperf", 0, 0);
576 	if (!rxperf_workqueue)
577 		goto error_workqueue;
578 
579 	keyring = keyring_alloc("rxperf_server",
580 				GLOBAL_ROOT_UID, GLOBAL_ROOT_GID, current_cred(),
581 				KEY_POS_VIEW | KEY_POS_READ | KEY_POS_SEARCH |
582 				KEY_POS_WRITE |
583 				KEY_USR_VIEW | KEY_USR_READ | KEY_USR_SEARCH |
584 				KEY_USR_WRITE |
585 				KEY_OTH_VIEW | KEY_OTH_READ | KEY_OTH_SEARCH,
586 				KEY_ALLOC_NOT_IN_QUOTA,
587 				NULL, NULL);
588 	if (IS_ERR(keyring)) {
589 		pr_err("Can't allocate rxperf server keyring: %ld\n",
590 		       PTR_ERR(keyring));
591 		goto error_keyring;
592 	}
593 	rxperf_sec_keyring = keyring;
594 	ret = rxperf_add_key(keyring);
595 	if (ret < 0)
596 		goto error_key;
597 
598 	ret = rxperf_open_socket();
599 	if (ret < 0)
600 		goto error_socket;
601 	return 0;
602 
603 error_socket:
604 error_key:
605 	key_put(rxperf_sec_keyring);
606 error_keyring:
607 	destroy_workqueue(rxperf_workqueue);
608 	rcu_barrier();
609 error_workqueue:
610 	pr_err("Failed to register: %d\n", ret);
611 	return ret;
612 }
613 late_initcall(rxperf_init); /* Must be called after net/ to create socket */
614 
615 static void __exit rxperf_exit(void)
616 {
617 	pr_info("Server unregistering.\n");
618 
619 	rxperf_close_socket();
620 	key_put(rxperf_sec_keyring);
621 	destroy_workqueue(rxperf_workqueue);
622 	rcu_barrier();
623 }
624 module_exit(rxperf_exit);
625 
626