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 dlm_lock *tmplock=NULL; 127 128 assert_spin_locked(&res->spinlock); 129 130 mlog(0, "type=%d, convert_type=%d, new convert_type=%d\n", 131 lock->ml.type, lock->ml.convert_type, type); 132 133 spin_lock(&lock->spinlock); 134 135 /* already converting? */ 136 if (lock->ml.convert_type != LKM_IVMODE) { 137 mlog(ML_ERROR, "attempted to convert a lock with a lock " 138 "conversion pending\n"); 139 status = DLM_DENIED; 140 goto unlock_exit; 141 } 142 143 /* must be on grant queue to convert */ 144 if (!dlm_lock_on_list(&res->granted, lock)) { 145 mlog(ML_ERROR, "attempted to convert a lock not on grant " 146 "queue\n"); 147 status = DLM_DENIED; 148 goto unlock_exit; 149 } 150 151 if (flags & LKM_VALBLK) { 152 switch (lock->ml.type) { 153 case LKM_EXMODE: 154 /* EX + LKM_VALBLK + convert == set lvb */ 155 mlog(0, "will set lvb: converting %s->%s\n", 156 dlm_lock_mode_name(lock->ml.type), 157 dlm_lock_mode_name(type)); 158 lock->lksb->flags |= DLM_LKSB_PUT_LVB; 159 break; 160 case LKM_PRMODE: 161 case LKM_NLMODE: 162 /* refetch if new level is not NL */ 163 if (type > LKM_NLMODE) { 164 mlog(0, "will fetch new value into " 165 "lvb: converting %s->%s\n", 166 dlm_lock_mode_name(lock->ml.type), 167 dlm_lock_mode_name(type)); 168 lock->lksb->flags |= DLM_LKSB_GET_LVB; 169 } else { 170 mlog(0, "will NOT fetch new value " 171 "into lvb: converting %s->%s\n", 172 dlm_lock_mode_name(lock->ml.type), 173 dlm_lock_mode_name(type)); 174 flags &= ~(LKM_VALBLK); 175 } 176 break; 177 } 178 } 179 180 181 /* in-place downconvert? */ 182 if (type <= lock->ml.type) 183 goto grant; 184 185 /* upconvert from here on */ 186 status = DLM_NORMAL; 187 list_for_each_entry(tmplock, &res->granted, list) { 188 if (tmplock == lock) 189 continue; 190 if (!dlm_lock_compatible(tmplock->ml.type, type)) 191 goto switch_queues; 192 } 193 194 list_for_each_entry(tmplock, &res->converting, list) { 195 if (!dlm_lock_compatible(tmplock->ml.type, type)) 196 goto switch_queues; 197 /* existing conversion requests take precedence */ 198 if (!dlm_lock_compatible(tmplock->ml.convert_type, type)) 199 goto switch_queues; 200 } 201 202 /* fall thru to grant */ 203 204 grant: 205 mlog(0, "res %.*s, granting %s lock\n", res->lockname.len, 206 res->lockname.name, dlm_lock_mode_name(type)); 207 /* immediately grant the new lock type */ 208 lock->lksb->status = DLM_NORMAL; 209 if (lock->ml.node == dlm->node_num) 210 mlog(0, "doing in-place convert for nonlocal lock\n"); 211 lock->ml.type = type; 212 if (lock->lksb->flags & DLM_LKSB_PUT_LVB) 213 memcpy(res->lvb, lock->lksb->lvb, DLM_LVB_LEN); 214 215 status = DLM_NORMAL; 216 *call_ast = 1; 217 goto unlock_exit; 218 219 switch_queues: 220 if (flags & LKM_NOQUEUE) { 221 mlog(0, "failed to convert NOQUEUE lock %.*s from " 222 "%d to %d...\n", res->lockname.len, res->lockname.name, 223 lock->ml.type, type); 224 status = DLM_NOTQUEUED; 225 goto unlock_exit; 226 } 227 mlog(0, "res %.*s, queueing...\n", res->lockname.len, 228 res->lockname.name); 229 230 lock->ml.convert_type = type; 231 /* do not alter lock refcount. switching lists. */ 232 list_move_tail(&lock->list, &res->converting); 233 234 unlock_exit: 235 spin_unlock(&lock->spinlock); 236 if (status == DLM_DENIED) { 237 __dlm_print_one_lock_resource(res); 238 } 239 if (status == DLM_NORMAL) 240 *kick_thread = 1; 241 return status; 242 } 243 244 void dlm_revert_pending_convert(struct dlm_lock_resource *res, 245 struct dlm_lock *lock) 246 { 247 /* do not alter lock refcount. switching lists. */ 248 list_move_tail(&lock->list, &res->granted); 249 lock->ml.convert_type = LKM_IVMODE; 250 lock->lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB); 251 } 252 253 /* messages the master site to do lock conversion 254 * locking: 255 * caller needs: none 256 * taken: takes and drops res->spinlock, uses DLM_LOCK_RES_IN_PROGRESS 257 * held on exit: none 258 * returns: DLM_NORMAL, DLM_RECOVERING, status from remote node 259 */ 260 enum dlm_status dlmconvert_remote(struct dlm_ctxt *dlm, 261 struct dlm_lock_resource *res, 262 struct dlm_lock *lock, int flags, int type) 263 { 264 enum dlm_status status; 265 266 mlog(0, "type=%d, convert_type=%d, busy=%d\n", lock->ml.type, 267 lock->ml.convert_type, res->state & DLM_LOCK_RES_IN_PROGRESS); 268 269 spin_lock(&res->spinlock); 270 if (res->state & DLM_LOCK_RES_RECOVERING) { 271 mlog(0, "bailing out early since res is RECOVERING " 272 "on secondary queue\n"); 273 /* __dlm_print_one_lock_resource(res); */ 274 status = DLM_RECOVERING; 275 goto bail; 276 } 277 /* will exit this call with spinlock held */ 278 __dlm_wait_on_lockres(res); 279 280 if (lock->ml.convert_type != LKM_IVMODE) { 281 __dlm_print_one_lock_resource(res); 282 mlog(ML_ERROR, "converting a remote lock that is already " 283 "converting! (cookie=%u:%llu, conv=%d)\n", 284 dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)), 285 dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)), 286 lock->ml.convert_type); 287 status = DLM_DENIED; 288 goto bail; 289 } 290 res->state |= DLM_LOCK_RES_IN_PROGRESS; 291 /* move lock to local convert queue */ 292 /* do not alter lock refcount. switching lists. */ 293 list_move_tail(&lock->list, &res->converting); 294 lock->convert_pending = 1; 295 lock->ml.convert_type = type; 296 297 if (flags & LKM_VALBLK) { 298 if (lock->ml.type == LKM_EXMODE) { 299 flags |= LKM_PUT_LVB; 300 lock->lksb->flags |= DLM_LKSB_PUT_LVB; 301 } else { 302 if (lock->ml.convert_type == LKM_NLMODE) 303 flags &= ~LKM_VALBLK; 304 else { 305 flags |= LKM_GET_LVB; 306 lock->lksb->flags |= DLM_LKSB_GET_LVB; 307 } 308 } 309 } 310 spin_unlock(&res->spinlock); 311 312 /* no locks held here. 313 * need to wait for a reply as to whether it got queued or not. */ 314 status = dlm_send_remote_convert_request(dlm, res, lock, flags, type); 315 316 spin_lock(&res->spinlock); 317 res->state &= ~DLM_LOCK_RES_IN_PROGRESS; 318 lock->convert_pending = 0; 319 /* if it failed, move it back to granted queue */ 320 if (status != DLM_NORMAL) { 321 if (status != DLM_NOTQUEUED) 322 dlm_error(status); 323 dlm_revert_pending_convert(res, lock); 324 } 325 bail: 326 spin_unlock(&res->spinlock); 327 328 /* TODO: should this be a wake_one? */ 329 /* wake up any IN_PROGRESS waiters */ 330 wake_up(&res->wq); 331 332 return status; 333 } 334 335 /* sends DLM_CONVERT_LOCK_MSG to master site 336 * locking: 337 * caller needs: none 338 * taken: none 339 * held on exit: none 340 * returns: DLM_NOLOCKMGR, status from remote node 341 */ 342 static enum dlm_status dlm_send_remote_convert_request(struct dlm_ctxt *dlm, 343 struct dlm_lock_resource *res, 344 struct dlm_lock *lock, int flags, int type) 345 { 346 struct dlm_convert_lock convert; 347 int tmpret; 348 enum dlm_status ret; 349 int status = 0; 350 struct kvec vec[2]; 351 size_t veclen = 1; 352 353 mlog(0, "%.*s\n", res->lockname.len, res->lockname.name); 354 355 memset(&convert, 0, sizeof(struct dlm_convert_lock)); 356 convert.node_idx = dlm->node_num; 357 convert.requested_type = type; 358 convert.cookie = lock->ml.cookie; 359 convert.namelen = res->lockname.len; 360 convert.flags = cpu_to_be32(flags); 361 memcpy(convert.name, res->lockname.name, convert.namelen); 362 363 vec[0].iov_len = sizeof(struct dlm_convert_lock); 364 vec[0].iov_base = &convert; 365 366 if (flags & LKM_PUT_LVB) { 367 /* extra data to send if we are updating lvb */ 368 vec[1].iov_len = DLM_LVB_LEN; 369 vec[1].iov_base = lock->lksb->lvb; 370 veclen++; 371 } 372 373 tmpret = o2net_send_message_vec(DLM_CONVERT_LOCK_MSG, dlm->key, 374 vec, veclen, res->owner, &status); 375 if (tmpret >= 0) { 376 // successfully sent and received 377 ret = status; // this is already a dlm_status 378 if (ret == DLM_RECOVERING) { 379 mlog(0, "node %u returned DLM_RECOVERING from convert " 380 "message!\n", res->owner); 381 } else if (ret == DLM_MIGRATING) { 382 mlog(0, "node %u returned DLM_MIGRATING from convert " 383 "message!\n", res->owner); 384 } else if (ret == DLM_FORWARD) { 385 mlog(0, "node %u returned DLM_FORWARD from convert " 386 "message!\n", res->owner); 387 } else if (ret != DLM_NORMAL && ret != DLM_NOTQUEUED) 388 dlm_error(ret); 389 } else { 390 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 391 "node %u\n", tmpret, DLM_CONVERT_LOCK_MSG, dlm->key, 392 res->owner); 393 if (dlm_is_host_down(tmpret)) { 394 /* instead of logging the same network error over 395 * and over, sleep here and wait for the heartbeat 396 * to notice the node is dead. times out after 5s. */ 397 dlm_wait_for_node_death(dlm, res->owner, 398 DLM_NODE_DEATH_WAIT_MAX); 399 ret = DLM_RECOVERING; 400 mlog(0, "node %u died so returning DLM_RECOVERING " 401 "from convert message!\n", res->owner); 402 } else { 403 ret = dlm_err_to_dlm_status(tmpret); 404 } 405 } 406 407 return ret; 408 } 409 410 /* handler for DLM_CONVERT_LOCK_MSG on master site 411 * locking: 412 * caller needs: none 413 * taken: takes and drop res->spinlock 414 * held on exit: none 415 * returns: DLM_NORMAL, DLM_IVLOCKID, DLM_BADARGS, 416 * status from __dlmconvert_master 417 */ 418 int dlm_convert_lock_handler(struct o2net_msg *msg, u32 len, void *data, 419 void **ret_data) 420 { 421 struct dlm_ctxt *dlm = data; 422 struct dlm_convert_lock *cnv = (struct dlm_convert_lock *)msg->buf; 423 struct dlm_lock_resource *res = NULL; 424 struct dlm_lock *lock = NULL; 425 struct dlm_lock *tmp_lock; 426 struct dlm_lockstatus *lksb; 427 enum dlm_status status = DLM_NORMAL; 428 u32 flags; 429 int call_ast = 0, kick_thread = 0, ast_reserved = 0, wake = 0; 430 431 if (!dlm_grab(dlm)) { 432 dlm_error(DLM_REJECTED); 433 return DLM_REJECTED; 434 } 435 436 mlog_bug_on_msg(!dlm_domain_fully_joined(dlm), 437 "Domain %s not fully joined!\n", dlm->name); 438 439 if (cnv->namelen > DLM_LOCKID_NAME_MAX) { 440 status = DLM_IVBUFLEN; 441 dlm_error(status); 442 goto leave; 443 } 444 445 flags = be32_to_cpu(cnv->flags); 446 447 if ((flags & (LKM_PUT_LVB|LKM_GET_LVB)) == 448 (LKM_PUT_LVB|LKM_GET_LVB)) { 449 mlog(ML_ERROR, "both PUT and GET lvb specified\n"); 450 status = DLM_BADARGS; 451 goto leave; 452 } 453 454 mlog(0, "lvb: %s\n", flags & LKM_PUT_LVB ? "put lvb" : 455 (flags & LKM_GET_LVB ? "get lvb" : "none")); 456 457 status = DLM_IVLOCKID; 458 res = dlm_lookup_lockres(dlm, cnv->name, cnv->namelen); 459 if (!res) { 460 dlm_error(status); 461 goto leave; 462 } 463 464 spin_lock(&res->spinlock); 465 status = __dlm_lockres_state_to_status(res); 466 if (status != DLM_NORMAL) { 467 spin_unlock(&res->spinlock); 468 dlm_error(status); 469 goto leave; 470 } 471 list_for_each_entry(tmp_lock, &res->granted, list) { 472 if (tmp_lock->ml.cookie == cnv->cookie && 473 tmp_lock->ml.node == cnv->node_idx) { 474 lock = tmp_lock; 475 dlm_lock_get(lock); 476 break; 477 } 478 } 479 spin_unlock(&res->spinlock); 480 if (!lock) { 481 status = DLM_IVLOCKID; 482 mlog(ML_ERROR, "did not find lock to convert on grant queue! " 483 "cookie=%u:%llu\n", 484 dlm_get_lock_cookie_node(be64_to_cpu(cnv->cookie)), 485 dlm_get_lock_cookie_seq(be64_to_cpu(cnv->cookie))); 486 dlm_print_one_lock_resource(res); 487 goto leave; 488 } 489 490 /* found the lock */ 491 lksb = lock->lksb; 492 493 /* see if caller needed to get/put lvb */ 494 if (flags & LKM_PUT_LVB) { 495 BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB)); 496 lksb->flags |= DLM_LKSB_PUT_LVB; 497 memcpy(&lksb->lvb[0], &cnv->lvb[0], DLM_LVB_LEN); 498 } else if (flags & LKM_GET_LVB) { 499 BUG_ON(lksb->flags & (DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB)); 500 lksb->flags |= DLM_LKSB_GET_LVB; 501 } 502 503 spin_lock(&res->spinlock); 504 status = __dlm_lockres_state_to_status(res); 505 if (status == DLM_NORMAL) { 506 __dlm_lockres_reserve_ast(res); 507 ast_reserved = 1; 508 res->state |= DLM_LOCK_RES_IN_PROGRESS; 509 status = __dlmconvert_master(dlm, res, lock, flags, 510 cnv->requested_type, 511 &call_ast, &kick_thread); 512 res->state &= ~DLM_LOCK_RES_IN_PROGRESS; 513 wake = 1; 514 } 515 spin_unlock(&res->spinlock); 516 if (wake) 517 wake_up(&res->wq); 518 519 if (status != DLM_NORMAL) { 520 if (status != DLM_NOTQUEUED) 521 dlm_error(status); 522 lksb->flags &= ~(DLM_LKSB_GET_LVB|DLM_LKSB_PUT_LVB); 523 } 524 525 leave: 526 if (lock) 527 dlm_lock_put(lock); 528 529 /* either queue the ast or release it, if reserved */ 530 if (call_ast) 531 dlm_queue_ast(dlm, lock); 532 else if (ast_reserved) 533 dlm_lockres_release_ast(dlm, res); 534 535 if (kick_thread) 536 dlm_kick_thread(dlm, res); 537 538 if (res) 539 dlm_lockres_put(res); 540 541 dlm_put(dlm); 542 543 return status; 544 } 545