sendmsg.c 19.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/* AF_RXRPC sendmsg() implementation.
 *
 * Copyright (C) 2007, 2016 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public Licence
 * as published by the Free Software Foundation; either version
 * 2 of the Licence, or (at your option) any later version.
 */

#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt

#include <linux/net.h>
#include <linux/gfp.h>
#include <linux/skbuff.h>
#include <linux/export.h>
18 19
#include <linux/sched/signal.h>

20 21 22 23
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include "ar-internal.h"

24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96
/*
 * Wait for space to appear in the Tx queue or a signal to occur.
 */
static int rxrpc_wait_for_tx_window_intr(struct rxrpc_sock *rx,
					 struct rxrpc_call *call,
					 long *timeo)
{
	for (;;) {
		set_current_state(TASK_INTERRUPTIBLE);
		if (call->tx_top - call->tx_hard_ack <
		    min_t(unsigned int, call->tx_winsize,
			  call->cong_cwnd + call->cong_extra))
			return 0;

		if (call->state >= RXRPC_CALL_COMPLETE)
			return call->error;

		if (signal_pending(current))
			return sock_intr_errno(*timeo);

		trace_rxrpc_transmit(call, rxrpc_transmit_wait);
		mutex_unlock(&call->user_mutex);
		*timeo = schedule_timeout(*timeo);
		if (mutex_lock_interruptible(&call->user_mutex) < 0)
			return sock_intr_errno(*timeo);
	}
}

/*
 * Wait for space to appear in the Tx queue uninterruptibly, but with
 * a timeout of 2*RTT if no progress was made and a signal occurred.
 */
static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
					    struct rxrpc_call *call)
{
	rxrpc_seq_t tx_start, tx_win;
	signed long rtt2, timeout;
	u64 rtt;

	rtt = READ_ONCE(call->peer->rtt);
	rtt2 = nsecs_to_jiffies64(rtt) * 2;
	if (rtt2 < 1)
		rtt2 = 1;

	timeout = rtt2;
	tx_start = READ_ONCE(call->tx_hard_ack);

	for (;;) {
		set_current_state(TASK_UNINTERRUPTIBLE);

		tx_win = READ_ONCE(call->tx_hard_ack);
		if (call->tx_top - tx_win <
		    min_t(unsigned int, call->tx_winsize,
			  call->cong_cwnd + call->cong_extra))
			return 0;

		if (call->state >= RXRPC_CALL_COMPLETE)
			return call->error;

		if (timeout == 0 &&
		    tx_win == tx_start && signal_pending(current))
			return -EINTR;

		if (tx_win != tx_start) {
			timeout = rtt2;
			tx_start = tx_win;
		}

		trace_rxrpc_transmit(call, rxrpc_transmit_wait);
		timeout = schedule_timeout(timeout);
	}
}

97
/*
98 99
 * wait for space to appear in the transmit/ACK window
 * - caller holds the socket locked
100
 */
101 102
static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx,
				    struct rxrpc_call *call,
103 104
				    long *timeo,
				    bool waitall)
105
{
106 107
	DECLARE_WAITQUEUE(myself, current);
	int ret;
108

109 110
	_enter(",{%u,%u,%u}",
	       call->tx_hard_ack, call->tx_top, call->tx_winsize);
111

112
	add_wait_queue(&call->waitq, &myself);
113

114 115 116 117
	if (waitall)
		ret = rxrpc_wait_for_tx_window_nonintr(rx, call);
	else
		ret = rxrpc_wait_for_tx_window_intr(rx, call, timeo);
118

119 120 121 122
	remove_wait_queue(&call->waitq, &myself);
	set_current_state(TASK_RUNNING);
	_leave(" = %d", ret);
	return ret;
123 124 125
}

/*
126
 * Schedule an instant Tx resend.
127
 */
128
static inline void rxrpc_instant_resend(struct rxrpc_call *call, int ix)
129
{
130 131 132 133 134
	spin_lock_bh(&call->lock);

	if (call->state < RXRPC_CALL_COMPLETE) {
		call->rxtx_annotations[ix] = RXRPC_TX_ANNO_RETRANS;
		if (!test_and_set_bit(RXRPC_CALL_EV_RESEND, &call->events))
135
			rxrpc_queue_call(call);
136
	}
137 138

	spin_unlock_bh(&call->lock);
139 140
}

