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/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 /* NOTE: __dlmconvert_master is the only function in here that 55 * needs a spinlock held on entry (res->spinlock) and it is the 56 * only one that holds a lock on exit (res->spinlock). 57 * All other functions in here need no locks and drop all of 58 * the locks that they acquire. */ 59 static 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); 64 static 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 * this is only called directly by dlmlock(), and only when the 70 * local node is the owner of the lockres 71 * locking: 72 * caller needs: none 73 * taken: takes and drops res->spinlock 74 * held on exit: none 75 * returns: see __dlmconvert_master 76 */ 77 enum 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 /* we are not in a network handler, this is fine */ 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 /* either queue the ast or release it */ 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 /* performs lock conversion at the lockres master site 112 * locking: 113 * caller needs: res->spinlock 114 * taken: takes and drops lock->spinlock 115 * held on exit: res->spinlock 116 * returns: DLM_NORMAL, DLM_NOTQUEUED, DLM_DENIED 117 * call_ast: whether ast should be called for this lock 118 * kick_thread: whether dlm_kick_thread should be called 119 */ 120 static 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 /* already converting? */ 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 /* must be on grant queue to convert */ 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 /* EX + LKM_VALBLK + convert == set lvb */ 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 /* refetch if new level is not NL */ 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 /* in-place downconvert? */ 184 if (type <= lock->ml.type) 185 goto grant; 186 187 /* upconvert from here on */ 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 /* existing conversion requests take precedence */ 202 if (!dlm_lock_compatible(tmplock->ml.convert_type, type)) 203 goto switch_queues; 204 } 205 206 /* fall thru to grant */ 207 208 grant: 209 mlog(0, "res %.*s, granting %s lock\n", res->lockname.len, 210 res->lockname.name, dlm_lock_mode_name(type)); 211 /* immediately grant the new lock type */ 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 223 switch_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 /* do not alter lock refcount. switching lists. */ 236 list_move_tail(&lock->list, &res->converting); 237 238 unlock_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 248 void dlm_revert_pending_convert(struct dlm_lock_resource *res, 249 struct dlm_lock *lock) 250 { 251 /* do not alter lock refcount. switching lists. */ 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 /* messages the master site to do lock conversion 258 * locking: 259 * caller needs: none 260 * taken: takes and drops res->spinlock, uses DLM_LOCK_RES_IN_PROGRESS 261 * held on exit: none 262 * returns: DLM_NORMAL, DLM_RECOVERING, status from remote node 263 */ 264 enum 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 /* __dlm_print_one_lock_resource(res); */ 278 status = DLM_RECOVERING; 279 goto bail; 280 } 281 /* will exit this call with spinlock held */ 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 /* move lock to local convert queue */ 296 /* do not alter lock refcount. switching lists. */ 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 /* no locks held here. 317 * need to wait for a reply as to whether it got queued or not. */ 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 /* if it failed, move it back to granted queue */ 324 if (status != DLM_NORMAL) { 325 if (status != DLM_NOTQUEUED) 326 dlm_error(status); 327 dlm_revert_pending_convert(res, lock); 328 } 329 bail: 330 spin_unlock(&res->spinlock); 331 332 /* TODO: should this be a wake_one? */ 333 /* wake up any IN_PROGRESS waiters */ 334 wake_up(&res->wq); 335 336 return status; 337 } 338 339 /* sends DLM_CONVERT_LOCK_MSG to master site 340 * locking: 341 * caller needs: none 342 * taken: none 343 * held on exit: none 344 * returns: DLM_NOLOCKMGR, status from remote node 345 */ 346 static 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 /* extra data to send if we are updating lvb */ 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 // successfully sent and received 381 ret = status; // this is already a dlm_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 /* instead of logging the same network error over 397 * and over, sleep here and wait for the heartbeat 398 * to notice the node is dead. times out after 5s. */ 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 /* handler for DLM_CONVERT_LOCK_MSG on master site 413 * locking: 414 * caller needs: none 415 * taken: takes and drop res->spinlock 416 * held on exit: none 417 * returns: DLM_NORMAL, DLM_IVLOCKID, DLM_BADARGS, 418 * status from __dlmconvert_master 419 */ 420 int 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 /* found the lock */ 494 lksb = lock->lksb; 495 496 /* see if caller needed to get/put lvb */ 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 528 leave: 529 if (lock) 530 dlm_lock_put(lock); 531 532 /* either queue the ast or release it, if reserved */ 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