Skip to content

Commit

Permalink
rxrpc: Fix call interruptibility handling
Browse files Browse the repository at this point in the history 
Fix the interruptibility of kernel-initiated client calls so that they're
either only interruptible when they're waiting for a call slot to come
available or they're not interruptible at all.  Either way, they're not
interruptible during transmission.

This should help prevent StoreData calls from being interrupted when
writeback is in progress.  It doesn't, however, handle interruption during
the receive phase.

Userspace-initiated calls are still interruptable.  After the signal has
been handled, sendmsg() will return the amount of data copied out of the
buffer and userspace can perform another sendmsg() call to continue
transmission.

Fixes: bc5e3a5 ("rxrpc: Use MSG_WAITALL to tell sendmsg() to temporarily ignore signals")
Signed-off-by: David Howells <[email protected]>
  • Loading branch information
@dhowells
dhowells committed Mar 13, 2020
1 parent 158fe66 commit e138aa7
Show file tree
Hide file tree
Showing 7 changed files with 60 additions and 19 deletions.
3 changes: 2 additions & 1 deletion fs/afs/rxrpc.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -413,7 +413,8 @@ void afs_make_call(struct afs_addr_cursor *ac, struct afs_call *call, gfp_t gfp)
afs_wake_up_async_call :
afs_wake_up_call_waiter),
call->upgrade,
call->intr,
(call->intr ? RXRPC_PREINTERRUPTIBLE :
RXRPC_UNINTERRUPTIBLE),
call->debug_id);
if (IS_ERR(rxcall)) {
ret = PTR_ERR(rxcall);
Expand Down
8 changes: 7 additions & 1 deletion include/net/af_rxrpc.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,12 @@ struct sock;
struct socket;
struct rxrpc_call;

enum rxrpc_interruptibility {
RXRPC_INTERRUPTIBLE, /* Call is interruptible */
RXRPC_PREINTERRUPTIBLE, /* Call can be cancelled whilst waiting for a slot */
RXRPC_UNINTERRUPTIBLE, /* Call should not be interruptible at all */
};

/*
* Debug ID counter for tracing.
*/
Expand All @@ -41,7 +47,7 @@ struct rxrpc_call *rxrpc_kernel_begin_call(struct socket *,
gfp_t,
rxrpc_notify_rx_t,
bool,
bool,
enum rxrpc_interruptibility,
unsigned int);
int rxrpc_kernel_send_data(struct socket *, struct rxrpc_call *,
struct msghdr *, size_t,
Expand Down
4 changes: 2 additions & 2 deletions net/rxrpc/af_rxrpc.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -285,7 +285,7 @@ struct rxrpc_call *rxrpc_kernel_begin_call(struct socket *sock,
gfp_t gfp,
rxrpc_notify_rx_t notify_rx,
bool upgrade,
bool intr,
enum rxrpc_interruptibility interruptibility,
unsigned int debug_id)
{
struct rxrpc_conn_parameters cp;
Expand All @@ -310,7 +310,7 @@ struct rxrpc_call *rxrpc_kernel_begin_call(struct socket *sock,
memset(&p, 0, sizeof(p));
p.user_call_ID = user_call_ID;
p.tx_total_len = tx_total_len;
p.intr = intr;
p.interruptibility = interruptibility;

memset(&cp, 0, sizeof(cp));
cp.local = rx->local;
Expand Down
4 changes: 2 additions & 2 deletions net/rxrpc/ar-internal.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -489,7 +489,6 @@ enum rxrpc_call_flag {
RXRPC_CALL_BEGAN_RX_TIMER, /* We began the expect_rx_by timer */
RXRPC_CALL_RX_HEARD, /* The peer responded at least once to this call */
RXRPC_CALL_RX_UNDERRUN, /* Got data underrun */
RXRPC_CALL_IS_INTR, /* The call is interruptible */
RXRPC_CALL_DISCONNECTED, /* The call has been disconnected */
};

Expand Down Expand Up @@ -598,6 +597,7 @@ struct rxrpc_call {
atomic_t usage;
u16 service_id; /* service ID */
u8 security_ix; /* Security type */
enum rxrpc_interruptibility interruptibility; /* At what point call may be interrupted */
u32 call_id; /* call ID on connection */
u32 cid; /* connection ID plus channel index */
int debug_id; /* debug ID for printks */
Expand Down Expand Up @@ -721,7 +721,7 @@ struct rxrpc_call_params {
u32 normal; /* Max time since last call packet (msec) */
} timeouts;
u8 nr_timeouts; /* Number of timeouts specified */
bool intr; /* The call is interruptible */
enum rxrpc_interruptibility interruptibility; /* How is interruptible is the call? */
};

struct rxrpc_send_params {
Expand Down
3 changes: 1 addition & 2 deletions net/rxrpc/call_object.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -237,8 +237,7 @@ struct rxrpc_call *rxrpc_new_client_call(struct rxrpc_sock *rx,
return call;
}

if (p->intr)
__set_bit(RXRPC_CALL_IS_INTR, &call->flags);
call->interruptibility = p->interruptibility;
call->tx_total_len = p->tx_total_len;
trace_rxrpc_call(call->debug_id, rxrpc_call_new_client,
atomic_read(&call->usage),
Expand Down
13 changes: 10 additions & 3 deletions net/rxrpc/conn_client.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -655,13 +655,20 @@ static int rxrpc_wait_for_channel(struct rxrpc_call *call, gfp_t gfp)

add_wait_queue_exclusive(&call->waitq, &myself);
for (;;) {
if (test_bit(RXRPC_CALL_IS_INTR, &call->flags))
switch (call->interruptibility) {
case RXRPC_INTERRUPTIBLE:
case RXRPC_PREINTERRUPTIBLE:
set_current_state(TASK_INTERRUPTIBLE);
else
break;
case RXRPC_UNINTERRUPTIBLE:
default:
set_current_state(TASK_UNINTERRUPTIBLE);
break;
}
if (call->call_id)
break;
if (test_bit(RXRPC_CALL_IS_INTR, &call->flags) &&
if ((call->interruptibility == RXRPC_INTERRUPTIBLE ||
call->interruptibility == RXRPC_PREINTERRUPTIBLE) &&
signal_pending(current)) {
ret = -ERESTARTSYS;
break;
Expand Down
44 changes: 36 additions & 8 deletions net/rxrpc/sendmsg.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ static int rxrpc_wait_for_tx_window_intr(struct rxrpc_sock *rx,
* 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,
static int rxrpc_wait_for_tx_window_waitall(struct rxrpc_sock *rx,
struct rxrpc_call *call)
{
rxrpc_seq_t tx_start, tx_win;
Expand All @@ -87,8 +87,7 @@ static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
if (call->state >= RXRPC_CALL_COMPLETE)
return call->error;

if (test_bit(RXRPC_CALL_IS_INTR, &call->flags) &&
timeout == 0 &&
if (timeout == 0 &&
tx_win == tx_start && signal_pending(current))
return -EINTR;

Expand All @@ -102,6 +101,26 @@ static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
}
}

/*
* Wait for space to appear in the Tx queue uninterruptibly.
*/
static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
struct rxrpc_call *call,
long *timeo)
{
for (;;) {
set_current_state(TASK_UNINTERRUPTIBLE);
if (rxrpc_check_tx_space(call, NULL))
return 0;

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

trace_rxrpc_transmit(call, rxrpc_transmit_wait);
*timeo = schedule_timeout(*timeo);
}
}

/*
* wait for space to appear in the transmit/ACK window
* - caller holds the socket locked
Expand All @@ -119,10 +138,19 @@ static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx,

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

if (waitall)
ret = rxrpc_wait_for_tx_window_nonintr(rx, call);
else
ret = rxrpc_wait_for_tx_window_intr(rx, call, timeo);
switch (call->interruptibility) {
case RXRPC_INTERRUPTIBLE:
if (waitall)
ret = rxrpc_wait_for_tx_window_waitall(rx, call);
else
ret = rxrpc_wait_for_tx_window_intr(rx, call, timeo);
break;
case RXRPC_PREINTERRUPTIBLE:
case RXRPC_UNINTERRUPTIBLE:
default:
ret = rxrpc_wait_for_tx_window_nonintr(rx, call, timeo);
break;
}

remove_wait_queue(&call->waitq, &myself);
set_current_state(TASK_RUNNING);
Expand Down Expand Up @@ -628,7 +656,7 @@ int rxrpc_do_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg, size_t len)
.call.tx_total_len = -1,
.call.user_call_ID = 0,
.call.nr_timeouts = 0,
.call.intr = true,
.call.interruptibility = RXRPC_INTERRUPTIBLE,
.abort_code = 0,
.command = RXRPC_CMD_SEND_DATA,
.exclusive = false,
Expand Down

0 comments on commit e138aa7

Please sign in to comment.