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 EXPORT_SYMTAB
  13 #define DEBUG_SUBSYSTEM S_LDLM
  14
  15 #include <linux/lustre_dlm.h>
  16
  17 int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
  18                      struct ldlm_namespace *ns,
  19                      struct ldlm_handle *parent_lock_handle,
  20                      __u64 *res_id,
  21                      __u32 type,
  22                      struct ldlm_extent *req_ex,
  23                      ldlm_mode_t mode,
  24                      int *flags,
  25                      void *data,
  26                      __u32 data_len,
  27                      struct ldlm_handle *lockh)
  28 {
  29         struct ldlm_lock *lock;
  30         struct ldlm_request *body;
  31         struct ldlm_reply *reply;
  32         struct ptlrpc_request *req;
  33         char *bufs[2] = {NULL, data};
  34         int rc, size[2] = {sizeof(*body), data_len};
  35         ENTRY;
  36
  37         *flags = 0;
  38         rc = ldlm_local_lock_create(ns, parent_lock_handle, res_id, type, mode,
  39                                     NULL, 0, lockh);
  40         if (rc != ELDLM_OK)
  41                 GOTO(out, rc);
  42
  43         lock = ldlm_handle2object(lockh);
  44
  45         spin_unlock(&lock->l_lock);
  46         req = ptlrpc_prep_req(cl, conn, LDLM_ENQUEUE, 2, size, bufs);
  47         if (!req)
  48                 GOTO(out, rc = -ENOMEM);
  49
  50         /* Dump all of this data into the request buffer */
  51         body = lustre_msg_buf(req->rq_reqmsg, 0);
  52         body->lock_desc.l_resource.lr_type = type;
  53         memcpy(body->lock_desc.l_resource.lr_name, res_id,
  54                sizeof(body->lock_desc.l_resource.lr_name));
  55
  56         body->lock_desc.l_req_mode = mode;
  57         if (req_ex)
  58                 memcpy(&body->lock_desc.l_extent, req_ex,
  59                        sizeof(body->lock_desc.l_extent));
  60         body->flags = *flags;
  61
  62         memcpy(&body->lock_handle1, lockh, sizeof(body->lock_handle1));
  63
  64         if (parent_lock_handle)
  65                 memcpy(&body->lock_handle2, parent_lock_handle,
  66                        sizeof(body->lock_handle2));
  67
  68         /* Continue as normal. */
  69         size[0] = sizeof(*reply);
  70         req->rq_replen = lustre_msg_size(1, size);
  71
  72         rc = ptlrpc_queue_wait(req);
  73         rc = ptlrpc_check_status(req, rc);
  74         if (rc != ELDLM_OK) {
  75                 spin_lock(&lock->l_resource->lr_lock);
  76                 ldlm_resource_put(lock->l_resource);
  77                 spin_unlock(&lock->l_resource->lr_lock);
  78                 ldlm_lock_free(lock);
  79                 GOTO(out, rc);
  80         }
  81
  82         lock->l_connection = conn;
  83         lock->l_client = cl;
  84         reply = lustre_msg_buf(req->rq_repmsg, 0);
  85         memcpy(&lock->l_remote_handle, &reply->lock_handle,
  86                sizeof(lock->l_remote_handle));
  87         memcpy(req_ex, &reply->lock_extent, sizeof(*req_ex));
  88         *flags = reply->flags;
  89
  90         CDEBUG(D_INFO, "remote handle: %p, flags: %d\n",
  91                (void *)(unsigned long)reply->lock_handle.addr, *flags);
  92         CDEBUG(D_INFO, "extent: %Lu -> %Lu\n",
  93                (unsigned long long)reply->lock_extent.start,
  94                (unsigned long long)reply->lock_extent.end);
  95
  96         ptlrpc_free_req(req);
  97
  98         rc = ldlm_local_lock_enqueue(lockh, req_ex, flags, NULL, NULL);
  99
 100         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 101                       LDLM_FL_BLOCK_CONV)) {
 102                 /* Go to sleep until the lock is granted. */
 103                 /* FIXME: or cancelled. */
 104                 CDEBUG(D_NET, "enqueue returned a blocked lock (%p), "
 105                        "going to sleep.\n", lock);
 106                 ldlm_lock_dump(lock);
 107                 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
 108                                          lock->l_granted_mode);
 109                 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
 110         }
 111         EXIT;
 112  out:
 113         return rc;
 114 }
 115
 116 int ldlm_cli_callback(struct ldlm_lock *lock, struct ldlm_lock *new,
 117                       void *data, __u32 data_len)
 118 {
 119         struct ldlm_request *body;
 120         struct ptlrpc_request *req;
 121         struct obd_device *obddev = lock->l_resource->lr_namespace->ns_obddev;
 122         struct ptlrpc_client *cl = obddev->u.ldlm.ldlm_client;
 123         int rc, size[2] = {sizeof(*body), data_len};
 124         char *bufs[2] = {NULL, data};
 125         ENTRY;
 126
 127         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CALLBACK, 2, size,
 128                               bufs);
 129         if (!req)
 130                 GOTO(out, rc = -ENOMEM);
 131
 132         body = lustre_msg_buf(req->rq_reqmsg, 0);
 133         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 134                sizeof(body->lock_handle1));
 135
 136         if (new == NULL) {
 137                 CDEBUG(D_NET, "Sending granted AST\n");
 138                 ldlm_lock2desc(lock, &body->lock_desc);
 139         } else {
 140                 CDEBUG(D_NET, "Sending blocked AST\n");
 141                 ldlm_lock2desc(new, &body->lock_desc);
 142                 ldlm_object2handle(new, &body->lock_handle2);
 143         }
 144
 145         req->rq_replen = lustre_msg_size(0, NULL);
 146
 147         rc = ptlrpc_queue_wait(req);
 148         rc = ptlrpc_check_status(req, rc);
 149         ptlrpc_free_req(req);
 150
 151         EXIT;
 152  out:
 153         return rc;
 154 }
 155
 156 int ldlm_cli_convert(struct ptlrpc_client *cl, struct ldlm_handle *lockh,
 157                      int new_mode, int *flags)
 158 {
 159         struct ldlm_request *body;
 160         struct ldlm_lock *lock;
 161         struct ldlm_resource *res;
 162         struct ptlrpc_request *req;
 163         int rc, size[2] = {sizeof(*body), 0};
 164         char *bufs[2] = {NULL, NULL};
 165         ENTRY;
 166
 167         lock = ldlm_handle2object(lockh);
 168         *flags = 0;
 169
 170         size[1] = lock->l_data_len;
 171         bufs[1] = lock->l_data;
 172         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CONVERT, 2, size,
 173                               bufs);
 174         if (!req)
 175                 GOTO(out, rc = -ENOMEM);
 176
 177         body = lustre_msg_buf(req->rq_reqmsg, 0);
 178         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 179                sizeof(body->lock_handle1));
 180
 181         body->lock_desc.l_req_mode = new_mode;
 182         body->flags = *flags;
 183
 184         req->rq_replen = lustre_msg_size(1, size);
 185
 186         rc = ptlrpc_queue_wait(req);
 187         rc = ptlrpc_check_status(req, rc);
 188         if (rc != ELDLM_OK)
 189                 GOTO(out, rc);
 190
 191         body = lustre_msg_buf(req->rq_repmsg, 0);
 192         res = ldlm_local_lock_convert(lockh, new_mode, &body->flags);
 193         if (res != NULL)
 194                 ldlm_reprocess_all(res);
 195         if (lock->l_req_mode != lock->l_granted_mode) {
 196                 /* Go to sleep until the lock is granted. */
 197                 /* FIXME: or cancelled. */
 198                 CDEBUG(D_NET, "convert returned a blocked lock, "
 199                        "going to sleep.\n");
 200                 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
 201                                          lock->l_granted_mode);
 202                 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
 203         }
 204         EXIT;
 205  out:
 206         ptlrpc_free_req(req);
 207         return rc;
 208 }
 209
 210 int ldlm_cli_cancel(struct ptlrpc_client *cl, struct ldlm_lock *lock)
 211 {
 212         struct ldlm_request *body;
 213         struct ptlrpc_request *req;
 214         struct ldlm_resource *res;
 215         int rc, size[2] = {sizeof(*body), 0};
 216         char *bufs[2] = {NULL, NULL};
 217         ENTRY;
 218
 219         if (lock->l_data_len == sizeof(struct inode)) {
 220                 /* FIXME: do something better than throwing away everything */
 221                 struct inode *inode = lock->l_data;
 222                 if (inode == NULL)
 223                         LBUG();
 224                 down(&inode->i_sem);
 225                 invalidate_inode_pages(inode);
 226                 up(&inode->i_sem);
 227         }
 228
 229         size[1] = lock->l_data_len;
 230         bufs[1] = lock->l_data;
 231         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CANCEL, 2, size,
 232                               bufs);
 233         if (!req)
 234                 GOTO(out, rc = -ENOMEM);
 235
 236         body = lustre_msg_buf(req->rq_reqmsg, 0);
 237         memcpy(&body->lock_handle1, &lock->l_remote_handle,
 238                sizeof(body->lock_handle1));
 239
 240         req->rq_replen = lustre_msg_size(0, NULL);
 241
 242         rc = ptlrpc_queue_wait(req);
 243         rc = ptlrpc_check_status(req, rc);
 244         ptlrpc_free_req(req);
 245         if (rc != ELDLM_OK)
 246                 GOTO(out, rc);
 247
 248         res = ldlm_local_lock_cancel(lock);
 249         if (res != NULL)
 250                 ldlm_reprocess_all(res);
 251         EXIT;
 252  out:
 253         return rc;
 254 }