[PATCH net-next 1/3] net/rds: don't use unpin_user_pages_dirty_lock() from atomic context

@@ -541,6 +541,13 @@ void rds_ib_exit(void)
 #endif
 	rds_ib_unregister_client();
 	rds_ib_destroy_nodev_conns();
+
+	/* The completion tasklets may have deferred message frees to
+	 * rds_wq, and those can call back into this module through the
+	 * op's MR.  Drain them before the module goes away.
+	 */
+	flush_workqueue(rds_wq);
+
 	rds_ib_sysctl_exit();
 	rds_ib_recv_exit();
 	rds_trans_unregister(&rds_ib_transport);

@@ -182,6 +182,29 @@ 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_purge_worker(struct work_struct *work)
+{
+	struct rds_message *rm = container_of(work, struct rds_message,
+					      m_purge_work);
+
+	rds_message_purge(rm);
+
+	kfree(rm);
+}
+
+/* Purging a message with an RDMA or atomic op whose pages were possibly
+ * written by the remote side must dirty those pages, which can sleep
+ * (and the pages may be file-backed, so dirtying them under a non-irq-safe
+ * filesystem lock must not happen from atomic context at all).
+ */
+static bool rds_message_dirties_pages(struct rds_message *rm)
+{
+	if (rm->rdma.op_active && rm->rdma.op_nents && !rm->rdma.op_write &&
+	    !rm->rdma.op_odp_mr)
+		return true;
+	return rm->atomic.op_active;
+}
+
 void rds_message_put(struct rds_message *rm)
 {
 	rdsdebug("put rm %p ref %d\n", rm, refcount_read(&rm->m_refcount));

@@ -189,6 +212,17 @@ 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));
+
+		/* The final put may come from the IB send completion
+		 * tasklet; page dirtying must be deferred to process
+		 * context then.
+		 */
+		if (!in_task() && rds_message_dirties_pages(rm)) {
+			INIT_WORK(&rm->m_purge_work, rds_message_purge_worker);
+			queue_work(rds_wq, &rm->m_purge_work);
+			return;
+		}
+
 		rds_message_purge(rm);
 
 		kfree(rm);

@@ -495,7 +495,9 @@ void rds_rdma_free_op(struct rm_rdma_op *ro)
 
 			/* Mark page dirty if it was possibly modified, which
 			 * is the case for a RDMA_READ which copies from remote
-			 * to local memory
+			 * to local memory.  Dirtying the page can sleep, so
+			 * rds_message_put() defers the purge of such ops to
+			 * process context.
 			 */
 			unpin_user_pages_dirty_lock(&page, 1, !ro->op_write);
 		}

@@ -513,7 +515,10 @@ void rds_atomic_free_op(struct rm_atomic_op *ao)
 
 	/* Mark page dirty if it was possibly modified, which
 	 * is the case for a RDMA_READ which copies from remote
-	 * to local memory */
+	 * to local memory.  Dirtying the page can sleep, so
+	 * rds_message_put() defers the purge of such ops to
+	 * process context.
+	 */
 	unpin_user_pages_dirty_lock(&page, 1, true);
 
 	kfree(ao->op_notifier);

@@ -445,6 +445,12 @@ struct rds_message {
 
 	void			*m_final_op;
 
+	/* Frees the message from process context when the final put
+	 * happens in atomic context but purging needs to dirty user
+	 * pages, which can sleep.
+	 */
+	struct work_struct	m_purge_work;
+
 	struct {
 		struct rm_atomic_op {
 			int			op_type;