lustre/ldlm/ldlm_request.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 DEBUG_SUBSYSTEM S_LDLM
  13
  14 #include <linux/lustre_dlm.h>
  15
  16 int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
  17                      struct ptlrpc_request *req,
  18                      struct ldlm_namespace *ns,
  19                      struct lustre_handle *parent_lock_handle,
  20                      __u64 *res_id,
  21                      __u32 type,
  22                      void *cookie, int cookielen,
  23                      ldlm_mode_t mode,
  24                      int *flags,
  25                      ldlm_lock_callback callback,
  26                      void *data,
  27                      __u32 data_len,
  28                      struct lustre_handle *lockh)
  29 {
  30         struct ldlm_lock *lock;
  31         struct ldlm_request *body;
  32         struct ldlm_reply *reply;
  33         int rc, size = sizeof(*body), req_passed_in = 1;
  34         ENTRY;
  35
  36         *flags = 0;
  37         lock = ldlm_local_lock_create(ns, parent_lock_handle, res_id, type,
  38                                       mode, data, data_len);
  39         if (lock == NULL)
  40                 GOTO(out, rc = -ENOMEM);
  41
  42         ldlm_lock2handle(lock, &lockh);
  43         LDLM_DEBUG(lock, "client-side enqueue START");
  44
  45         if (req == NULL) {
  46                 req = ptlrpc_prep_req(cl, conn, LDLM_ENQUEUE, 1, &size, NULL);
  47                 if (!req)
  48                         GOTO(out, rc = -ENOMEM);
  49                 req_passed_in = 0;
  50         } else if (req->rq_reqmsg->buflens[0] != sizeof(*body))
  51                 LBUG();
  52
  53         /* Dump all of this data into the request buffer */
  54         body = lustre_msg_buf(req->rq_reqmsg, 0);
  55         body->lock_desc.l_resource.lr_type = type;
  56         memcpy(body->lock_desc.l_resource.lr_name, res_id,
  57                sizeof(body->lock_desc.l_resource.lr_name));
  58
  59         body->lock_desc.l_req_mode = mode;
  60         if (type == LDLM_EXTENT)
  61                 memcpy(&body->lock_desc.l_extent, cookie,
  62                        sizeof(body->lock_desc.l_extent));
  63         body->lock_flags = *flags;
  64
  65         memcpy(&body->lock_handle1, lockh, sizeof(body->lock_handle1));
  66
  67         if (parent_lock_handle)
  68                 memcpy(&body->lock_handle2, parent_lock_handle,
  69                        sizeof(body->lock_handle2));
  70
  71         /* Continue as normal. */
  72         if (!req_passed_in) {
  73                 size = sizeof(*reply);
  74                 req->rq_replen = lustre_msg_size(1, &size);
  75         }
  76
  77         lock->l_connection = conn;
  78         lock->l_client = cl;
  79
  80         rc = ptlrpc_queue_wait(req);
  81         rc = ptlrpc_check_status(req, rc);
  82
  83         spin_lock(&lock->l_lock);
  84         if (rc != ELDLM_OK) {
  85                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
  86                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
  87                 spin_lock(&lock->l_resource->lr_lock);
  88                 if (!ldlm_resource_put(lock->l_resource))
  89                         spin_unlock(&lock->l_resource->lr_lock);
  90                 ldlm_lock_free(lock);
  91                 GOTO(out, rc);
  92         }
  93
  94         reply = lustre_msg_buf(req->rq_repmsg, 0);
  95         memcpy(&lock->l_remote_handle, &reply->lock_handle,
  96                sizeof(lock->l_remote_handle));
  97         if (type == LDLM_EXTENT)
  98                 memcpy(cookie, &reply->lock_extent, sizeof(reply->lock_extent));
  99         *flags = reply->lock_flags;
 100
 101         CDEBUG(D_INFO, "remote handle: %p, flags: %d\n",
 102                (void *)(unsigned long)reply->lock_handle.addr, *flags);
 103         CDEBUG(D_INFO, "extent: %Lu -> %Lu\n",
 104                (unsigned long long)reply->lock_extent.start,
 105                (unsigned long long)reply->lock_extent.end);
 106
 107         /* If enqueue returned a blocked lock but the completion handler has
 108          * already run, then it fixed up the resource and we don't need to do it
 109          * again. */
 110         if (*flags & LDLM_FL_LOCK_CHANGED &&
 111             lock->l_req_mode != lock->l_granted_mode) {
 112                 CDEBUG(D_INFO, "remote intent success, locking %ld instead of"
 113                        "%ld\n", (long)reply->lock_resource_name[0],
 114                        (long)lock->l_resource->lr_name[0]);
 115                 spin_lock(&lock->l_resource->lr_lock);
 116                 if (!ldlm_resource_put(lock->l_resource))
 117                         spin_unlock(&lock->l_resource->lr_lock);
 118
 119                 lock->l_resource =
 120                         ldlm_resource_get(ns, NULL, reply->lock_resource_name,
 121                                           type, 1);
 122                 if (lock->l_resource == NULL) {
 123                         LBUG();
 124                         RETURN(-ENOMEM);
 125                 }
 126                 LDLM_DEBUG(lock, "client-side enqueue, new resource");
 127         }
 128
 129         if (!req_passed_in)
 130                 ptlrpc_free_req(req);
 131
 132         spin_unlock(&lock->l_lock);
 133         rc = ldlm_local_lock_enqueue(lockh, cookie, cookielen, flags, callback,
 134                                      callback);
 135
 136         LDLM_DEBUG(lock, "client-side enqueue END");
 137         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 138                       LDLM_FL_BLOCK_CONV)) {
 139                 /* Go to sleep until the lock is granted. */
 140                 /* FIXME: or cancelled. */
 141                 CDEBUG(D_NET, "enqueue returned a blocked lock (%p), "
 142                        "going to sleep.\n", lock);
 143                 ldlm_lock_dump(lock);
 144                 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
 145                                          lock->l_granted_mode);
 146                 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
 147         }
 148         EXIT;
 149  out:
 150         return rc;
 151 }
 152
 153 int ldlm_cli_callback(struct ldlm_lock *lock, struct ldlm_lock *new,
 154                       void *data, __u32 data_len, struct ptlrpc_request **reqp)
 155 {
 156         struct ldlm_request *body;
 157         struct ptlrpc_request *req;
 158         struct ptlrpc_client *cl =
 159                 &lock->l_resource->lr_namespace->ns_rpc_client;
 160         int rc = 0, size = sizeof(*body);
 161         ENTRY;
 162
 163         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CALLBACK, 1,
 164                               &size, NULL);
 165         if (!req)
 166                 GOTO(out, rc = -ENOMEM);
 167
 168         body = lustre_msg_buf(req->rq_reqmsg, 0);
 169         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 170                sizeof(body->lock_handle1));
 171
 172         if (new == NULL) {
 173                 CDEBUG(D_NET, "Sending granted AST\n");
 174                 ldlm_lock2desc(lock, &body->lock_desc);
 175         } else {
 176                 CDEBUG(D_NET, "Sending blocked AST\n");
 177                 ldlm_lock2desc(new, &body->lock_desc);
 178                 ldlm_object2handle(new, &body->lock_handle2);
 179         }
 180
 181         LDLM_DEBUG(lock, "server preparing %s AST",
 182                    new == NULL ? "completion" : "blocked");
 183
 184         req->rq_replen = lustre_msg_size(0, NULL);
 185
 186         if (reqp == NULL) {
 187                 LBUG();
 188                 rc = ptlrpc_queue_wait(req);
 189                 rc = ptlrpc_check_status(req, rc);
 190                 ptlrpc_free_req(req);
 191         } else
 192                 *reqp = req;
 193
 194         EXIT;
 195  out:
 196         return rc;
 197 }
 198
 199 int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
 200                      int new_mode, int *flags)
 201 {
 202         struct ldlm_request *body;
 203         struct ldlm_reply *reply;
 204         struct ldlm_lock *lock;
 205         struct ldlm_resource *res;
 206         struct ptlrpc_request *req;
 207         int rc, size = sizeof(*body);
 208         ENTRY;
 209
 210         lock = lustre_handle2object(lockh);
 211         *flags = 0;
 212
 213         LDLM_DEBUG(lock, "client-side convert");
 214
 215         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CONVERT, 1, &size,
 216                               NULL);
 217         if (!req)
 218                 GOTO(out, rc = -ENOMEM);
 219
 220         body = lustre_msg_buf(req->rq_reqmsg, 0);
 221         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 222                sizeof(body->lock_handle1));
 223
 224         body->lock_desc.l_req_mode = new_mode;
 225         body->lock_flags = *flags;
 226
 227         size = sizeof(*reply);
 228         req->rq_replen = lustre_msg_size(1, &size);
 229
 230         rc = ptlrpc_queue_wait(req);
 231         rc = ptlrpc_check_status(req, rc);
 232         if (rc != ELDLM_OK)
 233                 GOTO(out, rc);
 234
 235         reply = lustre_msg_buf(req->rq_repmsg, 0);
 236         res = ldlm_local_lock_convert(lockh, new_mode, &reply->lock_flags);
 237         if (res != NULL)
 238                 ldlm_reprocess_all(res);
 239         if (lock->l_req_mode != lock->l_granted_mode) {
 240                 /* Go to sleep until the lock is granted. */
 241                 /* FIXME: or cancelled. */
 242                 CDEBUG(D_NET, "convert returned a blocked lock, "
 243                        "going to sleep.\n");
 244                 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
 245                                          lock->l_granted_mode);
 246                 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
 247         }
 248         EXIT;
 249  out:
 250         ptlrpc_free_req(req);
 251         return rc;
 252 }
 253
 254 int ldlm_cli_cancel(struct ptlrpc_client *cl, struct ldlm_lock *lock)
 255 {
 256         struct ptlrpc_request *req;
 257         struct ldlm_request *body;
 258         struct ldlm_resource *res;
 259         int rc, size = sizeof(*body);
 260         ENTRY;
 261
 262         LDLM_DEBUG(lock, "client-side cancel");
 263         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CANCEL, 1, &size,
 264                               NULL);
 265         if (!req)
 266                 GOTO(out, rc = -ENOMEM);
 267
 268         body = lustre_msg_buf(req->rq_reqmsg, 0);
 269         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 270                sizeof(body->lock_handle1));
 271
 272         req->rq_replen = lustre_msg_size(0, NULL);
 273
 274         rc = ptlrpc_queue_wait(req);
 275         rc = ptlrpc_check_status(req, rc);
 276         ptlrpc_free_req(req);
 277         if (rc != ELDLM_OK)
 278                 GOTO(out, rc);
 279
 280         res = ldlm_local_lock_cancel(lock);
 281         if (res != NULL)
 282                 ldlm_reprocess_all(res);
 283         else
 284                 rc = ELDLM_RESOURCE_FREED;
 285         EXIT;
 286  out:
 287         return rc;
 288 }