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