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 (c) 2002, 2010, Oracle and/or its affiliates. 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_seq = mdt_to_obd_objseq(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 + MDS_LOV_ALLOC_SIZE - 1) / MDS_LOV_ALLOC_SIZE;
 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                                         MDS_LOV_ALLOC_SIZE, &off);
 413                 if (rc < 0) {
 414                         CERROR("Error reading objids %d\n", rc);
 415                         GOTO(out, rc);
 416                 }
 417                 if (off == off_old) /* hole is read */
 418                         off += MDS_LOV_ALLOC_SIZE;
 419
 420                 count = (off - off_old) / sizeof(obd_id);
 421                 if (mds_lov_update_from_read(mds, data, count)) {
 422                         CERROR("Can't update mds data\n");
 423                         GOTO(out, rc = -EIO);
 424                 }
 425         }
 426         mds->mds_lov_objid_lastpage = page - 1;
 427         mds->mds_lov_objid_lastidx = count - 1;
 428
 429         CDEBUG(D_INFO, "Read %u - %u %u objid\n", mds->mds_lov_objid_count,
 430                mds->mds_lov_objid_lastpage, mds->mds_lov_objid_lastidx);
 431 out:
 432         mds_lov_dump_objids("read",obd);
 433
 434         RETURN(rc);
 435 }
 436
 437 int mds_lov_write_objids(struct obd_device *obd)
 438 {
 439         struct mds_obd *mds = &obd->u.mds;
 440         int i = 0, rc = 0;
 441         ENTRY;
 442
 443         if (cfs_bitmap_check_empty(mds->mds_lov_page_dirty))
 444                 RETURN(0);
 445
 446         mds_lov_dump_objids("write", obd);
 447
 448         cfs_foreach_bit(mds->mds_lov_page_dirty, i) {
 449                 obd_id *data =  mds->mds_lov_page_array[i];
 450                 unsigned int size = MDS_LOV_ALLOC_SIZE;
 451                 loff_t off = i * size;
 452
 453                 LASSERT(data != NULL);
 454
 455                 if (!cfs_bitmap_test_and_clear(mds->mds_lov_page_dirty, i))
 456                         continue;
 457
 458                 /* check for particaly filled last page */
 459                 if (i == mds->mds_lov_objid_lastpage)
 460                         size = (mds->mds_lov_objid_lastidx+1) * sizeof(obd_id);
 461
 462                 CDEBUG(D_INFO, "write %lld - %u\n", off, size);
 463                 rc = fsfilt_write_record(obd, mds->mds_lov_objid_filp, data,
 464                                          size, &off, 0);
 465                 if (rc < 0) {
 466                         cfs_bitmap_set(mds->mds_lov_page_dirty, i);
 467                         break;
 468                 }
 469         }
 470         if (rc >= 0)
 471                 rc = 0;
 472
 473         RETURN(rc);
 474 }
 475 EXPORT_SYMBOL(mds_lov_write_objids);
 476
 477 static int mds_lov_get_objid(struct obd_device * obd,
 478                              obd_id idx)
 479 {
 480         struct mds_obd *mds = &obd->u.mds;
 481         unsigned int page;
 482         unsigned int off;
 483         obd_id *data;
 484         int rc = 0;
 485         ENTRY;
 486
 487         page = idx / OBJID_PER_PAGE();
 488         off = idx % OBJID_PER_PAGE();
 489         data = mds->mds_lov_page_array[page];
 490         if (data[off] < 2) {
 491                 /* We never read this lastid; ask the osc */
 492                 struct obd_id_info lastid;
 493                 __u32 size = sizeof(lastid);
 494
 495                 lastid.idx = idx;
 496                 lastid.data = &data[off];
 497                 rc = obd_get_info(mds->mds_osc_exp, sizeof(KEY_LAST_ID),
 498                                   KEY_LAST_ID, &size, &lastid, NULL);
 499                 if (rc)
 500                         GOTO(out, rc);
 501
 502                 /* workaround for clean filter */
 503                 if (data[off] == 0)
 504                         data[off] = 1;
 505
 506                 cfs_bitmap_set(mds->mds_lov_page_dirty, page);
 507         }
 508         CDEBUG(D_INFO, "idx "LPU64" - %p - %d/%d - "LPU64"\n",
 509                idx, data, page, off, data[off]);
 510 out:
 511         RETURN(rc);
 512 }
 513
 514 int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid)
 515 {
 516         int rc;
 517         struct obdo oa = { 0 };
 518         struct obd_trans_info oti = {0};
 519         struct lov_stripe_md  *empty_ea = NULL;
 520         ENTRY;
 521
 522         LASSERT(mds->mds_lov_page_array != NULL);
 523
 524         /* This create will in fact either create or destroy:  If the OST is
 525          * missing objects below this ID, they will be created.  If it finds
 526          * objects above this ID, they will be removed. */
 527         memset(&oa, 0, sizeof(oa));
 528         oa.o_flags = OBD_FL_DELORPHAN;
 529         oa.o_seq = mdt_to_obd_objseq(mds->mds_id);
 530         oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
 531         if (ost_uuid != NULL)
 532                 oti.oti_ost_uuid = ost_uuid;
 533
 534         rc = obd_create(mds->mds_osc_exp, &oa, &empty_ea, &oti);
 535
 536         RETURN(rc);
 537 }
 538
 539 /* for one target */
 540 static int mds_lov_set_one_nextid(struct obd_device *obd, __u32 idx, obd_id *id)
 541 {
 542         struct mds_obd *mds = &obd->u.mds;
 543         int rc;
 544         struct obd_id_info info;
 545         ENTRY;
 546
 547         LASSERT(!obd->obd_recovering);
 548
 549         info.idx = idx;
 550         info.data = id;
 551         rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_NEXT_ID),
 552                                 KEY_NEXT_ID, sizeof(info), &info, NULL);
 553         if (rc)
 554                 CERROR ("%s: mds_lov_set_nextid failed (%d)\n",
 555                         obd->obd_name, rc);
 556
 557         RETURN(rc);
 558 }
 559
 560 /* Update the lov desc for a new size lov. */
 561 static int mds_lov_update_desc(struct obd_device *obd, int idx,
 562                                struct obd_uuid *uuid, enum obd_notify_event ev)
 563 {
 564         struct mds_obd *mds = &obd->u.mds;
 565         struct lov_desc *ld;
 566         __u32 valsize = sizeof(mds->mds_lov_desc);
 567         int rc = 0;
 568         ENTRY;
 569
 570         OBD_ALLOC(ld, sizeof(*ld));
 571         if (!ld)
 572                 RETURN(-ENOMEM);
 573
 574         rc = obd_get_info(mds->mds_osc_exp, sizeof(KEY_LOVDESC), KEY_LOVDESC,
 575                           &valsize, ld, NULL);
 576         if (rc)
 577                 GOTO(out, rc);
 578
 579         /* Don't change the mds_lov_desc until the objids size matches the
 580            count (paranoia) */
 581         mds->mds_lov_desc = *ld;
 582         CDEBUG(D_CONFIG, "updated lov_desc, tgt_count: %d - idx %d / uuid %s\n",
 583                mds->mds_lov_desc.ld_tgt_count, idx, uuid->uuid);
 584
 585         cfs_mutex_down(&obd->obd_dev_sem);
 586         rc = mds_lov_update_max_ost(mds, idx);
 587         cfs_mutex_up(&obd->obd_dev_sem);
 588         if (rc != 0)
 589                 GOTO(out, rc );
 590
 591         /* If we added a target we have to reconnect the llogs */
 592         /* We only _need_ to do this at first add (idx), or the first time
 593            after recovery.  However, it should now be safe to call anytime. */
 594         rc = obd_llog_init(obd, &obd->obd_olg, obd, &idx);
 595         if (rc)
 596                 GOTO(out, rc);
 597
 598         /*XXX this notifies the MDD until lov handling use old mds code */
 599         if (obd->obd_upcall.onu_owner) {
 600                  LASSERT(obd->obd_upcall.onu_upcall != NULL);
 601                  rc = obd->obd_upcall.onu_upcall(obd, NULL, ev,
 602                                                  obd->obd_upcall.onu_owner,
 603                                                  &mds->mds_obt.obt_mount_count);
 604         }
 605 out:
 606         OBD_FREE(ld, sizeof(*ld));
 607         RETURN(rc);
 608 }
 609
 610 /* Inform MDS about new/updated target */
 611 static int mds_lov_update_mds(struct obd_device *obd,
 612                               struct obd_device *watched,
 613                               __u32 idx, enum obd_notify_event ev)
 614 {
 615         struct mds_obd *mds = &obd->u.mds;
 616         int rc = 0;
 617         int page;
 618         int off;
 619         obd_id *data;
 620
 621         ENTRY;
 622
 623         /* Don't let anyone else mess with mds_lov_objids now */
 624         rc = mds_lov_update_desc(obd, idx, &watched->u.cli.cl_target_uuid, ev);
 625         if (rc)
 626                 GOTO(out, rc);
 627
 628         CDEBUG(D_CONFIG, "idx=%d, recov=%d/%d, cnt=%d\n",
 629                idx, obd->obd_recovering, obd->obd_async_recov,
 630                mds->mds_lov_desc.ld_tgt_count);
 631
 632         /* idx is set as data from lov_notify. */
 633         if (obd->obd_recovering)
 634                 GOTO(out, rc);
 635
 636         if (idx >= mds->mds_lov_desc.ld_tgt_count) {
 637                 CERROR("index %d > count %d!\n", idx,
 638                        mds->mds_lov_desc.ld_tgt_count);
 639                 GOTO(out, rc = -EINVAL);
 640         }
 641
 642         rc = mds_lov_get_objid(obd, idx);
 643         if (rc)
 644                 GOTO(out, rc);
 645
 646         page = idx / OBJID_PER_PAGE();
 647         off = idx % OBJID_PER_PAGE();
 648         data = mds->mds_lov_page_array[page];
 649
 650         /* We have read this lastid from disk; tell the osc.
 651            Don't call this during recovery. */
 652         rc = mds_lov_set_one_nextid(obd, idx, &data[off]);
 653         if (rc) {
 654                 CERROR("Failed to set next id, idx=%d rc=%d\n", idx,rc);
 655                 /* Don't abort the rest of the sync */
 656                 rc = 0;
 657         } else {
 658                 CDEBUG(D_CONFIG, "last object "LPU64" from OST %d rc=%d\n",
 659                         data[off], idx, rc);
 660         }
 661 out:
 662         RETURN(rc);
 663 }
 664
 665 /* update the LOV-OSC knowledge of the last used object id's */
 666 int mds_lov_connect(struct obd_device *obd, char * lov_name)
 667 {
 668         struct mds_obd *mds = &obd->u.mds;
 669         struct obd_connect_data *data;
 670         int rc;
 671         ENTRY;
 672
 673         if (IS_ERR(mds->mds_osc_obd))
 674                 RETURN(PTR_ERR(mds->mds_osc_obd));
 675
 676         if (mds->mds_osc_obd)
 677                 RETURN(0);
 678
 679         mds->mds_osc_obd = class_name2obd(lov_name);
 680         if (!mds->mds_osc_obd) {
 681                 CERROR("MDS cannot locate LOV %s\n", lov_name);
 682                 mds->mds_osc_obd = ERR_PTR(-ENOTCONN);
 683                 RETURN(-ENOTCONN);
 684         }
 685
 686         cfs_mutex_down(&obd->obd_dev_sem);
 687         rc = mds_lov_read_objids(obd);
 688         cfs_mutex_up(&obd->obd_dev_sem);
 689         if (rc) {
 690                 CERROR("cannot read %s: rc = %d\n", "lov_objids", rc);
 691                 GOTO(err_exit, rc);
 692         }
 693
 694         rc = obd_register_observer(mds->mds_osc_obd, obd);
 695         if (rc) {
 696                 CERROR("MDS cannot register as observer of LOV %s (%d)\n",
 697                        lov_name, rc);
 698                 GOTO(err_exit, rc);
 699         }
 700
 701         /* try init too early */
 702         rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
 703         if (rc)
 704                 GOTO(err_exit, rc);
 705
 706         mds->mds_osc_obd->u.lov.lov_sp_me = LUSTRE_SP_MDT;
 707
 708         OBD_ALLOC(data, sizeof(*data));
 709         if (data == NULL)
 710                 GOTO(err_exit, rc = -ENOMEM);
 711
 712         data->ocd_connect_flags = OBD_CONNECT_VERSION   | OBD_CONNECT_INDEX   |
 713                                   OBD_CONNECT_REQPORTAL | OBD_CONNECT_QUOTA64 |
 714                                   OBD_CONNECT_OSS_CAPA  | OBD_CONNECT_FID     |
 715                                   OBD_CONNECT_BRW_SIZE  | OBD_CONNECT_CKSUM   |
 716                                   OBD_CONNECT_CHANGE_QS | OBD_CONNECT_AT      |
 717                                   OBD_CONNECT_MDS | OBD_CONNECT_SKIP_ORPHAN   |
 718                                   OBD_CONNECT_SOM | OBD_CONNECT_FULL20;
 719 #ifdef HAVE_LRU_RESIZE_SUPPORT
 720         data->ocd_connect_flags |= OBD_CONNECT_LRU_RESIZE;
 721 #endif
 722         data->ocd_version = LUSTRE_VERSION_CODE;
 723         data->ocd_group = mdt_to_obd_objseq(mds->mds_id);
 724         /* send max bytes per rpc */
 725         data->ocd_brw_size = PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT;
 726         /* send the list of supported checksum types */
 727         data->ocd_cksum_types = OBD_CKSUM_ALL;
 728         /* NB: lov_connect() needs to fill in .ocd_index for each OST */
 729         rc = obd_connect(NULL, &mds->mds_osc_exp, mds->mds_osc_obd, &obd->obd_uuid, data, NULL);
 730         OBD_FREE(data, sizeof(*data));
 731         if (rc) {
 732                 CERROR("MDS cannot connect to LOV %s (%d)\n", lov_name, rc);
 733                 mds->mds_osc_obd = ERR_PTR(rc);
 734                 RETURN(rc);
 735         }
 736
 737         /* I want to see a callback happen when the OBD moves to a
 738          * "For General Use" state, and that's when we'll call
 739          * set_nextid().  The class driver can help us here, because
 740          * it can use the obd_recovering flag to determine when the
 741          * the OBD is full available. */
 742         /* MDD device will care about that
 743         if (!obd->obd_recovering)
 744                 rc = mds_postrecov(obd);
 745          */
 746         RETURN(rc);
 747
 748 err_exit:
 749         mds->mds_osc_exp = NULL;
 750         mds->mds_osc_obd = ERR_PTR(rc);
 751         RETURN(rc);
 752 }
 753
 754 int mds_lov_disconnect(struct obd_device *obd)
 755 {
 756         struct mds_obd *mds = &obd->u.mds;
 757         int rc = 0;
 758         ENTRY;
 759
 760         if (!IS_ERR(mds->mds_osc_obd) && mds->mds_osc_exp != NULL) {
 761                 obd_register_observer(mds->mds_osc_obd, NULL);
 762
 763                 /* The actual disconnect of the mds_lov will be called from
 764                  * class_disconnect_exports from mds_lov_clean. So we have to
 765                  * ensure that class_cleanup doesn't fail due to the extra ref
 766                  * we're holding now. The mechanism to do that already exists -
 767                  * the obd_force flag. We'll drop the final ref to the
 768                  * mds_osc_exp in mds_cleanup. */
 769                 mds->mds_osc_obd->obd_force = 1;
 770         }
 771
 772         RETURN(rc);
 773 }
 774
 775 struct mds_lov_sync_info {
 776         struct obd_device    *mlsi_obd;     /* the lov device to sync */
 777         struct obd_device    *mlsi_watched; /* target osc */
 778         __u32                 mlsi_index;   /* index of target */
 779         enum obd_notify_event mlsi_ev;      /* event type */
 780 };
 781
 782 static int mds_propagate_capa_keys(struct mds_obd *mds, struct obd_uuid *uuid)
 783 {
 784         struct mds_capa_info    info = { .uuid = uuid };
 785         struct lustre_capa_key *key;
 786         int i, rc = 0;
 787
 788         ENTRY;
 789
 790         if (!mds->mds_capa_keys)
 791                 RETURN(0);
 792
 793         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_SYNC_CAPA_SL, 5);
 794         for (i = 0; i < 2; i++) {
 795                 key = &mds->mds_capa_keys[i];
 796                 DEBUG_CAPA_KEY(D_SEC, key, "propagate");
 797
 798                 info.capa = key;
 799                 rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_CAPA_KEY),
 800                                         KEY_CAPA_KEY, sizeof(info), &info, NULL);
 801                 if (rc) {
 802                         DEBUG_CAPA_KEY(D_ERROR, key,
 803                                        "propagate failed (rc = %d) for", rc);
 804                         RETURN(rc);
 805                 }
 806         }
 807
 808         RETURN(0);
 809 }
 810
 811 /* We only sync one osc at a time, so that we don't have to hold
 812    any kind of lock on the whole mds_lov_desc, which may change
 813    (grow) as a result of mds_lov_add_ost.  This also avoids any
 814    kind of mismatch between the lov_desc and the mds_lov_desc,
 815    which are not in lock-step during lov_add_obd */
 816 static int __mds_lov_synchronize(void *data)
 817 {
 818         struct mds_lov_sync_info *mlsi = data;
 819         struct obd_device *obd = mlsi->mlsi_obd;
 820         struct obd_device *watched = mlsi->mlsi_watched;
 821         struct mds_obd *mds = &obd->u.mds;
 822         struct obd_uuid *uuid;
 823         __u32  idx = mlsi->mlsi_index;
 824         enum obd_notify_event ev = mlsi->mlsi_ev;
 825         struct mds_group_info mgi;
 826         struct llog_ctxt *ctxt;
 827         int rc = 0;
 828         ENTRY;
 829
 830         OBD_FREE_PTR(mlsi);
 831
 832         LASSERT(obd);
 833         LASSERT(watched);
 834         uuid = &watched->u.cli.cl_target_uuid;
 835         LASSERT(uuid);
 836
 837         cfs_down_read(&mds->mds_notify_lock);
 838         if (obd->obd_stopping || obd->obd_fail)
 839                 GOTO(out, rc = -ENODEV);
 840
 841         OBD_RACE(OBD_FAIL_MDS_LOV_SYNC_RACE);
 842         rc = mds_lov_update_mds(obd, watched, idx, ev);
 843         if (rc != 0) {
 844                 CERROR("%s failed at update_mds: %d\n", obd_uuid2str(uuid), rc);
 845                 GOTO(out, rc);
 846         }
 847         mgi.group = mdt_to_obd_objseq(mds->mds_id);
 848         mgi.uuid = uuid;
 849
 850         rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_MDS_CONN),
 851                                 KEY_MDS_CONN, sizeof(mgi), &mgi, NULL);
 852         if (rc != 0)
 853                 GOTO(out, rc);
 854         /* propagate capability keys */
 855         rc = mds_propagate_capa_keys(mds, uuid);
 856         if (rc)
 857                 GOTO(out, rc);
 858
 859         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 860         if (!ctxt)
 861                 GOTO(out, rc = -ENODEV);
 862
 863         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_LLOG_SYNC_TIMEOUT, 60);
 864         rc = llog_connect(ctxt, NULL, NULL, uuid);
 865         llog_ctxt_put(ctxt);
 866         if (rc != 0) {
 867                 CERROR("%s failed at llog_origin_connect: %d\n",
 868                        obd_uuid2str(uuid), rc);
 869                 GOTO(out, rc);
 870         }
 871
 872         LCONSOLE_INFO("MDS %s: %s now active, resetting orphans\n",
 873               obd->obd_name, obd_uuid2str(uuid));
 874         rc = mds_lov_clear_orphans(mds, uuid);
 875         if (rc != 0) {
 876                 CERROR("%s failed at mds_lov_clear_orphans: %d\n",
 877                        obd_uuid2str(uuid), rc);
 878                 GOTO(out, rc);
 879         }
 880
 881 #ifdef HAVE_QUOTA_SUPPORT
 882         if (obd->obd_upcall.onu_owner) {
 883                 /*
 884                  * This is a hack for mds_notify->mdd_notify. When the mds obd
 885                  * in mdd is removed, This hack should be removed.
 886                  */
 887                 LASSERT(obd->obd_upcall.onu_upcall != NULL);
 888                 rc = obd->obd_upcall.onu_upcall(obd, NULL, OBD_NOTIFY_QUOTA,
 889                                                 obd->obd_upcall.onu_owner,NULL);
 890         }
 891 #endif
 892         EXIT;
 893 out:
 894         cfs_up_read(&mds->mds_notify_lock);
 895         if (rc) {
 896                 /* Deactivate it for safety */
 897                 CERROR("%s sync failed %d, deactivating\n", obd_uuid2str(uuid),
 898                        rc);
 899                 if (!obd->obd_stopping && mds->mds_osc_obd &&
 900                     !mds->mds_osc_obd->obd_stopping && !watched->obd_stopping)
 901                         obd_notify(mds->mds_osc_obd, watched,
 902                                    OBD_NOTIFY_INACTIVE, NULL);
 903         }
 904
 905         class_decref(obd, "mds_lov_synchronize", obd);
 906         return rc;
 907 }
 908
 909 int mds_lov_synchronize(void *data)
 910 {
 911         struct mds_lov_sync_info *mlsi = data;
 912         char name[20];
 913
 914         snprintf(name, sizeof(name), "ll_sync_%02u", mlsi->mlsi_index);
 915         cfs_daemonize_ctxt(name);
 916
 917         RETURN(__mds_lov_synchronize(data));
 918 }
 919
 920 int mds_lov_start_synchronize(struct obd_device *obd,
 921                               struct obd_device *watched,
 922                               void *data, enum obd_notify_event ev)
 923 {
 924         struct mds_lov_sync_info *mlsi;
 925         int rc;
 926         struct obd_uuid *uuid;
 927         ENTRY;
 928
 929         LASSERT(watched);
 930         uuid = &watched->u.cli.cl_target_uuid;
 931
 932         OBD_ALLOC(mlsi, sizeof(*mlsi));
 933         if (mlsi == NULL)
 934                 RETURN(-ENOMEM);
 935
 936         LASSERT(data);
 937         mlsi->mlsi_obd = obd;
 938         mlsi->mlsi_watched = watched;
 939         mlsi->mlsi_index = *(__u32 *)data;
 940         mlsi->mlsi_ev = ev;
 941
 942         /* Although class_export_get(obd->obd_self_export) would lock
 943            the MDS in place, since it's only a self-export
 944            it doesn't lock the LOV in place.  The LOV can be disconnected
 945            during MDS precleanup, leaving nothing for __mds_lov_synchronize.
 946            Simply taking an export ref on the LOV doesn't help, because it's
 947            still disconnected. Taking an obd reference insures that we don't
 948            disconnect the LOV.  This of course means a cleanup won't
 949            finish for as long as the sync is blocking. */
 950         class_incref(obd, "mds_lov_synchronize", obd);
 951
 952         if (ev != OBD_NOTIFY_SYNC) {
 953                 /* Synchronize in the background */
 954                 rc = cfs_kernel_thread(mds_lov_synchronize, mlsi,
 955                                        CLONE_VM | CLONE_FILES);
 956                 if (rc < 0) {
 957                         CERROR("%s: error starting mds_lov_synchronize: %d\n",
 958                                obd->obd_name, rc);
 959                         class_decref(obd, "mds_lov_synchronize", obd);
 960                 } else {
 961                         CDEBUG(D_HA, "%s: mds_lov_synchronize idx=%d "
 962                                "thread=%d\n", obd->obd_name,
 963                                mlsi->mlsi_index, rc);
 964                         rc = 0;
 965                 }
 966         } else {
 967                 rc = __mds_lov_synchronize((void *)mlsi);
 968         }
 969
 970         RETURN(rc);
 971 }
 972
 973 int mds_notify(struct obd_device *obd, struct obd_device *watched,
 974                enum obd_notify_event ev, void *data)
 975 {
 976         int rc = 0;
 977         ENTRY;
 978
 979         CDEBUG(D_CONFIG, "notify %s ev=%d\n", watched->obd_name, ev);
 980
 981         switch (ev) {
 982         /* We only handle these: */
 983         case OBD_NOTIFY_ACTIVE:
 984                 /* lov want one or more _active_ targets for work */
 985                 /* activate event should be pass lov idx as argument */
 986         case OBD_NOTIFY_SYNC:
 987         case OBD_NOTIFY_SYNC_NONBLOCK:
 988                 /* sync event should be pass lov idx as argument */
 989                 break;
 990         default:
 991                 RETURN(0);
 992         }
 993
 994         if (strcmp(watched->obd_type->typ_name, LUSTRE_OSC_NAME) != 0) {
 995                 CERROR("unexpected notification of %s %s!\n",
 996                        watched->obd_type->typ_name, watched->obd_name);
 997                 RETURN(-EINVAL);
 998         }
 999
1000         if (obd->obd_recovering) {
1001                 CWARN("MDS %s: in recovery, not resetting orphans on %s\n",
1002                       obd->obd_name,
1003                       obd_uuid2str(&watched->u.cli.cl_target_uuid));
1004                 /* We still have to fix the lov descriptor for ost's added
1005                    after the mdt in the config log.  They didn't make it into
1006                    mds_lov_connect. */
1007                 rc = mds_lov_update_desc(obd, *(__u32 *)data,
1008                                          &watched->u.cli.cl_target_uuid, ev);
1009         } else {
1010                 rc = mds_lov_start_synchronize(obd, watched, data, ev);
1011         }
1012         RETURN(rc);
1013 }