rxrpc: Allow the kernel to mark a call as being non-interruptible

Allow kernel services using AF_RXRPC to indicate that a call should be
non-interruptible.  This allows kafs to make things like lock-extension and
writeback data storage calls non-interruptible.

If this is set, signals will be ignored for operations on that call where
possible - such as waiting to get a call channel on an rxrpc connection.

It doesn't prevent UDP sendmsg from being interrupted, but that will be
handled by packet retransmission.

rxrpc_kernel_recv_data() isn't affected by this since that never waits,
preferring instead to return -EAGAIN and leave the waiting to the caller.

Userspace initiated calls can't be set to be uninterruptible at this time.

Signed-off-by: David Howells <dhowells@redhat.com>

5 files changed, 16 insertions, 3 deletions

diff --git a/net/rxrpc/af_rxrpc.c b/net/rxrpc/af_rxrpc.c
index 213935fbbbf7..ffde5b187f5d 100644
--- a/net/rxrpc/af_rxrpc.c
+++ b/net/rxrpc/af_rxrpc.c
@@ -270,6 +270,7 @@ static int rxrpc_listen(struct socket *sock, int backlog)
  * @gfp: The allocation constraints
  * @notify_rx: Where to send notifications instead of socket queue
  * @upgrade: Request service upgrade for call
+ * @intr: The call is interruptible
  * @debug_id: The debug ID for tracing to be assigned to the call
  *
  * Allow a kernel service to begin a call on the nominated socket.  This just
@@ -287,6 +288,7 @@ struct rxrpc_call *rxrpc_kernel_begin_call(struct socket *sock,
                                            gfp_t gfp,
                                            rxrpc_notify_rx_t notify_rx,
                                            bool upgrade,
+                                           bool intr,
                                            unsigned int debug_id)
 {
         struct rxrpc_conn_parameters cp;
@@ -311,6 +313,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;
 
         memset(&cp, 0, sizeof(cp));
         cp.local                = rx->local;
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index 062ca9dc29b8..07fc1dfa4878 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -482,6 +482,7 @@ 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 */
 };
 
 /*
@@ -711,6 +712,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 */
 };
 
 struct rxrpc_send_params {
diff --git a/net/rxrpc/call_object.c b/net/rxrpc/call_object.c
index fe96881a334d..d0ca98d7aef5 100644
--- a/net/rxrpc/call_object.c
+++ b/net/rxrpc/call_object.c
@@ -241,6 +241,8 @@ 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->tx_total_len = p->tx_total_len;
         trace_rxrpc_call(call, rxrpc_call_new_client, atomic_read(&call->usage),
                          here, (const void *)p->user_call_ID);
diff --git a/net/rxrpc/conn_client.c b/net/rxrpc/conn_client.c
index 83797b3949e2..5cf5595a14d8 100644
--- a/net/rxrpc/conn_client.c
+++ b/net/rxrpc/conn_client.c
@@ -656,10 +656,14 @@ static int rxrpc_wait_for_channel(struct rxrpc_call *call, gfp_t gfp)
 
                 add_wait_queue_exclusive(&call->waitq, &myself);
                 for (;;) {
-                        set_current_state(TASK_INTERRUPTIBLE);
+                        if (test_bit(RXRPC_CALL_IS_INTR, &call->flags))
+                                set_current_state(TASK_INTERRUPTIBLE);
+                        else
+                                set_current_state(TASK_UNINTERRUPTIBLE);
                         if (call->call_id)
                                 break;
-                        if (signal_pending(current)) {
+                        if (test_bit(RXRPC_CALL_IS_INTR, &call->flags) &&
+                            signal_pending(current)) {
                                 ret = -ERESTARTSYS;
                                 break;
                         }
diff --git a/net/rxrpc/sendmsg.c b/net/rxrpc/sendmsg.c
index bec64deb7b0a..45a05d9a27fa 100644
--- a/net/rxrpc/sendmsg.c
+++ b/net/rxrpc/sendmsg.c
@@ -80,7 +80,8 @@ static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
                 if (call->state >= RXRPC_CALL_COMPLETE)
                         return call->error;
 
-                if (timeout == 0 &&
+                if (test_bit(RXRPC_CALL_IS_INTR, &call->flags) &&
+                    timeout == 0 &&
                     tx_win == tx_start && signal_pending(current))
                         return -EINTR;
 
@@ -620,6 +621,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,
                 .abort_code             = 0,
                 .command                = RXRPC_CMD_SEND_DATA,
                 .exclusive              = false,