lustre/mgs/mgs_nids.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/mgs/mgs_nids.c
  32  *
  33  * NID table management for lustre.
  34  *
  35  * Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_MGS
  39 #define D_MGS D_CONFIG
  40
  41 #include <linux/kthread.h>
  42 #include <linux/pagemap.h>
  43
  44 #include <obd.h>
  45 #include <obd_class.h>
  46 #include <lustre_disk.h>
  47
  48 #include "mgs_internal.h"
  49
  50 static time64_t ir_timeout;
  51
  52 static int nidtbl_is_sane(struct mgs_nidtbl *tbl)
  53 {
  54         struct mgs_nidtbl_target *tgt;
  55         int version = 0;
  56
  57         LASSERT(mutex_is_locked(&tbl->mn_lock));
  58         list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
  59                 if (!tgt->mnt_version)
  60                         continue;
  61
  62                 if (version >= tgt->mnt_version)
  63                         return 0;
  64
  65                 version = tgt->mnt_version;
  66         }
  67         return 1;
  68 }
  69
  70 /**
  71  * Fetch nidtbl entries whose version are not less than @version
  72  * nidtbl entries will be packed in @pages by @unit_size units - entries
  73  * shouldn't cross unit boundaries.
  74  */
  75 static int mgs_nidtbl_read(struct obd_export *exp, struct mgs_nidtbl *tbl,
  76                            struct mgs_config_res *res, u8 nid_size,
  77                            struct page **pages, int nrpages,
  78                            int units_total, int unit_size)
  79 {
  80         struct mgs_nidtbl_target *tgt;
  81         struct mgs_nidtbl_entry *entry;
  82         struct mgs_nidtbl_entry *last_in_unit = NULL;
  83         struct mgs_target_info *mti;
  84         __u64 version = res->mcr_offset;
  85         bool nobuf = false;
  86         void *buf = NULL;
  87         int bytes_in_unit = 0;
  88         int units_in_page = 0;
  89         int index = 0;
  90         int rc = 0;
  91
  92         ENTRY;
  93
  94         /* make sure unit_size is power 2 */
  95         LASSERT((unit_size & (unit_size - 1)) == 0);
  96         LASSERT(nrpages << PAGE_SHIFT >= units_total * unit_size);
  97
  98         mutex_lock(&tbl->mn_lock);
  99         LASSERT(nidtbl_is_sane(tbl));
 100
 101         /* no more entries ? */
 102         if (version > tbl->mn_version) {
 103                 version = tbl->mn_version;
 104                 goto out;
 105         }
 106
 107         /*
 108          * iterate over all targets to compose a bitmap by the type of llog.
 109          * If the llog is for MDTs, llog entries for OSTs will be returned;
 110          * otherwise, it's for clients, then llog entries for both OSTs and
 111          * MDTs will be returned.
 112          */
 113         list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
 114                 int entry_len = sizeof(*entry);
 115                 int i;
 116
 117                 if (tgt->mnt_version < version)
 118                         continue;
 119
 120                 /* write target recover information */
 121                 mti  = &tgt->mnt_mti;
 122                 if (!nid_size)
 123                         entry_len += mti->mti_nid_count * sizeof(lnet_nid_t);
 124                 else
 125                         entry_len += mti->mti_nid_count * nid_size;
 126
 127                 if (entry_len > unit_size) {
 128                         CWARN("nidtbl: too large entry: entry length %d, unit size: %d\n",
 129                               entry_len, unit_size);
 130                         GOTO(out, rc = -EOVERFLOW);
 131                 }
 132
 133                 if (bytes_in_unit < entry_len) {
 134                         if (units_total == 0) {
 135                                 nobuf = true;
 136                                 break;
 137                         }
 138
 139                         /* check if we need to consume remaining bytes. */
 140                         if (last_in_unit && bytes_in_unit) {
 141                                 last_in_unit->mne_length += bytes_in_unit;
 142                                 rc  += bytes_in_unit;
 143                                 buf += bytes_in_unit;
 144                                 last_in_unit = NULL;
 145                         }
 146                         LASSERT((rc & (unit_size - 1)) == 0);
 147
 148                         if (units_in_page == 0) {
 149                                 /* allocate a new page */
 150                                 pages[index] = alloc_page(GFP_KERNEL);
 151                                 if (!pages[index]) {
 152                                         rc = -ENOMEM;
 153                                         break;
 154                                 }
 155
 156                                 /* destroy previous map */
 157                                 if (index > 0)
 158                                         kunmap(pages[index - 1]);
 159
 160                                 /* reassign buffer */
 161                                 buf = kmap(pages[index]);
 162                                 ++index;
 163
 164                                 units_in_page = PAGE_SIZE / unit_size;
 165                                 LASSERT(units_in_page > 0);
 166                         }
 167
 168                         /* allocate an unit */
 169                         LASSERT(((long)buf & (unit_size - 1)) == 0);
 170                         bytes_in_unit = unit_size;
 171                         --units_in_page;
 172                         --units_total;
 173                 }
 174
 175                 /* fill in entry. */
 176                 entry = (struct mgs_nidtbl_entry *)buf;
 177                 entry->mne_version = tgt->mnt_version;
 178                 entry->mne_instance = mti->mti_instance;
 179                 entry->mne_index = mti->mti_stripe_index;
 180                 entry->mne_length = entry_len;
 181                 entry->mne_type = tgt->mnt_type;
 182                 if (nid_size) {
 183                         entry->mne_nid_size = nid_size;
 184                         entry->mne_nid_type = 1;
 185                 } else {
 186                         entry->mne_nid_size = sizeof(lnet_nid_t);
 187                         entry->mne_nid_type = 0;
 188                 }
 189                 entry->mne_nid_count = 0;
 190                 /* We have been sent the newer larger NID format but the
 191                  * current nidtbl doesn't support it. So filter the NIDs
 192                  * sent to reject any real larger size NIDS.
 193                  */
 194                 if (target_supports_large_nid(mti)) {
 195                         for (i = 0; i < mti->mti_nid_count; i++) {
 196                                 struct lnet_nid nid;
 197                                 int err;
 198
 199                                 err = libcfs_strnid(&nid, mti->mti_nidlist[i]);
 200                                 if (err < 0)
 201                                         GOTO(out, rc = err);
 202
 203                                 if (nid_size == 0) {
 204                                         if (!nid_is_nid4(&nid))
 205                                                 continue;
 206
 207                                         entry->u.nids[entry->mne_nid_count] =
 208                                                 lnet_nid_to_nid4(&nid);
 209                                 } else {
 210                                         /* If the mgs_target_info NIDs are
 211                                          * struct lnet_nid that have been
 212                                          * expanded in size we still can
 213                                          * use the nid if it fits in what
 214                                          * the client supports.
 215                                          */
 216                                         if (NID_BYTES(&nid) > nid_size)
 217                                                 continue;
 218
 219                                         entry->u.nidlist[entry->mne_nid_count] =
 220                                                 nid;
 221                                 }
 222                                 entry->mne_nid_count++;
 223                         }
 224                 } else {
 225                         if (nid_size) {
 226                                 for (i = 0; i < mti->mti_nid_count; i++)
 227                                         lnet_nid4_to_nid(mti->mti_nids[i],
 228                                                          &entry->u.nidlist[i]);
 229                         } else {
 230                                 memcpy(entry->u.nids, mti->mti_nids,
 231                                        mti->mti_nid_count * sizeof(lnet_nid_t));
 232                         }
 233                         entry->mne_nid_count = mti->mti_nid_count;
 234                 }
 235
 236                 version = tgt->mnt_version;
 237                 rc     += entry_len;
 238                 buf    += entry_len;
 239
 240                 bytes_in_unit -= entry_len;
 241                 last_in_unit   = entry;
 242
 243                 CDEBUG(D_MGS, "fsname %s, entry size %d, pages %d/%d/%d/%d.\n",
 244                        tbl->mn_fsdb->fsdb_name, entry_len,
 245                        bytes_in_unit, index, nrpages, units_total);
 246         }
 247         if (index > 0)
 248                 kunmap(pages[index - 1]);
 249 out:
 250         LASSERT(version <= tbl->mn_version);
 251         res->mcr_size = tbl->mn_version;
 252         res->mcr_offset = nobuf ? version : tbl->mn_version;
 253         mutex_unlock(&tbl->mn_lock);
 254         LASSERT(ergo(version == 1, rc == 0)); /* get the log first time */
 255
 256         CDEBUG(D_MGS, "Read IR logs %s return with %d, version %llu\n",
 257                tbl->mn_fsdb->fsdb_name, rc, version);
 258         RETURN(rc);
 259 }
 260
 261 static int nidtbl_update_version(const struct lu_env *env,
 262                                  struct mgs_device *mgs,
 263                                  struct mgs_nidtbl *tbl)
 264 {
 265         struct dt_object *fsdb;
 266         struct thandle *th;
 267         u64 version;
 268         struct lu_buf buf = {
 269                         .lb_buf = &version,
 270                         .lb_len = sizeof(version)
 271         };
 272         loff_t off = 0;
 273         int rc;
 274
 275         ENTRY;
 276
 277         if (mgs->mgs_bottom->dd_rdonly)
 278                 RETURN(0);
 279
 280         LASSERT(mutex_is_locked(&tbl->mn_lock));
 281
 282         fsdb = local_file_find_or_create(env, mgs->mgs_los, mgs->mgs_nidtbl_dir,
 283                                          tbl->mn_fsdb->fsdb_name,
 284                                          S_IFREG | S_IRUGO | S_IWUSR);
 285         if (IS_ERR(fsdb))
 286                 RETURN(PTR_ERR(fsdb));
 287
 288         th = dt_trans_create(env, mgs->mgs_bottom);
 289         if (IS_ERR(th))
 290                 GOTO(out_put, rc = PTR_ERR(th));
 291
 292         th->th_sync = 1; /* update table synchronously */
 293         rc = dt_declare_record_write(env, fsdb, &buf, off, th);
 294         if (rc)
 295                 GOTO(out, rc);
 296
 297         rc = dt_trans_start_local(env, mgs->mgs_bottom, th);
 298         if (rc)
 299                 GOTO(out, rc);
 300
 301         version = cpu_to_le64(tbl->mn_version);
 302         rc = dt_record_write(env, fsdb, &buf, &off, th);
 303
 304 out:
 305         dt_trans_stop(env, mgs->mgs_bottom, th);
 306 out_put:
 307         dt_object_put(env, fsdb);
 308         RETURN(rc);
 309 }
 310
 311 #define MGS_NIDTBL_VERSION_INIT 2
 312
 313 static int nidtbl_read_version(const struct lu_env *env,
 314                                struct mgs_device *mgs, struct mgs_nidtbl *tbl,
 315                                u64 *version)
 316 {
 317         struct dt_object *fsdb;
 318         struct lu_fid fid;
 319         u64 tmpver;
 320         struct lu_buf buf = {
 321                 .lb_buf = &tmpver,
 322                 .lb_len = sizeof(tmpver)
 323         };
 324         loff_t off = 0;
 325         int rc;
 326
 327         ENTRY;
 328
 329         LASSERT(mutex_is_locked(&tbl->mn_lock));
 330
 331         LASSERT(mgs->mgs_nidtbl_dir);
 332         rc = dt_lookup_dir(env, mgs->mgs_nidtbl_dir, tbl->mn_fsdb->fsdb_name,
 333                            &fid);
 334         if (rc == -ENOENT) {
 335                 *version = MGS_NIDTBL_VERSION_INIT;
 336                 RETURN(0);
 337         } else if (rc < 0) {
 338                 RETURN(rc);
 339         }
 340
 341         fsdb = dt_locate_at(env, mgs->mgs_bottom, &fid,
 342                             &mgs->mgs_dt_dev.dd_lu_dev, NULL);
 343         if (IS_ERR(fsdb))
 344                 RETURN(PTR_ERR(fsdb));
 345
 346         rc = dt_read(env, fsdb, &buf, &off);
 347         if (rc == buf.lb_len) {
 348                 *version = le64_to_cpu(tmpver);
 349                 rc = 0;
 350         } else if (rc == 0) {
 351                 *version = MGS_NIDTBL_VERSION_INIT;
 352         } else {
 353                 CERROR("%s: read version file %s error %d\n",
 354                        mgs->mgs_obd->obd_name, tbl->mn_fsdb->fsdb_name, rc);
 355         }
 356         dt_object_put(env, fsdb);
 357         RETURN(rc);
 358 }
 359
 360 static int mgs_nidtbl_write(const struct lu_env *env, struct fs_db *fsdb,
 361                             struct mgs_target_info *mti)
 362 {
 363         struct mgs_nidtbl *tbl;
 364         struct mgs_nidtbl_target *tgt;
 365         bool found = false;
 366         int type = mti->mti_flags & LDD_F_SV_TYPE_MASK;
 367         size_t mti_len = 0;
 368         int rc = 0;
 369
 370         ENTRY;
 371         type &= ~LDD_F_SV_TYPE_MGS;
 372         LASSERT(type != 0);
 373
 374         tbl = &fsdb->fsdb_nidtbl;
 375         mutex_lock(&tbl->mn_lock);
 376         list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
 377                 struct mgs_target_info *info = &tgt->mnt_mti;
 378
 379                 if (type == tgt->mnt_type &&
 380                     mti->mti_stripe_index == info->mti_stripe_index) {
 381                         found = true;
 382                         break;
 383                 }
 384         }
 385
 386         if (target_supports_large_nid(mti))
 387                 mti_len = mti->mti_nid_count * LNET_NIDSTR_SIZE;
 388         if (!found) {
 389                 size_t len = offsetof(struct mgs_nidtbl_target,
 390                                       mnt_mti.mti_nidlist);
 391
 392                 OBD_ALLOC(tgt, len + mti_len);
 393                 if (!tgt)
 394                         GOTO(out, rc = -ENOMEM);
 395
 396                 INIT_LIST_HEAD(&tgt->mnt_list);
 397                 tgt->mnt_fs = tbl;
 398                 tgt->mnt_version = 0; /* 0 means invalid */
 399                 tgt->mnt_type = type;
 400
 401                 ++tbl->mn_nr_targets;
 402         }
 403
 404         tgt->mnt_version = ++tbl->mn_version;
 405         tgt->mnt_mti = *mti;
 406         if (target_supports_large_nid(mti))
 407                 memcpy(tgt->mnt_mti.mti_nidlist, mti->mti_nidlist, mti_len);
 408
 409         list_move_tail(&tgt->mnt_list, &tbl->mn_targets);
 410
 411         rc = nidtbl_update_version(env, fsdb->fsdb_mgs, tbl);
 412         EXIT;
 413
 414 out:
 415         mutex_unlock(&tbl->mn_lock);
 416         if (rc)
 417                 CERROR("Write NID table version for file system %s error %d\n",
 418                        fsdb->fsdb_name, rc);
 419         return rc;
 420 }
 421
 422 static void mgs_nidtbl_fini_fs(struct fs_db *fsdb)
 423 {
 424         struct mgs_nidtbl *tbl = &fsdb->fsdb_nidtbl;
 425         size_t len = offsetof(struct mgs_nidtbl_target,
 426                               mnt_mti.mti_nidlist);
 427         LIST_HEAD(head);
 428
 429         mutex_lock(&tbl->mn_lock);
 430         tbl->mn_nr_targets = 0;
 431         list_splice_init(&tbl->mn_targets, &head);
 432         mutex_unlock(&tbl->mn_lock);
 433
 434         while (!list_empty(&head)) {
 435                 struct mgs_nidtbl_target *tgt;
 436                 size_t mti_len = 0;
 437
 438                 tgt = list_first_entry(&head, struct mgs_nidtbl_target,
 439                                        mnt_list);
 440                 if (target_supports_large_nid(&tgt->mnt_mti))
 441                         mti_len += tgt->mnt_mti.mti_nid_count * LNET_NIDSTR_SIZE;
 442                 list_del(&tgt->mnt_list);
 443                 OBD_FREE(tgt, len + mti_len);
 444         }
 445 }
 446
 447 static int mgs_nidtbl_init_fs(const struct lu_env *env, struct fs_db *fsdb)
 448 {
 449         struct mgs_nidtbl *tbl = &fsdb->fsdb_nidtbl;
 450         int rc;
 451
 452         INIT_LIST_HEAD(&tbl->mn_targets);
 453         mutex_init(&tbl->mn_lock);
 454         tbl->mn_nr_targets = 0;
 455         tbl->mn_fsdb = fsdb;
 456         mutex_lock(&tbl->mn_lock);
 457         rc = nidtbl_read_version(env, fsdb->fsdb_mgs, tbl, &tbl->mn_version);
 458         mutex_unlock(&tbl->mn_lock);
 459         if (rc < 0)
 460                 CERROR("%s: IR: failed to read current version, rc = %d\n",
 461                        fsdb->fsdb_mgs->mgs_obd->obd_name, rc);
 462         else
 463                 CDEBUG(D_MGS, "IR: current version is %llu\n",
 464                        tbl->mn_version);
 465
 466         return rc;
 467 }
 468
 469 /* --------- Imperative Recovery relies on nidtbl stuff ------- */
 470 void mgs_ir_notify_complete(struct fs_db *fsdb)
 471 {
 472         struct timespec64 ts;
 473         ktime_t delta;
 474
 475         atomic_set(&fsdb->fsdb_notify_phase, 0);
 476
 477         /* do statistic */
 478         fsdb->fsdb_notify_count++;
 479         delta = ktime_sub(ktime_get(), fsdb->fsdb_notify_start);
 480         fsdb->fsdb_notify_total = ktime_add(fsdb->fsdb_notify_total, delta);
 481         if (ktime_after(delta, fsdb->fsdb_notify_max))
 482                 fsdb->fsdb_notify_max = delta;
 483
 484         ts = ktime_to_timespec64(fsdb->fsdb_notify_max);
 485         CDEBUG(D_MGS, "Revoke recover lock of %s completed after %lld.%09lds\n",
 486                fsdb->fsdb_name, (s64)ts.tv_sec, ts.tv_nsec);
 487 }
 488
 489 static int mgs_ir_notify(void *arg)
 490 {
 491         struct fs_db *fsdb = arg;
 492         struct ldlm_res_id resid;
 493         char name[sizeof(fsdb->fsdb_name) + 16];
 494
 495         BUILD_BUG_ON(sizeof(name) >= 40); /* name is too large to be on stack */
 496
 497         snprintf(name, sizeof(name) - 1, "mgs_%s_notify", fsdb->fsdb_name);
 498         complete(&fsdb->fsdb_notify_comp);
 499         set_user_nice(current, -2);
 500         mgc_fsname2resid(fsdb->fsdb_name, &resid, MGS_CFG_T_RECOVER);
 501         while (1) {
 502                 wait_event_idle(fsdb->fsdb_notify_waitq,
 503                                 fsdb->fsdb_notify_stop ||
 504                                 atomic_read(&fsdb->fsdb_notify_phase));
 505
 506                 if (fsdb->fsdb_notify_stop)
 507                         break;
 508
 509                 CDEBUG(D_MGS, "%s woken up, phase is %d\n",
 510                        name, atomic_read(&fsdb->fsdb_notify_phase));
 511
 512                 fsdb->fsdb_notify_start = ktime_get();
 513                 mgs_revoke_lock(fsdb->fsdb_mgs, fsdb, MGS_CFG_T_RECOVER);
 514         }
 515
 516         complete(&fsdb->fsdb_notify_comp);
 517         return 0;
 518 }
 519
 520 int mgs_ir_init_fs(const struct lu_env *env, struct mgs_device *mgs,
 521                    struct fs_db *fsdb)
 522 {
 523         struct task_struct *task;
 524
 525         if (!ir_timeout)
 526                 ir_timeout = (time64_t)OBD_IR_MGS_TIMEOUT;
 527
 528         fsdb->fsdb_ir_state = IR_FULL;
 529         if (mgs->mgs_start_time + ir_timeout > ktime_get_real_seconds())
 530                 fsdb->fsdb_ir_state = IR_STARTUP;
 531         fsdb->fsdb_nonir_clients = 0;
 532         /* start notify thread */
 533         fsdb->fsdb_mgs = mgs;
 534         task = kthread_run(mgs_ir_notify, fsdb,
 535                                "mgs_%s_notify", fsdb->fsdb_name);
 536         if (!IS_ERR(task))
 537                 wait_for_completion(&fsdb->fsdb_notify_comp);
 538         else
 539                 CERROR("Start notify thread error %ld\n", PTR_ERR(task));
 540
 541         mgs_nidtbl_init_fs(env, fsdb);
 542         return 0;
 543 }
 544
 545 void mgs_ir_fini_fs(struct mgs_device *mgs, struct fs_db *fsdb)
 546 {
 547         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags))
 548                 return;
 549
 550         mgs_fsc_cleanup_by_fsdb(fsdb);
 551
 552         mgs_nidtbl_fini_fs(fsdb);
 553
 554         LASSERT(list_empty(&fsdb->fsdb_clients));
 555
 556         fsdb->fsdb_notify_stop = 1;
 557         wake_up(&fsdb->fsdb_notify_waitq);
 558         wait_for_completion(&fsdb->fsdb_notify_comp);
 559 }
 560
 561 /* caller must have held fsdb_mutex */
 562 static inline void ir_state_graduate(struct fs_db *fsdb)
 563 {
 564         if (fsdb->fsdb_ir_state == IR_STARTUP) {
 565                 if (ktime_get_real_seconds() >
 566                     fsdb->fsdb_mgs->mgs_start_time + ir_timeout) {
 567                         fsdb->fsdb_ir_state = IR_FULL;
 568                         if (fsdb->fsdb_nonir_clients)
 569                                 fsdb->fsdb_ir_state = IR_PARTIAL;
 570                 }
 571         }
 572 }
 573
 574 int mgs_ir_update(const struct lu_env *env, struct mgs_device *mgs,
 575                   struct mgs_target_info *mti)
 576 {
 577         struct fs_db *fsdb;
 578         bool notify = true;
 579         int rc;
 580
 581         if (mti->mti_instance == 0)
 582                 return -EINVAL;
 583
 584         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 585         if (rc)
 586                 return rc;
 587
 588         rc = mgs_nidtbl_write(env, fsdb, mti);
 589         if (rc)
 590                 GOTO(out, rc);
 591
 592         /* check ir state */
 593         mutex_lock(&fsdb->fsdb_mutex);
 594         ir_state_graduate(fsdb);
 595         switch (fsdb->fsdb_ir_state) {
 596         case IR_FULL:
 597                 mti->mti_flags |= LDD_F_IR_CAPABLE;
 598                 break;
 599         case IR_DISABLED:
 600                 notify = false;
 601         case IR_STARTUP:
 602         case IR_PARTIAL:
 603                 break;
 604         default:
 605                 LBUG();
 606         }
 607         mutex_unlock(&fsdb->fsdb_mutex);
 608
 609         LASSERT(ergo(mti->mti_flags & LDD_F_IR_CAPABLE, notify));
 610         if (notify) {
 611                 CDEBUG(D_MGS, "Try to revoke recover lock of %s\n",
 612                        fsdb->fsdb_name);
 613                 atomic_inc(&fsdb->fsdb_notify_phase);
 614                 wake_up(&fsdb->fsdb_notify_waitq);
 615         }
 616
 617 out:
 618         mgs_put_fsdb(mgs, fsdb);
 619         return rc;
 620 }
 621
 622 /* NID table can be cached by two entities: Clients and MDTs */
 623 enum {
 624         IR_CLIENT  = 1,
 625         IR_MDT     = 2
 626 };
 627
 628 static int delogname(char *logname, char *fsname, int *typ)
 629 {
 630         char *ptr;
 631         int type;
 632         int len;
 633
 634         ptr = strrchr(logname, '-');
 635         if (!ptr)
 636                 return -EINVAL;
 637
 638         /*
 639          * decouple file system name. The llog name may be:
 640          * - "prefix-fsname", prefix is "cliir" or "mdtir"
 641          */
 642         if (strncmp(ptr, "-mdtir", 6) == 0)
 643                 type = IR_MDT;
 644         else if (strncmp(ptr, "-cliir", 6) == 0)
 645                 type = IR_CLIENT;
 646         else
 647                 return -EINVAL;
 648
 649         len = ptr - logname;
 650         if (len == 0)
 651                 return -EINVAL;
 652
 653         memcpy(fsname, logname, len);
 654         fsname[len] = 0;
 655         if (typ)
 656                 *typ = type;
 657         return 0;
 658 }
 659
 660 int mgs_get_ir_logs(struct ptlrpc_request *req)
 661 {
 662         struct lu_env *env = req->rq_svc_thread->t_env;
 663         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 664         struct fs_db *fsdb = NULL;
 665         struct mgs_config_body *body;
 666         struct mgs_config_res *res;
 667         struct ptlrpc_bulk_desc *desc;
 668         char fsname[16];
 669         long bufsize;
 670         int unit_size;
 671         int type;
 672         int rc = 0;
 673         int i;
 674         int bytes;
 675         int page_count;
 676         int nrpages;
 677         struct page **pages = NULL;
 678
 679         ENTRY;
 680
 681         body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
 682         if (!body)
 683                 RETURN(-EINVAL);
 684
 685         if (body->mcb_type != MGS_CFG_T_RECOVER)
 686                 RETURN(-EINVAL);
 687
 688         rc = delogname(body->mcb_name, fsname, &type);
 689         if (rc)
 690                 RETURN(rc);
 691
 692         bufsize = body->mcb_units << body->mcb_bits;
 693         nrpages = (bufsize + PAGE_SIZE - 1) >> PAGE_SHIFT;
 694         if (nrpages > PTLRPC_MAX_BRW_PAGES)
 695                 RETURN(-EINVAL);
 696
 697         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
 698         if (rc)
 699                 RETURN(rc);
 700
 701         CDEBUG(D_MGS, "Reading IR log %s bufsize %ld.\n",
 702                body->mcb_name, bufsize);
 703
 704         OBD_ALLOC_PTR_ARRAY_LARGE(pages, nrpages);
 705         if (!pages)
 706                 GOTO(out, rc = -ENOMEM);
 707
 708         res = req_capsule_server_get(&req->rq_pill, &RMF_MGS_CONFIG_RES);
 709         if (!res)
 710                 GOTO(out, rc = -EINVAL);
 711
 712         res->mcr_offset = body->mcb_offset;
 713         unit_size = min_t(int, 1 << body->mcb_bits, PAGE_SIZE);
 714         bytes = mgs_nidtbl_read(req->rq_export, &fsdb->fsdb_nidtbl, res,
 715                                 body->mcb_rec_nid_size, pages, nrpages,
 716                                 bufsize / unit_size, unit_size);
 717         if (bytes < 0)
 718                 GOTO(out, rc = bytes);
 719
 720         /* start bulk transfer */
 721         page_count = (bytes + PAGE_SIZE - 1) >> PAGE_SHIFT;
 722         LASSERT(page_count <= nrpages);
 723         desc = ptlrpc_prep_bulk_exp(req, page_count, 1,
 724                                     PTLRPC_BULK_PUT_SOURCE,
 725                                     MGS_BULK_PORTAL,
 726                                     &ptlrpc_bulk_kiov_pin_ops);
 727         if (!desc)
 728                 GOTO(out, rc = -ENOMEM);
 729
 730         for (i = 0; i < page_count && bytes > 0; i++) {
 731                 desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
 732                                                  min_t(int, bytes,
 733                                                       PAGE_SIZE));
 734                 bytes -= PAGE_SIZE;
 735         }
 736
 737         rc = target_bulk_io(req->rq_export, desc);
 738         ptlrpc_free_bulk(desc);
 739
 740         GOTO(out, rc);
 741
 742 out:
 743         if (pages) {
 744                 for (i = 0; i < nrpages; i++) {
 745                         if (!pages[i])
 746                                 break;
 747
 748                         __free_page(pages[i]);
 749                 }
 750
 751                 OBD_FREE_PTR_ARRAY_LARGE(pages, nrpages);
 752         }
 753
 754         if (fsdb)
 755                 mgs_put_fsdb(mgs, fsdb);
 756
 757         return rc;
 758 }
 759
 760 static int lprocfs_ir_set_state(struct fs_db *fsdb, const char *buf)
 761 {
 762         const char *const strings[] = IR_STRINGS;
 763         int state = -1;
 764         int i;
 765
 766         for (i = 0; i < ARRAY_SIZE(strings); i++) {
 767                 if (strcmp(strings[i], buf) == 0) {
 768                         state = i;
 769                         break;
 770                 }
 771         }
 772         if (state < 0)
 773                 return -EINVAL;
 774
 775         CDEBUG(D_MGS, "change fsr state of %s from %s to %s\n",
 776                fsdb->fsdb_name, strings[fsdb->fsdb_ir_state], strings[state]);
 777         mutex_lock(&fsdb->fsdb_mutex);
 778         if (state == IR_FULL && fsdb->fsdb_nonir_clients)
 779                 state = IR_PARTIAL;
 780         fsdb->fsdb_ir_state = state;
 781         mutex_unlock(&fsdb->fsdb_mutex);
 782
 783         return 0;
 784 }
 785
 786 static int lprocfs_ir_set_timeout(struct fs_db *fsdb, const char *buf)
 787 {
 788         return -EINVAL;
 789 }
 790
 791 static int lprocfs_ir_clear_stats(struct fs_db *fsdb, const char *buf)
 792 {
 793         if (*buf)
 794                 return -EINVAL;
 795
 796         fsdb->fsdb_notify_total = ktime_set(0, 0);
 797         fsdb->fsdb_notify_max = ktime_set(0, 0);
 798         fsdb->fsdb_notify_count = 0;
 799         return 0;
 800 }
 801
 802 static struct lproc_ir_cmd {
 803         char *name;
 804         int namelen;
 805         int (*handler)(struct fs_db *, const char *);
 806 } ir_cmds[] = {
 807         { "state=",   6, lprocfs_ir_set_state },
 808         { "timeout=", 8, lprocfs_ir_set_timeout },
 809         { "0",        1, lprocfs_ir_clear_stats }
 810 };
 811
 812 int lprocfs_wr_ir_state(struct file *file, const char __user *buffer,
 813                         size_t count, void *data)
 814 {
 815         struct fs_db *fsdb = data;
 816         char *kbuf;
 817         char *ptr;
 818         int rc = 0;
 819
 820         if (count == 0 || count >= PAGE_SIZE)
 821                 return -EINVAL;
 822
 823         OBD_ALLOC(kbuf, count + 1);
 824         if (!kbuf)
 825                 return -ENOMEM;
 826
 827         if (copy_from_user(kbuf, buffer, count)) {
 828                 OBD_FREE(kbuf, count + 1);
 829                 return -EFAULT;
 830         }
 831
 832         kbuf[count] = 0; /* buffer is supposed to end with 0 */
 833         if (kbuf[count - 1] == '\n')
 834                 kbuf[count - 1] = 0;
 835         ptr = kbuf;
 836
 837         /* fsname=<file system name> must be the 1st entry */
 838         while (ptr) {
 839                 char *tmpptr;
 840                 int i;
 841
 842                 tmpptr = strchr(ptr, ';');
 843                 if (tmpptr)
 844                         *tmpptr++ = 0;
 845
 846                 rc = -EINVAL;
 847                 for (i = 0; i < ARRAY_SIZE(ir_cmds); i++) {
 848                         struct lproc_ir_cmd *cmd;
 849                         int cmdlen;
 850
 851                         cmd    = &ir_cmds[i];
 852                         cmdlen = cmd->namelen;
 853                         if (strncmp(cmd->name, ptr, cmdlen) == 0) {
 854                                 ptr += cmdlen;
 855                                 rc = cmd->handler(fsdb, ptr);
 856                                 break;
 857                         }
 858                 }
 859                 if (rc)
 860                         break;
 861
 862                 ptr = tmpptr;
 863         }
 864         if (rc)
 865                 CERROR("Unable to process command: %s(%d)\n", ptr, rc);
 866         OBD_FREE(kbuf, count + 1);
 867         return rc ?: count;
 868 }
 869
 870 int lprocfs_rd_ir_state(struct seq_file *seq, void *data)
 871 {
 872         struct fs_db *fsdb = data;
 873         struct mgs_nidtbl *tbl = &fsdb->fsdb_nidtbl;
 874         const char *const ir_strings[] = IR_STRINGS;
 875         struct timespec64 ts_max;
 876         struct timespec64 ts;
 877
 878         /* mgs_live_seq_show() already holds fsdb_mutex. */
 879         ir_state_graduate(fsdb);
 880
 881         seq_printf(seq, "\nimperative_recovery_state:\n");
 882         seq_printf(seq,
 883                    "    state: %s\n"
 884                    "    nonir_clients: %d\n"
 885                    "    nidtbl_version: %lld\n",
 886                    ir_strings[fsdb->fsdb_ir_state], fsdb->fsdb_nonir_clients,
 887                    tbl->mn_version);
 888
 889         ts = ktime_to_timespec64(fsdb->fsdb_notify_total);
 890         ts_max = ktime_to_timespec64(fsdb->fsdb_notify_max);
 891
 892         seq_printf(seq, "    notify_duration_total: %lld.%09ld\n"
 893                         "    notify_duation_max: %lld.%09ld\n"
 894                         "    notify_count: %u\n",
 895                    (s64)ts.tv_sec, ts.tv_nsec,
 896                    (s64)ts_max.tv_sec, ts_max.tv_nsec,
 897                    fsdb->fsdb_notify_count);
 898
 899         return 0;
 900 }
 901
 902 int lprocfs_ir_timeout_seq_show(struct seq_file *m, void *data)
 903 {
 904         seq_printf(m, "%lld\n", ir_timeout);
 905         return 0;
 906 }
 907
 908 ssize_t lprocfs_ir_timeout_seq_write(struct file *file,
 909                                      const char __user *buffer,
 910                                      size_t count, loff_t *off)
 911 {
 912         return kstrtoll_from_user(buffer, count, 0, &ir_timeout);
 913 }
 914
 915 /* --------------- Handle non IR support clients --------------- */
 916 /* attach a lustre file system to an export */
 917 int mgs_fsc_attach(const struct lu_env *env, struct obd_export *exp,
 918                    char *fsname)
 919 {
 920         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 921         struct mgs_device *mgs = exp2mgs_dev(exp);
 922         struct fs_db *fsdb = NULL;
 923         struct mgs_fsc *fsc = NULL;
 924         struct mgs_fsc *new_fsc = NULL;
 925         bool found = false;
 926         int rc;
 927
 928         ENTRY;
 929
 930         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
 931         if (rc)
 932                 RETURN(rc);
 933
 934         /* allocate a new fsc in case we need it in spinlock. */
 935         OBD_ALLOC_PTR(new_fsc);
 936         if (!new_fsc)
 937                 GOTO(out, rc = -ENOMEM);
 938
 939         INIT_LIST_HEAD(&new_fsc->mfc_export_list);
 940         INIT_LIST_HEAD(&new_fsc->mfc_fsdb_list);
 941         new_fsc->mfc_fsdb       = fsdb;
 942         new_fsc->mfc_export     = class_export_get(exp);
 943         new_fsc->mfc_ir_capable = !!(exp_connect_flags(exp) &
 944                                      OBD_CONNECT_IMP_RECOV);
 945
 946         rc = -EEXIST;
 947         mutex_lock(&fsdb->fsdb_mutex);
 948
 949         /* tend to find it in export list because this list is shorter. */
 950         spin_lock(&data->med_lock);
 951         list_for_each_entry(fsc, &data->med_clients, mfc_export_list) {
 952                 if (strcmp(fsname, fsc->mfc_fsdb->fsdb_name) == 0) {
 953                         found = true;
 954                         break;
 955                 }
 956         }
 957         if (!found) {
 958                 fsc = new_fsc;
 959                 new_fsc = NULL;
 960
 961                 /* add it into export list. */
 962                 list_add(&fsc->mfc_export_list, &data->med_clients);
 963
 964                 /* add into fsdb list. */
 965                 list_add(&fsc->mfc_fsdb_list, &fsdb->fsdb_clients);
 966                 if (!fsc->mfc_ir_capable) {
 967                         ++fsdb->fsdb_nonir_clients;
 968                         if (fsdb->fsdb_ir_state == IR_FULL)
 969                                 fsdb->fsdb_ir_state = IR_PARTIAL;
 970                 }
 971                 rc = 0;
 972         }
 973         spin_unlock(&data->med_lock);
 974         mutex_unlock(&fsdb->fsdb_mutex);
 975
 976         if (new_fsc) {
 977                 class_export_put(new_fsc->mfc_export);
 978                 OBD_FREE_PTR(new_fsc);
 979         }
 980
 981 out:
 982         mgs_put_fsdb(mgs, fsdb);
 983         RETURN(rc);
 984 }
 985
 986 void mgs_fsc_cleanup(struct obd_export *exp)
 987 {
 988         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 989         struct mgs_fsc *fsc, *tmp;
 990         LIST_HEAD(head);
 991
 992         spin_lock(&data->med_lock);
 993         list_splice_init(&data->med_clients, &head);
 994         spin_unlock(&data->med_lock);
 995
 996         list_for_each_entry_safe(fsc, tmp, &head, mfc_export_list) {
 997                 struct fs_db *fsdb = fsc->mfc_fsdb;
 998
 999                 LASSERT(fsc->mfc_export == exp);
1000
1001                 mutex_lock(&fsdb->fsdb_mutex);
1002                 list_del_init(&fsc->mfc_fsdb_list);
1003                 if (fsc->mfc_ir_capable == 0) {
1004                         --fsdb->fsdb_nonir_clients;
1005                         LASSERT(fsdb->fsdb_ir_state != IR_FULL);
1006                         if (fsdb->fsdb_nonir_clients == 0 &&
1007                             fsdb->fsdb_ir_state == IR_PARTIAL)
1008                                 fsdb->fsdb_ir_state = IR_FULL;
1009                 }
1010                 mutex_unlock(&fsdb->fsdb_mutex);
1011                 list_del_init(&fsc->mfc_export_list);
1012                 class_export_put(fsc->mfc_export);
1013                 OBD_FREE_PTR(fsc);
1014         }
1015 }
1016
1017 /* must be called with fsdb->fsdb_mutex held */
1018 void mgs_fsc_cleanup_by_fsdb(struct fs_db *fsdb)
1019 {
1020         struct mgs_fsc *fsc, *tmp;
1021
1022         list_for_each_entry_safe(fsc, tmp, &fsdb->fsdb_clients,
1023                                      mfc_fsdb_list) {
1024                 struct mgs_export_data *data = &fsc->mfc_export->u.eu_mgs_data;
1025
1026                 LASSERT(fsdb == fsc->mfc_fsdb);
1027                 list_del_init(&fsc->mfc_fsdb_list);
1028
1029                 spin_lock(&data->med_lock);
1030                 list_del_init(&fsc->mfc_export_list);
1031                 spin_unlock(&data->med_lock);
1032                 class_export_put(fsc->mfc_export);
1033                 OBD_FREE_PTR(fsc);
1034         }
1035
1036         fsdb->fsdb_nonir_clients = 0;
1037         if (fsdb->fsdb_ir_state == IR_PARTIAL)
1038                 fsdb->fsdb_ir_state = IR_FULL;
1039 }