libfc: Do not invoke the response handler after fc_exch_done()

While the FCoE initiator driver invokes fc_exch_done() from inside
the libfc response handler, FCoE target drivers typically invoke
fc_exch_done() from outside the libfc response handler. The object
fc_exch.arg points at may disappear as soon as fc_exch_done() has
finished. So it's important not to invoke the response handler
function after fc_exch_done() has finished. Modify libfc such that
this guarantee is provided if fc_exch_done() is invoked from
outside a response handler. This patch fixes a sporadic crash in
FCoE target implementations after a command has been aborted.

Signed-off-by: Bart Van Assche <[e-mail]>
Cc: Neil Horman <[e-mail]>
Signed-off-by: Robert Love <[e-mail]>

drivers/scsi/libfc/fc_exch.c

···
 /**
  * fc_exch_done_locked() - Complete an exchange with the exchange lock held
  * @ep: The exchange that is complete
+ *
+ * Note: May sleep if invoked from outside a response handler.
  */
 static int fc_exch_done_locked(struct fc_exch *ep)
 {
···
 	 * ep, and in that case we only clear the resp and set it as
 	 * complete, so it can be reused by the timer to send the rrq.
 	 */
-	ep->resp = NULL;
 	if (ep->state & FC_EX_DONE)
 		return rc;
 	ep->esb_stat |= ESB_ST_COMPLETE;
···
 
 /*
  * Set the response handler for the exchange associated with a sequence.
+ *
+ * Note: May sleep if invoked from outside a response handler.
  */
 static void fc_seq_set_resp(struct fc_seq *sp,
 			    void (*resp)(struct fc_seq *, struct fc_frame *,
···
 			    void *arg)
 {
 	struct fc_exch *ep = fc_seq_exch(sp);
+	DEFINE_WAIT(wait);
 
 	spin_lock_bh(&ep->ex_lock);
+	while (ep->resp_active && ep->resp_task != current) {
+		prepare_to_wait(&ep->resp_wq, &wait, TASK_UNINTERRUPTIBLE);
+		spin_unlock_bh(&ep->ex_lock);
+
+		schedule();
+
+		spin_lock_bh(&ep->ex_lock);
+	}
+	finish_wait(&ep->resp_wq, &wait);
 	ep->resp = resp;
 	ep->arg = arg;
 	spin_unlock_bh(&ep->ex_lock);
···
 }
 
 /**
+ * fc_invoke_resp() - invoke ep->resp()
+ *
+ * Notes:
+ * It is assumed that after initialization finished (this means the
+ * first unlock of ex_lock after fc_exch_alloc()) ep->resp and ep->arg are
+ * modified only via fc_seq_set_resp(). This guarantees that none of these
+ * two variables changes if ep->resp_active > 0.
+ *
+ * If an fc_seq_set_resp() call is busy modifying ep->resp and ep->arg when
+ * this function is invoked, the first spin_lock_bh() call in this function
+ * will wait until fc_seq_set_resp() has finished modifying these variables.
+ *
+ * Since fc_exch_done() invokes fc_seq_set_resp() it is guaranteed that that
+ * ep->resp() won't be invoked after fc_exch_done() has returned.
+ *
+ * The response handler itself may invoke fc_exch_done(), which will clear the
+ * ep->resp pointer.
+ *
+ * Return value:
+ * Returns true if and only if ep->resp has been invoked.
+ */
+static bool fc_invoke_resp(struct fc_exch *ep, struct fc_seq *sp,
+			   struct fc_frame *fp)
+{
+	void (*resp)(struct fc_seq *, struct fc_frame *fp, void *arg);
+	void *arg;
+	bool res = false;
+
+	spin_lock_bh(&ep->ex_lock);
+	ep->resp_active++;
+	if (ep->resp_task != current)
+		ep->resp_task = !ep->resp_task ? current : NULL;
+	resp = ep->resp;
+	arg = ep->arg;
+	spin_unlock_bh(&ep->ex_lock);
+
+	if (resp) {
+		resp(sp, fp, arg);
+		res = true;
+	} else if (!IS_ERR(fp)) {
+		fc_frame_free(fp);
+	}
+
+	spin_lock_bh(&ep->ex_lock);
+	if (--ep->resp_active == 0)
+		ep->resp_task = NULL;
+	spin_unlock_bh(&ep->ex_lock);
+
+	if (ep->resp_active == 0)
+		wake_up(&ep->resp_wq);
+
+	return res;
+}
+
+/**
  * fc_exch_timeout() - Handle exchange timer expiration
  * @work: The work_struct identifying the exchange that timed out
  */
···
 	struct fc_exch *ep = container_of(work, struct fc_exch,
 					  timeout_work.work);
 	struct fc_seq *sp = &ep->seq;
-	void (*resp)(struct fc_seq *, struct fc_frame *fp, void *arg);
-	void *arg;
 	u32 e_stat;
 	int rc = 1;
 
···
 			fc_exch_rrq(ep);
 		goto done;
 	} else {
-		resp = ep->resp;
-		arg = ep->arg;
-		ep->resp = NULL;
 		if (e_stat & ESB_ST_ABNORMAL)
 			rc = fc_exch_done_locked(ep);
 		spin_unlock_bh(&ep->ex_lock);
 		if (!rc)
 			fc_exch_delete(ep);
-		if (resp)
-			resp(sp, ERR_PTR(-FC_EX_TIMEOUT), arg);
+		fc_invoke_resp(ep, sp, ERR_PTR(-FC_EX_TIMEOUT));
+		fc_seq_set_resp(sp, NULL, ep->arg);
 		fc_seq_exch_abort(sp, 2 * ep->r_a_tov);
 		goto done;
 	}
···
 	ep->f_ctl = FC_FC_FIRST_SEQ;	/* next seq is first seq */
 	ep->rxid = FC_XID_UNKNOWN;
 	ep->class = mp->class;
+	ep->resp_active = 0;
+	init_waitqueue_head(&ep->resp_wq);
 	INIT_DELAYED_WORK(&ep->timeout_work, fc_exch_timeout);
 out:
 	return ep;
···
  * fc_exch_done() - Indicate that an exchange/sequence tuple is complete and
  *		    the memory allocated for the related objects may be freed.
  * @sp: The sequence that has completed
+ *
+ * Note: May sleep if invoked from outside a response handler.
  */
 static void fc_exch_done(struct fc_seq *sp)
 {
···
 	spin_lock_bh(&ep->ex_lock);
 	rc = fc_exch_done_locked(ep);
 	spin_unlock_bh(&ep->ex_lock);
+
+	fc_seq_set_resp(sp, NULL, ep->arg);
 	if (!rc)
 		fc_exch_delete(ep);
 }
···
 		 * If new exch resp handler is valid then call that
 		 * first.
 		 */
-		if (ep->resp)
-			ep->resp(sp, fp, ep->arg);
-		else
+		if (!fc_invoke_resp(ep, sp, fp))
 			lport->tt.lport_recv(lport, fp);
 		fc_exch_release(ep);	/* release from lookup */
 	} else {
···
 	struct fc_exch *ep;
 	enum fc_sof sof;
 	u32 f_ctl;
-	void (*resp)(struct fc_seq *, struct fc_frame *fp, void *arg);
-	void *ex_resp_arg;
 	int rc;
 
 	ep = fc_exch_find(mp, ntohs(fh->fh_ox_id));
···
 
 	if (fc_sof_needs_ack(sof))
 		fc_seq_send_ack(sp, fp);
-	resp = ep->resp;
-	ex_resp_arg = ep->arg;
 
 	if (fh->fh_type != FC_TYPE_FCP && fr_eof(fp) == FC_EOF_T &&
 	    (f_ctl & (FC_FC_LAST_SEQ | FC_FC_END_SEQ)) ==
 	    (FC_FC_LAST_SEQ | FC_FC_END_SEQ)) {
 		spin_lock_bh(&ep->ex_lock);
-		resp = ep->resp;
 		rc = fc_exch_done_locked(ep);
 		WARN_ON(fc_seq_exch(sp) != ep);
 		spin_unlock_bh(&ep->ex_lock);
···
 	 * If new exch resp handler is valid then call that
 	 * first.
 	 */
-	if (resp)
-		resp(sp, fp, ex_resp_arg);
-	else
-		fc_frame_free(fp);
+	fc_invoke_resp(ep, sp, fp);
+
 	fc_exch_release(ep);
 	return;
 rel:
···
  */
 static void fc_exch_abts_resp(struct fc_exch *ep, struct fc_frame *fp)
 {
-	void (*resp)(struct fc_seq *, struct fc_frame *fp, void *arg);
-	void *ex_resp_arg;
 	struct fc_frame_header *fh;
 	struct fc_ba_acc *ap;
 	struct fc_seq *sp;
···
 		break;
 	}
 
-	resp = ep->resp;
-	ex_resp_arg = ep->arg;
-
 	/* do we need to do some other checks here. Can we reuse more of
 	 * fc_exch_recv_seq_resp
 	 */
···
 	    ntoh24(fh->fh_f_ctl) & FC_FC_LAST_SEQ)
 		rc = fc_exch_done_locked(ep);
 	spin_unlock_bh(&ep->ex_lock);
+
+	fc_exch_hold(ep);
 	if (!rc)
 		fc_exch_delete(ep);
-
-	if (resp)
-		resp(sp, fp, ex_resp_arg);
-	else
-		fc_frame_free(fp);
-
+	fc_invoke_resp(ep, sp, fp);
 	if (has_rec)
 		fc_exch_timer_set(ep, ep->r_a_tov);
-
+	fc_exch_release(ep);
 }
 
 /**
···
 /**
  * fc_exch_reset() - Reset an exchange
  * @ep: The exchange to be reset
+ *
+ * Note: May sleep if invoked from outside a response handler.
  */
 static void fc_exch_reset(struct fc_exch *ep)
 {
 	struct fc_seq *sp;
-	void (*resp)(struct fc_seq *, struct fc_frame *, void *);
-	void *arg;
 	int rc = 1;
 
 	spin_lock_bh(&ep->ex_lock);
 	fc_exch_abort_locked(ep, 0);
 	ep->state |= FC_EX_RST_CLEANUP;
 	fc_exch_timer_cancel(ep);
-	resp = ep->resp;
-	ep->resp = NULL;
 	if (ep->esb_stat & ESB_ST_REC_QUAL)
 		atomic_dec(&ep->ex_refcnt);	/* drop hold for rec_qual */
 	ep->esb_stat &= ~ESB_ST_REC_QUAL;
-	arg = ep->arg;
 	sp = &ep->seq;
 	rc = fc_exch_done_locked(ep);
 	spin_unlock_bh(&ep->ex_lock);
+
+	fc_exch_hold(ep);
+
 	if (!rc)
 		fc_exch_delete(ep);
 
-	if (resp)
-		resp(sp, ERR_PTR(-FC_EX_CLOSED), arg);
+	fc_invoke_resp(ep, sp, ERR_PTR(-FC_EX_CLOSED));
+	fc_seq_set_resp(sp, NULL, ep->arg);
+	fc_exch_release(ep);
 }
 
 /**

include/scsi/libfc.h

···
  * @fh_type:      The frame type
  * @class:        The class of service
  * @seq:          The sequence in use on this exchange
+ * @resp_active:  Number of tasks that are concurrently executing @resp().
+ * @resp_task:    If @resp_active > 0, either the task executing @resp(), the
+ *                task that has been interrupted to execute the soft-IRQ
+ *                executing @resp() or NULL if more than one task is executing
+ *                @resp concurrently.
+ * @resp_wq:      Waitqueue for the tasks waiting on @resp_active.
  * @resp:         Callback for responses on this exchange
  * @destructor:   Called when destroying the exchange
  * @arg:          Passed as a void pointer to the resp() callback
···
 	u32		    r_a_tov;
 	u32		    f_ctl;
 	struct fc_seq       seq;
+	int		    resp_active;
+	struct task_struct  *resp_task;
+	wait_queue_head_t   resp_wq;
 	void		    (*resp)(struct fc_seq *, struct fc_frame *, void *);
 	void		    *arg;
 	void		    (*destructor)(struct fc_seq *, void *);