lustre/mds/mds_lov.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  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/mds/mds_lov.c
  37  *
  38  * Lustre Metadata Server (mds) handling of striped file data
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_MDS
  44
  45 #include <linux/module.h>
  46 #include <lustre_mds.h>
  47 #include <lustre/lustre_idl.h>
  48 #include <obd_class.h>
  49 #include <obd_lov.h>
  50 #include <lustre_lib.h>
  51 #include <lustre_fsfilt.h>
  52 #include <obd_cksum.h>
  53
  54 #include "mds_internal.h"
  55
  56 static void mds_lov_dump_objids(const char *label, struct obd_device *obd)
  57 {
  58         struct mds_obd *mds = &obd->u.mds;
  59         unsigned int i=0, j;
  60
  61         CDEBUG(D_INFO, "dump from %s\n", label);
  62         if (mds->mds_lov_page_dirty == NULL) {
  63                 CERROR("NULL bitmap!\n");
  64                 GOTO(skip_bitmap, i);
  65         }
  66
  67         for(i = 0; i < mds->mds_lov_page_dirty->size / BITS_PER_LONG + 1; i++)
  68                 CDEBUG(D_INFO, "%u - %lx\n", i,
  69                        mds->mds_lov_page_dirty->data[i]);
  70 skip_bitmap:
  71         if (mds->mds_lov_page_array == NULL) {
  72                 CERROR("not init page array!\n");
  73                 GOTO(skip_array, i);
  74
  75         }
  76         for(i = 0;i < MDS_LOV_OBJID_PAGES_COUNT; i++) {
  77                 obd_id *data = mds->mds_lov_page_array[i];
  78
  79                 if (data == NULL)
  80                         continue;
  81
  82                 for(j=0; j < OBJID_PER_PAGE(); j++) {
  83                         if (data[j] == 0)
  84                                 continue;
  85                         CDEBUG(D_INFO,"objid page %u idx %u - "LPU64" \n",
  86                                i, j, data[j]);
  87                 }
  88         }
  89 skip_array:
  90         EXIT;
  91 }
  92
  93 int mds_lov_init_objids(struct obd_device *obd)
  94 {
  95         struct mds_obd *mds = &obd->u.mds;
  96         int size = MDS_LOV_OBJID_PAGES_COUNT*sizeof(void *);
  97         struct file *file;
  98         int rc;
  99         ENTRY;
 100
 101         CLASSERT(((MDS_LOV_ALLOC_SIZE % sizeof(obd_id)) == 0));
 102
 103         mds->mds_lov_page_dirty =
 104                 CFS_ALLOCATE_BITMAP(MDS_LOV_OBJID_PAGES_COUNT);
 105         if (mds->mds_lov_page_dirty == NULL)
 106                 RETURN(-ENOMEM);
 107
 108
 109         OBD_ALLOC(mds->mds_lov_page_array, size);
 110         if (mds->mds_lov_page_array == NULL)
 111                 GOTO(err_free_bitmap, rc = -ENOMEM);
 112
 113         /* open and test the lov objd file */
 114         file = filp_open(LOV_OBJID, O_RDWR | O_CREAT, 0644);
 115         if (IS_ERR(file)) {
 116                 rc = PTR_ERR(file);
 117                 CERROR("cannot open/create %s file: rc = %d\n", LOV_OBJID, rc);
 118                 GOTO(err_free, rc = PTR_ERR(file));
 119         }
 120         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
 121                 CERROR("%s is not a regular file!: mode = %o\n", LOV_OBJID,
 122                        file->f_dentry->d_inode->i_mode);
 123                 GOTO(err_open, rc = -ENOENT);
 124         }
 125         mds->mds_lov_objid_filp = file;
 126
 127         RETURN (0);
 128 err_open:
 129         if (filp_close((struct file *)file, 0))
 130                 CERROR("can't close %s after error\n", LOV_OBJID);
 131 err_free:
 132         OBD_FREE(mds->mds_lov_page_array, size);
 133 err_free_bitmap:
 134         CFS_FREE_BITMAP(mds->mds_lov_page_dirty);
 135
 136         RETURN(rc);
 137 }
 138
 139 void mds_lov_destroy_objids(struct obd_device *obd)
 140 {
 141         struct mds_obd *mds = &obd->u.mds;
 142         int i, rc;
 143         ENTRY;
 144
 145         if (mds->mds_lov_page_array != NULL) {
 146                 for(i=0;i<MDS_LOV_OBJID_PAGES_COUNT;i++) {
 147                         obd_id *data = mds->mds_lov_page_array[i];
 148                         if (data != NULL)
 149                                 OBD_FREE(data, MDS_LOV_ALLOC_SIZE);
 150                 }
 151                 OBD_FREE(mds->mds_lov_page_array,
 152                          MDS_LOV_OBJID_PAGES_COUNT*sizeof(void *));
 153         }
 154
 155         if (mds->mds_lov_objid_filp) {
 156                 rc = filp_close((struct file *)mds->mds_lov_objid_filp, NULL);
 157                 mds->mds_lov_objid_filp = NULL;
 158                 if (rc)
 159                         CERROR("%s file won't close, rc=%d\n", LOV_OBJID, rc);
 160         }
 161
 162         CFS_FREE_BITMAP(mds->mds_lov_page_dirty);
 163         EXIT;
 164 }
 165
 166 /**
 167  * currently exist two ways for know about ost count and max ost index.
 168  * first - after ost is connected to mds and sync process finished
 169  * second - get from lmm in recovery process, in case when mds not have configs,
 170  * and ost isn't registered in mgs.
 171  *
 172  * \param mds pointer to mds structure
 173  * \param index maxium ost index
 174  *
 175  * \retval -ENOMEM is not hame memory for new page
 176  * \retval 0 is update passed
 177  */
 178 static int mds_lov_update_max_ost(struct mds_obd *mds, obd_id index)
 179 {
 180         __u32 page = index / OBJID_PER_PAGE();
 181         __u32 off = index % OBJID_PER_PAGE();
 182         obd_id *data =  mds->mds_lov_page_array[page];
 183
 184         if (data == NULL) {
 185                 OBD_ALLOC(data, MDS_LOV_ALLOC_SIZE);
 186                 if (data == NULL)
 187                         RETURN(-ENOMEM);
 188
 189                 mds->mds_lov_page_array[page] = data;
 190         }
 191
 192         if (index > mds->mds_lov_objid_max_index) {
 193                 mds->mds_lov_objid_lastpage = page;
 194                 mds->mds_lov_objid_lastidx = off;
 195                 mds->mds_lov_objid_max_index = index;
 196         }
 197
 198         /* workaround - New target not in objids file; increase mdsize */
 199         /* ld_tgt_count is used as the max index everywhere, despite its name. */
 200         if (data[off] == 0) {
 201                 __u32 stripes;
 202
 203                 data[off] = 1;
 204                 mds->mds_lov_objid_count++;
 205                 stripes = min_t(__u32, LOV_MAX_STRIPE_COUNT,
 206                                 mds->mds_lov_objid_count);
 207
 208                 mds->mds_max_mdsize = lov_mds_md_size(stripes, LOV_MAGIC_V3);
 209                 mds->mds_max_cookiesize = stripes * sizeof(struct llog_cookie);
 210
 211                 CDEBUG(D_CONFIG, "updated max_mdsize/max_cookiesize for %d"
 212                        " stripes: %d/%d\n", stripes, mds->mds_max_mdsize,
 213                        mds->mds_max_cookiesize);
 214         }
 215
 216         EXIT;
 217         return 0;
 218 }
 219
 220 int mds_lov_prepare_objids(struct obd_device *obd, struct lov_mds_md *lmm)
 221 {
 222         struct lov_ost_data_v1 *data;
 223         __u32 count;
 224         int rc = 0;
 225         __u32 j;
 226
 227         /* if we create file without objects - lmm is NULL */
 228         if (lmm == NULL)
 229                 return 0;
 230
 231         switch (le32_to_cpu(lmm->lmm_magic)) {
 232                 case LOV_MAGIC_V1:
 233                         count = le32_to_cpu(((struct lov_mds_md_v1*)lmm)->lmm_stripe_count);
 234                         data = &(((struct lov_mds_md_v1*)lmm)->lmm_objects[0]);
 235                         break;
 236                 case LOV_MAGIC_V3:
 237                         count = le32_to_cpu(((struct lov_mds_md_v3*)lmm)->lmm_stripe_count);
 238                         data = &(((struct lov_mds_md_v3*)lmm)->lmm_objects[0]);
 239                         break;
 240                 default:
 241                         CERROR("Unknow lmm type %X!\n", le32_to_cpu(lmm->lmm_magic));
 242                         RETURN(-EINVAL);
 243         }
 244
 245
 246         cfs_mutex_down(&obd->obd_dev_sem);
 247         for (j = 0; j < count; j++) {
 248                 __u32 i = le32_to_cpu(data[j].l_ost_idx);
 249                 if (mds_lov_update_max_ost(&obd->u.mds, i)) {
 250                         rc = -ENOMEM;
 251                         break;
 252                 }
 253         }
 254         cfs_mutex_up(&obd->obd_dev_sem);
 255
 256         RETURN(rc);
 257 }
 258 EXPORT_SYMBOL(mds_lov_prepare_objids);
 259
 260 /*
 261  * write llog orphan record about lost ost object,
 262  * Special lsm is allocated with single stripe, caller should deallocated it
 263  * after use
 264  */
 265 static int mds_log_lost_precreated(struct obd_device *obd,
 266                                    struct lov_stripe_md **lsmp, int *stripes,
 267                                    obd_id id, obd_count count, int idx)
 268 {
 269         struct lov_stripe_md *lsm = *lsmp;
 270         int rc;
 271         ENTRY;
 272
 273         if (*lsmp == NULL) {
 274                 rc = obd_alloc_memmd(obd->u.mds.mds_osc_exp, &lsm);
 275                 if (rc < 0)
 276                         RETURN(rc);
 277                 /* need only one stripe, save old value */
 278                 *stripes = lsm->lsm_stripe_count;
 279                 lsm->lsm_stripe_count = 1;
 280                 *lsmp = lsm;
 281         }
 282
 283         lsm->lsm_oinfo[0]->loi_id = id;
 284         lsm->lsm_oinfo[0]->loi_gr = mdt_to_obd_objgrp(obd->u.mds.mds_id);
 285         lsm->lsm_oinfo[0]->loi_ost_idx = idx;
 286
 287         rc = mds_log_op_orphan(obd, lsm, count);
 288         RETURN(rc);
 289 }
 290
 291 void mds_lov_update_objids(struct obd_device *obd, struct lov_mds_md *lmm)
 292 {
 293         struct mds_obd *mds = &obd->u.mds;
 294         int j;
 295         struct lov_ost_data_v1 *obj;
 296         struct lov_stripe_md *lsm = NULL;
 297         int stripes = 0;
 298         int count;
 299         ENTRY;
 300
 301         /* if we create file without objects - lmm is NULL */
 302         if (lmm == NULL)
 303                 return;
 304
 305         switch (le32_to_cpu(lmm->lmm_magic)) {
 306                 case LOV_MAGIC_V1:
 307                         count = le32_to_cpu(((struct lov_mds_md_v1*)lmm)->lmm_stripe_count);
 308                         obj = ((struct lov_mds_md_v1*)lmm)->lmm_objects;
 309                         break;
 310                 case LOV_MAGIC_V3:
 311                         count = le32_to_cpu(((struct lov_mds_md_v3*)lmm)->lmm_stripe_count);
 312                         obj = ((struct lov_mds_md_v3*)lmm)->lmm_objects;
 313                         break;
 314                 default:
 315                         CERROR("Unknow lmm type %X !\n",
 316                                le32_to_cpu(lmm->lmm_magic));
 317                         return;
 318         }
 319
 320         for (j = 0; j < count; j++) {
 321                 __u32 i = le32_to_cpu(obj[j].l_ost_idx);
 322                 obd_id id = le64_to_cpu(obj[j].l_object_id);
 323                 __u32 page = i / OBJID_PER_PAGE();
 324                 __u32 idx = i % OBJID_PER_PAGE();
 325                 obd_id *data;
 326
 327                 data = mds->mds_lov_page_array[page];
 328
 329                 CDEBUG(D_INODE,"update last object for ost %u"
 330                        " - new "LPU64" old "LPU64"\n", i, id, data[idx]);
 331                 if (id > data[idx]) {
 332                         int lost = id - data[idx] - 1;
 333                         /* we might have lost precreated objects due to VBR */
 334                         if (lost > 0 && obd->obd_recovering) {
 335                                 CDEBUG(D_HA, "Gap in objids is %u\n", lost);
 336                                 if (!obd->obd_version_recov)
 337                                         CERROR("Unexpected gap in objids\n");
 338                                 /* lsm is allocated if NULL */
 339                                 mds_log_lost_precreated(obd, &lsm, &stripes,
 340                                                         data[idx]+1, lost, i);
 341                         }
 342                         data[idx] = id;
 343                         cfs_bitmap_set(mds->mds_lov_page_dirty, page);
 344                 }
 345         }
 346         if (lsm) {
 347                 /* restore stripes number */
 348                 lsm->lsm_stripe_count = stripes;
 349                 obd_free_memmd(mds->mds_osc_exp, &lsm);
 350         }
 351         EXIT;
 352         return;
 353 }
 354 EXPORT_SYMBOL(mds_lov_update_objids);
 355
 356 static int mds_lov_update_from_read(struct mds_obd *mds, obd_id *data,
 357                                     __u32 count)
 358 {
 359         __u32 i;
 360         __u32 stripes;
 361
 362         for (i = 0; i < count; i++) {
 363                 if (data[i] == 0)
 364                         continue;
 365
 366                 mds->mds_lov_objid_count++;
 367         }
 368
 369         stripes = min_t(__u32, LOV_MAX_STRIPE_COUNT,
 370                          mds->mds_lov_objid_count);
 371
 372         mds->mds_max_mdsize = lov_mds_md_size(stripes, LOV_MAGIC_V3);
 373         mds->mds_max_cookiesize = stripes * sizeof(struct llog_cookie);
 374
 375         CDEBUG(D_CONFIG, "updated max_mdsize/max_cookiesize for %d stripes: "
 376                "%d/%d\n", stripes, mds->mds_max_mdsize,
 377                mds->mds_max_cookiesize);
 378
 379         EXIT;
 380         return 0;
 381 }
 382
 383 static int mds_lov_read_objids(struct obd_device *obd)
 384 {
 385         struct mds_obd *mds = &obd->u.mds;
 386         loff_t off = 0;
 387         int i, rc = 0, count = 0, page = 0;
 388         unsigned long size;
 389         ENTRY;
 390
 391         /* Read everything in the file, even if our current lov desc
 392            has fewer targets. Old targets not in the lov descriptor
 393            during mds setup may still have valid objids. */
 394         size = i_size_read(mds->mds_lov_objid_filp->f_dentry->d_inode);
 395         if (size == 0)
 396                 RETURN(0);
 397
 398         page = (size / (OBJID_PER_PAGE() * sizeof(obd_id))) + 1;
 399         CDEBUG(D_INFO, "file size %lu pages %d\n", size, page);
 400         for (i = 0; i < page; i++) {
 401                 obd_id *data;
 402                 loff_t off_old = off;
 403
 404                 LASSERT(mds->mds_lov_page_array[i] == NULL);
 405                 OBD_ALLOC(mds->mds_lov_page_array[i], MDS_LOV_ALLOC_SIZE);
 406                 if (mds->mds_lov_page_array[i] == NULL)
 407                         GOTO(out, rc = -ENOMEM);
 408
 409                 data = mds->mds_lov_page_array[i];
 410
 411                 rc = fsfilt_read_record(obd, mds->mds_lov_objid_filp, data,
 412                                         OBJID_PER_PAGE()*sizeof(obd_id), &off);
 413                 if (rc < 0) {
 414                         CERROR("Error reading objids %d\n", rc);
 415                         GOTO(out, rc);
 416                 }
 417
 418                 count += (off - off_old) / sizeof(obd_id);
 419                 if (mds_lov_update_from_read(mds, data, count)) {
 420                         CERROR("Can't update mds data\n");
 421                         GOTO(out, rc = -EIO);
 422                 }
 423
 424                 if (off == off_old)
 425                         break; /* eof */
 426         }
 427         mds->mds_lov_objid_lastpage = i;
 428         mds->mds_lov_objid_lastidx = count % OBJID_PER_PAGE();
 429
 430         CDEBUG(D_INFO, "Read %u - %u %u objid\n", mds->mds_lov_objid_count,
 431                mds->mds_lov_objid_lastpage, mds->mds_lov_objid_lastidx);
 432 out:
 433         mds_lov_dump_objids("read",obd);
 434
 435         RETURN(rc);
 436 }
 437
 438 int mds_lov_write_objids(struct obd_device *obd)
 439 {
 440         struct mds_obd *mds = &obd->u.mds;
 441         int i = 0, rc = 0;
 442         ENTRY;
 443
 444         if (cfs_bitmap_check_empty(mds->mds_lov_page_dirty))
 445                 RETURN(0);
 446
 447         mds_lov_dump_objids("write", obd);
 448
 449         cfs_foreach_bit(mds->mds_lov_page_dirty, i) {
 450                 obd_id *data =  mds->mds_lov_page_array[i];
 451                 unsigned int size = OBJID_PER_PAGE()*sizeof(obd_id);
 452                 loff_t off = i * size;
 453
 454                 LASSERT(data != NULL);
 455
 456                 if (!cfs_bitmap_test_and_clear(mds->mds_lov_page_dirty, i))
 457                         continue;
 458
 459                 /* check for particaly filled last page */
 460                 if (i == mds->mds_lov_objid_lastpage)
 461                         size = (mds->mds_lov_objid_lastidx+1) * sizeof(obd_id);
 462
 463                 CDEBUG(D_INFO, "write %lld - %u\n", off, size);
 464                 rc = fsfilt_write_record(obd, mds->mds_lov_objid_filp, data,
 465                                          size, &off, 0);
 466                 if (rc < 0) {
 467                         cfs_bitmap_set(mds->mds_lov_page_dirty, i);
 468                         break;
 469                 }
 470         }
 471         if (rc >= 0)
 472                 rc = 0;
 473
 474         RETURN(rc);
 475 }
 476 EXPORT_SYMBOL(mds_lov_write_objids);
 477
 478 static int mds_lov_get_objid(struct obd_device * obd,
 479                              obd_id idx)
 480 {
 481         struct mds_obd *mds = &obd->u.mds;
 482         unsigned int page;
 483         unsigned int off;
 484         obd_id *data;
 485         int rc = 0;
 486         ENTRY;
 487
 488         page = idx / OBJID_PER_PAGE();
 489         off = idx % OBJID_PER_PAGE();
 490         data = mds->mds_lov_page_array[page];
 491         if (data[off] < 2) {
 492                 /* We never read this lastid; ask the osc */
 493                 struct obd_id_info lastid;
 494                 __u32 size = sizeof(lastid);
 495
 496                 lastid.idx = idx;
 497                 lastid.data = &data[off];
 498                 rc = obd_get_info(mds->mds_osc_exp, sizeof(KEY_LAST_ID),
 499                                   KEY_LAST_ID, &size, &lastid, NULL);
 500                 if (rc)
 501                         GOTO(out, rc);
 502
 503                 /* workaround for clean filter */
 504                 if (data[off] == 0)
 505                         data[off] = 1;
 506
 507                 cfs_bitmap_set(mds->mds_lov_page_dirty, page);
 508         }
 509         CDEBUG(D_INFO, "idx "LPU64" - %p - %d/%d - "LPU64"\n",
 510                idx, data, page, off, data[off]);
 511 out:
 512         RETURN(rc);
 513 }
 514
 515 int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid)
 516 {
 517         int rc;
 518         struct obdo oa = { 0 };
 519         struct obd_trans_info oti = {0};
 520         struct lov_stripe_md  *empty_ea = NULL;
 521         ENTRY;
 522
 523         LASSERT(mds->mds_lov_page_array != NULL);
 524
 525         /* This create will in fact either create or destroy:  If the OST is
 526          * missing objects below this ID, they will be created.  If it finds
 527          * objects above this ID, they will be removed. */
 528         memset(&oa, 0, sizeof(oa));
 529         oa.o_flags = OBD_FL_DELORPHAN;
 530         oa.o_gr = mdt_to_obd_objgrp(mds->mds_id);
 531         oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
 532         if (ost_uuid != NULL)
 533                 oti.oti_ost_uuid = ost_uuid;
 534
 535         rc = obd_create(mds->mds_osc_exp, &oa, &empty_ea, &oti);
 536
 537         RETURN(rc);
 538 }
 539
 540 /* for one target */
 541 static int mds_lov_set_one_nextid(struct obd_device *obd, __u32 idx, obd_id *id)
 542 {
 543         struct mds_obd *mds = &obd->u.mds;
 544         int rc;
 545         struct obd_id_info info;
 546         ENTRY;
 547
 548         LASSERT(!obd->obd_recovering);
 549
 550         info.idx = idx;
 551         info.data = id;
 552         rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_NEXT_ID),
 553                                 KEY_NEXT_ID, sizeof(info), &info, NULL);
 554         if (rc)
 555                 CERROR ("%s: mds_lov_set_nextid failed (%d)\n",
 556                         obd->obd_name, rc);
 557
 558         RETURN(rc);
 559 }
 560
 561 /* Update the lov desc for a new size lov. */
 562 static int mds_lov_update_desc(struct obd_device *obd, int idx,
 563                                struct obd_uuid *uuid, enum obd_notify_event ev)
 564 {
 565         struct mds_obd *mds = &obd->u.mds;
 566         struct lov_desc *ld;
 567         __u32 valsize = sizeof(mds->mds_lov_desc);
 568         int rc = 0;
 569         ENTRY;
 570
 571         OBD_ALLOC(ld, sizeof(*ld));
 572         if (!ld)
 573                 RETURN(-ENOMEM);
 574
 575         rc = obd_get_info(mds->mds_osc_exp, sizeof(KEY_LOVDESC), KEY_LOVDESC,
 576                           &valsize, ld, NULL);
 577         if (rc)
 578                 GOTO(out, rc);
 579
 580         /* Don't change the mds_lov_desc until the objids size matches the
 581            count (paranoia) */
 582         mds->mds_lov_desc = *ld;
 583         CDEBUG(D_CONFIG, "updated lov_desc, tgt_count: %d - idx %d / uuid %s\n",
 584                mds->mds_lov_desc.ld_tgt_count, idx, uuid->uuid);
 585
 586         cfs_mutex_down(&obd->obd_dev_sem);
 587         rc = mds_lov_update_max_ost(mds, idx);
 588         cfs_mutex_up(&obd->obd_dev_sem);
 589         if (rc != 0)
 590                 GOTO(out, rc );
 591
 592         /* If we added a target we have to reconnect the llogs */
 593         /* We only _need_ to do this at first add (idx), or the first time
 594            after recovery.  However, it should now be safe to call anytime. */
 595         rc = obd_llog_init(obd, &obd->obd_olg, obd, &idx);
 596         if (rc)
 597                 GOTO(out, rc);
 598
 599         /*XXX this notifies the MDD until lov handling use old mds code */
 600         if (obd->obd_upcall.onu_owner) {
 601                  LASSERT(obd->obd_upcall.onu_upcall != NULL);
 602                  rc = obd->obd_upcall.onu_upcall(obd, NULL, ev,
 603                                                  obd->obd_upcall.onu_owner,
 604                                                  &mds->mds_obt.obt_mount_count);
 605         }
 606 out:
 607         OBD_FREE(ld, sizeof(*ld));
 608         RETURN(rc);
 609 }
 610
 611 /* Inform MDS about new/updated target */
 612 static int mds_lov_update_mds(struct obd_device *obd,
 613                               struct obd_device *watched,
 614                               __u32 idx, enum obd_notify_event ev)
 615 {
 616         struct mds_obd *mds = &obd->u.mds;
 617         int rc = 0;
 618         int page;
 619         int off;
 620         obd_id *data;
 621
 622         ENTRY;
 623
 624         /* Don't let anyone else mess with mds_lov_objids now */
 625         rc = mds_lov_update_desc(obd, idx, &watched->u.cli.cl_target_uuid, ev);
 626         if (rc)
 627                 GOTO(out, rc);
 628
 629         CDEBUG(D_CONFIG, "idx=%d, recov=%d/%d, cnt=%d\n",
 630                idx, obd->obd_recovering, obd->obd_async_recov,
 631                mds->mds_lov_desc.ld_tgt_count);
 632
 633         /* idx is set as data from lov_notify. */
 634         if (obd->obd_recovering)
 635                 GOTO(out, rc);
 636
 637         if (idx >= mds->mds_lov_desc.ld_tgt_count) {
 638                 CERROR("index %d > count %d!\n", idx,
 639                        mds->mds_lov_desc.ld_tgt_count);
 640                 GOTO(out, rc = -EINVAL);
 641         }
 642
 643         rc = mds_lov_get_objid(obd, idx);
 644         if (rc)
 645                 GOTO(out, rc);
 646
 647         page = idx / OBJID_PER_PAGE();
 648         off = idx % OBJID_PER_PAGE();
 649         data = mds->mds_lov_page_array[page];
 650
 651         /* We have read this lastid from disk; tell the osc.
 652            Don't call this during recovery. */
 653         rc = mds_lov_set_one_nextid(obd, idx, &data[off]);
 654         if (rc) {
 655                 CERROR("Failed to set next id, idx=%d rc=%d\n", idx,rc);
 656                 /* Don't abort the rest of the sync */
 657                 rc = 0;
 658         } else {
 659                 CDEBUG(D_CONFIG, "last object "LPU64" from OST %d rc=%d\n",
 660                         data[off], idx, rc);
 661         }
 662 out:
 663         RETURN(rc);
 664 }
 665
 666 /* update the LOV-OSC knowledge of the last used object id's */
 667 int mds_lov_connect(struct obd_device *obd, char * lov_name)
 668 {
 669         struct mds_obd *mds = &obd->u.mds;
 670         struct obd_connect_data *data;
 671         int rc;
 672         ENTRY;
 673
 674         if (IS_ERR(mds->mds_osc_obd))
 675                 RETURN(PTR_ERR(mds->mds_osc_obd));
 676
 677         if (mds->mds_osc_obd)
 678                 RETURN(0);
 679
 680         mds->mds_osc_obd = class_name2obd(lov_name);
 681         if (!mds->mds_osc_obd) {
 682                 CERROR("MDS cannot locate LOV %s\n", lov_name);
 683                 mds->mds_osc_obd = ERR_PTR(-ENOTCONN);
 684                 RETURN(-ENOTCONN);
 685         }
 686
 687         cfs_mutex_down(&obd->obd_dev_sem);
 688         rc = mds_lov_read_objids(obd);
 689         cfs_mutex_up(&obd->obd_dev_sem);
 690         if (rc) {
 691                 CERROR("cannot read %s: rc = %d\n", "lov_objids", rc);
 692                 GOTO(err_exit, rc);
 693         }
 694
 695         rc = obd_register_observer(mds->mds_osc_obd, obd);
 696         if (rc) {
 697                 CERROR("MDS cannot register as observer of LOV %s (%d)\n",
 698                        lov_name, rc);
 699                 GOTO(err_exit, rc);
 700         }
 701
 702         /* try init too early */
 703         rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
 704         if (rc)
 705                 GOTO(err_exit, rc);
 706
 707         mds->mds_osc_obd->u.lov.lov_sp_me = LUSTRE_SP_MDT;
 708
 709         OBD_ALLOC(data, sizeof(*data));
 710         if (data == NULL)
 711                 GOTO(err_exit, rc = -ENOMEM);
 712
 713         data->ocd_connect_flags = OBD_CONNECT_VERSION   | OBD_CONNECT_INDEX   |
 714                                   OBD_CONNECT_REQPORTAL | OBD_CONNECT_QUOTA64 |
 715                                   OBD_CONNECT_OSS_CAPA  | OBD_CONNECT_FID     |
 716                                   OBD_CONNECT_BRW_SIZE  | OBD_CONNECT_CKSUM   |
 717                                   OBD_CONNECT_CHANGE_QS | OBD_CONNECT_AT      |
 718                                   OBD_CONNECT_MDS | OBD_CONNECT_SKIP_ORPHAN   |
 719                                   OBD_CONNECT_SOM | OBD_CONNECT_FULL20;
 720 #ifdef HAVE_LRU_RESIZE_SUPPORT
 721         data->ocd_connect_flags |= OBD_CONNECT_LRU_RESIZE;
 722 #endif
 723         data->ocd_version = LUSTRE_VERSION_CODE;
 724         data->ocd_group = mdt_to_obd_objgrp(mds->mds_id);
 725         /* send max bytes per rpc */
 726         data->ocd_brw_size = PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT;
 727         /* send the list of supported checksum types */
 728         data->ocd_cksum_types = OBD_CKSUM_ALL;
 729         /* NB: lov_connect() needs to fill in .ocd_index for each OST */
 730         rc = obd_connect(NULL, &mds->mds_osc_exp, mds->mds_osc_obd, &obd->obd_uuid, data, NULL);
 731         OBD_FREE(data, sizeof(*data));
 732         if (rc) {
 733                 CERROR("MDS cannot connect to LOV %s (%d)\n", lov_name, rc);
 734                 mds->mds_osc_obd = ERR_PTR(rc);
 735                 RETURN(rc);
 736         }
 737
 738         /* I want to see a callback happen when the OBD moves to a
 739          * "For General Use" state, and that's when we'll call
 740          * set_nextid().  The class driver can help us here, because
 741          * it can use the obd_recovering flag to determine when the
 742          * the OBD is full available. */
 743         /* MDD device will care about that
 744         if (!obd->obd_recovering)
 745                 rc = mds_postrecov(obd);
 746          */
 747         RETURN(rc);
 748
 749 err_exit:
 750         mds->mds_osc_exp = NULL;
 751         mds->mds_osc_obd = ERR_PTR(rc);
 752         RETURN(rc);
 753 }
 754
 755 int mds_lov_disconnect(struct obd_device *obd)
 756 {
 757         struct mds_obd *mds = &obd->u.mds;
 758         int rc = 0;
 759         ENTRY;
 760
 761         if (!IS_ERR(mds->mds_osc_obd) && mds->mds_osc_exp != NULL) {
 762                 obd_register_observer(mds->mds_osc_obd, NULL);
 763
 764                 /* The actual disconnect of the mds_lov will be called from
 765                  * class_disconnect_exports from mds_lov_clean. So we have to
 766                  * ensure that class_cleanup doesn't fail due to the extra ref
 767                  * we're holding now. The mechanism to do that already exists -
 768                  * the obd_force flag. We'll drop the final ref to the
 769                  * mds_osc_exp in mds_cleanup. */
 770                 mds->mds_osc_obd->obd_force = 1;
 771         }
 772
 773         RETURN(rc);
 774 }
 775
 776 struct mds_lov_sync_info {
 777         struct obd_device    *mlsi_obd;     /* the lov device to sync */
 778         struct obd_device    *mlsi_watched; /* target osc */
 779         __u32                 mlsi_index;   /* index of target */
 780         enum obd_notify_event mlsi_ev;      /* event type */
 781 };
 782
 783 static int mds_propagate_capa_keys(struct mds_obd *mds, struct obd_uuid *uuid)
 784 {
 785         struct mds_capa_info    info = { .uuid = uuid };
 786         struct lustre_capa_key *key;
 787         int i, rc = 0;
 788
 789         ENTRY;
 790
 791         if (!mds->mds_capa_keys)
 792                 RETURN(0);
 793
 794         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_SYNC_CAPA_SL, 5);
 795         for (i = 0; i < 2; i++) {
 796                 key = &mds->mds_capa_keys[i];
 797                 DEBUG_CAPA_KEY(D_SEC, key, "propagate");
 798
 799                 info.capa = key;
 800                 rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_CAPA_KEY),
 801                                         KEY_CAPA_KEY, sizeof(info), &info, NULL);
 802                 if (rc) {
 803                         DEBUG_CAPA_KEY(D_ERROR, key,
 804                                        "propagate failed (rc = %d) for", rc);
 805                         RETURN(rc);
 806                 }
 807         }
 808
 809         RETURN(0);
 810 }
 811
 812 /* We only sync one osc at a time, so that we don't have to hold
 813    any kind of lock on the whole mds_lov_desc, which may change
 814    (grow) as a result of mds_lov_add_ost.  This also avoids any
 815    kind of mismatch between the lov_desc and the mds_lov_desc,
 816    which are not in lock-step during lov_add_obd */
 817 static int __mds_lov_synchronize(void *data)
 818 {
 819         struct mds_lov_sync_info *mlsi = data;
 820         struct obd_device *obd = mlsi->mlsi_obd;
 821         struct obd_device *watched = mlsi->mlsi_watched;
 822         struct mds_obd *mds = &obd->u.mds;
 823         struct obd_uuid *uuid;
 824         __u32  idx = mlsi->mlsi_index;
 825         enum obd_notify_event ev = mlsi->mlsi_ev;
 826         struct mds_group_info mgi;
 827         struct llog_ctxt *ctxt;
 828         int rc = 0;
 829         ENTRY;
 830
 831         OBD_FREE_PTR(mlsi);
 832
 833         LASSERT(obd);
 834         LASSERT(watched);
 835         uuid = &watched->u.cli.cl_target_uuid;
 836         LASSERT(uuid);
 837
 838         cfs_down_read(&mds->mds_notify_lock);
 839         if (obd->obd_stopping || obd->obd_fail)
 840                 GOTO(out, rc = -ENODEV);
 841
 842         OBD_RACE(OBD_FAIL_MDS_LOV_SYNC_RACE);
 843         rc = mds_lov_update_mds(obd, watched, idx, ev);
 844         if (rc != 0) {
 845                 CERROR("%s failed at update_mds: %d\n", obd_uuid2str(uuid), rc);
 846                 GOTO(out, rc);
 847         }
 848         mgi.group = mdt_to_obd_objgrp(mds->mds_id);
 849         mgi.uuid = uuid;
 850
 851         rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_MDS_CONN),
 852                                 KEY_MDS_CONN, sizeof(mgi), &mgi, NULL);
 853         if (rc != 0)
 854                 GOTO(out, rc);
 855         /* propagate capability keys */
 856         rc = mds_propagate_capa_keys(mds, uuid);
 857         if (rc)
 858                 GOTO(out, rc);
 859
 860         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 861         if (!ctxt)
 862                 GOTO(out, rc = -ENODEV);
 863
 864         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_LLOG_SYNC_TIMEOUT, 60);
 865         rc = llog_connect(ctxt, NULL, NULL, uuid);
 866         llog_ctxt_put(ctxt);
 867         if (rc != 0) {
 868                 CERROR("%s failed at llog_origin_connect: %d\n",
 869                        obd_uuid2str(uuid), rc);
 870                 GOTO(out, rc);
 871         }
 872
 873         LCONSOLE_INFO("MDS %s: %s now active, resetting orphans\n",
 874               obd->obd_name, obd_uuid2str(uuid));
 875         rc = mds_lov_clear_orphans(mds, uuid);
 876         if (rc != 0) {
 877                 CERROR("%s failed at mds_lov_clear_orphans: %d\n",
 878                        obd_uuid2str(uuid), rc);
 879                 GOTO(out, rc);
 880         }
 881
 882 #ifdef HAVE_QUOTA_SUPPORT
 883         if (obd->obd_upcall.onu_owner) {
 884                 /*
 885                  * This is a hack for mds_notify->mdd_notify. When the mds obd
 886                  * in mdd is removed, This hack should be removed.
 887                  */
 888                 LASSERT(obd->obd_upcall.onu_upcall != NULL);
 889                 rc = obd->obd_upcall.onu_upcall(obd, NULL, OBD_NOTIFY_QUOTA,
 890                                                 obd->obd_upcall.onu_owner,NULL);
 891         }
 892 #endif
 893         EXIT;
 894 out:
 895         cfs_up_read(&mds->mds_notify_lock);
 896         if (rc) {
 897                 /* Deactivate it for safety */
 898                 CERROR("%s sync failed %d, deactivating\n", obd_uuid2str(uuid),
 899                        rc);
 900                 if (!obd->obd_stopping && mds->mds_osc_obd &&
 901                     !mds->mds_osc_obd->obd_stopping && !watched->obd_stopping)
 902                         obd_notify(mds->mds_osc_obd, watched,
 903                                    OBD_NOTIFY_INACTIVE, NULL);
 904         }
 905
 906         class_decref(obd, "mds_lov_synchronize", obd);
 907         return rc;
 908 }
 909
 910 int mds_lov_synchronize(void *data)
 911 {
 912         struct mds_lov_sync_info *mlsi = data;
 913         char name[20];
 914
 915         snprintf(name, sizeof(name), "ll_sync_%02u", mlsi->mlsi_index);
 916         cfs_daemonize_ctxt(name);
 917
 918         RETURN(__mds_lov_synchronize(data));
 919 }
 920
 921 int mds_lov_start_synchronize(struct obd_device *obd,
 922                               struct obd_device *watched,
 923                               void *data, enum obd_notify_event ev)
 924 {
 925         struct mds_lov_sync_info *mlsi;
 926         int rc;
 927         struct obd_uuid *uuid;
 928         ENTRY;
 929
 930         LASSERT(watched);
 931         uuid = &watched->u.cli.cl_target_uuid;
 932
 933         OBD_ALLOC(mlsi, sizeof(*mlsi));
 934         if (mlsi == NULL)
 935                 RETURN(-ENOMEM);
 936
 937         LASSERT(data);
 938         mlsi->mlsi_obd = obd;
 939         mlsi->mlsi_watched = watched;
 940         mlsi->mlsi_index = *(__u32 *)data;
 941         mlsi->mlsi_ev = ev;
 942
 943         /* Although class_export_get(obd->obd_self_export) would lock
 944            the MDS in place, since it's only a self-export
 945            it doesn't lock the LOV in place.  The LOV can be disconnected
 946            during MDS precleanup, leaving nothing for __mds_lov_synchronize.
 947            Simply taking an export ref on the LOV doesn't help, because it's
 948            still disconnected. Taking an obd reference insures that we don't
 949            disconnect the LOV.  This of course means a cleanup won't
 950            finish for as long as the sync is blocking. */
 951         class_incref(obd, "mds_lov_synchronize", obd);
 952
 953         if (ev != OBD_NOTIFY_SYNC) {
 954                 /* Synchronize in the background */
 955                 rc = cfs_kernel_thread(mds_lov_synchronize, mlsi,
 956                                        CLONE_VM | CLONE_FILES);
 957                 if (rc < 0) {
 958                         CERROR("%s: error starting mds_lov_synchronize: %d\n",
 959                                obd->obd_name, rc);
 960                         class_decref(obd, "mds_lov_synchronize", obd);
 961                 } else {
 962                         CDEBUG(D_HA, "%s: mds_lov_synchronize idx=%d "
 963                                "thread=%d\n", obd->obd_name,
 964                                mlsi->mlsi_index, rc);
 965                         rc = 0;
 966                 }
 967         } else {
 968                 rc = __mds_lov_synchronize((void *)mlsi);
 969         }
 970
 971         RETURN(rc);
 972 }
 973
 974 int mds_notify(struct obd_device *obd, struct obd_device *watched,
 975                enum obd_notify_event ev, void *data)
 976 {
 977         int rc = 0;
 978         ENTRY;
 979
 980         CDEBUG(D_CONFIG, "notify %s ev=%d\n", watched->obd_name, ev);
 981
 982         switch (ev) {
 983         /* We only handle these: */
 984         case OBD_NOTIFY_ACTIVE:
 985                 /* lov want one or more _active_ targets for work */
 986                 /* activate event should be pass lov idx as argument */
 987         case OBD_NOTIFY_SYNC:
 988         case OBD_NOTIFY_SYNC_NONBLOCK:
 989                 /* sync event should be pass lov idx as argument */
 990                 break;
 991         default:
 992                 RETURN(0);
 993         }
 994
 995         if (strcmp(watched->obd_type->typ_name, LUSTRE_OSC_NAME) != 0) {
 996                 CERROR("unexpected notification of %s %s!\n",
 997                        watched->obd_type->typ_name, watched->obd_name);
 998                 RETURN(-EINVAL);
 999         }
1000
1001         if (obd->obd_recovering) {
1002                 CWARN("MDS %s: in recovery, not resetting orphans on %s\n",
1003                       obd->obd_name,
1004                       obd_uuid2str(&watched->u.cli.cl_target_uuid));
1005                 /* We still have to fix the lov descriptor for ost's added
1006                    after the mdt in the config log.  They didn't make it into
1007                    mds_lov_connect. */
1008                 rc = mds_lov_update_desc(obd, *(__u32 *)data,
1009                                          &watched->u.cli.cl_target_uuid, ev);
1010         } else {
1011                 rc = mds_lov_start_synchronize(obd, watched, data, ev);
1012         }
1013         RETURN(rc);
1014 }