LXR linux/fs/dlm/ast.c

   1/******************************************************************************
   2*******************************************************************************
   3**
   4**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
   5**  Copyright (C) 2004-2010 Red Hat, Inc.  All rights reserved.
   6**
   7**  This copyrighted material is made available to anyone wishing to use,
   8**  modify, copy, or redistribute it subject to the terms and conditions
   9**  of the GNU General Public License v.2.
  10**
  11*******************************************************************************
  12******************************************************************************/
  13
  14#include "dlm_internal.h"
  15#include "lock.h"
  16#include "user.h"
  17#include "ast.h"
  18
  19static uint64_t dlm_cb_seq;
  20static DEFINE_SPINLOCK(dlm_cb_seq_spin);
  21
  22static void dlm_dump_lkb_callbacks(struct dlm_lkb *lkb)
  23{
  24        int i;
  25
  26        log_print("last_bast %x %llu flags %x mode %d sb %d %x",
  27                  lkb->lkb_id,
  28                  (unsigned long long)lkb->lkb_last_bast.seq,
  29                  lkb->lkb_last_bast.flags,
  30                  lkb->lkb_last_bast.mode,
  31                  lkb->lkb_last_bast.sb_status,
  32                  lkb->lkb_last_bast.sb_flags);
  33
  34        log_print("last_cast %x %llu flags %x mode %d sb %d %x",
  35                  lkb->lkb_id,
  36                  (unsigned long long)lkb->lkb_last_cast.seq,
  37                  lkb->lkb_last_cast.flags,
  38                  lkb->lkb_last_cast.mode,
  39                  lkb->lkb_last_cast.sb_status,
  40                  lkb->lkb_last_cast.sb_flags);
  41
  42        for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
  43                log_print("cb %x %llu flags %x mode %d sb %d %x",
  44                          lkb->lkb_id,
  45                          (unsigned long long)lkb->lkb_callbacks[i].seq,
  46                          lkb->lkb_callbacks[i].flags,
  47                          lkb->lkb_callbacks[i].mode,
  48                          lkb->lkb_callbacks[i].sb_status,
  49                          lkb->lkb_callbacks[i].sb_flags);
  50        }
  51}
  52
  53int dlm_add_lkb_callback(struct dlm_lkb *lkb, uint32_t flags, int mode,
  54                         int status, uint32_t sbflags, uint64_t seq)
  55{
  56        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
  57        uint64_t prev_seq;
  58        int prev_mode;
  59        int i, rv;
  60
  61        for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
  62                if (lkb->lkb_callbacks[i].seq)
  63                        continue;
  64
  65                /*
  66                 * Suppress some redundant basts here, do more on removal.
  67                 * Don't even add a bast if the callback just before it
  68                 * is a bast for the same mode or a more restrictive mode.
  69                 * (the addional > PR check is needed for PR/CW inversion)
  70                 */
  71
  72                if ((i > 0) && (flags & DLM_CB_BAST) &&
  73                    (lkb->lkb_callbacks[i-1].flags & DLM_CB_BAST)) {
  74
  75                        prev_seq = lkb->lkb_callbacks[i-1].seq;
  76                        prev_mode = lkb->lkb_callbacks[i-1].mode;
  77
  78                        if ((prev_mode == mode) ||
  79                            (prev_mode > mode && prev_mode > DLM_LOCK_PR)) {
  80
  81                                log_debug(ls, "skip %x add bast %llu mode %d "
  82                                          "for bast %llu mode %d",
  83                                          lkb->lkb_id,
  84                                          (unsigned long long)seq,
  85                                          mode,
  86                                          (unsigned long long)prev_seq,
  87                                          prev_mode);
  88                                rv = 0;
  89                                goto out;
  90                        }
  91                }
  92
  93                lkb->lkb_callbacks[i].seq = seq;
  94                lkb->lkb_callbacks[i].flags = flags;
  95                lkb->lkb_callbacks[i].mode = mode;
  96                lkb->lkb_callbacks[i].sb_status = status;
  97                lkb->lkb_callbacks[i].sb_flags = (sbflags & 0x000000FF);
  98                rv = 0;
  99                break;
 100        }
 101
 102        if (i == DLM_CALLBACKS_SIZE) {
 103                log_error(ls, "no callbacks %x %llu flags %x mode %d sb %d %x",
 104                          lkb->lkb_id, (unsigned long long)seq,
 105                          flags, mode, status, sbflags);
 106                dlm_dump_lkb_callbacks(lkb);
 107                rv = -1;
 108                goto out;
 109        }
 110 out:
 111        return rv;
 112}
 113
 114int dlm_rem_lkb_callback(struct dlm_ls *ls, struct dlm_lkb *lkb,
 115                         struct dlm_callback *cb, int *resid)
 116{
 117        int i, rv;
 118
 119        *resid = 0;
 120
 121        if (!lkb->lkb_callbacks[0].seq) {
 122                rv = -ENOENT;
 123                goto out;
 124        }
 125
 126        /* oldest undelivered cb is callbacks[0] */
 127
 128        memcpy(cb, &lkb->lkb_callbacks[0], sizeof(struct dlm_callback));
 129        memset(&lkb->lkb_callbacks[0], 0, sizeof(struct dlm_callback));
 130
 131        /* shift others down */
 132
 133        for (i = 1; i < DLM_CALLBACKS_SIZE; i++) {
 134                if (!lkb->lkb_callbacks[i].seq)
 135                        break;
 136                memcpy(&lkb->lkb_callbacks[i-1], &lkb->lkb_callbacks[i],
 137                       sizeof(struct dlm_callback));
 138                memset(&lkb->lkb_callbacks[i], 0, sizeof(struct dlm_callback));
 139                (*resid)++;
 140        }
 141
 142        /* if cb is a bast, it should be skipped if the blocking mode is
 143           compatible with the last granted mode */
 144
 145        if ((cb->flags & DLM_CB_BAST) && lkb->lkb_last_cast.seq) {
 146                if (dlm_modes_compat(cb->mode, lkb->lkb_last_cast.mode)) {
 147                        cb->flags |= DLM_CB_SKIP;
 148
 149                        log_debug(ls, "skip %x bast %llu mode %d "
 150                                  "for cast %llu mode %d",
 151                                  lkb->lkb_id,
 152                                  (unsigned long long)cb->seq,
 153                                  cb->mode,
 154                                  (unsigned long long)lkb->lkb_last_cast.seq,
 155                                  lkb->lkb_last_cast.mode);
 156                        rv = 0;
 157                        goto out;
 158                }
 159        }
 160
 161        if (cb->flags & DLM_CB_CAST) {
 162                memcpy(&lkb->lkb_last_cast, cb, sizeof(struct dlm_callback));
 163                lkb->lkb_last_cast_time = ktime_get();
 164        }
 165
 166        if (cb->flags & DLM_CB_BAST) {
 167                memcpy(&lkb->lkb_last_bast, cb, sizeof(struct dlm_callback));
 168                lkb->lkb_last_bast_time = ktime_get();
 169        }
 170        rv = 0;
 171 out:
 172        return rv;
 173}
 174
 175void dlm_add_cb(struct dlm_lkb *lkb, uint32_t flags, int mode, int status,
 176                uint32_t sbflags)
 177{
 178        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
 179        uint64_t new_seq, prev_seq;
 180        int rv;
 181
 182        spin_lock(&dlm_cb_seq_spin);
 183        new_seq = ++dlm_cb_seq;
 184        spin_unlock(&dlm_cb_seq_spin);
 185
 186        if (lkb->lkb_flags & DLM_IFL_USER) {
 187                dlm_user_add_ast(lkb, flags, mode, status, sbflags, new_seq);
 188                return;
 189        }
 190
 191        mutex_lock(&lkb->lkb_cb_mutex);
 192        prev_seq = lkb->lkb_callbacks[0].seq;
 193
 194        rv = dlm_add_lkb_callback(lkb, flags, mode, status, sbflags, new_seq);
 195        if (rv < 0)
 196                goto out;
 197
 198        if (!prev_seq) {
 199                kref_get(&lkb->lkb_ref);
 200
 201                if (test_bit(LSFL_CB_DELAY, &ls->ls_flags)) {
 202                        mutex_lock(&ls->ls_cb_mutex);
 203                        list_add(&lkb->lkb_cb_list, &ls->ls_cb_delay);
 204                        mutex_unlock(&ls->ls_cb_mutex);
 205                } else {
 206                        queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work);
 207                }
 208        }
 209 out:
 210        mutex_unlock(&lkb->lkb_cb_mutex);
 211}
 212
 213void dlm_callback_work(struct work_struct *work)
 214{
 215        struct dlm_lkb *lkb = container_of(work, struct dlm_lkb, lkb_cb_work);
 216        struct dlm_ls *ls = lkb->lkb_resource->res_ls;
 217        void (*castfn) (void *astparam);
 218        void (*bastfn) (void *astparam, int mode);
 219        struct dlm_callback callbacks[DLM_CALLBACKS_SIZE];
 220        int i, rv, resid;
 221
 222        memset(&callbacks, 0, sizeof(callbacks));
 223
 224        mutex_lock(&lkb->lkb_cb_mutex);
 225        if (!lkb->lkb_callbacks[0].seq) {
 226                /* no callback work exists, shouldn't happen */
 227                log_error(ls, "dlm_callback_work %x no work", lkb->lkb_id);
 228                dlm_print_lkb(lkb);
 229                dlm_dump_lkb_callbacks(lkb);
 230        }
 231
 232        for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
 233                rv = dlm_rem_lkb_callback(ls, lkb, &callbacks[i], &resid);
 234                if (rv < 0)
 235                        break;
 236        }
 237
 238        if (resid) {
 239                /* cbs remain, loop should have removed all, shouldn't happen */
 240                log_error(ls, "dlm_callback_work %x resid %d", lkb->lkb_id,
 241                          resid);
 242                dlm_print_lkb(lkb);
 243                dlm_dump_lkb_callbacks(lkb);
 244        }
 245        mutex_unlock(&lkb->lkb_cb_mutex);
 246
 247        castfn = lkb->lkb_astfn;
 248        bastfn = lkb->lkb_bastfn;
 249
 250        for (i = 0; i < DLM_CALLBACKS_SIZE; i++) {
 251                if (!callbacks[i].seq)
 252                        break;
 253                if (callbacks[i].flags & DLM_CB_SKIP) {
 254                        continue;
 255                } else if (callbacks[i].flags & DLM_CB_BAST) {
 256                        bastfn(lkb->lkb_astparam, callbacks[i].mode);
 257                } else if (callbacks[i].flags & DLM_CB_CAST) {
 258                        lkb->lkb_lksb->sb_status = callbacks[i].sb_status;
 259                        lkb->lkb_lksb->sb_flags = callbacks[i].sb_flags;
 260                        castfn(lkb->lkb_astparam);
 261                }
 262        }
 263
 264        /* undo kref_get from dlm_add_callback, may cause lkb to be freed */
 265        dlm_put_lkb(lkb);
 266}
 267
 268int dlm_callback_start(struct dlm_ls *ls)
 269{
 270        ls->ls_callback_wq = alloc_workqueue("dlm_callback",
 271                                             WQ_UNBOUND | WQ_MEM_RECLAIM, 0);
 272        if (!ls->ls_callback_wq) {
 273                log_print("can't start dlm_callback workqueue");
 274                return -ENOMEM;
 275        }
 276        return 0;
 277}
 278
 279void dlm_callback_stop(struct dlm_ls *ls)
 280{
 281        if (ls->ls_callback_wq)
 282                destroy_workqueue(ls->ls_callback_wq);
 283}
 284
 285void dlm_callback_suspend(struct dlm_ls *ls)
 286{
 287        set_bit(LSFL_CB_DELAY, &ls->ls_flags);
 288
 289        if (ls->ls_callback_wq)
 290                flush_workqueue(ls->ls_callback_wq);
 291}
 292
 293void dlm_callback_resume(struct dlm_ls *ls)
 294{
 295        struct dlm_lkb *lkb, *safe;
 296        int count = 0;
 297
 298        clear_bit(LSFL_CB_DELAY, &ls->ls_flags);
 299
 300        if (!ls->ls_callback_wq)
 301                return;
 302
 303        mutex_lock(&ls->ls_cb_mutex);
 304        list_for_each_entry_safe(lkb, safe, &ls->ls_cb_delay, lkb_cb_list) {
 305                list_del_init(&lkb->lkb_cb_list);
 306                queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work);
 307                count++;
 308        }
 309        mutex_unlock(&ls->ls_cb_mutex);
 310
 311        if (count)
 312                log_rinfo(ls, "dlm_callback_resume %d", count);
 313}
 314
 315