lustre/ldlm/ldlm_lockd.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 code is issued under the GNU General Public License.
   7  * See the file COPYING in this distribution
   8  *
   9  * by Cluster File Systems, Inc.
  10  */
  11
  12 #define EXPORT_SYMTAB
  13
  14 #include <linux/version.h>
  15 #include <linux/module.h>
  16 #include <linux/slab.h>
  17 #include <asm/unistd.h>
  18
  19 #define DEBUG_SUBSYSTEM S_LDLM
  20
  21 #include <linux/obd_class.h>
  22 #include <linux/lustre_dlm.h>
  23 #include <linux/lustre_net.h>
  24
  25 extern kmem_cache_t *ldlm_resource_slab;
  26 extern kmem_cache_t *ldlm_lock_slab;
  27
  28 static int ldlm_client_callback(struct ldlm_lock *lock, struct ldlm_lock *new,
  29                                 void *data, __u32 data_len)
  30 {
  31         LBUG();
  32         return 0;
  33 }
  34
  35 static int ldlm_enqueue(struct ptlrpc_request *req)
  36 {
  37         struct ldlm_reply *dlm_rep;
  38         struct ldlm_request *dlm_req;
  39         ldlm_error_t err;
  40         int rc, bufsize = sizeof(*dlm_rep);
  41
  42         rc = lustre_pack_msg(1, &bufsize, NULL, &req->rq_replen,
  43                              &req->rq_repbuf);
  44         if (rc) {
  45                 CERROR("out of memory\n");
  46                 req->rq_status = -ENOMEM;
  47                 RETURN(0);
  48         }
  49         req->rq_repmsg = (struct lustre_msg *)req->rq_repbuf;
  50         dlm_rep = lustre_msg_buf(req->rq_repmsg, 0);
  51         dlm_req = lustre_msg_buf(req->rq_reqmsg, 0);
  52
  53         err = ldlm_local_lock_enqueue(req->rq_obd,
  54                                       dlm_req->ns_id,
  55                                       &dlm_req->parent_lock_handle,
  56                                       dlm_req->res_id,
  57                                       dlm_req->res_type,
  58                                       &dlm_req->lock_extent,
  59                                       dlm_req->mode,
  60                                       &dlm_req->flags,
  61                                       ldlm_client_callback,
  62                                       ldlm_client_callback,
  63                                       lustre_msg_buf(req->rq_reqmsg, 1),
  64                                       req->rq_reqmsg->buflens[1],
  65                                       &dlm_rep->lock_handle);
  66         req->rq_status = err;
  67
  68         CERROR("local_lock_enqueue: %d\n", err);
  69
  70         RETURN(0);
  71 }
  72
  73 static int ldlm_handle(struct obd_device *dev, struct ptlrpc_service *svc,
  74                        struct ptlrpc_request *req)
  75 {
  76         int rc;
  77         ENTRY;
  78
  79         rc = lustre_unpack_msg(req->rq_reqbuf, req->rq_reqlen);
  80         req->rq_reqmsg = (struct lustre_msg *)req->rq_reqbuf;
  81         if (rc) {
  82                 CERROR("lustre_ldlm: Invalid request\n");
  83                 GOTO(out, rc);
  84         }
  85
  86         if (req->rq_reqmsg->type != PTL_RPC_REQUEST) {
  87                 CERROR("lustre_ldlm: wrong packet type sent %d\n",
  88                        req->rq_reqmsg->type);
  89                 GOTO(out, rc = -EINVAL);
  90         }
  91
  92         switch (req->rq_reqmsg->opc) {
  93         case LDLM_ENQUEUE:
  94                 CDEBUG(D_INODE, "enqueue\n");
  95                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
  96                 rc = ldlm_enqueue(req);
  97                 break;
  98 #if 0
  99         case LDLM_CONVERT:
 100                 CDEBUG(D_INODE, "convert\n");
 101                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
 102                 rc = ldlm_convert(req);
 103                 break;
 104
 105         case LDLM_CANCEL:
 106                 CDEBUG(D_INODE, "cancel\n");
 107                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CANCEL, 0);
 108                 rc = ldlm_cancel(req);
 109                 break;
 110
 111         case LDLM_CALLBACK:
 112                 CDEBUG(D_INODE, "callback\n");
 113                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CALLBACK, 0);
 114                 rc = ldlm_callback(req);
 115                 break;
 116 #endif
 117
 118         default:
 119                 rc = ptlrpc_error(dev, svc, req);
 120                 RETURN(rc);
 121         }
 122
 123         EXIT;
 124 out:
 125         if (rc) {
 126                 CERROR("no header\n");
 127                 return 0;
 128         }
 129
 130         if( req->rq_status) {
 131                 ptlrpc_error(dev, svc, req);
 132         } else {
 133                 CDEBUG(D_NET, "sending reply\n");
 134                 ptlrpc_reply(dev, svc, req);
 135         }
 136
 137         return 0;
 138 }
 139
 140 static int ldlm_iocontrol(int cmd, struct obd_conn *conn, int len, void *karg,
 141                           void *uarg)
 142 {
 143         struct obd_device *obddev = conn->oc_dev;
 144         struct ptlrpc_client cl;
 145         int err;
 146
 147         ENTRY;
 148
 149         if ( _IOC_TYPE(cmd) != IOC_LDLM_TYPE ||
 150              _IOC_NR(cmd) < IOC_LDLM_MIN_NR  ||
 151              _IOC_NR(cmd) > IOC_LDLM_MAX_NR ) {
 152                 CDEBUG(D_IOCTL, "invalid ioctl ( type %d, nr %d, size %d )\n",
 153                                 _IOC_TYPE(cmd), _IOC_NR(cmd), _IOC_SIZE(cmd));
 154                 EXIT;
 155                 return -EINVAL;
 156         }
 157
 158         ptlrpc_init_client(-1, LDLM_REQUEST_PORTAL, LDLM_REPLY_PORTAL, &cl);
 159         err = ptlrpc_connect_client(-1, "ldlm", &cl);
 160         if (err) {
 161                 CERROR("cannot create client\n");
 162                 RETURN(-EINVAL);
 163         }
 164
 165         switch (cmd) {
 166         case IOC_LDLM_TEST: {
 167                 err = ldlm_test(obddev);
 168                 CERROR("-- done err %d\n", err);
 169                 EXIT;
 170                 break;
 171         }
 172         default:
 173                 err = -EINVAL;
 174                 EXIT;
 175                 break;
 176         }
 177
 178         return err;
 179 }
 180
 181 static int ldlm_setup(struct obd_device *obddev, obd_count len, void *data)
 182 {
 183         struct ldlm_obd *ldlm = &obddev->u.ldlm;
 184         int err;
 185         ENTRY;
 186
 187         INIT_LIST_HEAD(&obddev->u.ldlm.ldlm_namespaces);
 188         obddev->u.ldlm.ldlm_lock = SPIN_LOCK_UNLOCKED;
 189
 190         ldlm->ldlm_service = ptlrpc_init_svc(64 * 1024,
 191                                              LDLM_REQUEST_PORTAL,
 192                                              LDLM_REPLY_PORTAL,
 193                                              "self", ldlm_handle);
 194         if (!ldlm->ldlm_service)
 195                 LBUG();
 196
 197         err = ptlrpc_start_thread(obddev, ldlm->ldlm_service, "lustre_dlm");
 198         if (err)
 199                 CERROR("cannot start thread\n");
 200
 201         MOD_INC_USE_COUNT;
 202         RETURN(0);
 203 }
 204
 205 static int cleanup_resource(struct ldlm_resource *res, struct list_head *q)
 206 {
 207         struct list_head *tmp, *pos;
 208         int rc = 0;
 209
 210         list_for_each_safe(tmp, pos, q) {
 211                 struct ldlm_lock *lock;
 212
 213                 if (rc) {
 214                         /* Res was already cleaned up. */
 215                         LBUG();
 216                 }
 217
 218                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 219
 220                 ldlm_resource_del_lock(lock);
 221                 ldlm_lock_free(lock);
 222                 rc = ldlm_resource_put(res);
 223         }
 224
 225         return rc;
 226 }
 227
 228 static int do_free_namespace(struct ldlm_namespace *ns)
 229 {
 230         struct list_head *tmp, *pos;
 231         int i, rc;
 232
 233         for (i = 0; i < RES_HASH_SIZE; i++) {
 234                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
 235                         struct ldlm_resource *res;
 236                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
 237                         list_del_init(&res->lr_hash);
 238
 239                         rc = cleanup_resource(res, &res->lr_granted);
 240                         if (!rc)
 241                                 rc = cleanup_resource(res, &res->lr_converting);
 242                         if (!rc)
 243                                 rc = cleanup_resource(res, &res->lr_waiting);
 244
 245                         while (rc == 0)
 246                                 rc = ldlm_resource_put(res);
 247                 }
 248         }
 249
 250         return ldlm_namespace_free(ns);
 251 }
 252
 253 static int ldlm_free_all(struct obd_device *obddev)
 254 {
 255         struct list_head *tmp, *pos;
 256         int rc = 0;
 257
 258         ldlm_lock(obddev);
 259
 260         list_for_each_safe(tmp, pos, &obddev->u.ldlm.ldlm_namespaces) {
 261                 struct ldlm_namespace *ns;
 262                 ns = list_entry(tmp, struct ldlm_namespace, ns_link);
 263
 264                 rc |= do_free_namespace(ns);
 265         }
 266
 267         ldlm_unlock(obddev);
 268
 269         return rc;
 270 }
 271
 272 static int ldlm_cleanup(struct obd_device *obddev)
 273 {
 274         struct ldlm_obd *ldlm = &obddev->u.ldlm;
 275         ENTRY;
 276
 277         ptlrpc_stop_thread(ldlm->ldlm_service);
 278         rpc_unregister_service(ldlm->ldlm_service);
 279
 280         if (!list_empty(&ldlm->ldlm_service->srv_reqs)) {
 281                 // XXX reply with errors and clean up
 282                 CERROR("Request list not empty!\n");
 283         }
 284
 285         OBD_FREE(ldlm->ldlm_service, sizeof(*ldlm->ldlm_service));
 286
 287         if (ldlm_free_all(obddev)) {
 288                 CERROR("ldlm_free_all could not complete.\n");
 289                 RETURN(-1);
 290         }
 291
 292         MOD_DEC_USE_COUNT;
 293         RETURN(0);
 294 }
 295
 296 struct obd_ops ldlm_obd_ops = {
 297         o_iocontrol:   ldlm_iocontrol,
 298         o_setup:       ldlm_setup,
 299         o_cleanup:     ldlm_cleanup,
 300         o_connect:     gen_connect,
 301         o_disconnect:  gen_disconnect
 302 };
 303
 304
 305 static int __init ldlm_init(void)
 306 {
 307         int rc = obd_register_type(&ldlm_obd_ops, OBD_LDLM_DEVICENAME);
 308         if (rc != 0)
 309                 return rc;
 310
 311         ldlm_resource_slab = kmem_cache_create("ldlm_resources",
 312                                                sizeof(struct ldlm_resource), 0,
 313                                                SLAB_HWCACHE_ALIGN, NULL, NULL);
 314         if (ldlm_resource_slab == NULL)
 315                 return -ENOMEM;
 316
 317         ldlm_lock_slab = kmem_cache_create("ldlm_locks",
 318                                            sizeof(struct ldlm_lock), 0,
 319                                            SLAB_HWCACHE_ALIGN, NULL, NULL);
 320         if (ldlm_lock_slab == NULL) {
 321                 kmem_cache_destroy(ldlm_resource_slab);
 322                 return -ENOMEM;
 323         }
 324
 325         return 0;
 326 }
 327
 328 static void __exit ldlm_exit(void)
 329 {
 330         obd_unregister_type(OBD_LDLM_DEVICENAME);
 331         kmem_cache_destroy(ldlm_resource_slab);
 332         kmem_cache_destroy(ldlm_lock_slab);
 333 }
 334
 335 MODULE_AUTHOR("Cluster File Systems, Inc. <braam@clusterfs.com>");
 336 MODULE_DESCRIPTION("Lustre Lock Management Module v0.1");
 337 MODULE_LICENSE("GPL");
 338
 339 module_init(ldlm_init);
 340 module_exit(ldlm_exit);