diff --git a/net/rds/message.c b/net/rds/message.c index 7feb0eb6537d..f25f2592586f 100644 --- a/net/rds/message.c +++ b/net/rds/message.c @@ -182,6 +182,19 @@ static void rds_message_purge(struct rds_message *rm) kref_put(&rm->atomic.op_rdma_mr->r_kref, __rds_put_mr_final); } +static void rds_message_unpin_worker(struct work_struct *work) +{ + struct rds_message *rm = container_of(work, struct rds_message, + m_unpin_work); + + if (rm->rdma.op_unpin_deferred) + rds_rdma_op_unpin_pages(&rm->rdma); + if (rm->atomic.op_unpin_deferred) + rds_atomic_op_unpin_page(&rm->atomic); + + kfree(rm); +} + void rds_message_put(struct rds_message *rm) { rdsdebug("put rm %p ref %d\n", rm, refcount_read(&rm->m_refcount)); @@ -189,8 +202,21 @@ void rds_message_put(struct rds_message *rm) if (refcount_dec_and_test(&rm->m_refcount)) { BUG_ON(!list_empty(&rm->m_sock_item)); BUG_ON(!list_empty(&rm->m_conn_item)); + rds_message_purge(rm); + /* A final put in atomic context cannot dirty the ops' + * user pages on unpin, so rds_rdma_free_op() and + * rds_atomic_free_op() deferred it. Finish the unpin, + * and the free, from process context. + */ + if (rm->rdma.op_unpin_deferred || + rm->atomic.op_unpin_deferred) { + INIT_WORK(&rm->m_unpin_work, rds_message_unpin_worker); + queue_work(rds_wq, &rm->m_unpin_work); + return; + } + kfree(rm); } } diff --git a/net/rds/rdma.c b/net/rds/rdma.c index 61fb6e45281b..f360a7b3b5fe 100644 --- a/net/rds/rdma.c +++ b/net/rds/rdma.c @@ -483,22 +483,36 @@ void rds_rdma_unuse(struct rds_sock *rs, u32 r_key, int force) kref_put(&mr->r_kref, __rds_put_mr_final); } -void rds_rdma_free_op(struct rm_rdma_op *ro) +void rds_rdma_op_unpin_pages(struct rm_rdma_op *ro) { unsigned int i; + for (i = 0; i < ro->op_nents; i++) { + struct page *page = sg_page(&ro->op_sg[i]); + + /* Mark page dirty if it was possibly modified, which + * is the case for a RDMA_READ which copies from remote + * to local memory + */ + unpin_user_pages_dirty_lock(&page, 1, !ro->op_write); + } +} + +void rds_rdma_free_op(struct rm_rdma_op *ro) +{ if (ro->op_odp_mr) { kref_put(&ro->op_odp_mr->r_kref, __rds_put_mr_final); + } else if (in_task() || ro->op_write) { + /* An RDMA write's pages are only read by the remote + * side; unpinning without dirtying does not sleep. + */ + rds_rdma_op_unpin_pages(ro); } else { - for (i = 0; i < ro->op_nents; i++) { - struct page *page = sg_page(&ro->op_sg[i]); - - /* Mark page dirty if it was possibly modified, which - * is the case for a RDMA_READ which copies from remote - * to local memory - */ - unpin_user_pages_dirty_lock(&page, 1, !ro->op_write); - } + /* Dirtying the pages on unpin can sleep; leave them + * pinned and have rds_message_put() finish the unpin + * from process context. + */ + ro->op_unpin_deferred = 1; } kfree(ro->op_notifier); @@ -507,7 +521,7 @@ void rds_rdma_free_op(struct rm_rdma_op *ro) ro->op_odp_mr = NULL; } -void rds_atomic_free_op(struct rm_atomic_op *ao) +void rds_atomic_op_unpin_page(struct rm_atomic_op *ao) { struct page *page = sg_page(ao->op_sg); @@ -515,6 +529,19 @@ void rds_atomic_free_op(struct rm_atomic_op *ao) * is the case for a RDMA_READ which copies from remote * to local memory */ unpin_user_pages_dirty_lock(&page, 1, true); +} + +void rds_atomic_free_op(struct rm_atomic_op *ao) +{ + if (in_task()) { + rds_atomic_op_unpin_page(ao); + } else { + /* Dirtying the page on unpin can sleep; leave it + * pinned and have rds_message_put() finish the unpin + * from process context. + */ + ao->op_unpin_deferred = 1; + } kfree(ao->op_notifier); ao->op_notifier = NULL; diff --git a/net/rds/rds.h b/net/rds/rds.h index 6e0790e4b570..14bff7440b79 100644 --- a/net/rds/rds.h +++ b/net/rds/rds.h @@ -445,6 +445,12 @@ struct rds_message { void *m_final_op; + /* Unpins the ops' user pages and frees the message from + * process context when the final put happens in atomic + * context: dirtying the pages on unpin can sleep. + */ + struct work_struct m_unpin_work; + struct { struct rm_atomic_op { int op_type; @@ -468,6 +474,7 @@ struct rds_message { unsigned int op_mapped:1; unsigned int op_silent:1; unsigned int op_active:1; + unsigned int op_unpin_deferred:1; struct scatterlist *op_sg; struct rds_notifier *op_notifier; @@ -483,6 +490,7 @@ struct rds_message { unsigned int op_mapped:1; unsigned int op_silent:1; unsigned int op_active:1; + unsigned int op_unpin_deferred:1; unsigned int op_bytes; unsigned int op_nents; unsigned int op_count; @@ -972,6 +980,8 @@ int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm, struct cmsghdr *cmsg); void rds_rdma_free_op(struct rm_rdma_op *ro); void rds_atomic_free_op(struct rm_atomic_op *ao); +void rds_rdma_op_unpin_pages(struct rm_rdma_op *ro); +void rds_atomic_op_unpin_page(struct rm_atomic_op *ao); void rds_rdma_send_complete(struct rds_message *rm, int wc_status); void rds_atomic_send_complete(struct rds_message *rm, int wc_status); int rds_cmsg_atomic(struct rds_sock *rs, struct rds_message *rm,