141 142 143 144 145 146 147 148 149 150 151
/*
 * Notify the owner of the call that the transmit phase is ended and the last
 * packet has been queued.
 */
static void rxrpc_notify_end_tx(struct rxrpc_sock *rx, struct rxrpc_call *call,
				rxrpc_notify_end_tx_t notify_end_tx)
{
	if (notify_end_tx)
		notify_end_tx(&rx->sk, call, call->user_call_ID);
}

152
/*
153 154
 * Queue a DATA packet for transmission, set the resend timeout and send the
 * packet immediately
155
 */
156 157 158
static void rxrpc_queue_packet(struct rxrpc_sock *rx, struct rxrpc_call *call,
			       struct sk_buff *skb, bool last,
			       rxrpc_notify_end_tx_t notify_end_tx)
159
{
160
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
161 162
	rxrpc_seq_t seq = sp->hdr.seq;
	int ret, ix;
163
	u8 annotation = RXRPC_TX_ANNO_UNACK;
164 165

	_net("queue skb %p [%d]", skb, seq);
166

167
	ASSERTCMP(seq, ==, call->tx_top + 1);
168

169
	if (last) {
170
		annotation |= RXRPC_TX_ANNO_LAST;
171 172
		set_bit(RXRPC_CALL_TX_LASTQ, &call->flags);
	}
173

174 175 176 177 178
	/* We have to set the timestamp before queueing as the retransmit
	 * algorithm can see the packet as soon as we queue it.
	 */
	skb->tstamp = ktime_get_real();

179
	ix = seq & RXRPC_RXTX_BUFF_MASK;
D
David Howells 已提交
180
	rxrpc_get_skb(skb, rxrpc_skb_tx_got);
181
	call->rxtx_annotations[ix] = annotation;
182
	smp_wmb();
183 184
	call->rxtx_buffer[ix] = skb;
	call->tx_top = seq;
185
	if (last)
186
		trace_rxrpc_transmit(call, rxrpc_transmit_queue_last);
187
	else
188
		trace_rxrpc_transmit(call, rxrpc_transmit_queue);
189

190 191 192 193 194 195
	if (last || call->state == RXRPC_CALL_SERVER_ACK_REQUEST) {
		_debug("________awaiting reply/ACK__________");
		write_lock_bh(&call->state_lock);
		switch (call->state) {
		case RXRPC_CALL_CLIENT_SEND_REQUEST:
			call->state = RXRPC_CALL_CLIENT_AWAIT_REPLY;
196
			rxrpc_notify_end_tx(rx, call, notify_end_tx);
197 198 199
			break;
		case RXRPC_CALL_SERVER_ACK_REQUEST:
			call->state = RXRPC_CALL_SERVER_SEND_REPLY;
200 201 202 203 204
			call->ack_at = call->expire_at;
			if (call->ackr_reason == RXRPC_ACK_DELAY)
				call->ackr_reason = 0;
			__rxrpc_set_timer(call, rxrpc_timer_init_for_send_reply,
					  ktime_get_real());
205 206
			if (!last)
				break;
207
			/* Fall through */
208 209
		case RXRPC_CALL_SERVER_SEND_REPLY:
			call->state = RXRPC_CALL_SERVER_AWAIT_ACK;
210
			rxrpc_notify_end_tx(rx, call, notify_end_tx);
211 212 213 214 215 216
			break;
		default:
			break;
		}
		write_unlock_bh(&call->state_lock);
	}
217

218 219
	if (seq == 1 && rxrpc_is_client_call(call))
		rxrpc_expose_client_call(call);
220

221
	ret = rxrpc_send_data_packet(call, skb, false);
222 223
	if (ret < 0) {
		_debug("need instant resend %d", ret);
224
		rxrpc_instant_resend(call, ix);
225
	} else {
226
		ktime_t now = ktime_get_real(), resend_at;
227

228
		resend_at = ktime_add_ms(now, rxrpc_resend_timeout);
229

230
		if (ktime_before(resend_at, call->resend_at)) {
231
			call->resend_at = resend_at;
232
			rxrpc_set_timer(call, rxrpc_timer_set_for_send, now);
233
		}
234 235
	}

D
David Howells 已提交
236
	rxrpc_free_skb(skb, rxrpc_skb_tx_freed);
237
	_leave("");
238 239
}

