1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28#include <linux/module.h>
29#include <linux/fs.h>
30#include <linux/types.h>
31#include <linux/slab.h>
32#include <linux/highmem.h>
33#include <linux/init.h>
34#include <linux/sysctl.h>
35#include <linux/random.h>
36#include <linux/blkdev.h>
37#include <linux/socket.h>
38#include <linux/inet.h>
39#include <linux/spinlock.h>
40
41
42#include "cluster/heartbeat.h"
43#include "cluster/nodemanager.h"
44#include "cluster/tcp.h"
45
46#include "dlmapi.h"
47#include "dlmcommon.h"
48
49#include "dlmconvert.h"
50
51#define MLOG_MASK_PREFIX ML_DLM
52#include "cluster/masklog.h"
53
54
55
56
57
58
59static enum dlm_status __dlmconvert_master(struct dlm_ctxt *dlm,
60 struct dlm_lock_resource *res,
61 struct dlm_lock *lock, int flags,
62 int type, int *call_ast,
63 int *kick_thread);
64static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm,
65 struct dlm_lock_resource *res,
66 struct dlm_lock *lock, int flags, int type);
67
68
69
70
71
72
73
74
75
76
77enum dlm_status dlmconvert_master(struct dlm_ctxt *dlm,
78 struct dlm_lock_resource *res,
79 struct dlm_lock *lock, int flags, int type)
80{
81 int call_ast = 0, kick_thread = 0;
82 enum dlm_status status;
83
84 spin_lock(&res->spinlock);
85
86 __dlm_wait_on_lockres(res);
87 __dlm_lockres_reserve_ast(res);
88 res->state |= DLM_LOCK_RES_IN_PROGRESS;
89
90 status = __dlmconvert_master(dlm, res, lock, flags, type,
91 &call_ast, &kick_thread);
92
93 res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
94 spin_unlock(&res->spinlock);
95 wake_up(&res->wq);
96 if (status != DLM_NORMAL && status != DLM_NOTQUEUED)
97 dlm_error(status);
98
99
100 if (call_ast)
101 dlm_queue_ast(dlm, lock);
102 else
103 dlm_lockres_release_ast(dlm, res);
104
105 if (kick_thread)
106 dlm_kick_thread(dlm, res);
107
108 return status;
109}
110
111
112
113
114
115
116
117
118
119
120static enum dlm_status __dlmconvert_master(struct dlm_ctxt *dlm,
121 struct dlm_lock_resource *res,
122 struct dlm_lock *lock, int flags,
123 int type, int *call_ast,
124 int *kick_thread)
125{
126 enum dlm_status status = DLM_NORMAL;
127 struct list_head *iter;
128 struct dlm_lock *tmplock=NULL;
129
130 assert_spin_locked(&res->spinlock);
131
132 mlog_entry("type=%d, convert_type=%d, new convert_type=%d\n",
133 lock->ml.type, lock->ml.convert_type, type);
134
135 spin_lock(&lock->spinlock);
136
137
138 if (lock->ml.convert_type != LKM_IVMODE) {
139 mlog(ML_ERROR, "attempted to convert a lock with a lock "
140 "conversion pending\n");
141 status = DLM_DENIED;
142 goto unlock_exit;
143 }
144
145
146 if (!dlm_lock_on_list(&res->granted, lock)) {
147 mlog(ML_ERROR, "attempted to convert a lock not on grant "
148 "queue\n");
149 status = DLM_DENIED;
150 goto unlock_exit;
151 }
152
153 if (flags & LKM_VALBLK) {
154 switch (lock->ml.type) {
155 case LKM_EXMODE:
156
157 mlog(0, "will set lvb: converting %s->%s\n",
158 dlm_lock_mode_name(lock->ml.type),
159 dlm_lock_mode_name(type));
160 lock->lksb->flags |= DLM_LKSB_PUT_LVB;
161 break;
162 case LKM_PRMODE:
163 case LKM_NLMODE:
164
165 if (type > LKM_NLMODE) {
166 mlog(0, "will fetch new value into "
167 "lvb: converting %s->%s\n",
168 dlm_lock_mode_name(lock->ml.type),
169 dlm_lock_mode_name(type));
170 lock->lksb->flags |= DLM_LKSB_GET_LVB;
171 } else {
172 mlog(0, "will NOT fetch new value "
173 "into lvb: converting %s->%s\n",
174 dlm_lock_mode_name(lock->ml.type),
175 dlm_lock_mode_name(type));
176 flags &= ~(LKM_VALBLK);
177 }
178 break;
179 }
180 }
181
182
183
184 if (type <= lock->ml.type)
185 goto grant;
186
187
188 status = DLM_NORMAL;
189 list_for_each(iter, &res->granted) {
190 tmplock = list_entry(iter, struct dlm_lock, list);
191 if (tmplock == lock)
192 continue;
193 if (!dlm_lock_compatible(tmplock->ml.type, type))
194 goto switch_queues;
195 }
196
197 list_for_each(iter, &res->converting) {
198 tmplock = list_entry(iter, struct dlm_lock, list);
199 if (!dlm_lock_compatible(tmplock->ml.type, type))
200 goto switch_queues;
201
202 if (!dlm_lock_compatible(tmplock->ml.convert_type, type))
203 goto switch_queues;
204 }
205
206
207
208grant:
209 mlog(0, "res %.*s, granting %s lock\n", res->lockname.len,
210 res->lockname.name, dlm_lock_mode_name(type));
211
212 lock->lksb->status = DLM_NORMAL;
213 if (lock->ml.node == dlm->node_num)
214 mlog(0, "doing in-place convert for nonlocal lock\n");
215 lock->ml.type = type;
216 if (lock->lksb->flags & DLM_LKSB_PUT_LVB)
217 memcpy(res->lvb, lock->lksb->lvb, DLM_LVB_LEN);
218
219 status = DLM_NORMAL;
220 *call_ast = 1;
221 goto unlock_exit;
222
223switch_queues:
224 if (flags & LKM_NOQUEUE) {
225 mlog(0, "failed to convert NOQUEUE lock %.*s from "
226 "%d to %d...\n", res->lockname.len, res->lockname.name,
227 lock->ml.type, type);
228 status = DLM_NOTQUEUED;
229 goto unlock_exit;
230 }
231 mlog(0, "res %.*s, queueing...\n", res->lockname.len,
232 res->lockname.name);
233
234 lock->ml.convert_type = type;
235
236 list_move_tail(&lock->list, &res->converting);
237
238unlock_exit:
239 spin_unlock(&lock->spinlock);
240 if (status == DLM_DENIED) {
241 __dlm_print_one_lock_resource(res);
242 }
243 if (status == DLM_NORMAL)
244 *kick_thread = 1;
245 return status;
246}
247
248void dlm_revert_pending_convert(struct dlm_lock_resource *res,
249 struct dlm_lock *lock)
250{
251
252 list_move_tail(&lock->list, &res->granted);
253 lock->ml.convert_type = LKM_IVMODE;
254 lock->lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB);
255}
256
257
258
259
260
261
262
263
264enum dlm_status dlmconvert_remote(struct dlm_ctxt *dlm,
265 struct dlm_lock_resource *res,
266 struct dlm_lock *lock, int flags, int type)
267{
268 enum dlm_status status;
269
270 mlog(0, "type=%d, convert_type=%d, busy=%d\n", lock->ml.type,
271 lock->ml.convert_type, res->state & DLM_LOCK_RES_IN_PROGRESS);
272
273 spin_lock(&res->spinlock);
274 if (res->state & DLM_LOCK_RES_RECOVERING) {
275 mlog(0, "bailing out early since res is RECOVERING "
276 "on secondary queue\n");
277
278 status = DLM_RECOVERING;
279 goto bail;
280 }
281
282 __dlm_wait_on_lockres(res);
283
284 if (lock->ml.convert_type != LKM_IVMODE) {
285 __dlm_print_one_lock_resource(res);
286 mlog(ML_ERROR, "converting a remote lock that is already "
287 "converting! (cookie=%u:%llu, conv=%d)\n",
288 dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
289 dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
290 lock->ml.convert_type);
291 status = DLM_DENIED;
292 goto bail;
293 }
294 res->state |= DLM_LOCK_RES_IN_PROGRESS;
295
296
297 list_move_tail(&lock->list, &res->converting);
298 lock->convert_pending = 1;
299 lock->ml.convert_type = type;
300
301 if (flags & LKM_VALBLK) {
302 if (lock->ml.type == LKM_EXMODE) {
303 flags |= LKM_PUT_LVB;
304 lock->lksb->flags |= DLM_LKSB_PUT_LVB;
305 } else {
306 if (lock->ml.convert_type == LKM_NLMODE)
307 flags &= ~LKM_VALBLK;
308 else {
309 flags |= LKM_GET_LVB;
310 lock->lksb->flags |= DLM_LKSB_GET_LVB;
311 }
312 }
313 }
314 spin_unlock(&res->spinlock);
315
316
317
318 status = dlm_send_remote_convert_request(dlm, res, lock, flags, type);
319
320 spin_lock(&res->spinlock);
321 res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
322 lock->convert_pending = 0;
323
324 if (status != DLM_NORMAL) {
325 if (status != DLM_NOTQUEUED)
326 dlm_error(status);
327 dlm_revert_pending_convert(res, lock);
328 }
329bail:
330 spin_unlock(&res->spinlock);
331
332
333
334 wake_up(&res->wq);
335
336 return status;
337}
338
339
340
341
342
343
344
345
346static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm,
347 struct dlm_lock_resource *res,
348 struct dlm_lock *lock, int flags, int type)
349{
350 struct dlm_convert_lock convert;
351 int tmpret;
352 enum dlm_status ret;
353 int status = 0;
354 struct kvec vec[2];
355 size_t veclen = 1;
356
357 mlog_entry("%.*s\n", res->lockname.len, res->lockname.name);
358
359 memset(&convert, 0, sizeof(struct dlm_convert_lock));
360 convert.node_idx = dlm->node_num;
361 convert.requested_type = type;
362 convert.cookie = lock->ml.cookie;
363 convert.namelen = res->lockname.len;
364 convert.flags = cpu_to_be32(flags);
365 memcpy(convert.name, res->lockname.name, convert.namelen);
366
367 vec[0].iov_len = sizeof(struct dlm_convert_lock);
368 vec[0].iov_base = &convert;
369
370 if (flags & LKM_PUT_LVB) {
371
372 vec[1].iov_len = DLM_LVB_LEN;
373 vec[1].iov_base = lock->lksb->lvb;
374 veclen++;
375 }
376
377 tmpret = o2net_send_message_vec(DLM_CONVERT_LOCK_MSG, dlm->key,
378 vec, veclen, res->owner, &status);
379 if (tmpret >= 0) {
380
381 ret = status;
382 if (ret == DLM_RECOVERING) {
383 mlog(0, "node %u returned DLM_RECOVERING from convert "
384 "message!\n", res->owner);
385 } else if (ret == DLM_MIGRATING) {
386 mlog(0, "node %u returned DLM_MIGRATING from convert "
387 "message!\n", res->owner);
388 } else if (ret == DLM_FORWARD) {
389 mlog(0, "node %u returned DLM_FORWARD from convert "
390 "message!\n", res->owner);
391 } else if (ret != DLM_NORMAL && ret != DLM_NOTQUEUED)
392 dlm_error(ret);
393 } else {
394 mlog_errno(tmpret);
395 if (dlm_is_host_down(tmpret)) {
396
397
398
399 dlm_wait_for_node_death(dlm, res->owner,
400 DLM_NODE_DEATH_WAIT_MAX);
401 ret = DLM_RECOVERING;
402 mlog(0, "node %u died so returning DLM_RECOVERING "
403 "from convert message!\n", res->owner);
404 } else {
405 ret = dlm_err_to_dlm_status(tmpret);
406 }
407 }
408
409 return ret;
410}
411
412
413
414
415
416
417
418
419
420int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data,
421 void **ret_data)
422{
423 struct dlm_ctxt *dlm = data;
424 struct dlm_convert_lock *cnv = (struct dlm_convert_lock *)msg->buf;
425 struct dlm_lock_resource *res = NULL;
426 struct list_head *iter;
427 struct dlm_lock *lock = NULL;
428 struct dlm_lockstatus *lksb;
429 enum dlm_status status = DLM_NORMAL;
430 u32 flags;
431 int call_ast = 0, kick_thread = 0, ast_reserved = 0, wake = 0;
432
433 if (!dlm_grab(dlm)) {
434 dlm_error(DLM_REJECTED);
435 return DLM_REJECTED;
436 }
437
438 mlog_bug_on_msg(!dlm_domain_fully_joined(dlm),
439 "Domain %s not fully joined!\n", dlm->name);
440
441 if (cnv->namelen > DLM_LOCKID_NAME_MAX) {
442 status = DLM_IVBUFLEN;
443 dlm_error(status);
444 goto leave;
445 }
446
447 flags = be32_to_cpu(cnv->flags);
448
449 if ((flags & (LKM_PUT_LVB|LKM_GET_LVB)) ==
450 (LKM_PUT_LVB|LKM_GET_LVB)) {
451 mlog(ML_ERROR, "both PUT and GET lvb specified\n");
452 status = DLM_BADARGS;
453 goto leave;
454 }
455
456 mlog(0, "lvb: %s\n", flags & LKM_PUT_LVB ? "put lvb" :
457 (flags & LKM_GET_LVB ? "get lvb" : "none"));
458
459 status = DLM_IVLOCKID;
460 res = dlm_lookup_lockres(dlm, cnv->name, cnv->namelen);
461 if (!res) {
462 dlm_error(status);
463 goto leave;
464 }
465
466 spin_lock(&res->spinlock);
467 status = __dlm_lockres_state_to_status(res);
468 if (status != DLM_NORMAL) {
469 spin_unlock(&res->spinlock);
470 dlm_error(status);
471 goto leave;
472 }
473 list_for_each(iter, &res->granted) {
474 lock = list_entry(iter, struct dlm_lock, list);
475 if (lock->ml.cookie == cnv->cookie &&
476 lock->ml.node == cnv->node_idx) {
477 dlm_lock_get(lock);
478 break;
479 }
480 lock = NULL;
481 }
482 spin_unlock(&res->spinlock);
483 if (!lock) {
484 status = DLM_IVLOCKID;
485 mlog(ML_ERROR, "did not find lock to convert on grant queue! "
486 "cookie=%u:%llu\n",
487 dlm_get_lock_cookie_node(be64_to_cpu(cnv->cookie)),
488 dlm_get_lock_cookie_seq(be64_to_cpu(cnv->cookie)));
489 dlm_print_one_lock_resource(res);
490 goto leave;
491 }
492
493
494 lksb = lock->lksb;
495
496
497 if (flags & LKM_PUT_LVB) {
498 BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
499 lksb->flags |= DLM_LKSB_PUT_LVB;
500 memcpy(&lksb->lvb[0], &cnv->lvb[0], DLM_LVB_LEN);
501 } else if (flags & LKM_GET_LVB) {
502 BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB));
503 lksb->flags |= DLM_LKSB_GET_LVB;
504 }
505
506 spin_lock(&res->spinlock);
507 status = __dlm_lockres_state_to_status(res);
508 if (status == DLM_NORMAL) {
509 __dlm_lockres_reserve_ast(res);
510 ast_reserved = 1;
511 res->state |= DLM_LOCK_RES_IN_PROGRESS;
512 status = __dlmconvert_master(dlm, res, lock, flags,
513 cnv->requested_type,
514 &call_ast, &kick_thread);
515 res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
516 wake = 1;
517 }
518 spin_unlock(&res->spinlock);
519 if (wake)
520 wake_up(&res->wq);
521
522 if (status != DLM_NORMAL) {
523 if (status != DLM_NOTQUEUED)
524 dlm_error(status);
525 lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB);
526 }
527
528leave:
529 if (lock)
530 dlm_lock_put(lock);
531
532
533 if (call_ast)
534 dlm_queue_ast(dlm, lock);
535 else if (ast_reserved)
536 dlm_lockres_release_ast(dlm, res);
537
538 if (kick_thread)
539 dlm_kick_thread(dlm, res);
540
541 if (res)
542 dlm_lockres_put(res);
543
544 dlm_put(dlm);
545
546 return status;
547}
548