Skip to content

L
linux
Commit df423a4a authored by David Howells's avatar David Howells
Browse files
Options

rxrpc: Rearrange net/rxrpc/sendmsg.c 

Rearrange net/rxrpc/sendmsg.c to be in a more logical order.  This makes it
easier to follow and eliminates forward declarations.
Signed-off-by: default avatarDavid Howells <dhowells@redhat.com>
parent 0b58b8a1
Show whitespace changes
Inline Side-by-side
Showing with 501 additions and 505 deletions
net/rxrpc/sendmsg.c
View file @ df423a4a
...@@ -20,626 +20,622 @@ ...@@ -20,626 +20,622 @@
#include <net/af_rxrpc.h> #include <net/af_rxrpc.h>
#include "ar-internal.h" #include "ar-internal.h"
static int rxrpc_send_data(struct rxrpc_sock *rx,
struct rxrpc_call *call,
struct msghdr *msg, size_t len);
/* /*
* extract control messages from the sendmsg() control buffer * wait for space to appear in the transmit/ACK window
* - caller holds the socket locked
*/ */
static int rxrpc_sendmsg_cmsg(struct msghdr *msg, static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx,
unsigned long *user_call_ID, struct rxrpc_call *call,
enum rxrpc_command *command, long *timeo)
u32 *abort_code,
bool *_exclusive)
{ {
struct cmsghdr *cmsg; DECLARE_WAITQUEUE(myself, current);
bool got_user_ID = false; int ret;
int len;
*command = RXRPC_CMD_SEND_DATA;
if (msg->msg_controllen == 0)
return -EINVAL;
for_each_cmsghdr(cmsg, msg) {
if (!CMSG_OK(msg, cmsg))
return -EINVAL;
len = cmsg->cmsg_len - CMSG_ALIGN(sizeof(struct cmsghdr));
_debug("CMSG %d, %d, %d",
cmsg->cmsg_level, cmsg->cmsg_type, len);
if (cmsg->cmsg_level != SOL_RXRPC)
continue;
switch (cmsg->cmsg_type) { _enter(",{%d},%ld",
case RXRPC_USER_CALL_ID: CIRC_SPACE(call->acks_head, ACCESS_ONCE(call->acks_tail),
if (msg->msg_flags & MSG_CMSG_COMPAT) { call->acks_winsz),
if (len != sizeof(u32)) *timeo);
return -EINVAL;
*user_call_ID = *(u32 *) CMSG_DATA(cmsg);
} else {
if (len != sizeof(unsigned long))
return -EINVAL;
*user_call_ID = *(unsigned long *)
CMSG_DATA(cmsg);
}
_debug("User Call ID %lx", *user_call_ID);
got_user_ID = true;
break;
case RXRPC_ABORT: add_wait_queue(&call->waitq, &myself);
if (*command != RXRPC_CMD_SEND_DATA)
return -EINVAL;
*command = RXRPC_CMD_SEND_ABORT;
if (len != sizeof(*abort_code))
return -EINVAL;
*abort_code = *(unsigned int *) CMSG_DATA(cmsg);
_debug("Abort %x", *abort_code);
if (*abort_code == 0)
return -EINVAL;
break;
case RXRPC_ACCEPT: for (;;) {
if (*command != RXRPC_CMD_SEND_DATA) set_current_state(TASK_INTERRUPTIBLE);
return -EINVAL; ret = 0;
*command = RXRPC_CMD_ACCEPT; if (CIRC_SPACE(call->acks_head, ACCESS_ONCE(call->acks_tail),
if (len != 0) call->acks_winsz) > 0)
return -EINVAL;
break; break;
if (signal_pending(current)) {
case RXRPC_EXCLUSIVE_CALL: ret = sock_intr_errno(*timeo);
*_exclusive = true;
if (len != 0)
return -EINVAL;
break; break;
default:
return -EINVAL;
} }
release_sock(&rx->sk);
*timeo = schedule_timeout(*timeo);
lock_sock(&rx->sk);
} }
if (!got_user_ID) remove_wait_queue(&call->waitq, &myself);
return -EINVAL; set_current_state(TASK_RUNNING);
_leave(" = 0"); _leave(" = %d", ret);
return 0; return ret;
} }
/* /*
* abort a call, sending an ABORT packet to the peer * attempt to schedule an instant Tx resend
*/ */
static void rxrpc_send_abort(struct rxrpc_call *call, u32 abort_code) static inline void rxrpc_instant_resend(struct rxrpc_call *call)
{ {
if (call->state >= RXRPC_CALL_COMPLETE) read_lock_bh(&call->state_lock);
return; if (try_to_del_timer_sync(&call->resend_timer) >= 0) {
write_lock_bh(&call->state_lock);
if (__rxrpc_abort_call(call, abort_code, ECONNABORTED)) {
del_timer_sync(&call->resend_timer);
del_timer_sync(&call->ack_timer);
clear_bit(RXRPC_CALL_EV_RESEND_TIMER, &call->events);
clear_bit(RXRPC_CALL_EV_ACK, &call->events);
clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags); clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags);
if (call->state < RXRPC_CALL_COMPLETE &&
!test_and_set_bit(RXRPC_CALL_EV_RESEND_TIMER, &call->events))
rxrpc_queue_call(call); rxrpc_queue_call(call);
} }
read_unlock_bh(&call->state_lock);
write_unlock_bh(&call->state_lock);
} }
/* /*
* Create a new client call for sendmsg(). * queue a packet for transmission, set the resend timer and attempt
* to send the packet immediately
*/ */
static struct rxrpc_call * static void rxrpc_queue_packet(struct rxrpc_call *call, struct sk_buff *skb,
rxrpc_new_client_call_for_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg, bool last)
unsigned long user_call_ID, bool exclusive)
{ {
struct rxrpc_conn_parameters cp; struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
struct rxrpc_call *call; int ret;
struct key *key;
DECLARE_SOCKADDR(struct sockaddr_rxrpc *, srx, msg->msg_name); _net("queue skb %p [%d]", skb, call->acks_head);
_enter(""); ASSERT(call->acks_window != NULL);
call->acks_window[call->acks_head] = (unsigned long) skb;
smp_wmb();
call->acks_head = (call->acks_head + 1) & (call->acks_winsz - 1);
if (!msg->msg_name) if (last || call->state == RXRPC_CALL_SERVER_ACK_REQUEST) {
return ERR_PTR(-EDESTADDRREQ); _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;
break;
case RXRPC_CALL_SERVER_ACK_REQUEST:
call->state = RXRPC_CALL_SERVER_SEND_REPLY;
if (!last)
break;
case RXRPC_CALL_SERVER_SEND_REPLY:
call->state = RXRPC_CALL_SERVER_AWAIT_ACK;
break;
default:
break;
}
write_unlock_bh(&call->state_lock);
}
key = rx->key; _proto("Tx DATA %%%u { #%u }", sp->hdr.serial, sp->hdr.seq);
if (key && !rx->key->payload.data[0])
key = NULL;
memset(&cp, 0, sizeof(cp)); sp->need_resend = false;
cp.local = rx->local; sp->resend_at = jiffies + rxrpc_resend_timeout;
cp.key = rx->key; if (!test_and_set_bit(RXRPC_CALL_RUN_RTIMER, &call->flags)) {
cp.security_level = rx->min_sec_level; _debug("run timer");
cp.exclusive = rx->exclusive | exclusive; call->resend_timer.expires = sp->resend_at;
cp.service_id = srx->srx_service; add_timer(&call->resend_timer);
call = rxrpc_new_client_call(rx, &cp, srx, user_call_ID, GFP_KERNEL); }
_leave(" = %p\n", call); /* attempt to cancel the rx-ACK timer, deferring reply transmission if
return call; * we're ACK'ing the request phase of an incoming call */
ret = -EAGAIN;
if (try_to_del_timer_sync(&call->ack_timer) >= 0) {
/* the packet may be freed by rxrpc_process_call() before this
* returns */
if (rxrpc_is_client_call(call))
rxrpc_expose_client_call(call);
ret = rxrpc_send_data_packet(call->conn, skb);
_net("sent skb %p", skb);
} else {
_debug("failed to delete ACK timer");
}
if (ret < 0) {
_debug("need instant resend %d", ret);
sp->need_resend = true;
rxrpc_instant_resend(call);
}
_leave("");
} }
/* /*
* send a message forming part of a client call through an RxRPC socket * Convert a host-endian header into a network-endian header.
*/
static void rxrpc_insert_header(struct sk_buff *skb)
{
struct rxrpc_wire_header whdr;
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
whdr.epoch = htonl(sp->hdr.epoch);
whdr.cid = htonl(sp->hdr.cid);
whdr.callNumber = htonl(sp->hdr.callNumber);
whdr.seq = htonl(sp->hdr.seq);
whdr.serial = htonl(sp->hdr.serial);
whdr.type = sp->hdr.type;
whdr.flags = sp->hdr.flags;
whdr.userStatus = sp->hdr.userStatus;
whdr.securityIndex = sp->hdr.securityIndex;
whdr._rsvd = htons(sp->hdr._rsvd);
whdr.serviceId = htons(sp->hdr.serviceId);
memcpy(skb->head, &whdr, sizeof(whdr));
}
/*
* send data through a socket
* - must be called in process context
* - caller holds the socket locked * - 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) static int rxrpc_send_data(struct rxrpc_sock *rx,
struct rxrpc_call *call,
struct msghdr *msg, size_t len)
{ {
enum rxrpc_command cmd; struct rxrpc_skb_priv *sp;
struct rxrpc_call *call; struct sk_buff *skb;
unsigned long user_call_ID = 0; struct sock *sk = &rx->sk;
bool exclusive = false; long timeo;
u32 abort_code = 0; bool more;
int ret; int ret, copied;
_enter(""); timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);
ret = rxrpc_sendmsg_cmsg(msg, &user_call_ID, &cmd, &abort_code, /* this should be in poll */
&exclusive); sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
if (ret < 0)
return ret;
if (cmd == RXRPC_CMD_ACCEPT) { if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
if (rx->sk.sk_state != RXRPC_SERVER_LISTENING) return -EPIPE;
return -EINVAL;
call = rxrpc_accept_call(rx, user_call_ID, NULL);
if (IS_ERR(call))
return PTR_ERR(call);
rxrpc_put_call(call);
return 0;
}
call = rxrpc_find_call_by_user_ID(rx, user_call_ID); more = msg->msg_flags & MSG_MORE;
if (!call) {
if (cmd != RXRPC_CMD_SEND_DATA)
return -EBADSLT;
call = rxrpc_new_client_call_for_sendmsg(rx, msg, user_call_ID,
exclusive);
if (IS_ERR(call))
return PTR_ERR(call);
}
rxrpc_see_call(call); skb = call->tx_pending;
_debug("CALL %d USR %lx ST %d on CONN %p", call->tx_pending = NULL;
call->debug_id, call->user_call_ID, call->state, call->conn); rxrpc_see_skb(skb);
if (call->state >= RXRPC_CALL_COMPLETE) { copied = 0;
/* it's too late for this call */ do {
ret = -ESHUTDOWN; if (!skb) {
} else if (cmd == RXRPC_CMD_SEND_ABORT) { size_t size, chunk, max, space;
rxrpc_send_abort(call, abort_code);
ret = 0; _debug("alloc");
} else if (cmd != RXRPC_CMD_SEND_DATA) {
ret = -EINVAL; if (CIRC_SPACE(call->acks_head,
} else if (rxrpc_is_client_call(call) && ACCESS_ONCE(call->acks_tail),
call->state != RXRPC_CALL_CLIENT_SEND_REQUEST) { call->acks_winsz) <= 0) {
/* request phase complete for this client call */ ret = -EAGAIN;
ret = -EPROTO; if (msg->msg_flags & MSG_DONTWAIT)
} else if (rxrpc_is_service_call(call) && goto maybe_error;
call->state != RXRPC_CALL_SERVER_ACK_REQUEST && ret = rxrpc_wait_for_tx_window(rx, call,
call->state != RXRPC_CALL_SERVER_SEND_REPLY) { &timeo);
/* Reply phase not begun or not complete for service call. */ if (ret < 0)
ret = -EPROTO; goto maybe_error;
} else {
ret = rxrpc_send_data(rx, call, msg, len);
} }
rxrpc_put_call(call); max = call->conn->params.peer->maxdata;
_leave(" = %d", ret); max -= call->conn->security_size;
return ret; max &= ~(call->conn->size_align - 1UL);
}
/** chunk = max;
* rxrpc_kernel_send_data - Allow a kernel service to send data on a call if (chunk > msg_data_left(msg) && !more)
* @sock: The socket the call is on chunk = msg_data_left(msg);
* @call: The call to send data through
* @msg: The data to send
* @len: The amount of data to send
*
* 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,
struct msghdr *msg, size_t len)
{
int ret;
_enter("{%d,%s},", call->debug_id, rxrpc_call_states[call->state]); space = chunk + call->conn->size_align;
space &= ~(call->conn->size_align - 1UL);
ASSERTCMP(msg->msg_name, ==, NULL); size = space + call->conn->header_size;
ASSERTCMP(msg->msg_control, ==, NULL);
lock_sock(sock->sk); _debug("SIZE: %zu/%zu/%zu", chunk, space, size);
_debug("CALL %d USR %lx ST %d on CONN %p", /* create a buffer that we can retain until it's ACK'd */
call->debug_id, call->user_call_ID, call->state, call->conn); skb = sock_alloc_send_skb(
sk, size, msg->msg_flags & MSG_DONTWAIT, &ret);
if (!skb)
goto maybe_error;
if (call->state >= RXRPC_CALL_COMPLETE) { rxrpc_new_skb(skb);
ret = -ESHUTDOWN; /* it's too late for this call */
} else if (call->state != RXRPC_CALL_CLIENT_SEND_REQUEST &&
call->state != RXRPC_CALL_SERVER_ACK_REQUEST &&
call->state != RXRPC_CALL_SERVER_SEND_REPLY) {
ret = -EPROTO; /* request phase complete for this client call */
} else {
ret = rxrpc_send_data(rxrpc_sk(sock->sk), call, msg, len);
}
release_sock(sock->sk); _debug("ALLOC SEND %p", skb);
_leave(" = %d", ret);
return ret;
}
EXPORT_SYMBOL(rxrpc_kernel_send_data);
/** ASSERTCMP(skb->mark, ==, 0);
* 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
*
* Allow a kernel service to abort a call, if it's still in an abortable state.
*/
void rxrpc_kernel_abort_call(struct socket *sock, struct rxrpc_call *call,
u32 abort_code)
{
_enter("{%d},%d", call->debug_id, abort_code);
lock_sock(sock->sk); _debug("HS: %u", call->conn->header_size);
skb_reserve(skb, call->conn->header_size);
skb->len += call->conn->header_size;
_debug("CALL %d USR %lx ST %d on CONN %p", sp = rxrpc_skb(skb);
call->debug_id, call->user_call_ID, call->state, call->conn); sp->remain = chunk;
if (sp->remain > skb_tailroom(skb))
sp->remain = skb_tailroom(skb);
rxrpc_send_abort(call, abort_code); _net("skb: hr %d, tr %d, hl %d, rm %d",
skb_headroom(skb),
skb_tailroom(skb),
skb_headlen(skb),
sp->remain);
release_sock(sock->sk); skb->ip_summed = CHECKSUM_UNNECESSARY;
_leave(""); }
}
EXPORT_SYMBOL(rxrpc_kernel_abort_call); _debug("append");
sp = rxrpc_skb(skb);
/* /* append next segment of data to the current buffer */
* wait for space to appear in the transmit/ACK window if (msg_data_left(msg) > 0) {
* - caller holds the socket locked int copy = skb_tailroom(skb);
*/ ASSERTCMP(copy, >, 0);
static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx, if (copy > msg_data_left(msg))
struct rxrpc_call *call, copy = msg_data_left(msg);
long *timeo) if (copy > sp->remain)
{ copy = sp->remain;
DECLARE_WAITQUEUE(myself, current);
int ret;
_enter(",{%d},%ld", _debug("add");
CIRC_SPACE(call->acks_head, ACCESS_ONCE(call->acks_tail), ret = skb_add_data(skb, &msg->msg_iter, copy);
call->acks_winsz), _debug("added");
*timeo); if (ret < 0)
goto efault;
sp->remain -= copy;
skb->mark += copy;
copied += copy;
}
add_wait_queue(&call->waitq, &myself); /* check for the far side aborting the call or a network error
* occurring */
if (call->state == RXRPC_CALL_COMPLETE)
goto call_terminated;
for (;;) { /* add the packet to the send queue if it's now full */
set_current_state(TASK_INTERRUPTIBLE); if (sp->remain <= 0 ||
ret = 0; (msg_data_left(msg) == 0 && !more)) {
if (CIRC_SPACE(call->acks_head, ACCESS_ONCE(call->acks_tail), struct rxrpc_connection *conn = call->conn;
call->acks_winsz) > 0) uint32_t seq;
break; size_t pad;
if (signal_pending(current)) {
ret = sock_intr_errno(*timeo); /* pad out if we're using security */
break; 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)
memset(skb_put(skb, pad), 0, pad);
} }
release_sock(&rx->sk); seq = atomic_inc_return(&call->sequence);
*timeo = schedule_timeout(*timeo);
lock_sock(&rx->sk); sp->hdr.epoch = conn->proto.epoch;
sp->hdr.cid = call->cid;
sp->hdr.callNumber = call->call_id;
sp->hdr.seq = seq;
sp->hdr.serial = atomic_inc_return(&conn->serial);
sp->hdr.type = RXRPC_PACKET_TYPE_DATA;
sp->hdr.userStatus = 0;
sp->hdr.securityIndex = conn->security_ix;
sp->hdr._rsvd = 0;
sp->hdr.serviceId = call->service_id;
sp->hdr.flags = conn->out_clientflag;
if (msg_data_left(msg) == 0 && !more)
sp->hdr.flags |= RXRPC_LAST_PACKET;
else if (CIRC_SPACE(call->acks_head,
ACCESS_ONCE(call->acks_tail),
call->acks_winsz) > 1)
sp->hdr.flags |= RXRPC_MORE_PACKETS;
if (more && seq & 1)
sp->hdr.flags |= RXRPC_REQUEST_ACK;
ret = conn->security->secure_packet(
call, skb, skb->mark,
skb->head + sizeof(struct rxrpc_wire_header));
if (ret < 0)
goto out;
rxrpc_insert_header(skb);
rxrpc_queue_packet(call, skb, !msg_data_left(msg) && !more);
skb = NULL;
} }
} while (msg_data_left(msg) > 0);
remove_wait_queue(&call->waitq, &myself); success:
set_current_state(TASK_RUNNING); ret = copied;
out:
call->tx_pending = skb;
_leave(" = %d", ret); _leave(" = %d", ret);
return ret; return ret;
}
/* call_terminated:
* attempt to schedule an instant Tx resend rxrpc_free_skb(skb);
*/ _leave(" = %d", -call->error);
static inline void rxrpc_instant_resend(struct rxrpc_call *call) return ret;
{
read_lock_bh(&call->state_lock); maybe_error:
if (try_to_del_timer_sync(&call->resend_timer) >= 0) { if (copied)
clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags); goto success;
if (call->state < RXRPC_CALL_COMPLETE && goto out;
!test_and_set_bit(RXRPC_CALL_EV_RESEND_TIMER, &call->events))
rxrpc_queue_call(call); efault:
} ret = -EFAULT;
read_unlock_bh(&call->state_lock); goto out;
} }
/* /*
* queue a packet for transmission, set the resend timer and attempt * extract control messages from the sendmsg() control buffer
* to send the packet immediately
*/ */
static void rxrpc_queue_packet(struct rxrpc_call *call, struct sk_buff *skb, static int rxrpc_sendmsg_cmsg(struct msghdr *msg,
bool last) unsigned long *user_call_ID,
enum rxrpc_command *command,
u32 *abort_code,
bool *_exclusive)
{ {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb); struct cmsghdr *cmsg;
int ret; bool got_user_ID = false;
int len;
_net("queue skb %p [%d]", skb, call->acks_head); *command = RXRPC_CMD_SEND_DATA;
ASSERT(call->acks_window != NULL); if (msg->msg_controllen == 0)
call->acks_window[call->acks_head] = (unsigned long) skb; return -EINVAL;
smp_wmb();
call->acks_head = (call->acks_head + 1) & (call->acks_winsz - 1);
if (last || call->state == RXRPC_CALL_SERVER_ACK_REQUEST) { for_each_cmsghdr(cmsg, msg) {
_debug("________awaiting reply/ACK__________"); if (!CMSG_OK(msg, cmsg))
write_lock_bh(&call->state_lock); return -EINVAL;
switch (call->state) {
case RXRPC_CALL_CLIENT_SEND_REQUEST:
call->state = RXRPC_CALL_CLIENT_AWAIT_REPLY;
break;
case RXRPC_CALL_SERVER_ACK_REQUEST:
call->state = RXRPC_CALL_SERVER_SEND_REPLY;
if (!last)
break;
case RXRPC_CALL_SERVER_SEND_REPLY:
call->state = RXRPC_CALL_SERVER_AWAIT_ACK;
break;
default:
break;
}
write_unlock_bh(&call->state_lock);
}
_proto("Tx DATA %%%u { #%u }", sp->hdr.serial, sp->hdr.seq); len = cmsg->cmsg_len - CMSG_ALIGN(sizeof(struct cmsghdr));
_debug("CMSG %d, %d, %d",
cmsg->cmsg_level, cmsg->cmsg_type, len);
sp->need_resend = false; if (cmsg->cmsg_level != SOL_RXRPC)
sp->resend_at = jiffies + rxrpc_resend_timeout; continue;
if (!test_and_set_bit(RXRPC_CALL_RUN_RTIMER, &call->flags)) {
_debug("run timer");
call->resend_timer.expires = sp->resend_at;
add_timer(&call->resend_timer);
}
/* attempt to cancel the rx-ACK timer, deferring reply transmission if switch (cmsg->cmsg_type) {
* we're ACK'ing the request phase of an incoming call */ case RXRPC_USER_CALL_ID:
ret = -EAGAIN; if (msg->msg_flags & MSG_CMSG_COMPAT) {
if (try_to_del_timer_sync(&call->ack_timer) >= 0) { if (len != sizeof(u32))
/* the packet may be freed by rxrpc_process_call() before this return -EINVAL;
* returns */ *user_call_ID = *(u32 *) CMSG_DATA(cmsg);
if (rxrpc_is_client_call(call))
rxrpc_expose_client_call(call);
ret = rxrpc_send_data_packet(call->conn, skb);
_net("sent skb %p", skb);
} else { } else {
_debug("failed to delete ACK timer"); if (len != sizeof(unsigned long))
} return -EINVAL;
*user_call_ID = *(unsigned long *)
if (ret < 0) { CMSG_DATA(cmsg);
_debug("need instant resend %d", ret);
sp->need_resend = true;
rxrpc_instant_resend(call);
} }
_debug("User Call ID %lx", *user_call_ID);
got_user_ID = true;
break;
_leave(""); case RXRPC_ABORT:
} if (*command != RXRPC_CMD_SEND_DATA)
return -EINVAL;
*command = RXRPC_CMD_SEND_ABORT;
if (len != sizeof(*abort_code))
return -EINVAL;
*abort_code = *(unsigned int *) CMSG_DATA(cmsg);
_debug("Abort %x", *abort_code);
if (*abort_code == 0)
return -EINVAL;
break;
/* case RXRPC_ACCEPT:
* Convert a host-endian header into a network-endian header. if (*command != RXRPC_CMD_SEND_DATA)
*/ return -EINVAL;
static void rxrpc_insert_header(struct sk_buff *skb) *command = RXRPC_CMD_ACCEPT;
{ if (len != 0)
struct rxrpc_wire_header whdr; return -EINVAL;
struct rxrpc_skb_priv *sp = rxrpc_skb(skb); break;
whdr.epoch = htonl(sp->hdr.epoch); case RXRPC_EXCLUSIVE_CALL:
whdr.cid = htonl(sp->hdr.cid); *_exclusive = true;
whdr.callNumber = htonl(sp->hdr.callNumber); if (len != 0)
whdr.seq = htonl(sp->hdr.seq); return -EINVAL;
whdr.serial = htonl(sp->hdr.serial); break;
whdr.type = sp->hdr.type; default:
whdr.flags = sp->hdr.flags; return -EINVAL;
whdr.userStatus = sp->hdr.userStatus; }
whdr.securityIndex = sp->hdr.securityIndex; }
whdr._rsvd = htons(sp->hdr._rsvd);
whdr.serviceId = htons(sp->hdr.serviceId);
memcpy(skb->head, &whdr, sizeof(whdr)); if (!got_user_ID)
return -EINVAL;
_leave(" = 0");
return 0;
} }
/* /*
* send data through a socket * abort a call, sending an ABORT packet to the peer
* - must be called in process context
* - caller holds the socket locked
*/ */
static int rxrpc_send_data(struct rxrpc_sock *rx, static void rxrpc_send_abort(struct rxrpc_call *call, u32 abort_code)
struct rxrpc_call *call,
struct msghdr *msg, size_t len)
{ {
struct rxrpc_skb_priv *sp; if (call->state >= RXRPC_CALL_COMPLETE)
struct sk_buff *skb; return;
struct sock *sk = &rx->sk;
long timeo;
bool more;
int ret, copied;
timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);
/* this should be in poll */
sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
if (sk->sk_err || (sk->sk_shutdown & SEND_SHUTDOWN))
return -EPIPE;
more = msg->msg_flags & MSG_MORE;
skb = call->tx_pending;
call->tx_pending = NULL;
rxrpc_see_skb(skb);
copied = 0;
do {
if (!skb) {
size_t size, chunk, max, space;
_debug("alloc");
if (CIRC_SPACE(call->acks_head, write_lock_bh(&call->state_lock);
ACCESS_ONCE(call->acks_tail),
call->acks_winsz) <= 0) {
ret = -EAGAIN;
if (msg->msg_flags & MSG_DONTWAIT)
goto maybe_error;
ret = rxrpc_wait_for_tx_window(rx, call,
&timeo);
if (ret < 0)
goto maybe_error;
}
max = call->conn->params.peer->maxdata; if (__rxrpc_abort_call(call, abort_code, ECONNABORTED)) {
max -= call->conn->security_size; del_timer_sync(&call->resend_timer);
max &= ~(call->conn->size_align - 1UL); del_timer_sync(&call->ack_timer);
clear_bit(RXRPC_CALL_EV_RESEND_TIMER, &call->events);
clear_bit(RXRPC_CALL_EV_ACK, &call->events);
clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags);
rxrpc_queue_call(call);
}
chunk = max; write_unlock_bh(&call->state_lock);
if (chunk > msg_data_left(msg) && !more) }
chunk = msg_data_left(msg);
space = chunk + call->conn->size_align; /*
space &= ~(call->conn->size_align - 1UL); * Create a new client call for sendmsg().
*/
static struct rxrpc_call *
rxrpc_new_client_call_for_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg,
unsigned long user_call_ID, bool exclusive)
{
struct rxrpc_conn_parameters cp;
struct rxrpc_call *call;
struct key *key;
size = space + call->conn->header_size; DECLARE_SOCKADDR(struct sockaddr_rxrpc *, srx, msg->msg_name);
_debug("SIZE: %zu/%zu/%zu", chunk, space, size); _enter("");
/* create a buffer that we can retain until it's ACK'd */ if (!msg->msg_name)
skb = sock_alloc_send_skb( return ERR_PTR(-EDESTADDRREQ);
sk, size, msg->msg_flags & MSG_DONTWAIT, &ret);
if (!skb)
goto maybe_error;
rxrpc_new_skb(skb); key = rx->key;
if (key && !rx->key->payload.data[0])
key = NULL;
_debug("ALLOC SEND %p", skb); memset(&cp, 0, sizeof(cp));
cp.local = rx->local;
cp.key = rx->key;
cp.security_level = rx->min_sec_level;
cp.exclusive = rx->exclusive | exclusive;
cp.service_id = srx->srx_service;
call = rxrpc_new_client_call(rx, &cp, srx, user_call_ID, GFP_KERNEL);
ASSERTCMP(skb->mark, ==, 0); _leave(" = %p\n", call);
return call;
}
_debug("HS: %u", call->conn->header_size); /*
skb_reserve(skb, call->conn->header_size); * send a message forming part of a client call through an RxRPC socket
skb->len += call->conn->header_size; * - 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)
{
enum rxrpc_command cmd;
struct rxrpc_call *call;
unsigned long user_call_ID = 0;
bool exclusive = false;
u32 abort_code = 0;
int ret;
sp = rxrpc_skb(skb); _enter("");
sp->remain = chunk;
if (sp->remain > skb_tailroom(skb))
sp->remain = skb_tailroom(skb);
_net("skb: hr %d, tr %d, hl %d, rm %d", ret = rxrpc_sendmsg_cmsg(msg, &user_call_ID, &cmd, &abort_code,
skb_headroom(skb), &exclusive);
skb_tailroom(skb), if (ret < 0)
skb_headlen(skb), return ret;
sp->remain);
skb->ip_summed = CHECKSUM_UNNECESSARY; if (cmd == RXRPC_CMD_ACCEPT) {
if (rx->sk.sk_state != RXRPC_SERVER_LISTENING)
return -EINVAL;
call = rxrpc_accept_call(rx, user_call_ID, NULL);
if (IS_ERR(call))
return PTR_ERR(call);
rxrpc_put_call(call);
return 0;
} }
_debug("append"); call = rxrpc_find_call_by_user_ID(rx, user_call_ID);
sp = rxrpc_skb(skb); if (!call) {
if (cmd != RXRPC_CMD_SEND_DATA)
return -EBADSLT;
call = rxrpc_new_client_call_for_sendmsg(rx, msg, user_call_ID,
exclusive);
if (IS_ERR(call))
return PTR_ERR(call);
}
/* append next segment of data to the current buffer */ rxrpc_see_call(call);
if (msg_data_left(msg) > 0) { _debug("CALL %d USR %lx ST %d on CONN %p",
int copy = skb_tailroom(skb); call->debug_id, call->user_call_ID, call->state, call->conn);
ASSERTCMP(copy, >, 0);
if (copy > msg_data_left(msg))
copy = msg_data_left(msg);
if (copy > sp->remain)
copy = sp->remain;
_debug("add"); if (call->state >= RXRPC_CALL_COMPLETE) {
ret = skb_add_data(skb, &msg->msg_iter, copy); /* it's too late for this call */
_debug("added"); ret = -ESHUTDOWN;
if (ret < 0) } else if (cmd == RXRPC_CMD_SEND_ABORT) {
goto efault; rxrpc_send_abort(call, abort_code);
sp->remain -= copy; ret = 0;
skb->mark += copy; } else if (cmd != RXRPC_CMD_SEND_DATA) {
copied += copy; ret = -EINVAL;
} else if (rxrpc_is_client_call(call) &&
call->state != RXRPC_CALL_CLIENT_SEND_REQUEST) {
/* request phase complete for this client call */
ret = -EPROTO;
} else if (rxrpc_is_service_call(call) &&
call->state != RXRPC_CALL_SERVER_ACK_REQUEST &&
call->state != RXRPC_CALL_SERVER_SEND_REPLY) {
/* Reply phase not begun or not complete for service call. */
ret = -EPROTO;
} else {
ret = rxrpc_send_data(rx, call, msg, len);
} }
/* check for the far side aborting the call or a network error rxrpc_put_call(call);
* occurring */ _leave(" = %d", ret);
if (call->state == RXRPC_CALL_COMPLETE) return ret;
goto call_terminated; }
/* 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;
/* pad out if we're using security */ /**
if (conn->security_ix) { * rxrpc_kernel_send_data - Allow a kernel service to send data on a call
pad = conn->security_size + skb->mark; * @sock: The socket the call is on
pad = conn->size_align - pad; * @call: The call to send data through
pad &= conn->size_align - 1; * @msg: The data to send
_debug("pad %zu", pad); * @len: The amount of data to send
if (pad) *
memset(skb_put(skb, pad), 0, pad); * 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,
struct msghdr *msg, size_t len)
{
int ret;
seq = atomic_inc_return(&call->sequence); _enter("{%d,%s},", call->debug_id, rxrpc_call_states[call->state]);
sp->hdr.epoch = conn->proto.epoch; ASSERTCMP(msg->msg_name, ==, NULL);
sp->hdr.cid = call->cid; ASSERTCMP(msg->msg_control, ==, NULL);
sp->hdr.callNumber = call->call_id;
sp->hdr.seq = seq;
sp->hdr.serial = atomic_inc_return(&conn->serial);
sp->hdr.type = RXRPC_PACKET_TYPE_DATA;
sp->hdr.userStatus = 0;
sp->hdr.securityIndex = conn->security_ix;
sp->hdr._rsvd = 0;
sp->hdr.serviceId = call->service_id;
sp->hdr.flags = conn->out_clientflag; lock_sock(sock->sk);
if (msg_data_left(msg) == 0 && !more)
sp->hdr.flags |= RXRPC_LAST_PACKET;
else if (CIRC_SPACE(call->acks_head,
ACCESS_ONCE(call->acks_tail),
call->acks_winsz) > 1)
sp->hdr.flags |= RXRPC_MORE_PACKETS;
if (more && seq & 1)
sp->hdr.flags |= RXRPC_REQUEST_ACK;
ret = conn->security->secure_packet( _debug("CALL %d USR %lx ST %d on CONN %p",
call, skb, skb->mark, call->debug_id, call->user_call_ID, call->state, call->conn);
skb->head + sizeof(struct rxrpc_wire_header));
if (ret < 0)
goto out;
rxrpc_insert_header(skb); if (call->state >= RXRPC_CALL_COMPLETE) {
rxrpc_queue_packet(call, skb, !msg_data_left(msg) && !more); ret = -ESHUTDOWN; /* it's too late for this call */
skb = NULL; } else if (call->state != RXRPC_CALL_CLIENT_SEND_REQUEST &&
call->state != RXRPC_CALL_SERVER_ACK_REQUEST &&
call->state != RXRPC_CALL_SERVER_SEND_REPLY) {
ret = -EPROTO; /* request phase complete for this client call */
} else {
ret = rxrpc_send_data(rxrpc_sk(sock->sk), call, msg, len);
} }
} while (msg_data_left(msg) > 0);
success: release_sock(sock->sk);
ret = copied;
out:
call->tx_pending = skb;
_leave(" = %d", ret); _leave(" = %d", ret);
return ret; return ret;
}
EXPORT_SYMBOL(rxrpc_kernel_send_data);
call_terminated: /**
rxrpc_free_skb(skb); * rxrpc_kernel_abort_call - Allow a kernel service to abort a call
_leave(" = %d", -call->error); * @sock: The socket the call is on
return ret; * @call: The call to be aborted
* @abort_code: The abort code to stick into the ABORT packet
*
* Allow a kernel service to abort a call, if it's still in an abortable state.
*/
void rxrpc_kernel_abort_call(struct socket *sock, struct rxrpc_call *call,
u32 abort_code)
{
_enter("{%d},%d", call->debug_id, abort_code);
maybe_error: lock_sock(sock->sk);
if (copied)
goto success;
goto out;
efault: _debug("CALL %d USR %lx ST %d on CONN %p",
ret = -EFAULT; call->debug_id, call->user_call_ID, call->state, call->conn);
goto out;
rxrpc_send_abort(call, abort_code);
release_sock(sock->sk);
_leave("");
} }
EXPORT_SYMBOL(rxrpc_kernel_abort_call);
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7