From 86528bf0031fc7015bc29113160a572407d19b28 Mon Sep 17 00:00:00 2001
From: pschwan <pschwan>
Date: Wed, 4 Sep 2002 15:18:31 +0000
Subject: [PATCH] - I introduced a bug in the LOV, for the case where you're
 writing less than stripe_count stripes; fixed. - Match bulk and reply buffers
 by source NID, which doesn't hurt and may even fix a bug.

---
 lustre/lov/lov_obd.c   |  9 ++++++++-
 lustre/ptlrpc/events.c | 20 +++++++++++---------
 lustre/ptlrpc/niobuf.c | 12 ++++++++++--
 3 files changed, 29 insertions(+), 12 deletions(-)

diff --git a/lustre/lov/lov_obd.c b/lustre/lov/lov_obd.c
index a17171c..2fc0400 100644
--- a/lustre/lov/lov_obd.c
+++ b/lustre/lov/lov_obd.c
@@ -619,7 +619,14 @@ static inline int lov_brw(int cmd, struct lustre_handle *conn,
         }
 
         cbd->cb = callback;
-        atomic_set(&cbd->refcount, stripe_count);
+
+        /* This is the only race-free way I can think of to get the refcount
+         * correct. -phil */
+        atomic_set(&cbd->refcount, 0);
+        for (i = 0; i < stripe_count; i++)
+                if (stripeinfo[i].bufct)
+                        atomic_inc(&cbd->refcount);
+
         for (i = 0; i < stripe_count; i++) {
                 int shift = stripeinfo[i].index;
                 if (stripeinfo[i].bufct)
diff --git a/lustre/ptlrpc/events.c b/lustre/ptlrpc/events.c
index 1cb5e04..b52e1b0 100644
--- a/lustre/ptlrpc/events.c
+++ b/lustre/ptlrpc/events.c
@@ -87,7 +87,7 @@ static int reply_in_callback(ptl_event_t *ev)
 
         if (req->rq_xid != ev->match_bits) {
                 CERROR("Reply packet for wrong request\n");
-                LBUG(); 
+                LBUG();
         }
 
         if (ev->type == PTL_EVENT_PUT) {
@@ -108,7 +108,7 @@ int request_in_callback(ptl_event_t *ev)
         struct ptlrpc_service *service = ev->mem_desc.user_ptr;
 
         LASSERT ((ev->mem_desc.options & PTL_MD_IOV) == 0); /* requests always contiguous */
-        
+
         if (ev->rlength != ev->mlength)
                 CERROR("Warning: Possibly truncated rpc (%d/%d)\n",
                        ev->mlength, ev->rlength);
@@ -131,15 +131,16 @@ static int bulk_source_callback(ptl_event_t *ev)
 
         /* 1 fragment for each page always */
         LASSERT (ev->mem_desc.niov == desc->bd_page_count);
-        
+
         if (ev->type == PTL_EVENT_SENT) {
                 CDEBUG(D_NET, "got SENT event\n");
         } else if (ev->type == PTL_EVENT_ACK) {
                 CDEBUG(D_NET, "got ACK event\n");
-                
+
                 list_for_each_safe(tmp, next, &desc->bd_page_list) {
-                        bulk = list_entry(tmp, struct ptlrpc_bulk_page, bp_link);
-                        
+                        bulk = list_entry(tmp, struct ptlrpc_bulk_page,
+                                          bp_link);
+
                         if (bulk->bp_cb != NULL)
                                 bulk->bp_cb(bulk);
                 }
@@ -173,16 +174,17 @@ static int bulk_sink_callback(ptl_event_t *ev)
                 LASSERT (ev->mem_desc.niov == desc->bd_page_count);
 
                 list_for_each_safe (tmp, next, &desc->bd_page_list) {
-                        bulk = list_entry(tmp, struct ptlrpc_bulk_page, bp_link);
+                        bulk = list_entry(tmp, struct ptlrpc_bulk_page,
+                                          bp_link);
 
                         total += bulk->bp_buflen;
-                        
+
                         if (bulk->bp_cb != NULL)
                                 bulk->bp_cb(bulk);
                 }
 
                 LASSERT (ev->mem_desc.length == total);
-                
+
                 desc->bd_flags |= PTL_BULK_FL_RCVD;
                 wake_up(&desc->bd_waitq);
                 if (desc->bd_cb != NULL)
diff --git a/lustre/ptlrpc/niobuf.c b/lustre/ptlrpc/niobuf.c
index 1142380..41987e2 100644
--- a/lustre/ptlrpc/niobuf.c
+++ b/lustre/ptlrpc/niobuf.c
@@ -187,6 +187,7 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
         int rc;
         __u32 xid = 0;
         struct iovec *iov;
+        ptl_process_id_t source_id;
         ENTRY;
 
         iov = ptlrpc_get_bulk_iov (desc);
@@ -220,8 +221,11 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
         LASSERT (desc->bd_md.niov == desc->bd_page_count);
         LASSERT (desc->bd_md.niov != 0);
 
+        source_id.nid = desc->bd_connection->c_peer.peer_nid;
+        source_id.pid = PTL_PID_ANY;
+
         rc = PtlMEAttach(desc->bd_connection->c_peer.peer_ni,
-                         desc->bd_portal, local_id, xid, 0,
+                         desc->bd_portal, source_id, xid, 0,
                          PTL_UNLINK, PTL_INS_AFTER, &desc->bd_me_h);
 
         ptlrpc_put_bulk_iov (desc, iov);
@@ -303,6 +307,7 @@ int ptl_send_rpc(struct ptlrpc_request *request)
 {
         int rc;
         char *repbuf;
+        ptl_process_id_t source_id;
 
         ENTRY;
 
@@ -330,9 +335,12 @@ int ptl_send_rpc(struct ptlrpc_request *request)
 
         // down(&request->rq_client->cli_rpc_sem);
 
+        source_id.nid = request->rq_connection->c_peer.peer_nid;
+        source_id.pid = PTL_PID_ANY;
+
         rc = PtlMEAttach(request->rq_connection->c_peer.peer_ni,
                          request->rq_import->imp_client->cli_reply_portal,
-                         local_id, request->rq_xid, 0, PTL_UNLINK,
+                         source_id, request->rq_xid, 0, PTL_UNLINK,
                          PTL_INS_AFTER, &request->rq_reply_me_h);
         if (rc != PTL_OK) {
                 CERROR("PtlMEAttach failed: %d\n", rc);
-- 
1.8.3.1