fs/dlm/rcom.c

   1 /******************************************************************************
   2 *******************************************************************************
   3 **
   4 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
   5 **  Copyright (C) 2005-2008 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 "lockspace.h"
  16 #include "member.h"
  17 #include "lowcomms.h"
  18 #include "midcomms.h"
  19 #include "rcom.h"
  20 #include "recover.h"
  21 #include "dir.h"
  22 #include "config.h"
  23 #include "memory.h"
  24 #include "lock.h"
  25 #include "util.h"
  26 #include "member.h"
  27
  28
  29 static int rcom_response(struct dlm_ls *ls)
  30 {
  31         return test_bit(LSFL_RCOM_READY, &ls->ls_flags);
  32 }
  33
  34 static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
  35                        struct dlm_rcom **rc_ret, struct dlm_mhandle **mh_ret)
  36 {
  37         struct dlm_rcom *rc;
  38         struct dlm_mhandle *mh;
  39         char *mb;
  40         int mb_len = sizeof(struct dlm_rcom) + len;
  41
  42         mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
  43         if (!mh) {
  44                 log_print("create_rcom to %d type %d len %d ENOBUFS",
  45                           to_nodeid, type, len);
  46                 return -ENOBUFS;
  47         }
  48         memset(mb, 0, mb_len);
  49
  50         rc = (struct dlm_rcom *) mb;
  51
  52         rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
  53         rc->rc_header.h_lockspace = ls->ls_global_id;
  54         rc->rc_header.h_nodeid = dlm_our_nodeid();
  55         rc->rc_header.h_length = mb_len;
  56         rc->rc_header.h_cmd = DLM_RCOM;
  57
  58         rc->rc_type = type;
  59
  60         spin_lock(&ls->ls_recover_lock);
  61         rc->rc_seq = ls->ls_recover_seq;
  62         spin_unlock(&ls->ls_recover_lock);
  63
  64         *mh_ret = mh;
  65         *rc_ret = rc;
  66         return 0;
  67 }
  68
  69 static void send_rcom(struct dlm_ls *ls, struct dlm_mhandle *mh,
  70                       struct dlm_rcom *rc)
  71 {
  72         dlm_rcom_out(rc);
  73         dlm_lowcomms_commit_buffer(mh);
  74 }
  75
  76 static void set_rcom_status(struct dlm_ls *ls, struct rcom_status *rs,
  77                             uint32_t flags)
  78 {
  79         rs->rs_flags = cpu_to_le32(flags);
  80 }
  81
  82 /* When replying to a status request, a node also sends back its
  83    configuration values.  The requesting node then checks that the remote
  84    node is configured the same way as itself. */
  85
  86 static void set_rcom_config(struct dlm_ls *ls, struct rcom_config *rf,
  87                             uint32_t num_slots)
  88 {
  89         rf->rf_lvblen = cpu_to_le32(ls->ls_lvblen);
  90         rf->rf_lsflags = cpu_to_le32(ls->ls_exflags);
  91
  92         rf->rf_our_slot = cpu_to_le16(ls->ls_slot);
  93         rf->rf_num_slots = cpu_to_le16(num_slots);
  94         rf->rf_generation =  cpu_to_le32(ls->ls_generation);
  95 }
  96
  97 static int check_rcom_config(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
  98 {
  99         struct rcom_config *rf = (struct rcom_config *) rc->rc_buf;
 100
 101         if ((rc->rc_header.h_version & 0xFFFF0000) != DLM_HEADER_MAJOR) {
 102                 log_error(ls, "version mismatch: %x nodeid %d: %x",
 103                           DLM_HEADER_MAJOR | DLM_HEADER_MINOR, nodeid,
 104                           rc->rc_header.h_version);
 105                 return -EPROTO;
 106         }
 107
 108         if (le32_to_cpu(rf->rf_lvblen) != ls->ls_lvblen ||
 109             le32_to_cpu(rf->rf_lsflags) != ls->ls_exflags) {
 110                 log_error(ls, "config mismatch: %d,%x nodeid %d: %d,%x",
 111                           ls->ls_lvblen, ls->ls_exflags, nodeid,
 112                           le32_to_cpu(rf->rf_lvblen),
 113                           le32_to_cpu(rf->rf_lsflags));
 114                 return -EPROTO;
 115         }
 116         return 0;
 117 }
 118
 119 static void allow_sync_reply(struct dlm_ls *ls, uint64_t *new_seq)
 120 {
 121         spin_lock(&ls->ls_rcom_spin);
 122         *new_seq = ++ls->ls_rcom_seq;
 123         set_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
 124         spin_unlock(&ls->ls_rcom_spin);
 125 }
 126
 127 static void disallow_sync_reply(struct dlm_ls *ls)
 128 {
 129         spin_lock(&ls->ls_rcom_spin);
 130         clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
 131         clear_bit(LSFL_RCOM_READY, &ls->ls_flags);
 132         spin_unlock(&ls->ls_rcom_spin);
 133 }
 134
 135 /*
 136  * low nodeid gathers one slot value at a time from each node.
 137  * it sets need_slots=0, and saves rf_our_slot returned from each
 138  * rcom_config.
 139  *
 140  * other nodes gather all slot values at once from the low nodeid.
 141  * they set need_slots=1, and ignore the rf_our_slot returned from each
 142  * rcom_config.  they use the rf_num_slots returned from the low
 143  * node's rcom_config.
 144  */
 145
 146 int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags)
 147 {
 148         struct dlm_rcom *rc;
 149         struct dlm_mhandle *mh;
 150         int error = 0;
 151
 152         ls->ls_recover_nodeid = nodeid;
 153
 154         if (nodeid == dlm_our_nodeid()) {
 155                 rc = ls->ls_recover_buf;
 156                 rc->rc_result = dlm_recover_status(ls);
 157                 goto out;
 158         }
 159
 160         error = create_rcom(ls, nodeid, DLM_RCOM_STATUS,
 161                             sizeof(struct rcom_status), &rc, &mh);
 162         if (error)
 163                 goto out;
 164
 165         set_rcom_status(ls, (struct rcom_status *)rc->rc_buf, status_flags);
 166
 167         allow_sync_reply(ls, &rc->rc_id);
 168         memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
 169
 170         send_rcom(ls, mh, rc);
 171
 172         error = dlm_wait_function(ls, &rcom_response);
 173         disallow_sync_reply(ls);
 174         if (error)
 175                 goto out;
 176
 177         rc = ls->ls_recover_buf;
 178
 179         if (rc->rc_result == -ESRCH) {
 180                 /* we pretend the remote lockspace exists with 0 status */
 181                 log_debug(ls, "remote node %d not ready", nodeid);
 182                 rc->rc_result = 0;
 183                 error = 0;
 184         } else {
 185                 error = check_rcom_config(ls, rc, nodeid);
 186         }
 187
 188         /* the caller looks at rc_result for the remote recovery status */
 189  out:
 190         return error;
 191 }
 192
 193 static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 194 {
 195         struct dlm_rcom *rc;
 196         struct dlm_mhandle *mh;
 197         struct rcom_status *rs;
 198         uint32_t status;
 199         int nodeid = rc_in->rc_header.h_nodeid;
 200         int len = sizeof(struct rcom_config);
 201         int num_slots = 0;
 202         int error;
 203
 204         if (!dlm_slots_version(&rc_in->rc_header)) {
 205                 status = dlm_recover_status(ls);
 206                 goto do_create;
 207         }
 208
 209         rs = (struct rcom_status *)rc_in->rc_buf;
 210
 211         if (!(rs->rs_flags & DLM_RSF_NEED_SLOTS)) {
 212                 status = dlm_recover_status(ls);
 213                 goto do_create;
 214         }
 215
 216         spin_lock(&ls->ls_recover_lock);
 217         status = ls->ls_recover_status;
 218         num_slots = ls->ls_num_slots;
 219         spin_unlock(&ls->ls_recover_lock);
 220         len += num_slots * sizeof(struct rcom_slot);
 221
 222  do_create:
 223         error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY,
 224                             len, &rc, &mh);
 225         if (error)
 226                 return;
 227
 228         rc->rc_id = rc_in->rc_id;
 229         rc->rc_seq_reply = rc_in->rc_seq;
 230         rc->rc_result = status;
 231
 232         set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, num_slots);
 233
 234         if (!num_slots)
 235                 goto do_send;
 236
 237         spin_lock(&ls->ls_recover_lock);
 238         if (ls->ls_num_slots != num_slots) {
 239                 spin_unlock(&ls->ls_recover_lock);
 240                 log_debug(ls, "receive_rcom_status num_slots %d to %d",
 241                           num_slots, ls->ls_num_slots);
 242                 rc->rc_result = 0;
 243                 set_rcom_config(ls, (struct rcom_config *)rc->rc_buf, 0);
 244                 goto do_send;
 245         }
 246
 247         dlm_slots_copy_out(ls, rc);
 248         spin_unlock(&ls->ls_recover_lock);
 249
 250  do_send:
 251         send_rcom(ls, mh, rc);
 252 }
 253
 254 static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 255 {
 256         spin_lock(&ls->ls_rcom_spin);
 257         if (!test_bit(LSFL_RCOM_WAIT, &ls->ls_flags) ||
 258             rc_in->rc_id != ls->ls_rcom_seq) {
 259                 log_debug(ls, "reject reply %d from %d seq %llx expect %llx",
 260                           rc_in->rc_type, rc_in->rc_header.h_nodeid,
 261                           (unsigned long long)rc_in->rc_id,
 262                           (unsigned long long)ls->ls_rcom_seq);
 263                 goto out;
 264         }
 265         memcpy(ls->ls_recover_buf, rc_in, rc_in->rc_header.h_length);
 266         set_bit(LSFL_RCOM_READY, &ls->ls_flags);
 267         clear_bit(LSFL_RCOM_WAIT, &ls->ls_flags);
 268         wake_up(&ls->ls_wait_general);
 269  out:
 270         spin_unlock(&ls->ls_rcom_spin);
 271 }
 272
 273 int dlm_rcom_names(struct dlm_ls *ls, int nodeid, char *last_name, int last_len)
 274 {
 275         struct dlm_rcom *rc;
 276         struct dlm_mhandle *mh;
 277         int error = 0;
 278         int max_size = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
 279
 280         ls->ls_recover_nodeid = nodeid;
 281
 282         if (nodeid == dlm_our_nodeid()) {
 283                 ls->ls_recover_buf->rc_header.h_length =
 284                         dlm_config.ci_buffer_size;
 285                 dlm_copy_master_names(ls, last_name, last_len,
 286                                       ls->ls_recover_buf->rc_buf,
 287                                       max_size, nodeid);
 288                 goto out;
 289         }
 290
 291         error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh);
 292         if (error)
 293                 goto out;
 294         memcpy(rc->rc_buf, last_name, last_len);
 295
 296         allow_sync_reply(ls, &rc->rc_id);
 297         memset(ls->ls_recover_buf, 0, dlm_config.ci_buffer_size);
 298
 299         send_rcom(ls, mh, rc);
 300
 301         error = dlm_wait_function(ls, &rcom_response);
 302         disallow_sync_reply(ls);
 303  out:
 304         return error;
 305 }
 306
 307 static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 308 {
 309         struct dlm_rcom *rc;
 310         struct dlm_mhandle *mh;
 311         int error, inlen, outlen, nodeid;
 312
 313         nodeid = rc_in->rc_header.h_nodeid;
 314         inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
 315         outlen = dlm_config.ci_buffer_size - sizeof(struct dlm_rcom);
 316
 317         error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh);
 318         if (error)
 319                 return;
 320         rc->rc_id = rc_in->rc_id;
 321         rc->rc_seq_reply = rc_in->rc_seq;
 322
 323         dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen,
 324                               nodeid);
 325         send_rcom(ls, mh, rc);
 326 }
 327
 328 int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid)
 329 {
 330         struct dlm_rcom *rc;
 331         struct dlm_mhandle *mh;
 332         struct dlm_ls *ls = r->res_ls;
 333         int error;
 334
 335         error = create_rcom(ls, dir_nodeid, DLM_RCOM_LOOKUP, r->res_length,
 336                             &rc, &mh);
 337         if (error)
 338                 goto out;
 339         memcpy(rc->rc_buf, r->res_name, r->res_length);
 340         rc->rc_id = (unsigned long) r;
 341
 342         send_rcom(ls, mh, rc);
 343  out:
 344         return error;
 345 }
 346
 347 static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 348 {
 349         struct dlm_rcom *rc;
 350         struct dlm_mhandle *mh;
 351         int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid;
 352         int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom);
 353
 354         error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh);
 355         if (error)
 356                 return;
 357
 358         error = dlm_dir_lookup(ls, nodeid, rc_in->rc_buf, len, &ret_nodeid);
 359         if (error)
 360                 ret_nodeid = error;
 361         rc->rc_result = ret_nodeid;
 362         rc->rc_id = rc_in->rc_id;
 363         rc->rc_seq_reply = rc_in->rc_seq;
 364
 365         send_rcom(ls, mh, rc);
 366 }
 367
 368 static void receive_rcom_lookup_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 369 {
 370         dlm_recover_master_reply(ls, rc_in);
 371 }
 372
 373 static void pack_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb,
 374                            struct rcom_lock *rl)
 375 {
 376         memset(rl, 0, sizeof(*rl));
 377
 378         rl->rl_ownpid = cpu_to_le32(lkb->lkb_ownpid);
 379         rl->rl_lkid = cpu_to_le32(lkb->lkb_id);
 380         rl->rl_exflags = cpu_to_le32(lkb->lkb_exflags);
 381         rl->rl_flags = cpu_to_le32(lkb->lkb_flags);
 382         rl->rl_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
 383         rl->rl_rqmode = lkb->lkb_rqmode;
 384         rl->rl_grmode = lkb->lkb_grmode;
 385         rl->rl_status = lkb->lkb_status;
 386         rl->rl_wait_type = cpu_to_le16(lkb->lkb_wait_type);
 387
 388         if (lkb->lkb_bastfn)
 389                 rl->rl_asts |= DLM_CB_BAST;
 390         if (lkb->lkb_astfn)
 391                 rl->rl_asts |= DLM_CB_CAST;
 392
 393         rl->rl_namelen = cpu_to_le16(r->res_length);
 394         memcpy(rl->rl_name, r->res_name, r->res_length);
 395
 396         /* FIXME: might we have an lvb without DLM_LKF_VALBLK set ?
 397            If so, receive_rcom_lock_args() won't take this copy. */
 398
 399         if (lkb->lkb_lvbptr)
 400                 memcpy(rl->rl_lvb, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
 401 }
 402
 403 int dlm_send_rcom_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 404 {
 405         struct dlm_ls *ls = r->res_ls;
 406         struct dlm_rcom *rc;
 407         struct dlm_mhandle *mh;
 408         struct rcom_lock *rl;
 409         int error, len = sizeof(struct rcom_lock);
 410
 411         if (lkb->lkb_lvbptr)
 412                 len += ls->ls_lvblen;
 413
 414         error = create_rcom(ls, r->res_nodeid, DLM_RCOM_LOCK, len, &rc, &mh);
 415         if (error)
 416                 goto out;
 417
 418         rl = (struct rcom_lock *) rc->rc_buf;
 419         pack_rcom_lock(r, lkb, rl);
 420         rc->rc_id = (unsigned long) r;
 421
 422         send_rcom(ls, mh, rc);
 423  out:
 424         return error;
 425 }
 426
 427 /* needs at least dlm_rcom + rcom_lock */
 428 static void receive_rcom_lock(struct dlm_ls *ls, struct dlm_rcom *rc_in)
 429 {
 430         struct dlm_rcom *rc;
 431         struct dlm_mhandle *mh;
 432         int error, nodeid = rc_in->rc_header.h_nodeid;
 433
 434         dlm_recover_master_copy(ls, rc_in);
 435
 436         error = create_rcom(ls, nodeid, DLM_RCOM_LOCK_REPLY,
 437                             sizeof(struct rcom_lock), &rc, &mh);
 438         if (error)
 439                 return;
 440
 441         /* We send back the same rcom_lock struct we received, but
 442            dlm_recover_master_copy() has filled in rl_remid and rl_result */
 443
 444         memcpy(rc->rc_buf, rc_in->rc_buf, sizeof(struct rcom_lock));
 445         rc->rc_id = rc_in->rc_id;
 446         rc->rc_seq_reply = rc_in->rc_seq;
 447
 448         send_rcom(ls, mh, rc);
 449 }
 450
 451 /* If the lockspace doesn't exist then still send a status message
 452    back; it's possible that it just doesn't have its global_id yet. */
 453
 454 int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
 455 {
 456         struct dlm_rcom *rc;
 457         struct rcom_config *rf;
 458         struct dlm_mhandle *mh;
 459         char *mb;
 460         int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config);
 461
 462         mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
 463         if (!mh)
 464                 return -ENOBUFS;
 465         memset(mb, 0, mb_len);
 466
 467         rc = (struct dlm_rcom *) mb;
 468
 469         rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
 470         rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace;
 471         rc->rc_header.h_nodeid = dlm_our_nodeid();
 472         rc->rc_header.h_length = mb_len;
 473         rc->rc_header.h_cmd = DLM_RCOM;
 474
 475         rc->rc_type = DLM_RCOM_STATUS_REPLY;
 476         rc->rc_id = rc_in->rc_id;
 477         rc->rc_seq_reply = rc_in->rc_seq;
 478         rc->rc_result = -ESRCH;
 479
 480         rf = (struct rcom_config *) rc->rc_buf;
 481         rf->rf_lvblen = cpu_to_le32(~0U);
 482
 483         dlm_rcom_out(rc);
 484         dlm_lowcomms_commit_buffer(mh);
 485
 486         return 0;
 487 }
 488
 489 /* Called by dlm_recv; corresponds to dlm_receive_message() but special
 490    recovery-only comms are sent through here. */
 491
 492 void dlm_receive_rcom(struct dlm_ls *ls, struct dlm_rcom *rc, int nodeid)
 493 {
 494         int lock_size = sizeof(struct dlm_rcom) + sizeof(struct rcom_lock);
 495         int stop, reply = 0, lock = 0;
 496         uint32_t status;
 497         uint64_t seq;
 498
 499         switch (rc->rc_type) {
 500         case DLM_RCOM_LOCK:
 501                 lock = 1;
 502                 break;
 503         case DLM_RCOM_LOCK_REPLY:
 504                 lock = 1;
 505                 reply = 1;
 506                 break;
 507         case DLM_RCOM_STATUS_REPLY:
 508         case DLM_RCOM_NAMES_REPLY:
 509         case DLM_RCOM_LOOKUP_REPLY:
 510                 reply = 1;
 511         };
 512
 513         spin_lock(&ls->ls_recover_lock);
 514         status = ls->ls_recover_status;
 515         stop = test_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
 516         seq = ls->ls_recover_seq;
 517         spin_unlock(&ls->ls_recover_lock);
 518
 519         if ((stop && (rc->rc_type != DLM_RCOM_STATUS)) ||
 520             (reply && (rc->rc_seq_reply != seq)) ||
 521             (lock && !(status & DLM_RS_DIR))) {
 522                 log_limit(ls, "dlm_receive_rcom ignore msg %d "
 523                           "from %d %llu %llu recover seq %llu sts %x gen %u",
 524                            rc->rc_type,
 525                            nodeid,
 526                            (unsigned long long)rc->rc_seq,
 527                            (unsigned long long)rc->rc_seq_reply,
 528                            (unsigned long long)seq,
 529                            status, ls->ls_generation);
 530                 goto out;
 531         }
 532
 533         switch (rc->rc_type) {
 534         case DLM_RCOM_STATUS:
 535                 receive_rcom_status(ls, rc);
 536                 break;
 537
 538         case DLM_RCOM_NAMES:
 539                 receive_rcom_names(ls, rc);
 540                 break;
 541
 542         case DLM_RCOM_LOOKUP:
 543                 receive_rcom_lookup(ls, rc);
 544                 break;
 545
 546         case DLM_RCOM_LOCK:
 547                 if (rc->rc_header.h_length < lock_size)
 548                         goto Eshort;
 549                 receive_rcom_lock(ls, rc);
 550                 break;
 551
 552         case DLM_RCOM_STATUS_REPLY:
 553                 receive_sync_reply(ls, rc);
 554                 break;
 555
 556         case DLM_RCOM_NAMES_REPLY:
 557                 receive_sync_reply(ls, rc);
 558                 break;
 559
 560         case DLM_RCOM_LOOKUP_REPLY:
 561                 receive_rcom_lookup_reply(ls, rc);
 562                 break;
 563
 564         case DLM_RCOM_LOCK_REPLY:
 565                 if (rc->rc_header.h_length < lock_size)
 566                         goto Eshort;
 567                 dlm_recover_process_copy(ls, rc);
 568                 break;
 569
 570         default:
 571                 log_error(ls, "receive_rcom bad type %d", rc->rc_type);
 572         }
 573 out:
 574         return;
 575 Eshort:
 576         log_error(ls, "recovery message %x from %d is too short",
 577                           rc->rc_type, nodeid);
 578 }
 579