* PtlMDUnlink() can no longer return PTL_MD_INUSE, since it commits the MD

   for destruction.  If no network I/O is current at the time, a
   PTL_EVENT_UNLINK event is created.

*  The 'unlinked_me' field of an event has been replaced by a simple flag
   'unlinked' that is set if the event signals the destruction of the MD.

*  Events have a new 'status' field.  This is PTL_OK on successful
   completion, and any other portals errno on completion with failure.
   CWARN() messages in these callbacks log abnormal completion.

*  All event callbacks changed to handle the UNLINK event, completion
   status and unlinked flag.

*  All abnormal completions changed to work with PltMDUnlink and the new
   callbacks.

*  Removed bd_complete from ptlrpc_bulk_desc and added bd_success.
   Communications have completed when bd_network_rw gets cleared.   If
   bd_success is set, then bd_nob_transferred tells you how much data
   was sent/received.

*  Changed MDS and OST bulk completion to deal with failed bulk transfers.
   The PtlBD server just LASSERTS things went OK, so we can be reminded to
   implement better error handling there too.

*  ptlrpc_wake_client_req() inline helper.

*  Changed the lib/NAL interface as follows....

   .  cb_callback() is optional and defaults to calling the event queue's
      callback if it is left NULL.

   .  cb_read(), cb_write(), cb_map(), cb_map_pages(), return PTL_OK on
      success and another portals errno on failure.

   .  cb_send(), cb_send_pages(), cb_recv(), cb_recv_pages() return PTL_OK
      if and only if they can commit to calling lib_finalize() when the
      relevent message completes (possibly with error).

   .  cb_send(), cb_send_pages(), cb_recv(), cb_recv_pages() may not modify
      the iovec/ptl_kiov_t they are passed, and must do I/O on the
      subsection of this scatter/gather buffer starting at 'offset' for
      'mlen' bytes.  This greatly simplifies portals lib level descriptor
      management at minimal expense to the NAL.

   .  portals lib now exports lib_extract_iov(), lib_extract_kiov() and the
      other iov helpers take an additional 'offset' parameter, to simplify
      offset buffer coding in the NAL.

   .  lib_parse() is void (i.e. returns no value).

   .  lib_finalize() takes an addition ptl_errno_t completion status.

   ...note that NALs other than qswnal and socknal need to have these
   changes implemented properly and tested.

*  Swapped some loose fprintf()s for CERROR()

*  Dropped PORTAL_SLAB_ALLOC(); portals just uses PORTAL_ALLOC() now.
   Since there are no slabs now, I also changed #ifdef PTL_USE_SLAB_CACHE
   to #ifndef PTL_USE_LIB_FREELIST

*  Changed lib_msg_alloc() so it is _never_ called with the statelock held,
   just like all the other allocators.

*  Changed dynamic MD allocation to size the MD by the number of fragments.

*  Dropped a bunch of dross, plus the iovs from lib_msg_t so they become
   tiny again.

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index 8c40728..066c587 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -1045,65 +1045,56 @@ target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
         LASSERT (rs->rs_export == NULL);
         LASSERT (rs->rs_notified_callback == NULL);
         LASSERT (rs->rs_scheduled == 0 &&
-                 rs->rs_sent == 0 &&
-                 rs->rs_acked == 0 &&
-                 rs->rs_committed == 0 &&
-                 rs->rs_resent == 0);
+                 rs->rs_unlinked == 0 &&
+                 rs->rs_on_net == 0 &&
+                 list_empty(&rs->rs_obd_list) &&
+                 list_empty(&rs->rs_exp_list));
 
         exp = class_export_get (req->rq_export);
         obd = exp->exp_obd;
 
         /* disably reply scheduling onto srv_reply_queue while I'm setting up */
         rs->rs_scheduled = 1;
+        rs->rs_on_net    = 1;
         rs->rs_xid       = req->rq_xid;
         rs->rs_transno   = req->rq_transno;
         rs->rs_export    = exp;
+        
         /* Hack to get past circular module dependency */
         rs->rs_notified_callback = target_notified_reply_callback;
 
-        local_irq_save (flags);
+        spin_lock_irqsave (&obd->obd_uncommitted_replies_lock, flags);
 
-        spin_lock (&obd->obd_uncommitted_replies_lock);
-        if (rs->rs_transno <= obd->obd_last_committed) {
-                /* already committed */
-                /* No contention, so no need to hold srv_lock */
-                rs->rs_committed = 1;
-        } else {
+        if (rs->rs_transno > obd->obd_last_committed) {
+                /* not committed already */ 
                 list_add_tail (&rs->rs_obd_list, 
                                &obd->obd_uncommitted_replies);
         }
-        spin_unlock (&obd->obd_uncommitted_replies_lock);
 
+        spin_unlock (&obd->obd_uncommitted_replies_lock);
         spin_lock (&exp->exp_lock);
+
         list_add_tail (&rs->rs_exp_list, &exp->exp_outstanding_replies);
-        spin_unlock (&exp->exp_lock);
 
-        local_irq_restore (flags);
+        spin_unlock_irqrestore (&exp->exp_lock, flags);
 
         netrc = target_send_reply_msg (req, rc, fail_id);
 
         spin_lock_irqsave (&svc->srv_lock, flags);
         
-        if (netrc != 0) {
-                /* error sending: ptlrpc_handle_server_reply will believe
-                 * the reply is off the net */
-                rs->rs_sent = 1;
-                /* XXX maybe we want to hang on to this reply and retry
-                 * when the client resends? */
-                rs->rs_acked = 1;
-        }
+        if (netrc != 0) /* error sending: reply is off the net */
+                rs->rs_on_net = 0;
 
-        if (!(rs->rs_acked ||                   /* no notifiers */
-              rs->rs_committed ||               /* completed already */
-              rs->rs_resent ||
-              rs->rs_aborted)) {
-                list_add (&rs->rs_list, &sni->sni_active_replies);
-                rs->rs_scheduled = 0;           /* allow notifier to schedule */
-        } else {
+        if (!rs->rs_on_net ||                   /* some notifier */
+            list_empty(&rs->rs_exp_list) ||     /* completed already */
+            list_empty(&rs->rs_obd_list)) {
                 list_add_tail (&rs->rs_list, &svc->srv_reply_queue);
                 wake_up (&svc->srv_waitq);
+        } else {
+                list_add (&rs->rs_list, &sni->sni_active_replies);
+                rs->rs_scheduled = 0;           /* allow notifier to schedule */
         }
-        
+
         spin_unlock_irqrestore (&svc->srv_lock, flags);
 }