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