- remove the PTL_RPC_TYPE_* in favour of PTL_RPC_MSG_*

- request->rq_type is used to tell the rpc subsystem which buffers to use, so
  users of ptlrpc set that
- ptlrpc will fill in request->rq_reqmsg->type or repmsg->type, so it's now
  done in exactly one place
- it was impossible to get PTL_RPC_MSG_ERR before, because we clobbered it in
  ptlrpc_reply; fixed.
- move the incoming-request type verification into the service
- add some incoming-reply type verification to ptlrpc_queue_wait.  Just in case.

diff --git a/lustre/include/linux/lustre_idl.h b/lustre/include/linux/lustre_idl.h
index 2ed215e..f975ff1 100644 (file)
--- a/lustre/include/linux/lustre_idl.h
+++ b/lustre/include/linux/lustre_idl.h
@@ -87,11 +87,8 @@ typedef __u8 uuid_t[37];
 
 /* packet types */
 #define PTL_RPC_MSG_REQUEST 4711
-#define PTL_RPC_MSG_ERR 4712
-#define PTL_RPC_MSG_OK           0
-
-#define PTL_RPC_TYPE_REQUEST     2
-#define PTL_RPC_TYPE_REPLY       3
+#define PTL_RPC_MSG_ERR     4712
+#define PTL_RPC_MSG_REPLY   4713
 
 #define PTLRPC_MSG_MAGIC (cpu_to_le32(0x0BD00BD0))
 #define PTLRPC_MSG_VERSION (cpu_to_le32(0x00040001))

