- If a non-CONNECT request is made with an invalid export, we need to

instantiate a temporary lustre_connection with which to send a reply.  This fix
is untested, but Mike says that he can reproduce this scenario easily.
- make ptlrpc_get_connection happier about receiving a NULL uuid
- avoid a possible connection reference leak in target_handle_connect

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index a2e2857..08f285a 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -1,12 +1,22 @@
 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
  * vim:expandtab:shiftwidth=8:tabstop=8:
  *
- * Copyright (C) 2002 Cluster File Systems, Inc.
+ *  Copyright (C) 2002 Cluster File Systems, Inc.
  *
- * This code is issued under the GNU General Public License.
- * See the file COPYING in this distribution
+ *   This file is part of Lustre, http://www.lustre.org.
  *
- * by Cluster File Systems, Inc.
+ *   Lustre is free software; you can redistribute it and/or
+ *   modify it under the terms of version 2 of the GNU General Public
+ *   License as published by the Free Software Foundation.
+ *
+ *   Lustre is distributed in the hope that it will be useful,
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *   GNU General Public License for more details.
+ *
+ *   You should have received a copy of the GNU General Public License
+ *   along with Lustre; if not, write to the Free Software
+ *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
 #define DEBUG_SUBSYSTEM S_LDLM
@@ -59,7 +69,7 @@ static int expired_completion_wait(void *data)
 
 int ldlm_completion_ast(struct ldlm_lock *lock, int flags)
 {
-        struct l_wait_info lwi = 
+        struct l_wait_info lwi =
                 LWI_TIMEOUT_INTR(obd_timeout * HZ, expired_completion_wait,
                                  interrupted_completion_wait, lock);
         int rc = 0;
@@ -81,7 +91,7 @@ int ldlm_completion_ast(struct ldlm_lock *lock, int flags)
                                    "client-side enqueue waking up: failed (%d)",
                                    rc);
                 } else {
-                        LDLM_DEBUG(lock, 
+                        LDLM_DEBUG(lock,
                                    "client-side enqueue waking up: granted");
                 }
         } else if (flags == LDLM_FL_WAIT_NOREPROC) {
@@ -336,6 +346,8 @@ static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
         RETURN(0);
 }
 
+/* FIXME: one of ldlm_cli_convert or the server side should reject attempted
+ * conversion of locks which are on the waiting or converting queue */
 int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
 {
         struct ldlm_request *body;
@@ -422,7 +434,7 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
                 ldlm_cancel_callback(lock);
                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
-                req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh), 
+                req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh),
                                       LDLM_CANCEL, 1, &size, NULL);
                 if (!req)
                         GOTO(out, rc = -ENOMEM);

diff --git a/lustre/lib/l_net.c b/lustre/lib/l_net.c
index 89d5237..4e5b8dc 100644 (file)
--- a/lustre/lib/l_net.c
+++ b/lustre/lib/l_net.c
@@ -282,6 +282,8 @@ int target_handle_connect(struct ptlrpc_request *req)
 
         req->rq_export = export;
         export->exp_connection = ptlrpc_get_connection(&req->rq_peer, cluuid);
+        if (req->rq_connection != NULL)
+                ptlrpc_put_connection(req->rq_connection);
         req->rq_connection = export->exp_connection;
 
         spin_lock(&export->exp_connection->c_lock);

diff --git a/lustre/obdclass/lprocfs.c b/lustre/obdclass/lprocfs.c
index f5486f3..1e65889 100644 (file)
--- a/lustre/obdclass/lprocfs.c
+++ b/lustre/obdclass/lprocfs.c
@@ -3,15 +3,26 @@
  *
  * Copyright (C) 2002 Intel Corporation
  *
