summaryrefslogtreecommitdiff
path: root/fs/afs/rxrpc.c
diff options
context:
space:
mode:
Diffstat (limited to 'fs/afs/rxrpc.c')
-rw-r--r--fs/afs/rxrpc.c208
1 files changed, 111 insertions, 97 deletions
diff --git a/fs/afs/rxrpc.c b/fs/afs/rxrpc.c
index 25f05a8d21b1..95f42872b787 100644
--- a/fs/afs/rxrpc.c
+++ b/fs/afs/rxrpc.c
@@ -19,35 +19,16 @@
struct socket *afs_socket; /* my RxRPC socket */
static struct workqueue_struct *afs_async_calls;
static struct afs_call *afs_spare_incoming_call;
-static atomic_t afs_outstanding_calls;
+atomic_t afs_outstanding_calls;
-static void afs_free_call(struct afs_call *);
static void afs_wake_up_call_waiter(struct sock *, struct rxrpc_call *, unsigned long);
static int afs_wait_for_call_to_complete(struct afs_call *);
static void afs_wake_up_async_call(struct sock *, struct rxrpc_call *, unsigned long);
-static int afs_dont_wait_for_call_to_complete(struct afs_call *);
static void afs_process_async_call(struct work_struct *);
static void afs_rx_new_call(struct sock *, struct rxrpc_call *, unsigned long);
static void afs_rx_discard_new_call(struct rxrpc_call *, unsigned long);
static int afs_deliver_cm_op_id(struct afs_call *);
-/* synchronous call management */
-const struct afs_wait_mode afs_sync_call = {
- .notify_rx = afs_wake_up_call_waiter,
- .wait = afs_wait_for_call_to_complete,
-};
-
-/* asynchronous call management */
-const struct afs_wait_mode afs_async_call = {
- .notify_rx = afs_wake_up_async_call,
- .wait = afs_dont_wait_for_call_to_complete,
-};
-
-/* asynchronous incoming call management */
-static const struct afs_wait_mode afs_async_incoming_call = {
- .notify_rx = afs_wake_up_async_call,
-};
-
/* asynchronous incoming call initial processing */
static const struct afs_call_type afs_RXCMxxxx = {
.name = "CB.xxxx",
@@ -130,9 +111,11 @@ void afs_close_socket(void)
{
_enter("");
+ kernel_listen(afs_socket, 0);
+ flush_workqueue(afs_async_calls);
+
if (afs_spare_incoming_call) {
- atomic_inc(&afs_outstanding_calls);
- afs_free_call(afs_spare_incoming_call);
+ afs_put_call(afs_spare_incoming_call);
afs_spare_incoming_call = NULL;
}
@@ -141,7 +124,6 @@ void afs_close_socket(void)
TASK_UNINTERRUPTIBLE);
_debug("no outstanding calls");
- flush_workqueue(afs_async_calls);
kernel_sock_shutdown(afs_socket, SHUT_RDWR);
flush_workqueue(afs_async_calls);
sock_release(afs_socket);
@@ -152,44 +134,79 @@ void afs_close_socket(void)
}
/*
- * free a call
+ * Allocate a call.
*/
-static void afs_free_call(struct afs_call *call)
+static struct afs_call *afs_alloc_call(const struct afs_call_type *type,
+ gfp_t gfp)
{
- _debug("DONE %p{%s} [%d]",
- call, call->type->name, atomic_read(&afs_outstanding_calls));
+ struct afs_call *call;
+ int o;
- ASSERTCMP(call->rxcall, ==, NULL);
- ASSERT(!work_pending(&call->async_work));
- ASSERT(call->type->name != NULL);
+ call = kzalloc(sizeof(*call), gfp);
+ if (!call)
+ return NULL;
- kfree(call->request);
- kfree(call);
+ call->type = type;
+ atomic_set(&call->usage, 1);
+ INIT_WORK(&call->async_work, afs_process_async_call);
+ init_waitqueue_head(&call->waitq);
- if (atomic_dec_and_test(&afs_outstanding_calls))
- wake_up_atomic_t(&afs_outstanding_calls);
+ o = atomic_inc_return(&afs_outstanding_calls);
+ trace_afs_call(call, afs_call_trace_alloc, 1, o,
+ __builtin_return_address(0));
+ return call;
}
/*
- * End a call but do not free it
+ * Dispose of a reference on a call.
*/
-static void afs_end_call_nofree(struct afs_call *call)
+void afs_put_call(struct afs_call *call)
{
- if (call->rxcall) {
- rxrpc_kernel_end_call(afs_socket, call->rxcall);
- call->rxcall = NULL;
+ int n = atomic_dec_return(&call->usage);
+ int o = atomic_read(&afs_outstanding_calls);
+
+ trace_afs_call(call, afs_call_trace_put, n + 1, o,
+ __builtin_return_address(0));
+
+ ASSERTCMP(n, >=, 0);
+ if (n == 0) {
+ ASSERT(!work_pending(&call->async_work));
+ ASSERT(call->type->name != NULL);
+
+ if (call->rxcall) {
+ rxrpc_kernel_end_call(afs_socket, call->rxcall);
+ call->rxcall = NULL;
+ }
+ if (call->type->destructor)
+ call->type->destructor(call);
+
+ kfree(call->request);
+ kfree(call);
+
+ o = atomic_dec_return(&afs_outstanding_calls);
+ trace_afs_call(call, afs_call_trace_free, 0, o,
+ __builtin_return_address(0));
+ if (o == 0)
+ wake_up_atomic_t(&afs_outstanding_calls);
}
- if (call->type->destructor)
- call->type->destructor(call);
}
/*
- * End a call and free it
+ * Queue the call for actual work. Returns 0 unconditionally for convenience.
*/
-static void afs_end_call(struct afs_call *call)
+int afs_queue_call_work(struct afs_call *call)
{
- afs_end_call_nofree(call);
- afs_free_call(call);
+ int u = atomic_inc_return(&call->usage);
+
+ trace_afs_call(call, afs_call_trace_work, u,
+ atomic_read(&afs_outstanding_calls),
+ __builtin_return_address(0));
+
+ INIT_WORK(&call->work, call->type->work);
+
+ if (!queue_work(afs_wq, &call->work))
+ afs_put_call(call);
+ return 0;
}
/*
@@ -200,25 +217,19 @@ struct afs_call *afs_alloc_flat_call(const struct afs_call_type *type,
{
struct afs_call *call;
- call = kzalloc(sizeof(*call), GFP_NOFS);
+ call = afs_alloc_call(type, GFP_NOFS);
if (!call)
goto nomem_call;
- _debug("CALL %p{%s} [%d]",
- call, type->name, atomic_read(&afs_outstanding_calls));
- atomic_inc(&afs_outstanding_calls);
-
- call->type = type;
- call->request_size = request_size;
- call->reply_max = reply_max;
-
if (request_size) {
+ call->request_size = request_size;
call->request = kmalloc(request_size, GFP_NOFS);
if (!call->request)
goto nomem_free;
}
if (reply_max) {
+ call->reply_max = reply_max;
call->buffer = kmalloc(reply_max, GFP_NOFS);
if (!call->buffer)
goto nomem_free;
@@ -228,7 +239,7 @@ struct afs_call *afs_alloc_flat_call(const struct afs_call_type *type,
return call;
nomem_free:
- afs_free_call(call);
+ afs_put_call(call);
nomem_call:
return NULL;
}
@@ -315,7 +326,7 @@ static int afs_send_pages(struct afs_call *call, struct msghdr *msg,
* initiate a call
*/
int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
- const struct afs_wait_mode *wait_mode)
+ bool async)
{
struct sockaddr_rxrpc srx;
struct rxrpc_call *rxcall;
@@ -332,8 +343,7 @@ int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
call, call->type->name, key_serial(call->key),
atomic_read(&afs_outstanding_calls));
- call->wait_mode = wait_mode;
- INIT_WORK(&call->async_work, afs_process_async_call);
+ call->async = async;
memset(&srx, 0, sizeof(srx));
srx.srx_family = AF_RXRPC;
@@ -347,7 +357,9 @@ int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
/* create a call */
rxcall = rxrpc_kernel_begin_call(afs_socket, &srx, call->key,
(unsigned long) call, gfp,
- wait_mode->notify_rx);
+ (async ?
+ afs_wake_up_async_call :
+ afs_wake_up_call_waiter));
call->key = NULL;
if (IS_ERR(rxcall)) {
ret = PTR_ERR(rxcall);
@@ -386,12 +398,15 @@ int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
/* at this point, an async call may no longer exist as it may have
* already completed */
- return wait_mode->wait(call);
+ if (call->async)
+ return -EINPROGRESS;
+
+ return afs_wait_for_call_to_complete(call);
error_do_abort:
rxrpc_kernel_abort_call(afs_socket, rxcall, RX_USER_ABORT, -ret, "KSD");
error_kill_call:
- afs_end_call(call);
+ afs_put_call(call);
_leave(" = %d", ret);
return ret;
}
@@ -416,6 +431,8 @@ static void afs_deliver_to_call(struct afs_call *call)
ret = rxrpc_kernel_recv_data(afs_socket, call->rxcall,
NULL, 0, &offset, false,
&call->abort_code);
+ trace_afs_recv_data(call, 0, offset, false, ret);
+
if (ret == -EINPROGRESS || ret == -EAGAIN)
return;
if (ret == 1 || ret < 0) {
@@ -459,7 +476,7 @@ static void afs_deliver_to_call(struct afs_call *call)
done:
if (call->state == AFS_CALL_COMPLETE && call->incoming)
- afs_end_call(call);
+ afs_put_call(call);
out:
_leave("");
return;
@@ -516,7 +533,7 @@ static int afs_wait_for_call_to_complete(struct afs_call *call)
}
_debug("call complete");
- afs_end_call(call);
+ afs_put_call(call);
_leave(" = %d", ret);
return ret;
}
@@ -540,24 +557,25 @@ static void afs_wake_up_async_call(struct sock *sk, struct rxrpc_call *rxcall,
unsigned long call_user_ID)
{
struct afs_call *call = (struct afs_call *)call_user_ID;
+ int u;
+ trace_afs_notify_call(rxcall, call);
call->need_attention = true;
- queue_work(afs_async_calls, &call->async_work);
-}
-/*
- * put a call into asynchronous mode
- * - mustn't touch the call descriptor as the call my have completed by the
- * time we get here
- */
-static int afs_dont_wait_for_call_to_complete(struct afs_call *call)
-{
- _enter("");
- return -EINPROGRESS;
+ u = __atomic_add_unless(&call->usage, 1, 0);
+ if (u != 0) {
+ trace_afs_call(call, afs_call_trace_wake, u,
+ atomic_read(&afs_outstanding_calls),
+ __builtin_return_address(0));
+
+ if (!queue_work(afs_async_calls, &call->async_work))
+ afs_put_call(call);
+ }
}
/*
- * delete an asynchronous call
+ * Delete an asynchronous call. The work item carries a ref to the call struct
+ * that we need to release.
*/
static void afs_delete_async_call(struct work_struct *work)
{
@@ -565,13 +583,14 @@ static void afs_delete_async_call(struct work_struct *work)
_enter("");
- afs_free_call(call);
+ afs_put_call(call);
_leave("");
}
/*
- * perform processing on an asynchronous call
+ * Perform I/O processing on an asynchronous call. The work item carries a ref
+ * to the call struct that we either need to release or to pass on.
*/
static void afs_process_async_call(struct work_struct *work)
{
@@ -584,21 +603,19 @@ static void afs_process_async_call(struct work_struct *work)
afs_deliver_to_call(call);
}
- if (call->state == AFS_CALL_COMPLETE && call->wait_mode) {
- if (call->wait_mode->async_complete)
- call->wait_mode->async_complete(call->reply,
- call->error);
+ if (call->state == AFS_CALL_COMPLETE) {
call->reply = NULL;
- /* kill the call */
- afs_end_call_nofree(call);
-
- /* we can't just delete the call because the work item may be
- * queued */
+ /* We have two refs to release - one from the alloc and one
+ * queued with the work item - and we can't just deallocate the
+ * call because the work item may be queued again.
+ */
call->async_work.func = afs_delete_async_call;
- queue_work(afs_async_calls, &call->async_work);
+ if (!queue_work(afs_async_calls, &call->async_work))
+ afs_put_call(call);
}
+ afs_put_call(call);
_leave("");
}
@@ -618,15 +635,13 @@ static void afs_charge_preallocation(struct work_struct *work)
for (;;) {
if (!call) {
- call = kzalloc(sizeof(struct afs_call), GFP_KERNEL);
+ call = afs_alloc_call(&afs_RXCMxxxx, GFP_KERNEL);
if (!call)
break;
- INIT_WORK(&call->async_work, afs_process_async_call);
- call->wait_mode = &afs_async_incoming_call;
- call->type = &afs_RXCMxxxx;
- init_waitqueue_head(&call->waitq);
+ call->async = true;
call->state = AFS_CALL_AWAIT_OP_ID;
+ init_waitqueue_head(&call->waitq);
}
if (rxrpc_kernel_charge_accept(afs_socket,
@@ -648,9 +663,8 @@ static void afs_rx_discard_new_call(struct rxrpc_call *rxcall,
{
struct afs_call *call = (struct afs_call *)user_call_ID;
- atomic_inc(&afs_outstanding_calls);
call->rxcall = NULL;
- afs_free_call(call);
+ afs_put_call(call);
}
/*
@@ -659,7 +673,6 @@ static void afs_rx_discard_new_call(struct rxrpc_call *rxcall,
static void afs_rx_new_call(struct sock *sk, struct rxrpc_call *rxcall,
unsigned long user_call_ID)
{
- atomic_inc(&afs_outstanding_calls);
queue_work(afs_wq, &afs_charge_preallocation_work);
}
@@ -689,6 +702,8 @@ static int afs_deliver_cm_op_id(struct afs_call *call)
if (!afs_cm_incoming_call(call))
return -ENOTSUPP;
+ trace_afs_cb_call(call);
+
/* pass responsibility for the remainer of this message off to the
* cache manager op */
return call->type->deliver(call);
@@ -721,7 +736,6 @@ void afs_send_empty_reply(struct afs_call *call)
rxrpc_kernel_abort_call(afs_socket, call->rxcall,
RX_USER_ABORT, ENOMEM, "KOO");
default:
- afs_end_call(call);
_leave(" [error]");
return;
}
@@ -760,7 +774,6 @@ void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len)
rxrpc_kernel_abort_call(afs_socket, call->rxcall,
RX_USER_ABORT, ENOMEM, "KOO");
}
- afs_end_call(call);
_leave(" [error]");
}
@@ -780,6 +793,7 @@ int afs_extract_data(struct afs_call *call, void *buf, size_t count,
ret = rxrpc_kernel_recv_data(afs_socket, call->rxcall,
buf, count, &call->offset,
want_more, &call->abort_code);
+ trace_afs_recv_data(call, count, call->offset, want_more, ret);
if (ret == 0 || ret == -EAGAIN)
return ret;