240 241 242
/*
 * send data through a socket
 * - must be called in process context
243
 * - The caller holds the call user access mutex, but not the socket lock.
244
 */
245 246
static int rxrpc_send_data(struct rxrpc_sock *rx,
			   struct rxrpc_call *call,
247 248
			   struct msghdr *msg, size_t len,
			   rxrpc_notify_end_tx_t notify_end_tx)
249
{
250 251 252 253 254 255
	struct rxrpc_skb_priv *sp;
	struct sk_buff *skb;
	struct sock *sk = &rx->sk;
	long timeo;
	bool more;
	int ret, copied;
256

257
	timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);
258

259 260
	/* this should be in poll */
	sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
261

262 263
	if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
		return -EPIPE;
264

265
	more = msg->msg_flags & MSG_MORE;
266

267 268 269 270 271 272 273
	if (call->tx_total_len != -1) {
		if (len > call->tx_total_len)
			return -EMSGSIZE;
		if (!more && len != call->tx_total_len)
			return -EMSGSIZE;
	}

274 275
	skb = call->tx_pending;
	call->tx_pending = NULL;
D
David Howells 已提交
276
	rxrpc_see_skb(skb, rxrpc_skb_tx_seen);
277

278 279
	copied = 0;
	do {
280 281
		/* Check to see if there's a ping ACK to reply to. */
		if (call->ackr_reason == RXRPC_ACK_PING_RESPONSE)
282
			rxrpc_send_ack_packet(call, false);
283

284 285
		if (!skb) {
			size_t size, chunk, max, space;
286

287
			_debug("alloc");
288

289
			if (call->tx_top - call->tx_hard_ack >=
D
David Howells 已提交
290 291
			    min_t(unsigned int, call->tx_winsize,
				  call->cong_cwnd + call->cong_extra)) {
292 293 294 295
				ret = -EAGAIN;
				if (msg->msg_flags & MSG_DONTWAIT)
					goto maybe_error;
				ret = rxrpc_wait_for_tx_window(rx, call,
296 297
							       &timeo,
							       msg->msg_flags & MSG_WAITALL);
298 299 300
				if (ret < 0)
					goto maybe_error;
			}
301

302
			max = RXRPC_JUMBO_DATALEN;
303 304
			max -= call->conn->security_size;
			max &= ~(call->conn->size_align - 1UL);
305

306 307 308
			chunk = max;
			if (chunk > msg_data_left(msg) && !more)
				chunk = msg_data_left(msg);
309

310 311
			space = chunk + call->conn->size_align;
			space &= ~(call->conn->size_align - 1UL);
312

313
			size = space + call->conn->security_size;
314

315
			_debug("SIZE: %zu/%zu/%zu", chunk, space, size);
316

317 318 319 320 321
			/* create a buffer that we can retain until it's ACK'd */
			skb = sock_alloc_send_skb(
				sk, size, msg->msg_flags & MSG_DONTWAIT, &ret);
			if (!skb)
				goto maybe_error;
322

D
David Howells 已提交
323
			rxrpc_new_skb(skb, rxrpc_skb_tx_new);
324

325
			_debug("ALLOC SEND %p", skb);
326

327
			ASSERTCMP(skb->mark, ==, 0);
328

329 330 331
			_debug("HS: %u", call->conn->security_size);
			skb_reserve(skb, call->conn->security_size);
			skb->len += call->conn->security_size;
332

333 334 335 336
			sp = rxrpc_skb(skb);
			sp->remain = chunk;
			if (sp->remain > skb_tailroom(skb))
				sp->remain = skb_tailroom(skb);
337

338 339 340 341 342
			_net("skb: hr %d, tr %d, hl %d, rm %d",
			       skb_headroom(skb),
			       skb_tailroom(skb),
			       skb_headlen(skb),
			       sp->remain);
343

344 345
			skb->ip_summed = CHECKSUM_UNNECESSARY;
		}
346

347 348
		_debug("append");
		sp = rxrpc_skb(skb);
349

350 351 352 353 354 355 356 357
		/* append next segment of data to the current buffer */
		if (msg_data_left(msg) > 0) {
			int copy = skb_tailroom(skb);
			ASSERTCMP(copy, >, 0);
			if (copy > msg_data_left(msg))
				copy = msg_data_left(msg);
			if (copy > sp->remain)
				copy = sp->remain;
358

359 360 361 362 363 364 365 366
			_debug("add");
			ret = skb_add_data(skb, &msg->msg_iter, copy);
			_debug("added");
			if (ret < 0)
				goto efault;
			sp->remain -= copy;
			skb->mark += copy;
			copied += copy;
367 368
			if (call->tx_total_len != -1)
				call->tx_total_len -= copy;
369 370
		}

371 372 373 374 375 376
		/* add the packet to the send queue if it's now full */
		if (sp->remain <= 0 ||
		    (msg_data_left(msg) == 0 && !more)) {
			struct rxrpc_connection *conn = call->conn;
			uint32_t seq;
			size_t pad;
377

378 379 380 381 382 383 384
			/* pad out if we're using security */
			if (conn->security_ix) {
				pad = conn->security_size + skb->mark;
				pad = conn->size_align - pad;
				pad &= conn->size_align - 1;
				_debug("pad %zu", pad);
				if (pad)
385
					skb_put_zero(skb, pad);
386
			}
387

388
			seq = call->tx_top + 1;
389

390 391
			sp->hdr.seq	= seq;
			sp->hdr._rsvd	= 0;
392
			sp->hdr.flags	= conn->out_clientflag;
393

394 395
			if (msg_data_left(msg) == 0 && !more)
				sp->hdr.flags |= RXRPC_LAST_PACKET;
396 397
			else if (call->tx_top - call->tx_hard_ack <
				 call->tx_winsize)
398
				sp->hdr.flags |= RXRPC_MORE_PACKETS;
399

400
			ret = conn->security->secure_packet(
401
				call, skb, skb->mark, skb->head);
402 403
			if (ret < 0)
				goto out;
404

405 406 407
			rxrpc_queue_packet(rx, call, skb,
					   !msg_data_left(msg) && !more,
					   notify_end_tx);
408 409
			skb = NULL;
		}
410 411 412 413 414 415 416 417 418 419

		/* Check for the far side aborting the call or a network error
		 * occurring.  If this happens, save any packet that was under
		 * construction so that in the case of a network error, the
		 * call can be retried or redirected.
		 */
		if (call->state == RXRPC_CALL_COMPLETE) {
			ret = call->error;
			goto out;
		}
420
	} while (msg_data_left(msg) > 0);
