1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2002 Cluster File Systems, Inc.
6 * This code is issued under the GNU General Public License.
7 * See the file COPYING in this distribution
9 * by Cluster File Systems, Inc.
12 #define DEBUG_SUBSYSTEM S_LDLM
14 #include <linux/lustre_dlm.h>
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,
22 void *cookie, int cookielen,
25 ldlm_lock_callback callback,
28 struct lustre_handle *lockh)
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;
37 lock = ldlm_lock_create(ns, parent_lock_handle, res_id, type, mode,
40 GOTO(out, rc = -ENOMEM);
41 /* for the local lock, add the reference */
42 ldlm_lock_addref(lock, mode);
43 ldlm_lock2handle(lock, lockh);
45 LDLM_DEBUG(lock, "client-side enqueue START");
48 req = ptlrpc_prep_req(cl, conn, LDLM_ENQUEUE, 1, &size, NULL);
50 GOTO(out, rc = -ENOMEM);
52 } else if (req->rq_reqmsg->buflens[0] != sizeof(*body))
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;
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));
69 /* Continue as normal. */
71 size = sizeof(*reply);
72 req->rq_replen = lustre_msg_size(1, &size);
75 lock->l_connection = conn;
78 rc = ptlrpc_queue_wait(req);
79 /* FIXME: status check here? */
80 rc = ptlrpc_check_status(req, rc);
83 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
84 rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
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;
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);
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
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]);
111 ldlm_lock_change_resource(lock, reply->lock_resource_name);
112 if (lock->l_resource == NULL) {
116 LDLM_DEBUG(lock, "client-side enqueue, new resource");
120 ptlrpc_free_req(req);
122 rc = ldlm_lock_enqueue(lock, cookie, cookielen, flags, callback,
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");
143 int ldlm_cli_callback(struct lustre_handle *lockh, struct ldlm_lock_desc *desc,
144 void *data, __u32 data_len, struct ptlrpc_request **reqp)
146 struct ldlm_lock *lock;
147 struct ldlm_request *body;
148 struct ptlrpc_request *req;
149 struct ptlrpc_client *cl =
150 &lock->l_resource->lr_namespace->ns_rpc_client;
151 int rc = 0, size = sizeof(*body);
154 lock = ldlm_handle2lock(lockh);
158 req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CALLBACK, 1,
161 GOTO(out, rc = -ENOMEM);
163 body = lustre_msg_buf(req->rq_reqmsg, 0);
164 memcpy(&body->lock_handle1, &lock->l_remote_handle,
165 sizeof(body->lock_handle1));
168 CDEBUG(D_NET, "Sending granted AST\n");
169 ldlm_lock2desc(lock, &body->lock_desc);
171 CDEBUG(D_NET, "Sending blocked AST\n");
172 memcpy(&body->lock_desc, desc, sizeof(*desc));
175 LDLM_DEBUG(lock, "server preparing %s AST",
176 desc->l_req_mode == 0 ? "completion" : "blocked");
178 req->rq_replen = lustre_msg_size(0, NULL);
182 rc = ptlrpc_queue_wait(req);
183 rc = ptlrpc_check_status(req, rc);
184 ptlrpc_free_req(req);
195 int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
196 int new_mode, int *flags)
198 struct ldlm_request *body;
199 struct ldlm_reply *reply;
200 struct ldlm_lock *lock;
201 struct ldlm_resource *res;
202 struct ptlrpc_request *req;
203 int rc, size = sizeof(*body);
206 lock = ldlm_handle2lock(lockh);
211 LDLM_DEBUG(lock, "client-side convert");
213 req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CONVERT, 1, &size,
216 GOTO(out, rc = -ENOMEM);
218 body = lustre_msg_buf(req->rq_reqmsg, 0);
219 memcpy(&body->lock_handle1, &lock->l_remote_handle,
220 sizeof(body->lock_handle1));
222 body->lock_desc.l_req_mode = new_mode;
223 body->lock_flags = *flags;
225 size = sizeof(*reply);
226 req->rq_replen = lustre_msg_size(1, &size);
228 rc = ptlrpc_queue_wait(req);
229 rc = ptlrpc_check_status(req, rc);
233 reply = lustre_msg_buf(req->rq_repmsg, 0);
234 res = ldlm_convert(lock, new_mode, &reply->lock_flags);
236 ldlm_reprocess_all(res);
237 if (lock->l_req_mode != lock->l_granted_mode) {
238 /* Go to sleep until the lock is granted. */
239 /* FIXME: or cancelled. */
240 CDEBUG(D_NET, "convert returned a blocked lock, "
241 "going to sleep.\n");
242 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
243 lock->l_granted_mode);
244 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
249 ptlrpc_free_req(req);
253 int ldlm_cli_cancel(struct lustre_handle *lockh)
255 struct ptlrpc_request *req;
256 struct ldlm_lock *lock;
257 struct ldlm_request *body;
258 struct ldlm_resource *res;
259 int rc, size = sizeof(*body);
262 lock = ldlm_handle2lock(lockh);
266 LDLM_DEBUG(lock, "client-side cancel");
267 req = ptlrpc_prep_req(lock->l_client, lock->l_connection, LDLM_CANCEL,
270 GOTO(out, rc = -ENOMEM);
272 body = lustre_msg_buf(req->rq_reqmsg, 0);
273 memcpy(&body->lock_handle1, &lock->l_remote_handle,
274 sizeof(body->lock_handle1));
276 req->rq_replen = lustre_msg_size(0, NULL);
278 rc = ptlrpc_queue_wait(req);
279 rc = ptlrpc_check_status(req, rc);
280 ptlrpc_free_req(req);
284 ldlm_lock_cancel(lock);
285 ldlm_reprocess_all(res);