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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mgs/mgs_nids.c
  37  *
  38  * NID table management for lustre.
  39  *
  40  * Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_MGS
  44 #define D_MGS D_CONFIG
  45
  46 #ifdef __KERNEL__
  47 #include <linux/pagemap.h>
  48 #endif
  49
  50 #include <obd.h>
  51 #include <obd_class.h>
  52 #include <lustre_disk.h>
  53
  54 #include "mgs_internal.h"
  55
  56 static unsigned int ir_timeout;
  57
  58 static int nidtbl_is_sane(struct mgs_nidtbl *tbl)
  59 {
  60         struct mgs_nidtbl_target *tgt;
  61         int version = 0;
  62
  63         LASSERT(cfs_mutex_is_locked(&tbl->mn_lock));
  64         cfs_list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
  65                 if (!tgt->mnt_version)
  66                         continue;
  67
  68                 if (version >= tgt->mnt_version)
  69                         return 0;
  70
  71                 version = tgt->mnt_version;
  72         }
  73         return 1;
  74 }
  75
  76 /**
  77  * Fetch nidtbl entries whose version are not less than @version
  78  * nidtbl entries will be packed in @pages by @unit_size units - entries
  79  * shouldn't cross unit boundaries.
  80  */
  81 static int mgs_nidtbl_read(struct obd_export *exp, struct mgs_nidtbl *tbl,
  82                            struct mgs_config_res *res, cfs_page_t **pages,
  83                            int nrpages, int units_total, int unit_size)
  84 {
  85         struct mgs_nidtbl_target *tgt;
  86         struct mgs_nidtbl_entry  *entry;
  87         struct mgs_nidtbl_entry  *last_in_unit = NULL;
  88         struct mgs_target_info   *mti;
  89         __u64 version = res->mcr_offset;
  90         bool nobuf = false;
  91         void *buf = NULL;
  92         int bytes_in_unit = 0;
  93         int units_in_page = 0;
  94         int index = 0;
  95         int rc = 0;
  96         ENTRY;
  97
  98         /* make sure unit_size is power 2 */
  99         LASSERT((unit_size & (unit_size - 1)) == 0);
 100         LASSERT(nrpages << CFS_PAGE_SHIFT >= units_total * unit_size);
 101
 102         cfs_mutex_lock(&tbl->mn_lock);
 103         LASSERT(nidtbl_is_sane(tbl));
 104
 105         /* no more entries ? */
 106         if (version > tbl->mn_version) {
 107                 version = tbl->mn_version;
 108                 goto out;
 109         }
 110
 111         /* iterate over all targets to compose a bitmap by the type of llog.
 112          * If the llog is for MDTs, llog entries for OSTs will be returned;
 113          * otherwise, it's for clients, then llog entries for both OSTs and
 114          * MDTs will be returned.
 115          */
 116         cfs_list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
 117                 int entry_len = sizeof(*entry);
 118
 119                 if (tgt->mnt_version < version)
 120                         continue;
 121
 122                 /* write target recover information */
 123                 mti  = &tgt->mnt_mti;
 124                 LASSERT(mti->mti_nid_count < MTI_NIDS_MAX);
 125                 entry_len += mti->mti_nid_count * sizeof(lnet_nid_t);
 126
 127                 if (entry_len > unit_size) {
 128                         CWARN("nidtbl: too large entry: entry length %d,"
 129                               "unit size: %d\n", 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 != NULL && bytes_in_unit) {
 141 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 50, 0)
 142                                 /* May need to swab back to update the length.*/
 143                                 if (exp->exp_need_mne_swab)
 144                                         lustre_swab_mgs_nidtbl_entry(last_in_unit);
 145 #endif
 146                                 last_in_unit->mne_length += bytes_in_unit;
 147 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 50, 0)
 148                                 if (exp->exp_need_mne_swab)
 149                                         lustre_swab_mgs_nidtbl_entry(last_in_unit);
 150 #endif
 151                                 rc  += bytes_in_unit;
 152                                 buf += bytes_in_unit;
 153                                 last_in_unit = NULL;
 154                         }
 155                         LASSERT((rc & (unit_size - 1)) == 0);
 156
 157                         if (units_in_page == 0) {
 158                                 /* allocate a new page */
 159                                 pages[index] = cfs_alloc_page(CFS_ALLOC_STD);
 160                                 if (pages[index] == NULL) {
 161                                         rc = -ENOMEM;
 162                                         break;
 163                                 }
 164
 165                                 /* destroy previous map */
 166                                 if (index > 0)
 167                                         cfs_kunmap(pages[index - 1]);
 168
 169                                 /* reassign buffer */
 170                                 buf = cfs_kmap(pages[index]);
 171                                 ++index;
 172
 173                                 units_in_page = CFS_PAGE_SIZE / unit_size;
 174                                 LASSERT(units_in_page > 0);
 175                         }
 176
 177                         /* allocate an unit */
 178                         LASSERT(((long)buf & (unit_size - 1)) == 0);
 179                         bytes_in_unit = unit_size;
 180                         --units_in_page;
 181                         --units_total;
 182                 }
 183
 184                 /* fill in entry. */
 185                 entry = (struct mgs_nidtbl_entry *)buf;
 186                 entry->mne_version   = tgt->mnt_version;
 187                 entry->mne_instance  = mti->mti_instance;
 188                 entry->mne_index     = mti->mti_stripe_index;
 189                 entry->mne_length    = entry_len;
 190                 entry->mne_type      = tgt->mnt_type;
 191                 entry->mne_nid_type  = 0;
 192                 entry->mne_nid_size  = sizeof(lnet_nid_t);
 193                 entry->mne_nid_count = mti->mti_nid_count;
 194                 memcpy(entry->u.nids, mti->mti_nids,
 195                        mti->mti_nid_count * sizeof(lnet_nid_t));
 196
 197 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 50, 0)
 198                 /* For LU-1644, swab entry for 2.2 clients. */
 199                 if (exp->exp_need_mne_swab)
 200                         lustre_swab_mgs_nidtbl_entry(entry);
 201 #endif
 202
 203                 version = tgt->mnt_version;
 204                 rc     += entry_len;
 205                 buf    += entry_len;
 206
 207                 bytes_in_unit -= entry_len;
 208                 last_in_unit   = entry;
 209
 210                 CDEBUG(D_MGS, "fsname %s, entry size %d, pages %d/%d/%d/%d.\n",
 211                        tbl->mn_fsdb->fsdb_name, entry_len,
 212                        bytes_in_unit, index, nrpages, units_total);
 213         }
 214         if (index > 0)
 215                 cfs_kunmap(pages[index - 1]);
 216 out:
 217         LASSERT(version <= tbl->mn_version);
 218         res->mcr_size = tbl->mn_version;
 219         res->mcr_offset = nobuf ? version : tbl->mn_version;
 220         cfs_mutex_unlock(&tbl->mn_lock);
 221         LASSERT(ergo(version == 1, rc == 0)); /* get the log first time */
 222
 223         CDEBUG(D_MGS, "Read IR logs %s return with %d, version %llu\n",
 224                tbl->mn_fsdb->fsdb_name, rc, version);
 225         RETURN(rc);
 226 }
 227
 228 static int nidtbl_update_version(const struct lu_env *env,
 229                                  struct mgs_device *mgs,
 230                                  struct mgs_nidtbl *tbl)
 231 {
 232         struct dt_object *fsdb;
 233         struct thandle   *th;
 234         u64               version;
 235         struct lu_buf     buf = {
 236                                 .lb_buf = &version,
 237                                 .lb_len = sizeof(version)
 238                           };
 239         loff_t            off = 0;
 240         int               rc;
 241         ENTRY;
 242
 243         LASSERT(cfs_mutex_is_locked(&tbl->mn_lock));
 244
 245         fsdb = local_file_find_or_create(env, mgs->mgs_los, mgs->mgs_nidtbl_dir,
 246                                          tbl->mn_fsdb->fsdb_name,
 247                                          S_IFREG | S_IRUGO | S_IWUSR);
 248         if (IS_ERR(fsdb))
 249                 RETURN(PTR_ERR(fsdb));
 250
 251         th = dt_trans_create(env, mgs->mgs_bottom);
 252         if (IS_ERR(th))
 253                 GOTO(out_put, rc = PTR_ERR(th));
 254
 255         th->th_sync = 1; /* update table synchronously */
 256         rc = dt_declare_record_write(env, fsdb, buf.lb_len, off, th);
 257         if (rc)
 258                 GOTO(out, rc);
 259
 260         rc = dt_trans_start_local(env, mgs->mgs_bottom, th);
 261         if (rc)
 262                 GOTO(out, rc);
 263
 264         version = cpu_to_le64(tbl->mn_version);
 265         rc = dt_record_write(env, fsdb, &buf, &off, th);
 266
 267 out:
 268         dt_trans_stop(env, mgs->mgs_bottom, th);
 269 out_put:
 270         lu_object_put(env, &fsdb->do_lu);
 271         RETURN(rc);
 272 }
 273
 274 #define MGS_NIDTBL_VERSION_INIT 2
 275
 276 static int nidtbl_read_version(const struct lu_env *env,
 277                                struct mgs_device *mgs, struct mgs_nidtbl *tbl)
 278 {
 279         struct dt_object *fsdb;
 280         struct lu_fid     fid;
 281         u64               tmpver;
 282         struct lu_buf     buf = {
 283                                 .lb_buf = &tmpver,
 284                                 .lb_len = sizeof(tmpver)
 285                           };
 286         loff_t               off = 0;
 287         int                  rc;
 288         ENTRY;
 289
 290         LASSERT(cfs_mutex_is_locked(&tbl->mn_lock));
 291
 292         LASSERT(mgs->mgs_nidtbl_dir);
 293         rc = dt_lookup_dir(env, mgs->mgs_nidtbl_dir, tbl->mn_fsdb->fsdb_name,
 294                            &fid);
 295         if (rc == -ENOENT)
 296                 RETURN(MGS_NIDTBL_VERSION_INIT);
 297         else if (rc < 0)
 298                 RETURN(rc);
 299
 300         fsdb = dt_locate_at(env, mgs->mgs_bottom, &fid,
 301                             &mgs->mgs_dt_dev.dd_lu_dev);
 302         if (IS_ERR(fsdb))
 303                 RETURN(PTR_ERR(fsdb));
 304
 305         rc = dt_read(env, fsdb, &buf, &off);
 306         if (rc == buf.lb_len)
 307                 rc = le64_to_cpu(tmpver);
 308         else if (rc == 0)
 309                 rc = MGS_NIDTBL_VERSION_INIT;
 310         else
 311                 CERROR("%s: read version file %s error %d\n",
 312                        mgs->mgs_obd->obd_name, tbl->mn_fsdb->fsdb_name, rc);
 313         lu_object_put(env, &fsdb->do_lu);
 314         RETURN(rc);
 315 }
 316
 317 static int mgs_nidtbl_write(const struct lu_env *env, struct fs_db *fsdb,
 318                             struct mgs_target_info *mti)
 319 {
 320         struct mgs_nidtbl        *tbl;
 321         struct mgs_nidtbl_target *tgt;
 322         bool found = false;
 323         int type   = mti->mti_flags & LDD_F_SV_TYPE_MASK;
 324         int rc     = 0;
 325         ENTRY;
 326
 327         type &= ~LDD_F_SV_TYPE_MGS;
 328         LASSERT(type != 0);
 329
 330         tbl = &fsdb->fsdb_nidtbl;
 331         cfs_mutex_lock(&tbl->mn_lock);
 332         cfs_list_for_each_entry(tgt, &tbl->mn_targets, mnt_list) {
 333                 struct mgs_target_info *info = &tgt->mnt_mti;
 334                 if (type == tgt->mnt_type &&
 335                     mti->mti_stripe_index == info->mti_stripe_index) {
 336                         found = true;
 337                         break;
 338                 }
 339         }
 340         if (!found) {
 341                 OBD_ALLOC_PTR(tgt);
 342                 if (tgt == NULL)
 343                         GOTO(out, rc = -ENOMEM);
 344
 345                 CFS_INIT_LIST_HEAD(&tgt->mnt_list);
 346                 tgt->mnt_fs      = tbl;
 347                 tgt->mnt_version = 0;       /* 0 means invalid */
 348                 tgt->mnt_type    = type;
 349
 350                 ++tbl->mn_nr_targets;
 351         }
 352
 353         tgt->mnt_version = ++tbl->mn_version;
 354         tgt->mnt_mti     = *mti;
 355
 356         cfs_list_move_tail(&tgt->mnt_list, &tbl->mn_targets);
 357
 358         rc = nidtbl_update_version(env, fsdb->fsdb_mgs, tbl);
 359         EXIT;
 360
 361 out:
 362         cfs_mutex_unlock(&tbl->mn_lock);
 363         if (rc)
 364                 CERROR("Write NID table version for file system %s error %d\n",
 365                        fsdb->fsdb_name, rc);
 366         return rc;
 367 }
 368
 369 static void mgs_nidtbl_fini_fs(struct fs_db *fsdb)
 370 {
 371         struct mgs_nidtbl *tbl = &fsdb->fsdb_nidtbl;
 372         CFS_LIST_HEAD(head);
 373
 374         cfs_mutex_lock(&tbl->mn_lock);
 375         tbl->mn_nr_targets = 0;
 376         cfs_list_splice_init(&tbl->mn_targets, &head);
 377         cfs_mutex_unlock(&tbl->mn_lock);
 378
 379         while (!cfs_list_empty(&head)) {
 380                 struct mgs_nidtbl_target *tgt;
 381                 tgt = list_entry(head.next, struct mgs_nidtbl_target, mnt_list);
 382                 cfs_list_del(&tgt->mnt_list);
 383                 OBD_FREE_PTR(tgt);
 384         }
 385 }
 386
 387 static int mgs_nidtbl_init_fs(const struct lu_env *env, struct fs_db *fsdb)
 388 {
 389         struct mgs_nidtbl *tbl = &fsdb->fsdb_nidtbl;
 390
 391         CFS_INIT_LIST_HEAD(&tbl->mn_targets);
 392         cfs_mutex_init(&tbl->mn_lock);
 393         tbl->mn_nr_targets = 0;
 394         tbl->mn_fsdb = fsdb;
 395         cfs_mutex_lock(&tbl->mn_lock);
 396         tbl->mn_version = nidtbl_read_version(env, fsdb->fsdb_mgs, tbl);
 397         cfs_mutex_unlock(&tbl->mn_lock);
 398         CDEBUG(D_MGS, "IR: current version is %llu\n", tbl->mn_version);
 399
 400         return 0;
 401 }
 402
 403 /* --------- Imperative Recovery relies on nidtbl stuff ------- */
 404 void mgs_ir_notify_complete(struct fs_db *fsdb)
 405 {
 406         struct timeval tv;
 407         cfs_duration_t delta;
 408
 409         cfs_atomic_set(&fsdb->fsdb_notify_phase, 0);
 410
 411         /* do statistic */
 412         fsdb->fsdb_notify_count++;
 413         delta = cfs_time_sub(cfs_time_current(), fsdb->fsdb_notify_start);
 414         fsdb->fsdb_notify_total += delta;
 415         if (delta > fsdb->fsdb_notify_max)
 416                 fsdb->fsdb_notify_max = delta;
 417
 418         cfs_duration_usec(delta, &tv);
 419         CDEBUG(D_MGS, "Revoke recover lock of %s completed after %ld.%06lds\n",
 420                fsdb->fsdb_name, tv.tv_sec, tv.tv_usec);
 421 }
 422
 423 static int mgs_ir_notify(void *arg)
 424 {
 425         struct fs_db      *fsdb   = arg;
 426         struct ldlm_res_id resid;
 427
 428         char name[sizeof(fsdb->fsdb_name) + 20];
 429
 430         LASSERTF(sizeof(name) < 32, "name is too large to be in stack.\n");
 431         sprintf(name, "mgs_%s_notify", fsdb->fsdb_name);
 432         cfs_daemonize(name);
 433
 434         cfs_complete(&fsdb->fsdb_notify_comp);
 435
 436         set_user_nice(current, -2);
 437
 438         mgc_fsname2resid(fsdb->fsdb_name, &resid, CONFIG_T_RECOVER);
 439         while (1) {
 440                 struct l_wait_info   lwi = { 0 };
 441
 442                 l_wait_event(fsdb->fsdb_notify_waitq,
 443                              fsdb->fsdb_notify_stop ||
 444                              cfs_atomic_read(&fsdb->fsdb_notify_phase),
 445                              &lwi);
 446                 if (fsdb->fsdb_notify_stop)
 447                         break;
 448
 449                 CDEBUG(D_MGS, "%s woken up, phase is %d\n",
 450                        name, cfs_atomic_read(&fsdb->fsdb_notify_phase));
 451
 452                 fsdb->fsdb_notify_start = cfs_time_current();
 453                 mgs_revoke_lock(fsdb->fsdb_mgs, fsdb, CONFIG_T_RECOVER);
 454         }
 455
 456         cfs_complete(&fsdb->fsdb_notify_comp);
 457         return 0;
 458 }
 459
 460 int mgs_ir_init_fs(const struct lu_env *env, struct mgs_device *mgs,
 461                    struct fs_db *fsdb)
 462 {
 463         int rc;
 464
 465         if (!ir_timeout)
 466                 ir_timeout = OBD_IR_MGS_TIMEOUT;
 467
 468         fsdb->fsdb_ir_state = IR_FULL;
 469         if (cfs_time_before(cfs_time_current_sec(),
 470                             mgs->mgs_start_time + ir_timeout))
 471                 fsdb->fsdb_ir_state = IR_STARTUP;
 472         fsdb->fsdb_nonir_clients = 0;
 473         CFS_INIT_LIST_HEAD(&fsdb->fsdb_clients);
 474
 475         /* start notify thread */
 476         fsdb->fsdb_obd = mgs->mgs_obd;
 477         fsdb->fsdb_mgs = mgs;
 478         cfs_atomic_set(&fsdb->fsdb_notify_phase, 0);
 479         cfs_waitq_init(&fsdb->fsdb_notify_waitq);
 480         cfs_init_completion(&fsdb->fsdb_notify_comp);
 481         rc = cfs_create_thread(mgs_ir_notify, fsdb, CFS_DAEMON_FLAGS);
 482         if (rc > 0)
 483                 cfs_wait_for_completion(&fsdb->fsdb_notify_comp);
 484         else
 485                 CERROR("Start notify thread error %d\n", rc);
 486
 487         mgs_nidtbl_init_fs(env, fsdb);
 488         return 0;
 489 }
 490
 491 void mgs_ir_fini_fs(struct mgs_device *mgs, struct fs_db *fsdb)
 492 {
 493         if (cfs_test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags))
 494                 return;
 495
 496         mgs_fsc_cleanup_by_fsdb(fsdb);
 497
 498         mgs_nidtbl_fini_fs(fsdb);
 499
 500         LASSERT(cfs_list_empty(&fsdb->fsdb_clients));
 501
 502         fsdb->fsdb_notify_stop = 1;
 503         cfs_waitq_signal(&fsdb->fsdb_notify_waitq);
 504         cfs_wait_for_completion(&fsdb->fsdb_notify_comp);
 505 }
 506
 507 /* caller must have held fsdb_mutex */
 508 static inline void ir_state_graduate(struct fs_db *fsdb)
 509 {
 510         if (fsdb->fsdb_ir_state == IR_STARTUP) {
 511                 if (cfs_time_before(fsdb->fsdb_mgs->mgs_start_time + ir_timeout,
 512                                     cfs_time_current_sec())) {
 513                         fsdb->fsdb_ir_state = IR_FULL;
 514                         if (fsdb->fsdb_nonir_clients)
 515                                 fsdb->fsdb_ir_state = IR_PARTIAL;
 516                 }
 517         }
 518 }
 519
 520 int mgs_ir_update(const struct lu_env *env, struct mgs_device *mgs,
 521                   struct mgs_target_info *mti)
 522 {
 523         struct fs_db *fsdb;
 524         bool notify = true;
 525         int rc;
 526
 527         if (mti->mti_instance == 0)
 528                 return -EINVAL;
 529
 530         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 531         if (rc)
 532                 return rc;
 533
 534         rc = mgs_nidtbl_write(env, fsdb, mti);
 535         if (rc)
 536                 return rc;
 537
 538         /* check ir state */
 539         cfs_mutex_lock(&fsdb->fsdb_mutex);
 540         ir_state_graduate(fsdb);
 541         switch (fsdb->fsdb_ir_state) {
 542         case IR_FULL:
 543                 mti->mti_flags |= LDD_F_IR_CAPABLE;
 544                 break;
 545         case IR_DISABLED:
 546                 notify = false;
 547         case IR_STARTUP:
 548         case IR_PARTIAL:
 549                 break;
 550         default:
 551                 LBUG();
 552         }
 553         cfs_mutex_unlock(&fsdb->fsdb_mutex);
 554
 555         LASSERT(ergo(mti->mti_flags & LDD_F_IR_CAPABLE, notify));
 556         if (notify) {
 557                 CDEBUG(D_MGS, "Try to revoke recover lock of %s\n",
 558                        fsdb->fsdb_name);
 559                 cfs_atomic_inc(&fsdb->fsdb_notify_phase);
 560                 cfs_waitq_signal(&fsdb->fsdb_notify_waitq);
 561         }
 562         return 0;
 563 }
 564
 565 /* NID table can be cached by two entities: Clients and MDTs */
 566 enum {
 567         IR_CLIENT  = 1,
 568         IR_MDT     = 2
 569 };
 570
 571 static int delogname(char *logname, char *fsname, int *typ)
 572 {
 573         char *ptr;
 574         int   type;
 575         int   len;
 576
 577         ptr = strrchr(logname, '-');
 578         if (ptr == NULL)
 579                 return -EINVAL;
 580
 581         /* decouple file system name. The llog name may be:
 582          * - "prefix-fsname", prefix is "cliir" or "mdtir"
 583          */
 584         if (strncmp(ptr, "-mdtir", 6) == 0)
 585                 type = IR_MDT;
 586         else if (strncmp(ptr, "-cliir", 6) == 0)
 587                 type = IR_CLIENT;
 588         else
 589                 return -EINVAL;
 590
 591         len = ptr - logname;
 592         if (len == 0)
 593                 return -EINVAL;
 594
 595         memcpy(fsname, logname, len);
 596         fsname[len] = 0;
 597         if (typ)
 598                 *typ = type;
 599         return 0;
 600 }
 601
 602 int mgs_get_ir_logs(struct ptlrpc_request *req)
 603 {
 604         struct lu_env     *env = req->rq_svc_thread->t_env;
 605         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 606         struct fs_db      *fsdb;
 607         struct mgs_config_body  *body;
 608         struct mgs_config_res   *res;
 609         struct ptlrpc_bulk_desc *desc;
 610         struct l_wait_info lwi;
 611         char               fsname[16];
 612         long               bufsize;
 613         int                unit_size;
 614         int                type;
 615         int                rc = 0;
 616         int                i;
 617         int                bytes;
 618         int                page_count;
 619         int                nrpages;
 620         cfs_page_t       **pages = NULL;
 621         ENTRY;
 622
 623         body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
 624         if (body == NULL)
 625                 RETURN(-EINVAL);
 626
 627         if (body->mcb_type != CONFIG_T_RECOVER)
 628                 RETURN(-EINVAL);
 629
 630         rc = delogname(body->mcb_name, fsname, &type);
 631         if (rc)
 632                 RETURN(rc);
 633
 634         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
 635         if (rc)
 636                 RETURN(rc);
 637
 638         bufsize = body->mcb_units << body->mcb_bits;
 639         nrpages = (bufsize + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT;
 640         if (nrpages > PTLRPC_MAX_BRW_PAGES)
 641                 RETURN(-EINVAL);
 642
 643         CDEBUG(D_MGS, "Reading IR log %s bufsize %ld.\n",
 644                body->mcb_name, bufsize);
 645
 646         OBD_ALLOC(pages, sizeof(*pages) * nrpages);
 647         if (pages == NULL)
 648                 RETURN(-ENOMEM);
 649
 650         rc = req_capsule_server_pack(&req->rq_pill);
 651         if (rc)
 652                 GOTO(out, rc);
 653
 654         res = req_capsule_server_get(&req->rq_pill, &RMF_MGS_CONFIG_RES);
 655         if (res == NULL)
 656                 GOTO(out, rc = -EINVAL);
 657
 658         res->mcr_offset = body->mcb_offset;
 659         unit_size = min_t(int, 1 << body->mcb_bits, CFS_PAGE_SIZE);
 660         bytes = mgs_nidtbl_read(req->rq_export, &fsdb->fsdb_nidtbl, res,
 661                                 pages, nrpages, bufsize / unit_size, unit_size);
 662         if (bytes < 0)
 663                 GOTO(out, rc = bytes);
 664
 665         /* start bulk transfer */
 666         page_count = (bytes + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT;
 667         LASSERT(page_count <= nrpages);
 668         desc = ptlrpc_prep_bulk_exp(req, page_count,
 669                                     BULK_PUT_SOURCE, MGS_BULK_PORTAL);
 670         if (desc == NULL)
 671                 GOTO(out, rc = -ENOMEM);
 672
 673         for (i = 0; i < page_count && bytes > 0; i++) {
 674                 ptlrpc_prep_bulk_page(desc, pages[i], 0,
 675                                 min_t(int, bytes, CFS_PAGE_SIZE));
 676                 bytes -= CFS_PAGE_SIZE;
 677         }
 678
 679         rc = target_bulk_io(req->rq_export, desc, &lwi);
 680         ptlrpc_free_bulk(desc);
 681
 682 out:
 683         if (pages) {
 684                 for (i = 0; i < nrpages; i++) {
 685                         if (pages[i] == NULL)
 686                                 break;
 687                         cfs_free_page(pages[i]);
 688                 }
 689                 OBD_FREE(pages, sizeof(*pages) * nrpages);
 690         }
 691         return rc;
 692 }
 693
 694 static int lprocfs_ir_set_state(struct fs_db *fsdb, const char *buf)
 695 {
 696         const char *strings[] = IR_STRINGS;
 697         int         state = -1;
 698         int         i;
 699
 700         for (i = 0; i < ARRAY_SIZE(strings); i++) {
 701                 if (strcmp(strings[i], buf) == 0) {
 702                         state = i;
 703                         break;
 704                 }
 705         }
 706         if (state < 0)
 707                 return -EINVAL;
 708
 709         CDEBUG(D_MGS, "change fsr state of %s from %s to %s\n",
 710                fsdb->fsdb_name, strings[fsdb->fsdb_ir_state], strings[state]);
 711         cfs_mutex_lock(&fsdb->fsdb_mutex);
 712         if (state == IR_FULL && fsdb->fsdb_nonir_clients)
 713                 state = IR_PARTIAL;
 714         fsdb->fsdb_ir_state = state;
 715         cfs_mutex_unlock(&fsdb->fsdb_mutex);
 716
 717         return 0;
 718 }
 719
 720 static int lprocfs_ir_set_timeout(struct fs_db *fsdb, const char *buf)
 721 {
 722         return -EINVAL;
 723 }
 724
 725 static int lprocfs_ir_clear_stats(struct fs_db *fsdb, const char *buf)
 726 {
 727         if (*buf)
 728                 return -EINVAL;
 729
 730         fsdb->fsdb_notify_total = 0;
 731         fsdb->fsdb_notify_max   = 0;
 732         fsdb->fsdb_notify_count = 0;
 733         return 0;
 734 }
 735
 736 static struct lproc_ir_cmd {
 737         char *name;
 738         int   namelen;
 739         int (*handler)(struct fs_db *, const char *);
 740 } ir_cmds[] = {
 741         { "state=",   6, lprocfs_ir_set_state },
 742         { "timeout=", 8, lprocfs_ir_set_timeout },
 743         { "0",        1, lprocfs_ir_clear_stats }
 744 };
 745
 746 int lprocfs_wr_ir_state(struct file *file, const char *buffer,
 747                          unsigned long count, void *data)
 748 {
 749         struct fs_db *fsdb = data;
 750         char *kbuf;
 751         char *ptr;
 752         int rc = 0;
 753
 754         if (count > CFS_PAGE_SIZE)
 755                 return -EINVAL;
 756
 757         OBD_ALLOC(kbuf, count + 1);
 758         if (kbuf == NULL)
 759                 return -ENOMEM;
 760
 761         if (copy_from_user(kbuf, buffer, count)) {
 762                 OBD_FREE(kbuf, count);
 763                 return -EFAULT;
 764         }
 765
 766         kbuf[count] = 0; /* buffer is supposed to end with 0 */
 767         if (kbuf[count - 1] == '\n')
 768                 kbuf[count - 1] = 0;
 769         ptr = kbuf;
 770
 771         /* fsname=<file system name> must be the 1st entry */
 772         while (ptr != NULL) {
 773                 char *tmpptr;
 774                 int i;
 775
 776                 tmpptr = strchr(ptr, ';');
 777                 if (tmpptr)
 778                         *tmpptr++ = 0;
 779
 780                 rc = -EINVAL;
 781                 for (i = 0; i < ARRAY_SIZE(ir_cmds); i++) {
 782                         struct lproc_ir_cmd *cmd;
 783                         int cmdlen;
 784
 785                         cmd    = &ir_cmds[i];
 786                         cmdlen = cmd->namelen;
 787                         if (strncmp(cmd->name, ptr, cmdlen) == 0) {
 788                                 ptr += cmdlen;
 789                                 rc = cmd->handler(fsdb, ptr);
 790                                 break;
 791                         }
 792                 }
 793                 if (rc)
 794                         break;
 795
 796                 ptr = tmpptr;
 797         }
 798         if (rc)
 799                 CERROR("Unable to process command: %s(%d)\n", ptr, rc);
 800         OBD_FREE(kbuf, count + 1);
 801         return rc ?: count;
 802 }
 803
 804 int lprocfs_rd_ir_state(struct seq_file *seq, void *data)
 805 {
 806         struct fs_db      *fsdb = data;
 807         struct mgs_nidtbl *tbl  = &fsdb->fsdb_nidtbl;
 808         const char        *ir_strings[] = IR_STRINGS;
 809         struct timeval     tv_max;
 810         struct timeval     tv;
 811
 812         /* mgs_live_seq_show() already holds fsdb_mutex. */
 813         ir_state_graduate(fsdb);
 814
 815         seq_printf(seq, "\nimperative_recovery_state:\n");
 816         seq_printf(seq,
 817                    "    state: %s\n"
 818                    "    nonir_clients: %d\n"
 819                    "    nidtbl_version: %lld\n",
 820                    ir_strings[fsdb->fsdb_ir_state], fsdb->fsdb_nonir_clients,
 821                    tbl->mn_version);
 822
 823         cfs_duration_usec(fsdb->fsdb_notify_total, &tv);
 824         cfs_duration_usec(fsdb->fsdb_notify_max, &tv_max);
 825
 826         seq_printf(seq, "    notify_duration_total: %lu.%06lu\n"
 827                         "    notify_duation_max: %lu.%06lu\n"
 828                         "    notify_count: %u\n",
 829                    tv.tv_sec, tv.tv_usec,
 830                    tv_max.tv_sec, tv_max.tv_usec,
 831                    fsdb->fsdb_notify_count);
 832
 833         return 0;
 834 }
 835
 836 int lprocfs_rd_ir_timeout(char *page, char **start, off_t off, int count,
 837                           int *eof, void *data)
 838 {
 839         *eof = 1;
 840         return snprintf(page, count, "%d\n", ir_timeout);
 841 }
 842
 843 int lprocfs_wr_ir_timeout(struct file *file, const char *buffer,
 844                           unsigned long count, void *data)
 845 {
 846         return lprocfs_wr_uint(file, buffer, count, &ir_timeout);
 847 }
 848
 849 /* --------------- Handle non IR support clients --------------- */
 850 /* attach a lustre file system to an export */
 851 int mgs_fsc_attach(const struct lu_env *env, struct obd_export *exp,
 852                    char *fsname)
 853 {
 854         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 855         struct mgs_device *mgs = exp2mgs_dev(exp);
 856         struct fs_db      *fsdb;
 857         struct mgs_fsc    *fsc     = NULL;
 858         struct mgs_fsc    *new_fsc = NULL;
 859         bool               found   = false;
 860         int                rc;
 861         ENTRY;
 862
 863         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
 864         if (rc)
 865                 RETURN(rc);
 866
 867         /* allocate a new fsc in case we need it in spinlock. */
 868         OBD_ALLOC_PTR(new_fsc);
 869         if (new_fsc == NULL)
 870                 RETURN(-ENOMEM);
 871
 872         CFS_INIT_LIST_HEAD(&new_fsc->mfc_export_list);
 873         CFS_INIT_LIST_HEAD(&new_fsc->mfc_fsdb_list);
 874         new_fsc->mfc_fsdb       = fsdb;
 875         new_fsc->mfc_export     = class_export_get(exp);
 876         new_fsc->mfc_ir_capable =
 877                         !!(exp->exp_connect_flags & OBD_CONNECT_IMP_RECOV);
 878
 879         rc = -EEXIST;
 880         cfs_mutex_lock(&fsdb->fsdb_mutex);
 881
 882         /* tend to find it in export list because this list is shorter. */
 883         cfs_spin_lock(&data->med_lock);
 884         cfs_list_for_each_entry(fsc, &data->med_clients, mfc_export_list) {
 885                 if (strcmp(fsname, fsc->mfc_fsdb->fsdb_name) == 0) {
 886                         found = true;
 887                         break;
 888                 }
 889         }
 890         if (!found) {
 891                 fsc = new_fsc;
 892                 new_fsc = NULL;
 893
 894                 /* add it into export list. */
 895                 cfs_list_add(&fsc->mfc_export_list, &data->med_clients);
 896
 897                 /* add into fsdb list. */
 898                 cfs_list_add(&fsc->mfc_fsdb_list, &fsdb->fsdb_clients);
 899                 if (!fsc->mfc_ir_capable) {
 900                         ++fsdb->fsdb_nonir_clients;
 901                         if (fsdb->fsdb_ir_state == IR_FULL)
 902                                 fsdb->fsdb_ir_state = IR_PARTIAL;
 903                 }
 904                 rc = 0;
 905         }
 906         cfs_spin_unlock(&data->med_lock);
 907         cfs_mutex_unlock(&fsdb->fsdb_mutex);
 908
 909         if (new_fsc) {
 910                 class_export_put(new_fsc->mfc_export);
 911                 OBD_FREE_PTR(new_fsc);
 912         }
 913         RETURN(rc);
 914 }
 915
 916 void mgs_fsc_cleanup(struct obd_export *exp)
 917 {
 918         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 919         struct mgs_fsc *fsc, *tmp;
 920         CFS_LIST_HEAD(head);
 921
 922         cfs_spin_lock(&data->med_lock);
 923         cfs_list_splice_init(&data->med_clients, &head);
 924         cfs_spin_unlock(&data->med_lock);
 925
 926         cfs_list_for_each_entry_safe(fsc, tmp, &head, mfc_export_list) {
 927                 struct fs_db *fsdb = fsc->mfc_fsdb;
 928
 929                 LASSERT(fsc->mfc_export == exp);
 930
 931                 cfs_mutex_lock(&fsdb->fsdb_mutex);
 932                 cfs_list_del_init(&fsc->mfc_fsdb_list);
 933                 if (fsc->mfc_ir_capable == 0) {
 934                         --fsdb->fsdb_nonir_clients;
 935                         LASSERT(fsdb->fsdb_ir_state != IR_FULL);
 936                         if (fsdb->fsdb_nonir_clients == 0 &&
 937                             fsdb->fsdb_ir_state == IR_PARTIAL)
 938                                 fsdb->fsdb_ir_state = IR_FULL;
 939                 }
 940                 cfs_mutex_unlock(&fsdb->fsdb_mutex);
 941                 cfs_list_del_init(&fsc->mfc_export_list);
 942                 class_export_put(fsc->mfc_export);
 943                 OBD_FREE_PTR(fsc);
 944         }
 945 }
 946
 947 /* must be called with fsdb->fsdb_mutex held */
 948 void mgs_fsc_cleanup_by_fsdb(struct fs_db *fsdb)
 949 {
 950         struct mgs_fsc *fsc, *tmp;
 951
 952         cfs_list_for_each_entry_safe(fsc, tmp, &fsdb->fsdb_clients,
 953                                      mfc_fsdb_list) {
 954                 struct mgs_export_data *data = &fsc->mfc_export->u.eu_mgs_data;
 955
 956                 LASSERT(fsdb == fsc->mfc_fsdb);
 957                 cfs_list_del_init(&fsc->mfc_fsdb_list);
 958
 959                 cfs_spin_lock(&data->med_lock);
 960                 cfs_list_del_init(&fsc->mfc_export_list);
 961                 cfs_spin_unlock(&data->med_lock);
 962                 class_export_put(fsc->mfc_export);
 963                 OBD_FREE_PTR(fsc);
 964         }
 965
 966         fsdb->fsdb_nonir_clients = 0;
 967         if (fsdb->fsdb_ir_state == IR_PARTIAL)
 968                 fsdb->fsdb_ir_state = IR_FULL;
 969 }