421

422 423 424 425 426 427
success:
	ret = copied;
out:
	call->tx_pending = skb;
	_leave(" = %d", ret);
	return ret;
428

429 430 431 432
maybe_error:
	if (copied)
		goto success;
	goto out;
433

434 435 436
efault:
	ret = -EFAULT;
	goto out;
437 438 439
}

/*
440
 * extract control messages from the sendmsg() control buffer
441
 */
442
static int rxrpc_sendmsg_cmsg(struct msghdr *msg, struct rxrpc_send_params *p)
443
{
444 445 446
	struct cmsghdr *cmsg;
	bool got_user_ID = false;
	int len;
447

448 449
	if (msg->msg_controllen == 0)
		return -EINVAL;
450

451 452 453
	for_each_cmsghdr(cmsg, msg) {
		if (!CMSG_OK(msg, cmsg))
			return -EINVAL;
454

455
		len = cmsg->cmsg_len - sizeof(struct cmsghdr);
456 457
		_debug("CMSG %d, %d, %d",
		       cmsg->cmsg_level, cmsg->cmsg_type, len);
458

459 460
		if (cmsg->cmsg_level != SOL_RXRPC)
			continue;
461

462 463 464 465 466
		switch (cmsg->cmsg_type) {
		case RXRPC_USER_CALL_ID:
			if (msg->msg_flags & MSG_CMSG_COMPAT) {
				if (len != sizeof(u32))
					return -EINVAL;
467
				p->call.user_call_ID = *(u32 *)CMSG_DATA(cmsg);
468 469 470
			} else {
				if (len != sizeof(unsigned long))
					return -EINVAL;
471
				p->call.user_call_ID = *(unsigned long *)
472 473 474 475
					CMSG_DATA(cmsg);
			}
			got_user_ID = true;
			break;
476

477
		case RXRPC_ABORT:
478
			if (p->command != RXRPC_CMD_SEND_DATA)
479
				return -EINVAL;
480 481
			p->command = RXRPC_CMD_SEND_ABORT;
			if (len != sizeof(p->abort_code))
482
				return -EINVAL;
483 484
			p->abort_code = *(unsigned int *)CMSG_DATA(cmsg);
			if (p->abort_code == 0)
485 486
				return -EINVAL;
			break;
487

488
		case RXRPC_ACCEPT:
489
			if (p->command != RXRPC_CMD_SEND_DATA)
490
				return -EINVAL;
491
			p->command = RXRPC_CMD_ACCEPT;
492 493 494
			if (len != 0)
				return -EINVAL;
			break;
495

496
		case RXRPC_EXCLUSIVE_CALL:
497
			p->exclusive = true;
498 499 500
			if (len != 0)
				return -EINVAL;
			break;
501 502

		case RXRPC_UPGRADE_SERVICE:
503
			p->upgrade = true;
504 505 506 507
			if (len != 0)
				return -EINVAL;
			break;

508
		case RXRPC_TX_LENGTH:
509
			if (p->call.tx_total_len != -1 || len != sizeof(__s64))
510
				return -EINVAL;
511 512
			p->call.tx_total_len = *(__s64 *)CMSG_DATA(cmsg);
			if (p->call.tx_total_len < 0)
513 514 515
				return -EINVAL;
			break;

516 517 518 519
		default:
			return -EINVAL;
		}
	}
520

521 522
	if (!got_user_ID)
		return -EINVAL;
523
	if (p->call.tx_total_len != -1 && p->command != RXRPC_CMD_SEND_DATA)
524
		return -EINVAL;
525 526 527
	_leave(" = 0");
	return 0;
}
528

