lustre/ptlrpc/connection.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (C) 2002 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  *
  24  */
  25
  26 #define DEBUG_SUBSYSTEM S_RPC
  27 #ifdef __KERNEL__
  28 #include <obd_support.h>
  29 #include <obd_class.h>
  30 #include <lustre_net.h>
  31 #else
  32 #include <liblustre.h>
  33 #endif
  34
  35 #include "ptlrpc_internal.h"
  36 #include <class_hash.h>
  37
  38 static spinlock_t conn_lock;
  39 static struct list_head conn_list;
  40 static struct list_head conn_unused_list;
  41 static struct lustre_class_hash_body *conn_hash_body;
  42 static struct lustre_class_hash_body *conn_unused_hash_body;
  43
  44 extern struct lustre_hash_operations conn_hash_operations;
  45
  46 void ptlrpc_dump_connections(void)
  47 {
  48         struct list_head *tmp;
  49         struct ptlrpc_connection *c;
  50         ENTRY;
  51
  52         list_for_each(tmp, &conn_list) {
  53                 c = list_entry(tmp, struct ptlrpc_connection, c_link);
  54                 CERROR("Connection %p/%s has refcount %d (nid=%s->%s)\n",
  55                        c, c->c_remote_uuid.uuid, atomic_read(&c->c_refcount),
  56                        libcfs_nid2str(c->c_self),
  57                        libcfs_nid2str(c->c_peer.nid));
  58         }
  59         EXIT;
  60 }
  61
  62 struct ptlrpc_connection*
  63 ptlrpc_lookup_conn_locked (lnet_process_id_t peer)
  64 {
  65         struct ptlrpc_connection *c;
  66
  67         c = lustre_hash_get_object_by_key(conn_hash_body, &peer);
  68         if (c != NULL)
  69                 return c;
  70
  71         c = lustre_hash_get_object_by_key(conn_unused_hash_body, &peer);
  72         if (c != NULL)
  73                 return c;
  74
  75         return NULL;
  76 }
  77
  78
  79 struct ptlrpc_connection *ptlrpc_get_connection(lnet_process_id_t peer,
  80                                                 lnet_nid_t self, struct obd_uuid *uuid)
  81 {
  82         struct ptlrpc_connection *c;
  83         struct ptlrpc_connection *c2;
  84         int rc = 0;
  85         ENTRY;
  86
  87         CDEBUG(D_INFO, "self %s peer %s\n",
  88                libcfs_nid2str(self), libcfs_id2str(peer));
  89
  90         spin_lock(&conn_lock);
  91
  92         c = ptlrpc_lookup_conn_locked(peer);
  93
  94         spin_unlock(&conn_lock);
  95
  96         if (c != NULL)
  97                 RETURN (c);
  98
  99         OBD_ALLOC(c, sizeof(*c));
 100         if (c == NULL)
 101                 RETURN (NULL);
 102
 103         atomic_set(&c->c_refcount, 1);
 104         c->c_peer = peer;
 105         c->c_self = self;
 106         if (uuid != NULL)
 107                 obd_str2uuid(&c->c_remote_uuid, uuid->uuid);
 108
 109         spin_lock(&conn_lock);
 110
 111         c2 = ptlrpc_lookup_conn_locked(peer);
 112         if (c2 == NULL) {
 113                 list_add(&c->c_link, &conn_list);
 114                 rc = lustre_hash_additem_unique(conn_hash_body, &peer,
 115                                                 &c->c_hash);
 116                 if (rc != 0) {
 117                         list_del(&c->c_link);
 118                         CERROR("Cannot add connection to conn_hash_body\n");
 119                         goto out_conn;
 120                 }
 121         }
 122
 123 out_conn:
 124
 125         spin_unlock(&conn_lock);
 126
 127         if (c2 == NULL && rc == 0)
 128                 RETURN (c);
 129
 130         if (c != NULL)
 131                 OBD_FREE(c, sizeof(*c));
 132         RETURN (c2);
 133 }
 134
 135 int ptlrpc_put_connection(struct ptlrpc_connection *c)
 136 {
 137         int rc = 0;
 138         ENTRY;
 139
 140         if (c == NULL) {
 141                 CERROR("NULL connection\n");
 142                 RETURN(0);
 143         }
 144
 145         CDEBUG (D_INFO, "connection=%p refcount %d to %s\n",
 146                 c, atomic_read(&c->c_refcount) - 1,
 147                 libcfs_nid2str(c->c_peer.nid));
 148
 149         spin_lock(&conn_lock);
 150         LASSERT(!hlist_unhashed(&c->c_hash));
 151         spin_unlock(&conn_lock);
 152
 153         if (atomic_dec_return(&c->c_refcount) == 1) {
 154
 155                 spin_lock(&conn_lock);
 156
 157                 lustre_hash_delitem(conn_hash_body, &c->c_peer, &c->c_hash);
 158                 list_del(&c->c_link);
 159
 160                 list_add(&c->c_link, &conn_unused_list);
 161                 rc = lustre_hash_additem_unique(conn_unused_hash_body, &c->c_peer,
 162                                                 &c->c_hash);
 163                 if (rc != 0) {
 164                         spin_unlock(&conn_lock);
 165                         CERROR("Cannot hash connection to conn_hash_body\n");
 166                         GOTO(ret, rc);
 167                 }
 168
 169                 spin_unlock(&conn_lock);
 170                 rc = 1;
 171
 172         }
 173
 174         if (atomic_read(&c->c_refcount) < 0)
 175                 CERROR("connection %p refcount %d!\n",
 176                        c, atomic_read(&c->c_refcount));
 177 ret :
 178
 179         RETURN(rc);
 180 }
 181
 182 struct ptlrpc_connection *ptlrpc_connection_addref(struct ptlrpc_connection *c)
 183 {
 184         ENTRY;
 185         atomic_inc(&c->c_refcount);
 186         CDEBUG (D_INFO, "connection=%p refcount %d to %s\n",
 187                 c, atomic_read(&c->c_refcount),
 188                 libcfs_nid2str(c->c_peer.nid));
 189         RETURN(c);
 190 }
 191
 192 int ptlrpc_init_connection(void)
 193 {
 194         int rc = 0;
 195         CFS_INIT_LIST_HEAD(&conn_list);
 196         rc = lustre_hash_init(&conn_hash_body, "CONN_HASH",
 197                               128, &conn_hash_operations);
 198         if (rc)
 199                 GOTO(ret, rc);
 200
 201         CFS_INIT_LIST_HEAD(&conn_unused_list);
 202         rc = lustre_hash_init(&conn_unused_hash_body, "CONN_UNUSED_HASH",
 203                               128, &conn_hash_operations);
 204         if (rc)
 205                 GOTO(ret, rc);
 206
 207         spin_lock_init(&conn_lock);
 208 ret:
 209         if (rc) {
 210                 lustre_hash_exit(&conn_hash_body);
 211                 lustre_hash_exit(&conn_unused_hash_body);
 212         }
 213         RETURN(rc);
 214 }
 215
 216 void ptlrpc_cleanup_connection(void)
 217 {
 218         struct list_head *tmp, *pos;
 219         struct ptlrpc_connection *c;
 220
 221         spin_lock(&conn_lock);
 222
 223         lustre_hash_exit(&conn_unused_hash_body);
 224         list_for_each_safe(tmp, pos, &conn_unused_list) {
 225                 c = list_entry(tmp, struct ptlrpc_connection, c_link);
 226                 list_del(&c->c_link);
 227                 OBD_FREE(c, sizeof(*c));
 228         }
 229
 230         lustre_hash_exit(&conn_hash_body);
 231         list_for_each_safe(tmp, pos, &conn_list) {
 232                 c = list_entry(tmp, struct ptlrpc_connection, c_link);
 233                 CERROR("Connection %p/%s has refcount %d (nid=%s)\n",
 234                        c, c->c_remote_uuid.uuid, atomic_read(&c->c_refcount),
 235                        libcfs_nid2str(c->c_peer.nid));
 236                 list_del(&c->c_link);
 237                 OBD_FREE(c, sizeof(*c));
 238         }
 239         spin_unlock(&conn_lock);
 240 }