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 19 static uint64_t dlm_cb_seq; 20 static DEFINE_SPINLOCK(dlm_cb_seq_spin); 21 22 static 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 53 int 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 114 int 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 175 void 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 if (!dlm_cb_seq) 185 new_seq = ++dlm_cb_seq; 186 spin_unlock(&dlm_cb_seq_spin); 187 188 if (lkb->lkb_flags & DLM_IFL_USER) { 189 dlm_user_add_ast(lkb, flags, mode, status, sbflags, new_seq); 190 return; 191 } 192 193 mutex_lock(&lkb->lkb_cb_mutex); 194 prev_seq = lkb->lkb_callbacks[0].seq; 195 196 rv = dlm_add_lkb_callback(lkb, flags, mode, status, sbflags, new_seq); 197 if (rv < 0) 198 goto out; 199 200 if (!prev_seq) { 201 kref_get(&lkb->lkb_ref); 202 203 if (test_bit(LSFL_CB_DELAY, &ls->ls_flags)) { 204 mutex_lock(&ls->ls_cb_mutex); 205 list_add(&lkb->lkb_cb_list, &ls->ls_cb_delay); 206 mutex_unlock(&ls->ls_cb_mutex); 207 } else { 208 queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work); 209 } 210 } 211 out: 212 mutex_unlock(&lkb->lkb_cb_mutex); 213 } 214 215 void dlm_callback_work(struct work_struct *work) 216 { 217 struct dlm_lkb *lkb = container_of(work, struct dlm_lkb, lkb_cb_work); 218 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 219 void (*castfn) (void *astparam); 220 void (*bastfn) (void *astparam, int mode); 221 struct dlm_callback callbacks[DLM_CALLBACKS_SIZE]; 222 int i, rv, resid; 223 224 memset(&callbacks, 0, sizeof(callbacks)); 225 226 mutex_lock(&lkb->lkb_cb_mutex); 227 if (!lkb->lkb_callbacks[0].seq) { 228 /* no callback work exists, shouldn't happen */ 229 log_error(ls, "dlm_callback_work %x no work", lkb->lkb_id); 230 dlm_print_lkb(lkb); 231 dlm_dump_lkb_callbacks(lkb); 232 } 233 234 for (i = 0; i < DLM_CALLBACKS_SIZE; i++) { 235 rv = dlm_rem_lkb_callback(ls, lkb, &callbacks[i], &resid); 236 if (rv < 0) 237 break; 238 } 239 240 if (resid) { 241 /* cbs remain, loop should have removed all, shouldn't happen */ 242 log_error(ls, "dlm_callback_work %x resid %d", lkb->lkb_id, 243 resid); 244 dlm_print_lkb(lkb); 245 dlm_dump_lkb_callbacks(lkb); 246 } 247 mutex_unlock(&lkb->lkb_cb_mutex); 248 249 castfn = lkb->lkb_astfn; 250 bastfn = lkb->lkb_bastfn; 251 252 for (i = 0; i < DLM_CALLBACKS_SIZE; i++) { 253 if (!callbacks[i].seq) 254 break; 255 if (callbacks[i].flags & DLM_CB_SKIP) { 256 continue; 257 } else if (callbacks[i].flags & DLM_CB_BAST) { 258 bastfn(lkb->lkb_astparam, callbacks[i].mode); 259 } else if (callbacks[i].flags & DLM_CB_CAST) { 260 lkb->lkb_lksb->sb_status = callbacks[i].sb_status; 261 lkb->lkb_lksb->sb_flags = callbacks[i].sb_flags; 262 castfn(lkb->lkb_astparam); 263 } 264 } 265 266 /* undo kref_get from dlm_add_callback, may cause lkb to be freed */ 267 dlm_put_lkb(lkb); 268 } 269 270 int dlm_callback_start(struct dlm_ls *ls) 271 { 272 ls->ls_callback_wq = alloc_workqueue("dlm_callback", 273 WQ_HIGHPRI | WQ_MEM_RECLAIM, 0); 274 if (!ls->ls_callback_wq) { 275 log_print("can't start dlm_callback workqueue"); 276 return -ENOMEM; 277 } 278 return 0; 279 } 280 281 void dlm_callback_stop(struct dlm_ls *ls) 282 { 283 if (ls->ls_callback_wq) 284 destroy_workqueue(ls->ls_callback_wq); 285 } 286 287 void dlm_callback_suspend(struct dlm_ls *ls) 288 { 289 set_bit(LSFL_CB_DELAY, &ls->ls_flags); 290 291 if (ls->ls_callback_wq) 292 flush_workqueue(ls->ls_callback_wq); 293 } 294 295 #define MAX_CB_QUEUE 25 296 297 void dlm_callback_resume(struct dlm_ls *ls) 298 { 299 struct dlm_lkb *lkb, *safe; 300 int count = 0; 301 302 clear_bit(LSFL_CB_DELAY, &ls->ls_flags); 303 304 if (!ls->ls_callback_wq) 305 return; 306 307 more: 308 mutex_lock(&ls->ls_cb_mutex); 309 list_for_each_entry_safe(lkb, safe, &ls->ls_cb_delay, lkb_cb_list) { 310 list_del_init(&lkb->lkb_cb_list); 311 queue_work(ls->ls_callback_wq, &lkb->lkb_cb_work); 312 count++; 313 if (count == MAX_CB_QUEUE) 314 break; 315 } 316 mutex_unlock(&ls->ls_cb_mutex); 317 318 if (count) 319 log_rinfo(ls, "dlm_callback_resume %d", count); 320 if (count == MAX_CB_QUEUE) { 321 count = 0; 322 cond_resched(); 323 goto more; 324 } 325 } 326 327