529 530
/*
 * Create a new client call for sendmsg().
531 532
 * - Called with the socket lock held, which it must release.
 * - If it returns a call, the call's lock will need releasing by the caller.
533 534 535
 */
static struct rxrpc_call *
rxrpc_new_client_call_for_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg,
536
				  struct rxrpc_send_params *p)
537
	__releases(&rx->sk.sk_lock.slock)
538 539 540 541
{
	struct rxrpc_conn_parameters cp;
	struct rxrpc_call *call;
	struct key *key;
542

543
	DECLARE_SOCKADDR(struct sockaddr_rxrpc *, srx, msg->msg_name);
544

545
	_enter("");
546

547 548
	if (!msg->msg_name) {
		release_sock(&rx->sk);
549
		return ERR_PTR(-EDESTADDRREQ);
550
	}
551

552 553 554
	key = rx->key;
	if (key && !rx->key->payload.data[0])
		key = NULL;
555

556 557 558 559
	memset(&cp, 0, sizeof(cp));
	cp.local		= rx->local;
	cp.key			= rx->key;
	cp.security_level	= rx->min_sec_level;
560 561
	cp.exclusive		= rx->exclusive | p->exclusive;
	cp.upgrade		= p->upgrade;
562
	cp.service_id		= srx->srx_service;
563
	call = rxrpc_new_client_call(rx, &cp, srx, &p->call, GFP_KERNEL);
564
	/* The socket is now unlocked */
565

566 567 568
	_leave(" = %p\n", call);
	return call;
}
569

570 571 572 573 574 575
/*
 * send a message forming part of a client call through an RxRPC socket
 * - caller holds the socket locked
 * - the socket may be either a client socket or a server socket
 */
int rxrpc_do_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg, size_t len)
576
	__releases(&rx->sk.sk_lock.slock)