diff --git a/lustre/include/linux/lustre_net.h b/lustre/include/linux/lustre_net.h
index 037e555..19c9be9 100644 (file)
--- a/lustre/include/linux/lustre_net.h
+++ b/lustre/include/linux/lustre_net.h
@@ -94,7 +94,7 @@ struct ptlrpc_client {
 #define PTL_RPC_FL_ALLOCREP  (1 << 12) /* reply buffer allocated */
 
 struct ptlrpc_request { 
-        int rq_type; /* one of PTL_RPC_REQUEST, PTL_RPC_REPLY, PTL_RPC_BULK */
+        int rq_type; /* one of PTL_RPC_MSG_* */
         struct list_head rq_list;
         struct list_head rq_multi;
         struct obd_device *rq_obd;

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index b8d2293..0f36c0a 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -466,11 +466,6 @@ static int ldlm_callback_handler(struct ptlrpc_request *req)
                 GOTO(out, rc);
         }
 
-        if (req->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
-                CERROR("lustre_ldlm: wrong packet type sent %d\n",
-                       req->rq_reqmsg->type);
-                GOTO(out, rc = -EINVAL);
-        }
         switch (req->rq_reqmsg->opc) {
         case LDLM_BL_CALLBACK:
                 CDEBUG(D_INODE, "blocking ast\n");

diff --git a/lustre/lov/lov_obd.c b/lustre/lov/lov_obd.c
index 80278ad..3ee0e5b 100644 (file)
--- a/lustre/lov/lov_obd.c
+++ b/lustre/lov/lov_obd.c
@@ -306,7 +306,7 @@ static int lov_getattr(struct lustre_handle *conn, struct obdo *oa,
         ENTRY;
 
         if (!md) {
-                CERROR("LOV requires striping ea for desctruction\n");
+                CERROR("LOV requires striping ea for destruction\n");
                 RETURN(-EINVAL);
         }
 

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index e2a0f76..e5a4724 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -849,12 +849,6 @@ int mds_handle(struct ptlrpc_request *req)
                 GOTO(out, rc);
         }
 
-        if (req->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
-                CERROR("lustre_mds: wrong packet type sent %d\n",
-                       req->rq_reqmsg->type);
-                GOTO(out, rc = -EINVAL);
-        }
-
         if (req->rq_reqmsg->opc != MDS_CONNECT && req->rq_export == NULL)
                 GOTO(out, rc = -ENOTCONN);
 

diff --git a/lustre/ost/ost_handler.c b/lustre/ost/ost_handler.c
index dbe8ca9..d956d72 100644 (file)
--- a/lustre/ost/ost_handler.c
+++ b/lustre/ost/ost_handler.c
@@ -172,7 +172,7 @@ static int ost_punch(struct ptlrpc_request *req)
 
         repbody = lustre_msg_buf(req->rq_repmsg, 0);
         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
-        req->rq_status = obd_punch(conn, &repbody->oa, NULL, 
+        req->rq_status = obd_punch(conn, &repbody->oa, NULL,
                                    repbody->oa.o_blocks, repbody->oa.o_size);
         RETURN(0);
 }
@@ -394,7 +394,8 @@ static int ost_brw_write(struct ptlrpc_request *req)
         ptlrpc_reply(req->rq_svc, req);
 
         lwi = LWI_TIMEOUT(obd_timeout * HZ, ost_bulk_timeout, desc);
-        rc = l_wait_event(desc->bd_waitq, desc->bd_flags & PTL_BULK_FL_RCVD, &lwi);
+        rc = l_wait_event(desc->bd_waitq, desc->bd_flags & PTL_BULK_FL_RCVD,
+                          &lwi);
         if (rc) {
                 if (rc != -ETIMEDOUT)
                         LBUG();
@@ -434,12 +435,6 @@ static int ost_handle(struct ptlrpc_request *req)
                 GOTO(out, rc);
         }
 
-        if (req->rq_reqmsg->type != PTL_RPC_MSG_REQUEST) {
-                CERROR("lustre_ost: wrong packet type sent %d\n",
-                       req->rq_reqmsg->type);
-                GOTO(out, rc = -EINVAL);
-        }
-
         if (req->rq_reqmsg->opc != OST_CONNECT &&
             req->rq_export == NULL) {
                 CERROR("lustre_ost: operation %d on unconnected OST\n",
@@ -603,7 +598,7 @@ static int ost_setup(struct obd_device *obddev, obd_count len, void *buf)
         }
 
         ost->ost_service = ptlrpc_init_svc(64 * 1024, OST_REQUEST_PORTAL,
-                                           OSC_REPLY_PORTAL, "self",ost_handle, 
+                                           OSC_REPLY_PORTAL, "self", ost_handle,
                                            "ost");
         if (!ost->ost_service) {
                 CERROR("failed to start service\n");

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index b4b4596..437cbdc 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -167,7 +167,7 @@ struct ptlrpc_request *ptlrpc_prep_req(struct obd_import *imp, int opcode,
         }
 
         request->rq_level = LUSTRE_CONN_FULL;
-        request->rq_type = PTL_RPC_TYPE_REQUEST;
+        request->rq_type = PTL_RPC_MSG_REQUEST;
         request->rq_import = imp;
         request->rq_connection = ptlrpc_connection_addref(conn);
 
@@ -183,7 +183,6 @@ struct ptlrpc_request *ptlrpc_prep_req(struct obd_import *imp, int opcode,
         request->rq_reqmsg->magic = PTLRPC_MSG_MAGIC; 
         request->rq_reqmsg->version = PTLRPC_MSG_VERSION;
         request->rq_reqmsg->opc = HTON__u32(opcode);
-        request->rq_reqmsg->type = HTON__u32(PTL_RPC_MSG_REQUEST);
 
         ptlrpc_hdl2req(request, &imp->imp_handle);
         RETURN(request);
@@ -524,7 +523,7 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
         // up(&cli->cli_rpc_sem);
         if (req->rq_flags & PTL_RPC_FL_INTR) {
                 if (!(req->rq_flags & PTL_RPC_FL_TIMEOUT))
-                        LBUG(); /* should only be interrupted if we timed out. */
+                        LBUG(); /* should only be interrupted if we timed out */
                 /* Clean up the dangling reply buffers */
                 ptlrpc_abort(req);
                 GOTO(out, rc = -EINTR);
@@ -541,6 +540,14 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
                 CERROR("unpack_rep failed: %d\n", rc);
                 GOTO(out, rc);
         }
+        if (req->rq_repmsg->type != PTL_RPC_MSG_REPLY &&
+            req->rq_repmsg->type != PTL_RPC_MSG_ERR) {
+                CERROR("invalid packet type received (type=%u)\n",
+                       req->rq_repmsg->type);
+                LBUG();
+                GOTO(out, rc = -EINVAL);
+        }
+
         CDEBUG(D_NET, "got rep %Ld\n", req->rq_xid);
         if (req->rq_repmsg->status == 0)
                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repmsg,

diff --git a/lustre/ptlrpc/niobuf.c b/lustre/ptlrpc/niobuf.c
index 7de7922..872e8f5 100644 (file)
--- a/lustre/ptlrpc/niobuf.c
+++ b/lustre/ptlrpc/niobuf.c
@@ -36,34 +36,31 @@ static int ptl_send_buf(struct ptlrpc_request *request,
         int rc;
         ptl_process_id_t remote_id;
         ptl_handle_md_t md_h;
-        ptl_ack_req_t ack;
 
         request->rq_req_md.user_ptr = request;
 
         switch (request->rq_type) {
-        case PTL_RPC_TYPE_REQUEST:
+        case PTL_RPC_MSG_REQUEST:
+                request->rq_reqmsg->type = HTON__u32(request->rq_type);
                 request->rq_req_md.start = request->rq_reqmsg;
                 request->rq_req_md.length = request->rq_reqlen;
                 request->rq_req_md.eventq = request_out_eq;
-                request->rq_req_md.threshold = 1;
-                ack = PTL_NOACK_REQ;
                 break;
-        case PTL_RPC_TYPE_REPLY:
+        case PTL_RPC_MSG_REPLY:
+                request->rq_repmsg->type = HTON__u32(request->rq_type);
                 request->rq_req_md.start = request->rq_repmsg;
                 request->rq_req_md.length = request->rq_replen;
                 request->rq_req_md.eventq = reply_out_eq;
-                request->rq_req_md.threshold = 1;
-                ack = PTL_NOACK_REQ;
                 break;
         default:
                 LBUG();
                 return -1; /* notreached */
         }
+        request->rq_req_md.threshold = 1;
         request->rq_req_md.options = PTL_MD_OP_PUT;
         request->rq_req_md.user_ptr = request;
 
         rc = PtlMDBind(conn->c_peer.peer_ni, request->rq_req_md, &md_h);
-        //CERROR("MDBind (outgoing req/rep/bulk): %Lu\n", (__u64)md_h);
         if (rc != 0) {
                 CERROR("PtlMDBind failed: %d\n", rc);
                 LBUG();
@@ -76,7 +73,7 @@ static int ptl_send_buf(struct ptlrpc_request *request,
         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %Ld\n",
                request->rq_req_md.length, portal, request->rq_xid);
 
-        rc = PtlPut(md_h, ack, remote_id, portal, 0, request->rq_xid,
+        rc = PtlPut(md_h, PTL_NOACK_REQ, remote_id, portal, 0, request->rq_xid,
                     0, 0);
         if (rc != PTL_OK) {
                 CERROR("PtlPut(%Lu, %d, %Ld) failed: %d\n", remote_id.nid,
@@ -91,14 +88,14 @@ static inline struct iovec *
 ptlrpc_get_bulk_iov (struct ptlrpc_bulk_desc *desc)
 {
         struct iovec *iov;
-        
+
         if (desc->bd_page_count <= sizeof (desc->bd_iov)/sizeof (struct iovec))
                 return (desc->bd_iov);
-        
+
         OBD_ALLOC (iov, desc->bd_page_count * sizeof (struct iovec));
         if (iov == NULL)
                 LBUG();
-        
+
         return (iov);
 }
 
@@ -141,16 +138,16 @@ int ptlrpc_send_bulk(struct ptlrpc_bulk_desc *desc)
                 if (desc->bd_md.niov == 0)
                         xid = bulk->bp_xid;
                 LASSERT (xid == bulk->bp_xid);   /* should all be the same */
-                
+
                 iov[desc->bd_md.niov].iov_base = bulk->bp_buf;
                 iov[desc->bd_md.niov].iov_len = bulk->bp_buflen;
                 desc->bd_md.niov++;
                 desc->bd_md.length += bulk->bp_buflen;
         }
-        
+
         LASSERT (desc->bd_md.niov == desc->bd_page_count);
         LASSERT (desc->bd_md.niov != 0);
-        
+
         rc = PtlMDBind(desc->bd_connection->c_peer.peer_ni, desc->bd_md,
                        &desc->bd_md_h);
 
@@ -161,14 +158,14 @@ int ptlrpc_send_bulk(struct ptlrpc_bulk_desc *desc)
                 LBUG();
                 RETURN(rc);
         }
-        
+
         remote_id.nid = desc->bd_connection->c_peer.peer_nid;
         remote_id.pid = 0;
-        
+
         CDEBUG(D_NET, "Sending %u pages %u bytes to portal %d nid %Lx pid %d xid %d\n",
-               desc->bd_md.niov, desc->bd_md.length, 
+               desc->bd_md.niov, desc->bd_md.length,
                desc->bd_portal, remote_id.nid, remote_id.pid, xid);
-        
+
         rc = PtlPut(desc->bd_md_h, PTL_ACK_REQ, remote_id,
                     desc->bd_portal, 0, xid, 0, 0);
         if (rc != PTL_OK) {
@@ -211,7 +208,7 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
                 if (desc->bd_md.niov == 0)
                         xid = bulk->bp_xid;
                 LASSERT (xid == bulk->bp_xid);   /* should all be the same */
-                
+
                 iov[desc->bd_md.niov].iov_base = bulk->bp_buf;
                 iov[desc->bd_md.niov].iov_len = bulk->bp_buflen;
                 desc->bd_md.niov++;
@@ -220,7 +217,7 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
 
         LASSERT (desc->bd_md.niov == desc->bd_page_count);
         LASSERT (desc->bd_md.niov != 0);
-        
+
         rc = PtlMEAttach(desc->bd_connection->c_peer.peer_ni,
                          desc->bd_portal, local_id, xid, 0,
                          PTL_UNLINK, PTL_INS_AFTER, &desc->bd_me_h);
@@ -232,7 +229,7 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
                 LBUG();
                 GOTO(cleanup, rc);
         }
-        
+
         rc = PtlMDAttach(desc->bd_me_h, desc->bd_md, PTL_UNLINK,
                          &desc->bd_md_h);
         if (rc != PTL_OK) {
@@ -240,9 +237,9 @@ int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *desc)
                 LBUG();
                 GOTO(cleanup, rc);
         }
-        
+
         CDEBUG(D_NET, "Setup bulk sink buffers: %u pages %u bytes, xid %u, "
-               "portal %u\n", desc->bd_md.niov, desc->bd_md.length, 
+               "portal %u\n", desc->bd_md.niov, desc->bd_md.length,
                xid, desc->bd_portal);
 
         RETURN(0);
@@ -272,11 +269,11 @@ int ptlrpc_reply(struct ptlrpc_service *svc, struct ptlrpc_request *req)
         }
 
         /* FIXME: we need to increment the count of handled events */
-        req->rq_type = PTL_RPC_TYPE_REPLY;
+        if (req->rq_type != PTL_RPC_MSG_ERR)
+                req->rq_type = PTL_RPC_MSG_REPLY;
         //req->rq_repmsg->conn = req->rq_connection->c_remote_conn;
         //req->rq_repmsg->token = req->rq_connection->c_remote_token;
         req->rq_repmsg->status = HTON__u32(req->rq_status);
-        req->rq_repmsg->type = HTON__u32(req->rq_type);
         return ptl_send_buf(req, req->rq_connection, svc->srv_rep_portal);
 }
 
@@ -300,7 +297,6 @@ int ptlrpc_error(struct ptlrpc_service *svc, struct ptlrpc_request *req)
         RETURN(rc);
 }
 
-
 int ptl_send_rpc(struct ptlrpc_request *request)
 {
         int rc;
@@ -308,7 +304,7 @@ int ptl_send_rpc(struct ptlrpc_request *request)
 
         ENTRY;
 
-        if (NTOH__u32(request->rq_reqmsg->type) != PTL_RPC_MSG_REQUEST) {
+        if (request->rq_type != PTL_RPC_MSG_REQUEST) {
                 CERROR("wrong packet type sent %d\n",
                        NTOH__u32(request->rq_reqmsg->type));
                 LBUG();
@@ -342,7 +338,6 @@ int ptl_send_rpc(struct ptlrpc_request *request)
                 GOTO(cleanup, rc);
         }
 
-        request->rq_type = PTL_RPC_TYPE_REQUEST;
         request->rq_reply_md.start = repbuf;
         request->rq_reply_md.length = request->rq_replen;
         request->rq_reply_md.threshold = 1;
@@ -389,7 +384,7 @@ void ptlrpc_link_svc_me(struct ptlrpc_service *service, int i)
                 CERROR("PtlMEAttach failed: %d\n", rc);
                 LBUG();
         }
-        
+
         if (service->srv_ref_count[i])
                 LBUG();
 
@@ -401,26 +396,26 @@ void ptlrpc_link_svc_me(struct ptlrpc_service *service, int i)
         dummy.user_ptr      = service;
         dummy.eventq        = service->srv_eq_h;
         dummy.max_offset    = service->srv_buf_size;
-        
+
         rc = PtlMDAttach(service->srv_me_h[i], dummy, PTL_UNLINK, &md_h);
         if (rc != PTL_OK) {
                 /* cleanup */
                 CERROR("PtlMDAttach failed: %d\n", rc);
                 LBUG();
         }
-}        
+}
 
 /* ptl_handled_rpc() should be called by the sleeping process once
  * it finishes processing an event.  This ensures the ref count is
  * decremented and that the rpc ring buffer cycles properly.
- */ 
-int ptl_handled_rpc(struct ptlrpc_service *service, void *start) 
+ */
+int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 {
         int index;
 
         spin_lock(&service->srv_lock);
         for (index = 0; index < service->srv_ring_length; index++)
-                if (service->srv_buf[index] == start) 
+                if (service->srv_buf[index] == start)
                         break;
 
         if (index == service->srv_ring_length)
@@ -435,10 +430,10 @@ int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 
         if (service->srv_ref_count[index] == 0 &&
             !ptl_is_valid_handle(&(service->srv_me_h[index]))) {
-                CDEBUG(D_NET, "relinking %d\n", index); 
-                ptlrpc_link_svc_me(service, index); 
+                CDEBUG(D_NET, "relinking %d\n", index);
+                ptlrpc_link_svc_me(service, index);
         }
-        
+
         spin_unlock(&service->srv_lock);
         return 0;
 }

diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index 7fb4407..7bba07c 100644 (file)
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -133,7 +133,6 @@ static int handle_incoming_request(struct obd_device *obddev,
                                    ptl_event_t *event)
 {
         struct ptlrpc_request request;
-        struct lustre_peer peer;
         void *start;
         int rc;
 
@@ -157,6 +156,13 @@ static int handle_incoming_request(struct obd_device *obddev,
                 RETURN(-EINVAL);
         }
 
+        if (NTOH__u32(request.rq_reqmsg->type) != PTL_RPC_MSG_REQUEST) {
+                CERROR("wrong packet type received (type=%u)\n",
+                       request.rq_reqmsg->type);
+                LBUG();
+                RETURN(-EINVAL);
+        }
+
         if (request.rq_reqmsg->magic != PTLRPC_MSG_MAGIC) {
                 CERROR("wrong lustre_msg magic %d: ptl %d from %Lx xid %Ld\n",
                        request.rq_reqmsg->magic, svc->srv_req_portal,