linux/fs/ocfs2/dlm/dlmunlock.c
<<
>>
Prefs
   1/* -*- mode: c; c-basic-offset: 8; -*-
   2 * vim: noexpandtab sw=8 ts=8 sts=0:
   3 *
   4 * dlmunlock.c
   5 *
   6 * underlying calls for unlocking locks
   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#include <linux/delay.h>
  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#define MLOG_MASK_PREFIX ML_DLM
  49#include "cluster/masklog.h"
  50
  51#define DLM_UNLOCK_FREE_LOCK           0x00000001
  52#define DLM_UNLOCK_CALL_AST            0x00000002
  53#define DLM_UNLOCK_REMOVE_LOCK         0x00000004
  54#define DLM_UNLOCK_REGRANT_LOCK        0x00000008
  55#define DLM_UNLOCK_CLEAR_CONVERT_TYPE  0x00000010
  56
  57
  58static enum dlm_status dlm_get_cancel_actions(struct dlm_ctxt *dlm,
  59                                              struct dlm_lock_resource *res,
  60                                              struct dlm_lock *lock,
  61                                              struct dlm_lockstatus *lksb,
  62                                              int *actions);
  63static enum dlm_status dlm_get_unlock_actions(struct dlm_ctxt *dlm,
  64                                              struct dlm_lock_resource *res,
  65                                              struct dlm_lock *lock,
  66                                              struct dlm_lockstatus *lksb,
  67                                              int *actions);
  68
  69static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
  70                                                 struct dlm_lock_resource *res,
  71                                                 struct dlm_lock *lock,
  72                                                 struct dlm_lockstatus *lksb,
  73                                                 int flags,
  74                                                 u8 owner);
  75
  76
  77/*
  78 * according to the spec:
  79 * http://opendlm.sourceforge.net/cvsmirror/opendlm/docs/dlmbook_final.pdf
  80 *
  81 *  flags & LKM_CANCEL != 0: must be converting or blocked
  82 *  flags & LKM_CANCEL == 0: must be granted
  83 *
  84 * So to unlock a converting lock, you must first cancel the
  85 * convert (passing LKM_CANCEL in flags), then call the unlock
  86 * again (with no LKM_CANCEL in flags).
  87 */
  88
  89
  90/*
  91 * locking:
  92 *   caller needs:  none
  93 *   taken:         res->spinlock and lock->spinlock taken and dropped
  94 *   held on exit:  none
  95 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
  96 * all callers should have taken an extra ref on lock coming in
  97 */
  98static enum dlm_status dlmunlock_common(struct dlm_ctxt *dlm,
  99                                        struct dlm_lock_resource *res,
 100                                        struct dlm_lock *lock,
 101                                        struct dlm_lockstatus *lksb,
 102                                        int flags, int *call_ast,
 103                                        int master_node)
 104{
 105        enum dlm_status status;
 106        int actions = 0;
 107        int in_use;
 108        u8 owner;
 109
 110        mlog(0, "master_node = %d, valblk = %d\n", master_node,
 111             flags & LKM_VALBLK);
 112
 113        if (master_node)
 114                BUG_ON(res->owner != dlm->node_num);
 115        else
 116                BUG_ON(res->owner == dlm->node_num);
 117
 118        spin_lock(&dlm->ast_lock);
 119        /* We want to be sure that we're not freeing a lock
 120         * that still has AST's pending... */
 121        in_use = !list_empty(&lock->ast_list);
 122        spin_unlock(&dlm->ast_lock);
 123        if (in_use && !(flags & LKM_CANCEL)) {
 124               mlog(ML_ERROR, "lockres %.*s: Someone is calling dlmunlock "
 125                    "while waiting for an ast!", res->lockname.len,
 126                    res->lockname.name);
 127                return DLM_BADPARAM;
 128        }
 129
 130        spin_lock(&res->spinlock);
 131        if (res->state & DLM_LOCK_RES_IN_PROGRESS) {
 132                if (master_node && !(flags & LKM_CANCEL)) {
 133                        mlog(ML_ERROR, "lockres in progress!\n");
 134                        spin_unlock(&res->spinlock);
 135                        return DLM_FORWARD;
 136                }
 137                /* ok for this to sleep if not in a network handler */
 138                __dlm_wait_on_lockres(res);
 139                res->state |= DLM_LOCK_RES_IN_PROGRESS;
 140        }
 141        spin_lock(&lock->spinlock);
 142
 143        if (res->state & DLM_LOCK_RES_RECOVERING) {
 144                status = DLM_RECOVERING;
 145                goto leave;
 146        }
 147
 148        if (res->state & DLM_LOCK_RES_MIGRATING) {
 149                status = DLM_MIGRATING;
 150                goto leave;
 151        }
 152
 153        /* see above for what the spec says about
 154         * LKM_CANCEL and the lock queue state */
 155        if (flags & LKM_CANCEL)
 156                status = dlm_get_cancel_actions(dlm, res, lock, lksb, &actions);
 157        else
 158                status = dlm_get_unlock_actions(dlm, res, lock, lksb, &actions);
 159
 160        if (status != DLM_NORMAL && (status != DLM_CANCELGRANT || !master_node))
 161                goto leave;
 162
 163        /* By now this has been masked out of cancel requests. */
 164        if (flags & LKM_VALBLK) {
 165                /* make the final update to the lvb */
 166                if (master_node)
 167                        memcpy(res->lvb, lksb->lvb, DLM_LVB_LEN);
 168                else
 169                        flags |= LKM_PUT_LVB; /* let the send function
 170                                               * handle it. */
 171        }
 172
 173        if (!master_node) {
 174                owner = res->owner;
 175                /* drop locks and send message */
 176                if (flags & LKM_CANCEL)
 177                        lock->cancel_pending = 1;
 178                else
 179                        lock->unlock_pending = 1;
 180                spin_unlock(&lock->spinlock);
 181                spin_unlock(&res->spinlock);
 182                status = dlm_send_remote_unlock_request(dlm, res, lock, lksb,
 183                                                        flags, owner);
 184                spin_lock(&res->spinlock);
 185                spin_lock(&lock->spinlock);
 186                /* if the master told us the lock was already granted,
 187                 * let the ast handle all of these actions */
 188                if (status == DLM_CANCELGRANT) {
 189                        actions &= ~(DLM_UNLOCK_REMOVE_LOCK|
 190                                     DLM_UNLOCK_REGRANT_LOCK|
 191                                     DLM_UNLOCK_CLEAR_CONVERT_TYPE);
 192                } else if (status == DLM_RECOVERING ||
 193                           status == DLM_MIGRATING ||
 194                           status == DLM_FORWARD) {
 195                        /* must clear the actions because this unlock
 196                         * is about to be retried.  cannot free or do
 197                         * any list manipulation. */
 198                        mlog(0, "%s:%.*s: clearing actions, %s\n",
 199                             dlm->name, res->lockname.len,
 200                             res->lockname.name,
 201                             status==DLM_RECOVERING?"recovering":
 202                             (status==DLM_MIGRATING?"migrating":
 203                              "forward"));
 204                        actions = 0;
 205                }
 206                if (flags & LKM_CANCEL)
 207                        lock->cancel_pending = 0;
 208                else
 209                        lock->unlock_pending = 0;
 210
 211        }
 212
 213        /* get an extra ref on lock.  if we are just switching
 214         * lists here, we dont want the lock to go away. */
 215        dlm_lock_get(lock);
 216
 217        if (actions & DLM_UNLOCK_REMOVE_LOCK) {
 218                list_del_init(&lock->list);
 219                dlm_lock_put(lock);
 220        }
 221        if (actions & DLM_UNLOCK_REGRANT_LOCK) {
 222                dlm_lock_get(lock);
 223                list_add_tail(&lock->list, &res->granted);
 224        }
 225        if (actions & DLM_UNLOCK_CLEAR_CONVERT_TYPE) {
 226                mlog(0, "clearing convert_type at %smaster node\n",
 227                     master_node ? "" : "non-");
 228                lock->ml.convert_type = LKM_IVMODE;
 229        }
 230
 231        /* remove the extra ref on lock */
 232        dlm_lock_put(lock);
 233
 234leave:
 235        res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
 236        if (!dlm_lock_on_list(&res->converting, lock))
 237                BUG_ON(lock->ml.convert_type != LKM_IVMODE);
 238        else
 239                BUG_ON(lock->ml.convert_type == LKM_IVMODE);
 240        spin_unlock(&lock->spinlock);
 241        spin_unlock(&res->spinlock);
 242        wake_up(&res->wq);
 243
 244        /* let the caller's final dlm_lock_put handle the actual kfree */
 245        if (actions & DLM_UNLOCK_FREE_LOCK) {
 246                /* this should always be coupled with list removal */
 247                BUG_ON(!(actions & DLM_UNLOCK_REMOVE_LOCK));
 248                mlog(0, "lock %u:%llu should be gone now! refs=%d\n",
 249                     dlm_get_lock_cookie_node(be64_to_cpu(lock->ml.cookie)),
 250                     dlm_get_lock_cookie_seq(be64_to_cpu(lock->ml.cookie)),
 251                     atomic_read(&lock->lock_refs.refcount)-1);
 252                dlm_lock_put(lock);
 253        }
 254        if (actions & DLM_UNLOCK_CALL_AST)
 255                *call_ast = 1;
 256
 257        /* if cancel or unlock succeeded, lvb work is done */
 258        if (status == DLM_NORMAL)
 259                lksb->flags &= ~(DLM_LKSB_PUT_LVB|DLM_LKSB_GET_LVB);
 260
 261        return status;
 262}
 263
 264void dlm_commit_pending_unlock(struct dlm_lock_resource *res,
 265                               struct dlm_lock *lock)
 266{
 267        /* leave DLM_LKSB_PUT_LVB on the lksb so any final
 268         * update of the lvb will be sent to the new master */
 269        list_del_init(&lock->list);
 270}
 271
 272void dlm_commit_pending_cancel(struct dlm_lock_resource *res,
 273                               struct dlm_lock *lock)
 274{
 275        list_move_tail(&lock->list, &res->granted);
 276        lock->ml.convert_type = LKM_IVMODE;
 277}
 278
 279
 280static inline enum dlm_status dlmunlock_master(struct dlm_ctxt *dlm,
 281                                          struct dlm_lock_resource *res,
 282                                          struct dlm_lock *lock,
 283                                          struct dlm_lockstatus *lksb,
 284                                          int flags,
 285                                          int *call_ast)
 286{
 287        return dlmunlock_common(dlm, res, lock, lksb, flags, call_ast, 1);
 288}
 289
 290static inline enum dlm_status dlmunlock_remote(struct dlm_ctxt *dlm,
 291                                          struct dlm_lock_resource *res,
 292                                          struct dlm_lock *lock,
 293                                          struct dlm_lockstatus *lksb,
 294                                          int flags, int *call_ast)
 295{
 296        return dlmunlock_common(dlm, res, lock, lksb, flags, call_ast, 0);
 297}
 298
 299/*
 300 * locking:
 301 *   caller needs:  none
 302 *   taken:         none
 303 *   held on exit:  none
 304 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
 305 */
 306static enum dlm_status dlm_send_remote_unlock_request(struct dlm_ctxt *dlm,
 307                                                 struct dlm_lock_resource *res,
 308                                                 struct dlm_lock *lock,
 309                                                 struct dlm_lockstatus *lksb,
 310                                                 int flags,
 311                                                 u8 owner)
 312{
 313        struct dlm_unlock_lock unlock;
 314        int tmpret;
 315        enum dlm_status ret;
 316        int status = 0;
 317        struct kvec vec[2];
 318        size_t veclen = 1;
 319
 320        mlog(0, "%.*s\n", res->lockname.len, res->lockname.name);
 321
 322        if (owner == dlm->node_num) {
 323                /* ended up trying to contact ourself.  this means
 324                 * that the lockres had been remote but became local
 325                 * via a migration.  just retry it, now as local */
 326                mlog(0, "%s:%.*s: this node became the master due to a "
 327                     "migration, re-evaluate now\n", dlm->name,
 328                     res->lockname.len, res->lockname.name);
 329                return DLM_FORWARD;
 330        }
 331
 332        memset(&unlock, 0, sizeof(unlock));
 333        unlock.node_idx = dlm->node_num;
 334        unlock.flags = cpu_to_be32(flags);
 335        unlock.cookie = lock->ml.cookie;
 336        unlock.namelen = res->lockname.len;
 337        memcpy(unlock.name, res->lockname.name, unlock.namelen);
 338
 339        vec[0].iov_len = sizeof(struct dlm_unlock_lock);
 340        vec[0].iov_base = &unlock;
 341
 342        if (flags & LKM_PUT_LVB) {
 343                /* extra data to send if we are updating lvb */
 344                vec[1].iov_len = DLM_LVB_LEN;
 345                vec[1].iov_base = lock->lksb->lvb;
 346                veclen++;
 347        }
 348
 349        tmpret = o2net_send_message_vec(DLM_UNLOCK_LOCK_MSG, dlm->key,
 350                                        vec, veclen, owner, &status);
 351        if (tmpret >= 0) {
 352                // successfully sent and received
 353                if (status == DLM_FORWARD)
 354                        mlog(0, "master was in-progress.  retry\n");
 355                ret = status;
 356        } else {
 357                mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
 358                     "node %u\n", tmpret, DLM_UNLOCK_LOCK_MSG, dlm->key, owner);
 359                if (dlm_is_host_down(tmpret)) {
 360                        /* NOTE: this seems strange, but it is what we want.
 361                         * when the master goes down during a cancel or
 362                         * unlock, the recovery code completes the operation
 363                         * as if the master had not died, then passes the
 364                         * updated state to the recovery master.  this thread
 365                         * just needs to finish out the operation and call
 366                         * the unlockast. */
 367                        ret = DLM_NORMAL;
 368                } else {
 369                        /* something bad.  this will BUG in ocfs2 */
 370                        ret = dlm_err_to_dlm_status(tmpret);
 371                }
 372        }
 373
 374        return ret;
 375}
 376
 377/*
 378 * locking:
 379 *   caller needs:  none
 380 *   taken:         takes and drops res->spinlock
 381 *   held on exit:  none
 382 * returns: DLM_NORMAL, DLM_BADARGS, DLM_IVLOCKID,
 383 *          return value from dlmunlock_master
 384 */
 385int dlm_unlock_lock_handler(struct o2net_msg *msg, u32 len, void *data,
 386                            void **ret_data)
 387{
 388        struct dlm_ctxt *dlm = data;
 389        struct dlm_unlock_lock *unlock = (struct dlm_unlock_lock *)msg->buf;
 390        struct dlm_lock_resource *res = NULL;
 391        struct list_head *iter;
 392        struct dlm_lock *lock = NULL;
 393        enum dlm_status status = DLM_NORMAL;
 394        int found = 0, i;
 395        struct dlm_lockstatus *lksb = NULL;
 396        int ignore;
 397        u32 flags;
 398        struct list_head *queue;
 399
 400        flags = be32_to_cpu(unlock->flags);
 401
 402        if (flags & LKM_GET_LVB) {
 403                mlog(ML_ERROR, "bad args!  GET_LVB specified on unlock!\n");
 404                return DLM_BADARGS;
 405        }
 406
 407        if ((flags & (LKM_PUT_LVB|LKM_CANCEL)) == (LKM_PUT_LVB|LKM_CANCEL)) {
 408                mlog(ML_ERROR, "bad args!  cannot modify lvb on a CANCEL "
 409                     "request!\n");
 410                return DLM_BADARGS;
 411        }
 412
 413        if (unlock->namelen > DLM_LOCKID_NAME_MAX) {
 414                mlog(ML_ERROR, "Invalid name length in unlock handler!\n");
 415                return DLM_IVBUFLEN;
 416        }
 417
 418        if (!dlm_grab(dlm))
 419                return DLM_REJECTED;
 420
 421        mlog_bug_on_msg(!dlm_domain_fully_joined(dlm),
 422                        "Domain %s not fully joined!\n", dlm->name);
 423
 424        mlog(0, "lvb: %s\n", flags & LKM_PUT_LVB ? "put lvb" : "none");
 425
 426        res = dlm_lookup_lockres(dlm, unlock->name, unlock->namelen);
 427        if (!res) {
 428                /* We assume here that a no lock resource simply means
 429                 * it was migrated away and destroyed before the other
 430                 * node could detect it. */
 431                mlog(0, "returning DLM_FORWARD -- res no longer exists\n");
 432                status = DLM_FORWARD;
 433                goto not_found;
 434        }
 435
 436        queue=&res->granted;
 437        found = 0;
 438        spin_lock(&res->spinlock);
 439        if (res->state & DLM_LOCK_RES_RECOVERING) {
 440                spin_unlock(&res->spinlock);
 441                mlog(0, "returning DLM_RECOVERING\n");
 442                status = DLM_RECOVERING;
 443                goto leave;
 444        }
 445
 446        if (res->state & DLM_LOCK_RES_MIGRATING) {
 447                spin_unlock(&res->spinlock);
 448                mlog(0, "returning DLM_MIGRATING\n");
 449                status = DLM_MIGRATING;
 450                goto leave;
 451        }
 452
 453        if (res->owner != dlm->node_num) {
 454                spin_unlock(&res->spinlock);
 455                mlog(0, "returning DLM_FORWARD -- not master\n");
 456                status = DLM_FORWARD;
 457                goto leave;
 458        }
 459
 460        for (i=0; i<3; i++) {
 461                list_for_each(iter, queue) {
 462                        lock = list_entry(iter, struct dlm_lock, list);
 463                        if (lock->ml.cookie == unlock->cookie &&
 464                            lock->ml.node == unlock->node_idx) {
 465                                dlm_lock_get(lock);
 466                                found = 1;
 467                                break;
 468                        }
 469                }
 470                if (found)
 471                        break;
 472                /* scan granted -> converting -> blocked queues */
 473                queue++;
 474        }
 475        spin_unlock(&res->spinlock);
 476        if (!found) {
 477                status = DLM_IVLOCKID;
 478                goto not_found;
 479        }
 480
 481        /* lock was found on queue */
 482        lksb = lock->lksb;
 483        if (flags & (LKM_VALBLK|LKM_PUT_LVB) &&
 484            lock->ml.type != LKM_EXMODE)
 485                flags &= ~(LKM_VALBLK|LKM_PUT_LVB);
 486
 487        /* unlockast only called on originating node */
 488        if (flags & LKM_PUT_LVB) {
 489                lksb->flags |= DLM_LKSB_PUT_LVB;
 490                memcpy(&lksb->lvb[0], &unlock->lvb[0], DLM_LVB_LEN);
 491        }
 492
 493        /* if this is in-progress, propagate the DLM_FORWARD
 494         * all the way back out */
 495        status = dlmunlock_master(dlm, res, lock, lksb, flags, &ignore);
 496        if (status == DLM_FORWARD)
 497                mlog(0, "lockres is in progress\n");
 498
 499        if (flags & LKM_PUT_LVB)
 500                lksb->flags &= ~DLM_LKSB_PUT_LVB;
 501
 502        dlm_lockres_calc_usage(dlm, res);
 503        dlm_kick_thread(dlm, res);
 504
 505not_found:
 506        if (!found)
 507                mlog(ML_ERROR, "failed to find lock to unlock! "
 508                               "cookie=%u:%llu\n",
 509                     dlm_get_lock_cookie_node(be64_to_cpu(unlock->cookie)),
 510                     dlm_get_lock_cookie_seq(be64_to_cpu(unlock->cookie)));
 511        else
 512                dlm_lock_put(lock);
 513
 514leave:
 515        if (res)
 516                dlm_lockres_put(res);
 517
 518        dlm_put(dlm);
 519
 520        return status;
 521}
 522
 523
 524static enum dlm_status dlm_get_cancel_actions(struct dlm_ctxt *dlm,
 525                                              struct dlm_lock_resource *res,
 526                                              struct dlm_lock *lock,
 527                                              struct dlm_lockstatus *lksb,
 528                                              int *actions)
 529{
 530        enum dlm_status status;
 531
 532        if (dlm_lock_on_list(&res->blocked, lock)) {
 533                /* cancel this outright */
 534                status = DLM_NORMAL;
 535                *actions = (DLM_UNLOCK_CALL_AST |
 536                            DLM_UNLOCK_REMOVE_LOCK);
 537        } else if (dlm_lock_on_list(&res->converting, lock)) {
 538                /* cancel the request, put back on granted */
 539                status = DLM_NORMAL;
 540                *actions = (DLM_UNLOCK_CALL_AST |
 541                            DLM_UNLOCK_REMOVE_LOCK |
 542                            DLM_UNLOCK_REGRANT_LOCK |
 543                            DLM_UNLOCK_CLEAR_CONVERT_TYPE);
 544        } else if (dlm_lock_on_list(&res->granted, lock)) {
 545                /* too late, already granted. */
 546                status = DLM_CANCELGRANT;
 547                *actions = DLM_UNLOCK_CALL_AST;
 548        } else {
 549                mlog(ML_ERROR, "lock to cancel is not on any list!\n");
 550                status = DLM_IVLOCKID;
 551                *actions = 0;
 552        }
 553        return status;
 554}
 555
 556static enum dlm_status dlm_get_unlock_actions(struct dlm_ctxt *dlm,
 557                                              struct dlm_lock_resource *res,
 558                                              struct dlm_lock *lock,
 559                                              struct dlm_lockstatus *lksb,
 560                                              int *actions)
 561{
 562        enum dlm_status status;
 563
 564        /* unlock request */
 565        if (!dlm_lock_on_list(&res->granted, lock)) {
 566                status = DLM_DENIED;
 567                dlm_error(status);
 568                *actions = 0;
 569        } else {
 570                /* unlock granted lock */
 571                status = DLM_NORMAL;
 572                *actions = (DLM_UNLOCK_FREE_LOCK |
 573                            DLM_UNLOCK_CALL_AST |
 574                            DLM_UNLOCK_REMOVE_LOCK);
 575        }
 576        return status;
 577}
 578
 579/* there seems to be no point in doing this async
 580 * since (even for the remote case) there is really
 581 * no work to queue up... so just do it and fire the
 582 * unlockast by hand when done... */
 583enum dlm_status dlmunlock(struct dlm_ctxt *dlm, struct dlm_lockstatus *lksb,
 584                          int flags, dlm_astunlockfunc_t *unlockast, void *data)
 585{
 586        enum dlm_status status;
 587        struct dlm_lock_resource *res;
 588        struct dlm_lock *lock = NULL;
 589        int call_ast, is_master;
 590
 591        if (!lksb) {
 592                dlm_error(DLM_BADARGS);
 593                return DLM_BADARGS;
 594        }
 595
 596        if (flags & ~(LKM_CANCEL | LKM_VALBLK | LKM_INVVALBLK)) {
 597                dlm_error(DLM_BADPARAM);
 598                return DLM_BADPARAM;
 599        }
 600
 601        if ((flags & (LKM_VALBLK | LKM_CANCEL)) == (LKM_VALBLK | LKM_CANCEL)) {
 602                mlog(0, "VALBLK given with CANCEL: ignoring VALBLK\n");
 603                flags &= ~LKM_VALBLK;
 604        }
 605
 606        if (!lksb->lockid || !lksb->lockid->lockres) {
 607                dlm_error(DLM_BADPARAM);
 608                return DLM_BADPARAM;
 609        }
 610
 611        lock = lksb->lockid;
 612        BUG_ON(!lock);
 613        dlm_lock_get(lock);
 614
 615        res = lock->lockres;
 616        BUG_ON(!res);
 617        dlm_lockres_get(res);
 618retry:
 619        call_ast = 0;
 620        /* need to retry up here because owner may have changed */
 621        mlog(0, "lock=%p res=%p\n", lock, res);
 622
 623        spin_lock(&res->spinlock);
 624        is_master = (res->owner == dlm->node_num);
 625        if (flags & LKM_VALBLK && lock->ml.type != LKM_EXMODE)
 626                flags &= ~LKM_VALBLK;
 627        spin_unlock(&res->spinlock);
 628
 629        if (is_master) {
 630                status = dlmunlock_master(dlm, res, lock, lksb, flags,
 631                                          &call_ast);
 632                mlog(0, "done calling dlmunlock_master: returned %d, "
 633                     "call_ast is %d\n", status, call_ast);
 634        } else {
 635                status = dlmunlock_remote(dlm, res, lock, lksb, flags,
 636                                          &call_ast);
 637                mlog(0, "done calling dlmunlock_remote: returned %d, "
 638                     "call_ast is %d\n", status, call_ast);
 639        }
 640
 641        if (status == DLM_RECOVERING ||
 642            status == DLM_MIGRATING ||
 643            status == DLM_FORWARD) {
 644                /* We want to go away for a tiny bit to allow recovery
 645                 * / migration to complete on this resource. I don't
 646                 * know of any wait queue we could sleep on as this
 647                 * may be happening on another node. Perhaps the
 648                 * proper solution is to queue up requests on the
 649                 * other end? */
 650
 651                /* do we want to yield(); ?? */
 652                msleep(50);
 653
 654                mlog(0, "retrying unlock due to pending recovery/"
 655                     "migration/in-progress\n");
 656                goto retry;
 657        }
 658
 659        if (call_ast) {
 660                mlog(0, "calling unlockast(%p, %d)\n", data, status);
 661                if (is_master) {
 662                        /* it is possible that there is one last bast
 663                         * pending.  make sure it is flushed, then
 664                         * call the unlockast.
 665                         * not an issue if this is a mastered remotely,
 666                         * since this lock has been removed from the
 667                         * lockres queues and cannot be found. */
 668                        dlm_kick_thread(dlm, NULL);
 669                        wait_event(dlm->ast_wq,
 670                                   dlm_lock_basts_flushed(dlm, lock));
 671                }
 672                (*unlockast)(data, status);
 673        }
 674
 675        if (status == DLM_CANCELGRANT)
 676                status = DLM_NORMAL;
 677
 678        if (status == DLM_NORMAL) {
 679                mlog(0, "kicking the thread\n");
 680                dlm_kick_thread(dlm, res);
 681        } else
 682                dlm_error(status);
 683
 684        dlm_lockres_calc_usage(dlm, res);
 685        dlm_lockres_put(res);
 686        dlm_lock_put(lock);
 687
 688        mlog(0, "returning status=%d!\n", status);
 689        return status;
 690}
 691EXPORT_SYMBOL_GPL(dlmunlock);
 692
 693