577
{
578
	enum rxrpc_call_state state;
579 580
	struct rxrpc_call *call;
	int ret;
581

582
	struct rxrpc_send_params p = {
583 584 585 586 587 588
		.call.tx_total_len	= -1,
		.call.user_call_ID	= 0,
		.abort_code		= 0,
		.command		= RXRPC_CMD_SEND_DATA,
		.exclusive		= false,
		.upgrade		= false,
589 590
	};

591
	_enter("");
592

593
	ret = rxrpc_sendmsg_cmsg(msg, &p);
594
	if (ret < 0)
595
		goto error_release_sock;
596

597
	if (p.command == RXRPC_CMD_ACCEPT) {
598
		ret = -EINVAL;
599
		if (rx->sk.sk_state != RXRPC_SERVER_LISTENING)
600
			goto error_release_sock;
601
		call = rxrpc_accept_call(rx, p.call.user_call_ID, NULL);
602
		/* The socket is now unlocked. */
603 604
		if (IS_ERR(call))
			return PTR_ERR(call);
605 606
		ret = 0;
		goto out_put_unlock;
607
	}
608

609
	call = rxrpc_find_call_by_user_ID(rx, p.call.user_call_ID);
610
	if (!call) {
611
		ret = -EBADSLT;
612
		if (p.command != RXRPC_CMD_SEND_DATA)
613
			goto error_release_sock;
614
		call = rxrpc_new_client_call_for_sendmsg(rx, msg, &p);
615
		/* The socket is now unlocked... */
616 617
		if (IS_ERR(call))
			return PTR_ERR(call);
618 619
		/* ... and we have the call lock. */
	} else {
620 621 622 623 624 625 626
		switch (READ_ONCE(call->state)) {
		case RXRPC_CALL_UNINITIALISED:
		case RXRPC_CALL_CLIENT_AWAIT_CONN:
		case RXRPC_CALL_SERVER_PREALLOC:
		case RXRPC_CALL_SERVER_SECURING:
		case RXRPC_CALL_SERVER_ACCEPTING:
			ret = -EBUSY;
627
			goto error_release_sock;
628 629 630
		default:
			break;
		}
631

632 633 634 635 636 637
		ret = mutex_lock_interruptible(&call->user_mutex);
		release_sock(&rx->sk);
		if (ret < 0) {
			ret = -ERESTARTSYS;
			goto error_put;
		}
638

639
		if (p.call.tx_total_len != -1) {
640 641 642 643 644
			ret = -EINVAL;
			if (call->tx_total_len != -1 ||
			    call->tx_pending ||
			    call->tx_top != 0)
				goto error_put;
645
			call->tx_total_len = p.call.tx_total_len;
646
		}
647
	}
648

649
	state = READ_ONCE(call->state);
650
	_debug("CALL %d USR %lx ST %d on CONN %p",
651
	       call->debug_id, call->user_call_ID, state, call->conn);
652

653
	if (state >= RXRPC_CALL_COMPLETE) {
654 655
		/* it's too late for this call */
		ret = -ESHUTDOWN;
656
	} else if (p.command == RXRPC_CMD_SEND_ABORT) {
657
		ret = 0;
658
		if (rxrpc_abort_call("CMD", call, 0, p.abort_code, -ECONNABORTED))
659
			ret = rxrpc_send_abort_packet(call);
660
	} else if (p.command != RXRPC_CMD_SEND_DATA) {
661 662
		ret = -EINVAL;
	} else if (rxrpc_is_client_call(call) &&
663
		   state != RXRPC_CALL_CLIENT_SEND_REQUEST) {
664 665 666
		/* request phase complete for this client call */
		ret = -EPROTO;
	} else if (rxrpc_is_service_call(call) &&
667 668
		   state != RXRPC_CALL_SERVER_ACK_REQUEST &&
		   state != RXRPC_CALL_SERVER_SEND_REPLY) {
669 670 671
		/* Reply phase not begun or not complete for service call. */
		ret = -EPROTO;
	} else {
672
		ret = rxrpc_send_data(rx, call, msg, len, NULL);
673
	}
674

675
out_put_unlock:
676 677
	mutex_unlock(&call->user_mutex);
error_put:
678
	rxrpc_put_call(call, rxrpc_call_put);
679 680
	_leave(" = %d", ret);
	return ret;
681 682 683 684

error_release_sock:
	release_sock(&rx->sk);
	return ret;
685
}
686