+ *   This file is part of Lustre, http://www.lustre.org.
+ *
+ *   Lustre is free software; you can redistribute it and/or
+ *   modify it under the terms of version 2 of the GNU General Public
+ *   License as published by the Free Software Foundation.
+ *
+ *   Lustre is distributed in the hope that it will be useful,
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *   GNU General Public License for more details.
+ *
+ *   You should have received a copy of the GNU General Public License
+ *   along with Lustre; if not, write to the Free Software
+ *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
 /*
  * Author: Hariharan Thantry
  * File Name: lprocfs.c
  *
- * This code is issued under the GNU General Public License.
- * See the file COPYING in this distribution
- *
  * During initialization (of lustre), the following directory materializes
  *          /proc/lustre
  * When the first OBD device of a class is created (due to insmod)
@@ -612,20 +623,17 @@ int lprocfs_get_idx(struct namespace_index* class, const char* dir_name)
         return -1;
 }
 
-
-
-
 struct proc_dir_entry* lprocfs_mkdir(const char* dname,
                                      struct proc_dir_entry *parent)
 {
-       struct proc_dir_entry *child_dir_entry;
+        struct proc_dir_entry *child_dir_entry;
 
-       child_dir_entry = proc_mkdir(dname, parent);
+        child_dir_entry = proc_mkdir(dname, parent);
 
-       if (!child_dir_entry)
+        if (!child_dir_entry)
                 CERROR("lustre: failed to create /proc entry %s\n", dname);
 
-       return child_dir_entry;
+        return child_dir_entry;
 }
 
 /*
@@ -836,7 +844,7 @@ int lprocfs_ll_rd(char *page, char **start, off_t off,
         int len;
         __u64 *temp = (__u64 *)data;
 
-        len = snprintf(page, count, "%Lu\n", *temp);
+        len = snprintf(page, count, LPU64"\n", *temp);
 
         return len;
 }

diff --git a/lustre/ptlrpc/connection.c b/lustre/ptlrpc/connection.c
index ef4dd4f..7a1d47b 100644 (file)
--- a/lustre/ptlrpc/connection.c
+++ b/lustre/ptlrpc/connection.c
@@ -29,6 +29,18 @@ static spinlock_t conn_lock;
 static struct list_head conn_list;
 static struct list_head conn_unused_list;
 
+/* If UUID is NULL, c->c_remote_uuid must be all zeroes
+ * If UUID is non-NULL, c->c_remote_uuid must match. */
+static int match_connection_uuid(struct ptlrpc_connection *c, obd_uuid_t uuid)
+{
+        obd_uuid_t zero_uuid = {0};
+
+        if (uuid)
+                return memcmp(c->c_remote_uuid, uuid, sizeof(uuid));
+
+        return memcmp(c->c_remote_uuid, zero_uuid, sizeof(zero_uuid));
+}
+
 struct ptlrpc_connection *ptlrpc_get_connection(struct lustre_peer *peer,
                                                 obd_uuid_t uuid)
 {
@@ -43,7 +55,7 @@ struct ptlrpc_connection *ptlrpc_get_connection(struct lustre_peer *peer,
         list_for_each(tmp, &conn_list) {
                 c = list_entry(tmp, struct ptlrpc_connection, c_link);
                 if (memcmp(peer, &c->c_peer, sizeof(*peer)) == 0 &&
-                    (!uuid || !memcmp(c->c_remote_uuid, uuid, sizeof(uuid)))) {
+                    !match_connection_uuid(c, uuid)) {
                         ptlrpc_connection_addref(c);
                         GOTO(out, c);
                 }
@@ -52,7 +64,7 @@ struct ptlrpc_connection *ptlrpc_get_connection(struct lustre_peer *peer,
         list_for_each_safe(tmp, pos, &conn_unused_list) {
                 c = list_entry(tmp, struct ptlrpc_connection, c_link);
                 if (memcmp(peer, &c->c_peer, sizeof(*peer)) == 0 &&
-                    (!uuid || strcmp(c->c_remote_uuid, uuid) == 0)) {
+                    !match_connection_uuid(c, uuid)) {
                         ptlrpc_connection_addref(c);
                         list_del(&c->c_link);
                         list_add(&c->c_link, &conn_list);
@@ -72,7 +84,8 @@ struct ptlrpc_connection *ptlrpc_get_connection(struct lustre_peer *peer,
         c->c_generation = 1;
         c->c_epoch = 1;
         c->c_bootcount = 0;
-        strcpy(c->c_remote_uuid, uuid);
+        if (uuid)
+                strcpy(c->c_remote_uuid, uuid);
         INIT_LIST_HEAD(&c->c_delayed_head);
         INIT_LIST_HEAD(&c->c_sending_head);
         INIT_LIST_HEAD(&c->c_dying_head);

diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index cac9932..3bbe3d3 100644 (file)
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -64,8 +64,8 @@ static int ptlrpc_check_event(struct ptlrpc_service *svc,
 }
 
 struct ptlrpc_service *
-ptlrpc_init_svc(__u32 nevents, __u32 nbufs, 
-                __u32 bufsize, __u32 max_req_size, 
+ptlrpc_init_svc(__u32 nevents, __u32 nbufs,
+                __u32 bufsize, __u32 max_req_size,
                 int req_portal, int rep_portal,
                 obd_uuid_t uuid, svc_handler_t handler, char *name)
 {
@@ -115,8 +115,7 @@ ptlrpc_init_svc(__u32 nevents, __u32 nbufs,
                 struct ptlrpc_request_buffer_desc *rqbd;
 
                 OBD_ALLOC (rqbd, sizeof (*rqbd));
-                if (rqbd == NULL)
-                {
+                if (rqbd == NULL) {
                         CERROR ("no memory\n");
                         GOTO (failed, NULL);
                 }
@@ -132,7 +131,7 @@ ptlrpc_init_svc(__u32 nevents, __u32 nbufs,
                 }
                 list_add (&rqbd->rqbd_list, &service->srv_rqbds);
                 service->srv_nrqbds++;
-                
+
                 ptlrpc_link_svc_me(rqbd);
         }
 
@@ -172,7 +171,8 @@ static int handle_incoming_request(struct obd_device *obddev,
         rc = -EINVAL;
 
         if (request->rq_reqlen < sizeof(struct lustre_msg)) {
-                CERROR("incomplete request (%d): ptl %d from "LPX64" xid "LPD64"\n",
+                CERROR("incomplete request (%d): ptl %d from "LPX64" xid "
+                       LPD64"\n",
                        request->rq_reqlen, svc->srv_req_portal,
                        event->initiator.nid, request->rq_xid);
                 goto out;
@@ -185,14 +185,16 @@ static int handle_incoming_request(struct obd_device *obddev,
         }
 
         if (request->rq_reqmsg->magic != PTLRPC_MSG_MAGIC) {
-                CERROR("wrong lustre_msg magic %d: ptl %d from "LPX64" xid "LPD64"\n",
+                CERROR("wrong lustre_msg magic %d: ptl %d from "LPX64" xid "
+                       LPD64"\n",
                        request->rq_reqmsg->magic, svc->srv_req_portal,
                        event->initiator.nid, request->rq_xid);
                 goto out;
         }
 
         if (request->rq_reqmsg->version != PTLRPC_MSG_VERSION) {
-                CERROR("wrong lustre_msg version %d: ptl %d from "LPX64" xid "LPD64"\n",
+                CERROR("wrong lustre_msg version %d: ptl %d from "LPX64" xid "
+                       LPD64"\n",
                        request->rq_reqmsg->version, svc->srv_req_portal,
                        event->initiator.nid, request->rq_xid);
                 goto out;
@@ -211,6 +213,12 @@ static int handle_incoming_request(struct obd_device *obddev,
         if (request->rq_export) {
                 request->rq_connection = request->rq_export->exp_connection;
                 ptlrpc_connection_addref(request->rq_connection);
+        } else {
+                /* create a (hopefully temporary) connection that will be used
+                 * to send the reply if this call doesn't create an export.
+                 * XXX revisit this when we revamp ptlrpc */
+                request->rq_connection =
+                        ptlrpc_get_connection(&request->rq_peer, NULL);
         }
 
         rc = svc->srv_handler(request);
@@ -219,7 +227,7 @@ static int handle_incoming_request(struct obd_device *obddev,
  out:
         if (atomic_dec_and_test (&rqbd->rqbd_refcount)) /* last reference? */
                 ptlrpc_link_svc_me (rqbd);
-        
+
         return rc;
 }
 
@@ -374,10 +382,9 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
          * _after_ unlinking _all_ the request buffers, but _before_
          * freeing them.
          */
-        
-        while (!list_empty (&service->srv_rqbds))
-        {
-                struct ptlrpc_request_buffer_desc *rqbd = 
+
+        while (!list_empty (&service->srv_rqbds)) {
+                struct ptlrpc_request_buffer_desc *rqbd =
                         list_entry (service->srv_rqbds.next,
                                     struct ptlrpc_request_buffer_desc,
                                     rqbd_list);
@@ -402,7 +409,7 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
         }
 
         LASSERT (service->srv_nrqbds == 0);
-        
+
         rc = PtlEQFree(service->srv_eq_h);
         if (rc)
                 CERROR("PtlEQFree failed: %d\n", rc);