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>
15 #include <linux/obd.h>
17 int ldlm_completion_ast(struct ldlm_lock *lock, int flags)
21 if (flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
22 LDLM_FL_BLOCK_CONV)) {
23 /* Go to sleep until the lock is granted. */
24 /* FIXME: or cancelled. */
25 LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock,"
28 ldlm_reprocess_all(lock->l_resource);
29 wait_event(lock->l_waitq, (lock->l_req_mode ==
30 lock->l_granted_mode));
31 LDLM_DEBUG(lock, "client-side enqueue waking up: granted");
32 } else if (flags == LDLM_FL_WAIT_NOREPROC) {
33 wait_event(lock->l_waitq, (lock->l_req_mode ==
34 lock->l_granted_mode));
35 } else if (flags == 0) {
36 wake_up(&lock->l_waitq);
42 static int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
43 struct lustre_handle *parent_lockh,
46 void *cookie, int cookielen,
49 ldlm_completion_callback completion,
50 ldlm_blocking_callback blocking,
53 struct lustre_handle *lockh)
55 struct ldlm_lock *lock;
59 CERROR("Trying to cancel local lock\n");
63 lock = ldlm_lock_create(ns, parent_lockh, res_id, type, mode, NULL, 0);
65 GOTO(out_nolock, err = -ENOMEM);
66 LDLM_DEBUG(lock, "client-side local enqueue handler, new lock created");
68 ldlm_lock_addref_internal(lock, mode);
69 ldlm_lock2handle(lock, lockh);
72 err = ldlm_lock_enqueue(lock, cookie, cookielen, flags, completion,
77 if (type == LDLM_EXTENT)
78 memcpy(cookie, &lock->l_extent, sizeof(lock->l_extent));
79 if ((*flags) & LDLM_FL_LOCK_CHANGED)
80 memcpy(res_id, lock->l_resource->lr_name, sizeof(*res_id));
82 LDLM_DEBUG_NOLOCK("client-side local enqueue handler END (lock %p)",
85 if (lock->l_completion_ast)
86 lock->l_completion_ast(lock, *flags);
88 LDLM_DEBUG(lock, "client-side local enqueue END");
96 int ldlm_cli_enqueue(struct lustre_handle *connh,
97 struct ptlrpc_request *req,
98 struct ldlm_namespace *ns,
99 struct lustre_handle *parent_lock_handle,
102 void *cookie, int cookielen,
105 ldlm_completion_callback completion,
106 ldlm_blocking_callback blocking,
109 struct lustre_handle *lockh)
111 struct ldlm_lock *lock;
112 struct ldlm_request *body;
113 struct ldlm_reply *reply;
114 int rc, size = sizeof(*body), req_passed_in = 1;
118 return ldlm_cli_enqueue_local(ns, parent_lock_handle, res_id,
119 type, cookie, cookielen, mode,
120 flags, completion, blocking, data,
124 lock = ldlm_lock_create(ns, parent_lock_handle, res_id, type, mode,
127 GOTO(out_nolock, rc = -ENOMEM);
128 LDLM_DEBUG(lock, "client-side enqueue START");
129 /* for the local lock, add the reference */
130 ldlm_lock_addref_internal(lock, mode);
131 ldlm_lock2handle(lock, lockh);
134 req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_ENQUEUE, 1,
137 GOTO(out, rc = -ENOMEM);
139 } else if (req->rq_reqmsg->buflens[0] != sizeof(*body))
142 /* Dump all of this data into the request buffer */
143 body = lustre_msg_buf(req->rq_reqmsg, 0);
144 ldlm_lock2desc(lock, &body->lock_desc);
145 /* Phil: make this part of ldlm_lock2desc */
146 if (type == LDLM_EXTENT)
147 memcpy(&body->lock_desc.l_extent, cookie,
148 sizeof(body->lock_desc.l_extent));
149 body->lock_flags = *flags;
151 memcpy(&body->lock_handle1, lockh, sizeof(*lockh));
152 if (parent_lock_handle)
153 memcpy(&body->lock_handle2, parent_lock_handle,
154 sizeof(body->lock_handle2));
156 /* Continue as normal. */
157 if (!req_passed_in) {
158 size = sizeof(*reply);
159 req->rq_replen = lustre_msg_size(1, &size);
161 lock->l_connh = connh;
162 lock->l_export = NULL;
164 rc = ptlrpc_queue_wait(req);
165 /* FIXME: status check here? */
166 rc = ptlrpc_check_status(req, rc);
168 if (rc != ELDLM_OK) {
169 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
170 rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
171 ldlm_lock_decref(lockh, mode);
172 /* FIXME: if we've already received a completion AST, this will
174 ldlm_lock_destroy(lock);
178 reply = lustre_msg_buf(req->rq_repmsg, 0);
179 memcpy(&lock->l_remote_handle, &reply->lock_handle,
180 sizeof(lock->l_remote_handle));
181 if (type == LDLM_EXTENT)
182 memcpy(cookie, &reply->lock_extent, sizeof(reply->lock_extent));
183 *flags = reply->lock_flags;
185 CDEBUG(D_INFO, "remote handle: %p, flags: %d\n",
186 (void *)(unsigned long)reply->lock_handle.addr, *flags);
187 CDEBUG(D_INFO, "extent: %Lu -> %Lu\n",
188 (unsigned long long)reply->lock_extent.start,
189 (unsigned long long)reply->lock_extent.end);
191 /* If enqueue returned a blocked lock but the completion handler has
192 * already run, then it fixed up the resource and we don't need to do it
194 if ((*flags) & LDLM_FL_LOCK_CHANGED) {
195 int newmode = reply->lock_mode;
196 if (newmode && newmode != lock->l_req_mode) {
197 LDLM_DEBUG(lock, "server returned different mode %s",
198 ldlm_lockname[newmode]);
199 lock->l_req_mode = newmode;
202 if (reply->lock_resource_name[0] !=
203 lock->l_resource->lr_name[0]) {
204 CDEBUG(D_INFO, "remote intent success, locking %ld "
206 (long)reply->lock_resource_name[0],
207 (long)lock->l_resource->lr_name[0]);
209 ldlm_lock_change_resource(lock,
210 reply->lock_resource_name);
211 if (lock->l_resource == NULL) {
215 LDLM_DEBUG(lock, "client-side enqueue, new resource");
220 ptlrpc_free_req(req);
222 rc = ldlm_lock_enqueue(lock, cookie, cookielen, flags, completion,
224 if (lock->l_completion_ast)
225 lock->l_completion_ast(lock, *flags);
227 LDLM_DEBUG(lock, "client-side enqueue END");
235 int ldlm_match_or_enqueue(struct lustre_handle *connh,
236 struct ptlrpc_request *req,
237 struct ldlm_namespace *ns,
238 struct lustre_handle *parent_lock_handle,
241 void *cookie, int cookielen,
244 ldlm_completion_callback completion,
245 ldlm_blocking_callback blocking,
248 struct lustre_handle *lockh)
252 rc = ldlm_lock_match(ns, res_id, type, cookie, cookielen, mode, lockh);
254 rc = ldlm_cli_enqueue(connh, req, ns,
255 parent_lock_handle, res_id, type, cookie,
256 cookielen, mode, flags, completion,
257 blocking, data, data_len, lockh);
259 CERROR("ldlm_cli_enqueue: err: %d\n", rc);
265 static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
269 if (lock->l_resource->lr_namespace->ns_client) {
270 CERROR("Trying to cancel local lock\n");
273 LDLM_DEBUG(lock, "client-side local convert");
275 ldlm_lock_convert(lock, new_mode, flags);
276 ldlm_reprocess_all(lock->l_resource);
278 LDLM_DEBUG(lock, "client-side local convert handler END");
283 int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
285 struct ldlm_request *body;
286 struct lustre_handle *connh;
287 struct ldlm_reply *reply;
288 struct ldlm_lock *lock;
289 struct ldlm_resource *res;
290 struct ptlrpc_request *req;
291 int rc, size = sizeof(*body);
294 lock = ldlm_handle2lock(lockh);
300 connh = lock->l_connh;
303 return ldlm_cli_convert_local(lock, new_mode, flags);
305 LDLM_DEBUG(lock, "client-side convert");
307 req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_CONVERT, 1, &size,
310 GOTO(out, rc = -ENOMEM);
312 body = lustre_msg_buf(req->rq_reqmsg, 0);
313 memcpy(&body->lock_handle1, &lock->l_remote_handle,
314 sizeof(body->lock_handle1));
316 body->lock_desc.l_req_mode = new_mode;
317 body->lock_flags = *flags;
319 size = sizeof(*reply);
320 req->rq_replen = lustre_msg_size(1, &size);
322 rc = ptlrpc_queue_wait(req);
323 rc = ptlrpc_check_status(req, rc);
327 reply = lustre_msg_buf(req->rq_repmsg, 0);
328 res = ldlm_lock_convert(lock, new_mode, &reply->lock_flags);
330 ldlm_reprocess_all(res);
331 /* Go to sleep until the lock is granted. */
332 /* FIXME: or cancelled. */
333 if (lock->l_completion_ast)
334 lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC);
338 ptlrpc_free_req(req);
342 int ldlm_cli_cancel(struct lustre_handle *lockh)
344 struct ptlrpc_request *req;
345 struct ldlm_lock *lock;
346 struct ldlm_request *body;
347 int rc = 0, size = sizeof(*body);
350 lock = ldlm_handle2lock(lockh);
352 /* It's possible that the decref that we did just before this
353 * cancel was the last reader/writer, and caused a cancel before
354 * we could call this function. If we want to make this
355 * impossible (by adding a dec_and_cancel() or similar), then
356 * we can put the LBUG back. */
362 LDLM_DEBUG(lock, "client-side cancel");
363 /* Set this flag to prevent others from getting new references*/
364 l_lock(&lock->l_resource->lr_namespace->ns_lock);
365 lock->l_flags |= LDLM_FL_CBPENDING;
366 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
368 req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh),
369 LDLM_CANCEL, 1, &size, NULL);
371 GOTO(out, rc = -ENOMEM);
373 body = lustre_msg_buf(req->rq_reqmsg, 0);
374 memcpy(&body->lock_handle1, &lock->l_remote_handle,
375 sizeof(body->lock_handle1));
377 req->rq_replen = lustre_msg_size(0, NULL);
379 rc = ptlrpc_queue_wait(req);
380 rc = ptlrpc_check_status(req, rc);
381 ptlrpc_free_req(req);
385 ldlm_lock_cancel(lock);
387 LDLM_DEBUG(lock, "client-side local cancel");
388 if (lock->l_resource->lr_namespace->ns_client) {
389 CERROR("Trying to cancel local lock\n");
392 ldlm_lock_cancel(lock);
393 ldlm_reprocess_all(lock->l_resource);
394 LDLM_DEBUG(lock, "client-side local cancel handler END");
403 /* Cancel all locks on a given resource that have 0 readers/writers */
404 int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id)
406 struct ldlm_resource *res;
407 struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
408 struct ldlm_ast_work *w;
411 res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
415 l_lock(&ns->ns_lock);
416 list_for_each(tmp, &res->lr_granted) {
417 struct ldlm_lock *lock;
418 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
420 if (lock->l_readers || lock->l_writers)
423 /* Setting the CBPENDING flag is a little misleading, but
424 * prevents an important race; namely, once CBPENDING is set,
425 * the lock can accumulate no more readers/writers. Since
426 * readers and writers are already zero here, ldlm_lock_decref
427 * won't see this flag and call l_blocking_ast */
428 lock->l_flags |= LDLM_FL_CBPENDING;
430 OBD_ALLOC(w, sizeof(*w));
433 w->w_lock = LDLM_LOCK_GET(lock);
434 list_add(&w->w_list, &list);
436 l_unlock(&ns->ns_lock);
438 list_for_each_safe(tmp, next, &list) {
439 struct lustre_handle lockh;
441 w = list_entry(tmp, struct ldlm_ast_work, w_list);
443 ldlm_lock2handle(w->w_lock, &lockh);
444 rc = ldlm_cli_cancel(&lockh);
446 CERROR("ldlm_cli_cancel: %d\n", rc);
448 LDLM_LOCK_PUT(w->w_lock);
449 list_del(&w->w_list);
450 OBD_FREE(w, sizeof(*w));