687 688 689 690 691 692
/**
 * rxrpc_kernel_send_data - Allow a kernel service to send data on a call
 * @sock: The socket the call is on
 * @call: The call to send data through
 * @msg: The data to send
 * @len: The amount of data to send
693
 * @notify_end_tx: Notification that the last packet is queued.
694 695 696 697 698 699 700
 *
 * Allow a kernel service to send data on a call.  The call must be in an state
 * appropriate to sending data.  No control data should be supplied in @msg,
 * nor should an address be supplied.  MSG_MORE should be flagged if there's
 * more data to come, otherwise this data will end the transmission phase.
 */
int rxrpc_kernel_send_data(struct socket *sock, struct rxrpc_call *call,
701 702
			   struct msghdr *msg, size_t len,
			   rxrpc_notify_end_tx_t notify_end_tx)
703 704
{
	int ret;
705

706
	_enter("{%d,%s},", call->debug_id, rxrpc_call_states[call->state]);
707

708 709
	ASSERTCMP(msg->msg_name, ==, NULL);
	ASSERTCMP(msg->msg_control, ==, NULL);
710

711
	mutex_lock(&call->user_mutex);
712

713 714
	_debug("CALL %d USR %lx ST %d on CONN %p",
	       call->debug_id, call->user_call_ID, call->state, call->conn);
715

716 717 718 719
	switch (READ_ONCE(call->state)) {
	case RXRPC_CALL_CLIENT_SEND_REQUEST:
	case RXRPC_CALL_SERVER_ACK_REQUEST:
	case RXRPC_CALL_SERVER_SEND_REPLY:
720 721
		ret = rxrpc_send_data(rxrpc_sk(sock->sk), call, msg, len,
				      notify_end_tx);
722 723
		break;
	case RXRPC_CALL_COMPLETE:
724
		read_lock_bh(&call->state_lock);
725
		ret = call->error;
726
		read_unlock_bh(&call->state_lock);
727 728
		break;
	default:
729 730
		/* Request phase complete for this client call */
		trace_rxrpc_rx_eproto(call, 0, tracepoint_string("late_send"));
731 732
		ret = -EPROTO;
		break;
733 734
	}

735
	mutex_unlock(&call->user_mutex);
736 737
	_leave(" = %d", ret);
	return ret;
738 739
}
EXPORT_SYMBOL(rxrpc_kernel_send_data);
740

741 742 743 744 745
/**
 * rxrpc_kernel_abort_call - Allow a kernel service to abort a call
 * @sock: The socket the call is on
 * @call: The call to be aborted
 * @abort_code: The abort code to stick into the ABORT packet
746 747
 * @error: Local error value
 * @why: 3-char string indicating why.
748
 *
749 750
 * Allow a kernel service to abort a call, if it's still in an abortable state
 * and return true if the call was aborted, false if it was already complete.
751
 */
752
bool rxrpc_kernel_abort_call(struct socket *sock, struct rxrpc_call *call,
753
			     u32 abort_code, int error, const char *why)
754
{
755 756
	bool aborted;

757
	_enter("{%d},%d,%d,%s", call->debug_id, abort_code, error, why);
758

759
	mutex_lock(&call->user_mutex);
760

761 762
	aborted = rxrpc_abort_call(why, call, 0, abort_code, error);
	if (aborted)
763
		rxrpc_send_abort_packet(call);
764

765
	mutex_unlock(&call->user_mutex);
766
	return aborted;
767
}
768
EXPORT_SYMBOL(rxrpc_kernel_abort_call);
769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788

/**
 * rxrpc_kernel_set_tx_length - Set the total Tx length on a call
 * @sock: The socket the call is on
 * @call: The call to be informed
 * @tx_total_len: The amount of data to be transmitted for this call
 *
 * Allow a kernel service to set the total transmit length on a call.  This
 * allows buffer-to-packet encrypt-and-copy to be performed.
 *
 * This function is primarily for use for setting the reply length since the
 * request length can be set when beginning the call.
 */
void rxrpc_kernel_set_tx_length(struct socket *sock, struct rxrpc_call *call,
				s64 tx_total_len)
{
	WARN_ON(call->tx_total_len != -1);
	call->tx_total_len = tx_total_len;
}
EXPORT_SYMBOL(rxrpc_kernel_set_tx_length);