lustre/ofd/ofd_fs.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/ofd/ofd_fs.c
  32  *
  33  * This file provides helper functions to handle various data stored on disk.
  34  * It uses OSD API and works with any OSD.
  35  *
  36  * Note: this file contains also functions for sequence handling, they are
  37  * placed here improperly and will be moved to the ofd_dev.c and ofd_internal.h,
  38  * this comment is to be removed after that.
  39  *
  40  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.pershin@intel.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_FILTER
  45
  46 #include "ofd_internal.h"
  47
  48 /**
  49  * Restrict precreate batch count by its upper limit.
  50  *
  51  * The precreate batch count is a number of precreates to do in
  52  * single transaction. It has upper limit - ofd_device::ofd_precreate_batch
  53  * value which shouldn't be exceeded.
  54  *
  55  * \param[in] ofd       OFD device
  56  * \param[in] batch     number of updates in the batch
  57  *
  58  * \retval              \a batch limited by ofd_device::ofd_precreate_batch
  59  */
  60 int ofd_precreate_batch(struct ofd_device *ofd, int batch)
  61 {
  62         int count;
  63
  64         spin_lock(&ofd->ofd_batch_lock);
  65         count = min(ofd->ofd_precreate_batch, batch);
  66         spin_unlock(&ofd->ofd_batch_lock);
  67
  68         return count;
  69 }
  70
  71 /**
  72  * Get ofd_seq for \a seq.
  73  *
  74  * Function finds appropriate structure by \a seq number and
  75  * increases the reference counter of that structure.
  76  *
  77  * \param[in] ofd       OFD device
  78  * \param[in] seq       sequence number, FID sequence number usually
  79  *
  80  * \retval              pointer to the requested ofd_seq structure
  81  * \retval              NULL if ofd_seq is not found
  82  */
  83 struct ofd_seq *ofd_seq_get(struct ofd_device *ofd, u64 seq)
  84 {
  85         struct ofd_seq *oseq;
  86
  87         read_lock(&ofd->ofd_seq_list_lock);
  88         list_for_each_entry(oseq, &ofd->ofd_seq_list, os_list) {
  89                 if (ostid_seq(&oseq->os_oi) == seq) {
  90                         atomic_inc(&oseq->os_refc);
  91                         read_unlock(&ofd->ofd_seq_list_lock);
  92                         return oseq;
  93                 }
  94         }
  95         read_unlock(&ofd->ofd_seq_list_lock);
  96         return NULL;
  97 }
  98
  99 /**
 100  * Drop a reference to ofd_seq.
 101  *
 102  * The paired function to the ofd_seq_get(). It decrease the reference counter
 103  * of the ofd_seq structure and free it if that reference was last one.
 104  *
 105  * \param[in] env       execution environment
 106  * \param[in] oseq      ofd_seq structure to put
 107  */
 108 void ofd_seq_put(const struct lu_env *env, struct ofd_seq *oseq)
 109 {
 110         if (atomic_dec_and_test(&oseq->os_refc)) {
 111                 LASSERT(list_empty(&oseq->os_list));
 112                 LASSERT(oseq->os_lastid_obj != NULL);
 113                 dt_object_put(env, oseq->os_lastid_obj);
 114                 OBD_FREE_PTR(oseq);
 115         }
 116 }
 117
 118 /**
 119  * Add a new ofd_seq to the given OFD device.
 120  *
 121  * First it checks if there is already existent ofd_seq with the same
 122  * sequence number as used by \a new_seq.
 123  * If such ofd_seq is not found then the \a new_seq is added to the list
 124  * of all ofd_seq structures else the \a new_seq is dropped and the found
 125  * ofd_seq is returned back.
 126  *
 127  * \param[in] env       execution environment
 128  * \param[in] ofd       OFD device
 129  * \param[in] new_seq   new ofd_seq to be added
 130  *
 131  * \retval              ofd_seq structure
 132  */
 133 static struct ofd_seq *ofd_seq_add(const struct lu_env *env,
 134                                    struct ofd_device *ofd,
 135                                    struct ofd_seq *new_seq)
 136 {
 137         struct ofd_seq *os = NULL;
 138
 139         write_lock(&ofd->ofd_seq_list_lock);
 140         list_for_each_entry(os, &ofd->ofd_seq_list, os_list) {
 141                 if (ostid_seq(&os->os_oi) == ostid_seq(&new_seq->os_oi)) {
 142                         atomic_inc(&os->os_refc);
 143                         write_unlock(&ofd->ofd_seq_list_lock);
 144                         /* The seq has not been added to the list */
 145                         ofd_seq_put(env, new_seq);
 146                         return os;
 147                 }
 148         }
 149         atomic_inc(&new_seq->os_refc);
 150         list_add_tail(&new_seq->os_list, &ofd->ofd_seq_list);
 151         ofd->ofd_seq_count++;
 152         write_unlock(&ofd->ofd_seq_list_lock);
 153         return new_seq;
 154 }
 155
 156 /**
 157  * Get last object ID for the given sequence.
 158  *
 159  * \param[in] oseq      OFD sequence structure
 160  *
 161  * \retval              the last object ID for this sequence
 162  */
 163 u64 ofd_seq_last_oid(struct ofd_seq *oseq)
 164 {
 165         u64 id;
 166
 167         spin_lock(&oseq->os_last_oid_lock);
 168         id = ostid_id(&oseq->os_oi);
 169         spin_unlock(&oseq->os_last_oid_lock);
 170
 171         return id;
 172 }
 173
 174 /**
 175  * Set new last object ID for the given sequence.
 176  *
 177  * \param[in] oseq      OFD sequence
 178  * \param[in] id        the new OID to set
 179  */
 180 void ofd_seq_last_oid_set(struct ofd_seq *oseq, u64 id)
 181 {
 182         spin_lock(&oseq->os_last_oid_lock);
 183         if (likely(ostid_id(&oseq->os_oi) < id)) {
 184                 if (ostid_set_id(&oseq->os_oi, id)) {
 185                         CERROR("Bad %llu to set " DOSTID "\n",
 186                                (unsigned long long)id, POSTID(&oseq->os_oi));
 187                 }
 188         }
 189         spin_unlock(&oseq->os_last_oid_lock);
 190 }
 191
 192 /**
 193  * Update last used OID on disk for the given sequence.
 194  *
 195  * The last used object ID is stored persistently on disk and
 196  * must be written when updated. This function writes the sequence data.
 197  * The format is just an object ID of the latest used object FID.
 198  * Each ID is stored in per-sequence file.
 199  *
 200  * \param[in] env       execution environment
 201  * \param[in] ofd       OFD device
 202  * \param[in] oseq      ofd_seq structure with data to write
 203  *
 204  * \retval              0 on successful write of data from \a oseq
 205  * \retval              negative value on error
 206  */
 207 int ofd_seq_last_oid_write(const struct lu_env *env, struct ofd_device *ofd,
 208                            struct ofd_seq *oseq)
 209 {
 210         struct ofd_thread_info  *info = ofd_info(env);
 211         u64                      tmp;
 212         struct dt_object        *obj = oseq->os_lastid_obj;
 213         struct thandle          *th;
 214         int                      rc;
 215
 216         ENTRY;
 217
 218         if (ofd->ofd_osd->dd_rdonly)
 219                 RETURN(0);
 220
 221         tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
 222
 223         info->fti_buf.lb_buf = &tmp;
 224         info->fti_buf.lb_len = sizeof(tmp);
 225         info->fti_off = 0;
 226
 227         LASSERT(obj != NULL);
 228
 229         th = dt_trans_create(env, ofd->ofd_osd);
 230         if (IS_ERR(th))
 231                 RETURN(PTR_ERR(th));
 232
 233         rc = dt_declare_record_write(env, obj, &info->fti_buf,
 234                                      info->fti_off, th);
 235         if (rc < 0)
 236                 GOTO(out, rc);
 237         rc = dt_trans_start_local(env, ofd->ofd_osd, th);
 238         if (rc < 0)
 239                 GOTO(out, rc);
 240         rc = dt_record_write(env, obj, &info->fti_buf, &info->fti_off,
 241                              th);
 242         if (rc < 0)
 243                 GOTO(out, rc);
 244
 245         CDEBUG(D_INODE, "%s: write last_objid "DOSTID": rc = %d\n",
 246                ofd_name(ofd), POSTID(&oseq->os_oi), rc);
 247         EXIT;
 248 out:
 249         dt_trans_stop(env, ofd->ofd_osd, th);
 250         return rc;
 251 }
 252
 253 /**
 254  * Deregister LWP items for FLDB and SEQ client on OFD.
 255  *
 256  * LWP is lightweight proxy - simplified connection between
 257  * servers. It is used for FID Location Database (FLDB) and
 258  * sequence (SEQ) client-server interactions.
 259  *
 260  * This function is used during server cleanup process to free
 261  * LWP items that were previously set up upon OFD start.
 262  *
 263  * \param[in]     ofd   OFD device
 264  */
 265 static void ofd_deregister_seq_exp(struct ofd_device *ofd)
 266 {
 267         struct seq_server_site  *ss = &ofd->ofd_seq_site;
 268
 269         if (ss->ss_client_seq != NULL) {
 270                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
 271                 ss->ss_client_seq->lcs_exp = NULL;
 272         }
 273
 274         if (ss->ss_server_fld != NULL) {
 275                 lustre_deregister_lwp_item(&ss->ss_server_fld->lsf_control_exp);
 276                 ss->ss_server_fld->lsf_control_exp = NULL;
 277         }
 278 }
 279
 280 /**
 281  * Stop FLDB server on OFD.
 282  *
 283  * This function is part of OFD cleanup process.
 284  *
 285  * \param[in] env       execution environment
 286  * \param[in] ofd       OFD device
 287  *
 288  */
 289 static void ofd_fld_fini(const struct lu_env *env, struct ofd_device *ofd)
 290 {
 291         struct seq_server_site *ss = &ofd->ofd_seq_site;
 292
 293         if (ss != NULL && ss->ss_server_fld != NULL) {
 294                 fld_server_fini(env, ss->ss_server_fld);
 295                 OBD_FREE_PTR(ss->ss_server_fld);
 296                 ss->ss_server_fld = NULL;
 297         }
 298 }
 299
 300 /**
 301  * Free sequence structures on OFD.
 302  *
 303  * This function is part of OFD cleanup process, it goes through
 304  * the list of ofd_seq structures stored in ofd_device structure
 305  * and frees them.
 306  *
 307  * \param[in] env       execution environment
 308  * \param[in] ofd       OFD device
 309  */
 310 void ofd_seqs_free(const struct lu_env *env, struct ofd_device *ofd)
 311 {
 312         struct ofd_seq *oseq;
 313         struct ofd_seq *tmp;
 314         LIST_HEAD(dispose);
 315
 316         write_lock(&ofd->ofd_seq_list_lock);
 317         list_for_each_entry_safe(oseq, tmp, &ofd->ofd_seq_list, os_list)
 318                 list_move(&oseq->os_list, &dispose);
 319         write_unlock(&ofd->ofd_seq_list_lock);
 320
 321         while (!list_empty(&dispose)) {
 322                 oseq = container_of(dispose.next, struct ofd_seq, os_list);
 323                 list_del_init(&oseq->os_list);
 324                 ofd_seq_put(env, oseq);
 325         }
 326 }
 327
 328 /**
 329  * Stop FLDB and SEQ services on OFD.
 330  *
 331  * This function is part of OFD cleanup process.
 332  *
 333  * \param[in] env       execution environment
 334  * \param[in] ofd       OFD device
 335  *
 336  */
 337 void ofd_seqs_fini(const struct lu_env *env, struct ofd_device *ofd)
 338 {
 339         int rc;
 340
 341         ofd_deregister_seq_exp(ofd);
 342
 343         rc = ofd_fid_fini(env, ofd);
 344         if (rc != 0)
 345                 CERROR("%s: fid fini error: rc = %d\n", ofd_name(ofd), rc);
 346
 347         ofd_fld_fini(env, ofd);
 348
 349         ofd_seqs_free(env, ofd);
 350
 351         LASSERT(list_empty(&ofd->ofd_seq_list));
 352 }
 353
 354 /**
 355  * Return ofd_seq structure filled with valid data.
 356  *
 357  * This function gets the ofd_seq by sequence number and read
 358  * corresponding data from disk.
 359  *
 360  * \param[in] env       execution environment
 361  * \param[in] ofd       OFD device
 362  * \param[in] seq       sequence number
 363  *
 364  * \retval              ofd_seq structure filled with data
 365  * \retval              ERR_PTR pointer on error
 366  */
 367 struct ofd_seq *ofd_seq_load(const struct lu_env *env, struct ofd_device *ofd,
 368                              u64 seq)
 369 {
 370         struct ofd_thread_info  *info = ofd_info(env);
 371         struct ofd_seq          *oseq = NULL;
 372         struct dt_object        *dob;
 373         u64                      lastid;
 374         int                      rc;
 375
 376         ENTRY;
 377
 378         /* if seq is already initialized */
 379         oseq = ofd_seq_get(ofd, seq);
 380         if (oseq != NULL) {
 381                 CDEBUG(D_TRACE, "%s: got sequence %#llx "DOSTID"\n",
 382                        ofd_name(ofd), seq, POSTID(&oseq->os_oi));
 383                 RETURN(oseq);
 384         }
 385
 386         OBD_ALLOC_PTR(oseq);
 387         if (oseq == NULL)
 388                 RETURN(ERR_PTR(-ENOMEM));
 389
 390         lu_last_id_fid(&info->fti_fid, seq, ofd->ofd_lut.lut_lsd.lsd_osd_index);
 391         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
 392         info->fti_attr.la_valid = LA_MODE;
 393         info->fti_attr.la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
 394         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
 395
 396         /* create object tracking per-seq last created
 397          * id to be used by orphan recovery mechanism */
 398         dob = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
 399                                 &info->fti_dof, &info->fti_attr);
 400         if (IS_ERR(dob)) {
 401                 OBD_FREE_PTR(oseq);
 402                 RETURN((void *)dob);
 403         }
 404
 405         oseq->os_lastid_obj = dob;
 406
 407         INIT_LIST_HEAD(&oseq->os_list);
 408         mutex_init(&oseq->os_create_lock);
 409         spin_lock_init(&oseq->os_last_oid_lock);
 410         ostid_set_seq(&oseq->os_oi, seq);
 411         oseq->os_last_id_synced = 0;
 412
 413         atomic_set(&oseq->os_refc, 1);
 414         atomic_set(&oseq->os_precreate_in_progress, 0);
 415
 416         rc = dt_attr_get(env, dob, &info->fti_attr);
 417         if (rc)
 418                 GOTO(cleanup, rc);
 419
 420         if (info->fti_attr.la_size == 0) {
 421                 /* object is just created, initialize last id */
 422                 if (OBD_FAIL_CHECK(OBD_FAIL_OFD_SET_OID))
 423                         ofd_seq_last_oid_set(oseq, 0xffffff00);
 424                 else
 425                         ofd_seq_last_oid_set(oseq, OFD_INIT_OBJID);
 426                 ofd_seq_last_oid_write(env, ofd, oseq);
 427         } else if (info->fti_attr.la_size == sizeof(lastid)) {
 428                 info->fti_off = 0;
 429                 info->fti_buf.lb_buf = &lastid;
 430                 info->fti_buf.lb_len = sizeof(lastid);
 431
 432                 rc = dt_record_read(env, dob, &info->fti_buf, &info->fti_off);
 433                 if (rc) {
 434                         CERROR("%s: can't read last_id: rc = %d\n",
 435                                 ofd_name(ofd), rc);
 436                         GOTO(cleanup, rc);
 437                 }
 438                 ofd_seq_last_oid_set(oseq, le64_to_cpu(lastid));
 439         } else {
 440                 CERROR("%s: corrupted size %llu LAST_ID of seq %#llx\n",
 441                         ofd_name(ofd), (__u64)info->fti_attr.la_size, seq);
 442                 GOTO(cleanup, rc = -EINVAL);
 443         }
 444
 445         CDEBUG(D_HA, "%s: adding sequence %#llx\n", ofd_name(ofd), seq);
 446
 447         oseq = ofd_seq_add(env, ofd, oseq);
 448         RETURN((oseq != NULL) ? oseq : ERR_PTR(-ENOENT));
 449 cleanup:
 450         ofd_seq_put(env, oseq);
 451         return ERR_PTR(rc);
 452 }
 453
 454 /**
 455  * initialize local FLDB server.
 456  *
 457  * \param[in] env       execution environment
 458  * \param[in] uuid      unique name for this FLDS server
 459  * \param[in] ofd       OFD device
 460  *
 461  * \retval              0 on successful initialization
 462  * \retval              negative value on error
 463  */
 464 static int ofd_fld_init(const struct lu_env *env, const char *uuid,
 465                         struct ofd_device *ofd)
 466 {
 467         struct seq_server_site *ss = &ofd->ofd_seq_site;
 468         int rc;
 469
 470         ENTRY;
 471
 472         OBD_ALLOC_PTR(ss->ss_server_fld);
 473         if (ss->ss_server_fld == NULL)
 474                 RETURN(rc = -ENOMEM);
 475
 476         rc = fld_server_init(env, ss->ss_server_fld, ofd->ofd_osd, uuid,
 477                              LU_SEQ_RANGE_OST);
 478         if (rc < 0) {
 479                 OBD_FREE_PTR(ss->ss_server_fld);
 480                 ss->ss_server_fld = NULL;
 481                 RETURN(rc);
 482         }
 483         RETURN(0);
 484 }
 485
 486 /**
 487  * Update local FLDB copy from master server.
 488  *
 489  * This callback is called when LWP is connected to the server.
 490  * It retrieves its FLDB entries from MDT0, and it only happens
 491  * when upgrading the existing file system to 2.6.
 492  *
 493  * \param[in] data      OFD device
 494  *
 495  * \retval              0 on successful FLDB update
 496  * \retval              negative value in case if failure
 497  */
 498 static int ofd_register_lwp_callback(void *data)
 499 {
 500         struct lu_env           *env;
 501         struct ofd_device       *ofd = data;
 502         struct lu_server_fld    *fld = ofd->ofd_seq_site.ss_server_fld;
 503         int                     rc;
 504
 505         ENTRY;
 506
 507         if (!likely(fld->lsf_new))
 508                 RETURN(0);
 509
 510         OBD_ALLOC_PTR(env);
 511         if (env == NULL)
 512                 RETURN(-ENOMEM);
 513
 514         rc = lu_env_init(env, LCT_DT_THREAD);
 515         if (rc < 0)
 516                 GOTO(out, rc);
 517
 518         rc = fld_update_from_controller(env, fld);
 519         if (rc < 0) {
 520                 CERROR("%s: cannot update controller: rc = %d\n",
 521                        ofd_name(ofd), rc);
 522                 GOTO(out, rc);
 523         }
 524         EXIT;
 525 out:
 526         lu_env_fini(env);
 527         OBD_FREE_PTR(env);
 528         return rc;
 529 }
 530
 531 /**
 532  * Get LWP exports from LWP connection for local FLDB server and SEQ client.
 533  *
 534  * This function is part of setup process and initialize FLDB server and SEQ
 535  * client, so they may work with remote servers.
 536  *
 537  * \param[in] ofd       OFD device
 538  *
 539  * \retval              0 on successful export get
 540  * \retval              negative value on error
 541  */
 542 static int ofd_register_seq_exp(struct ofd_device *ofd)
 543 {
 544         struct seq_server_site  *ss = &ofd->ofd_seq_site;
 545         char                    *lwp_name = NULL;
 546         int                     rc;
 547
 548         OBD_ALLOC(lwp_name, MAX_OBD_NAME);
 549         if (lwp_name == NULL)
 550                 GOTO(out_free, rc = -ENOMEM);
 551
 552         rc = tgt_name2lwp_name(ofd_name(ofd), lwp_name, MAX_OBD_NAME, 0);
 553         if (rc != 0)
 554                 GOTO(out_free, rc);
 555
 556         rc = lustre_register_lwp_item(lwp_name, &ss->ss_client_seq->lcs_exp,
 557                                       NULL, NULL);
 558         if (rc != 0)
 559                 GOTO(out_free, rc);
 560
 561         rc = lustre_register_lwp_item(lwp_name,
 562                                       &ss->ss_server_fld->lsf_control_exp,
 563                                       ofd_register_lwp_callback, ofd);
 564         if (rc != 0) {
 565                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
 566                 ss->ss_client_seq->lcs_exp = NULL;
 567                 GOTO(out_free, rc);
 568         }
 569 out_free:
 570         if (lwp_name != NULL)
 571                 OBD_FREE(lwp_name, MAX_OBD_NAME);
 572
 573         return rc;
 574 }
 575
 576 /**
 577  * Initialize SEQ and FLD service on OFD.
 578  *
 579  * This is part of OFD setup process.
 580  *
 581  * \param[in] env       execution environment
 582  * \param[in] ofd       OFD device
 583  *
 584  * \retval              0 on successful services initialization
 585  * \retval              negative value on error
 586  */
 587 int ofd_seqs_init(const struct lu_env *env, struct ofd_device *ofd)
 588 {
 589         int rc;
 590
 591         rwlock_init(&ofd->ofd_seq_list_lock);
 592         INIT_LIST_HEAD(&ofd->ofd_seq_list);
 593         ofd->ofd_seq_count = 0;
 594
 595         rc = ofd_fid_init(env, ofd);
 596         if (rc != 0) {
 597                 CERROR("%s: fid init error: rc = %d\n", ofd_name(ofd), rc);
 598                 GOTO(out, rc);
 599         }
 600
 601         rc = ofd_fld_init(env, ofd_name(ofd), ofd);
 602         if (rc < 0) {
 603                 CERROR("%s: Can't init fld, rc %d\n", ofd_name(ofd), rc);
 604                 GOTO(out_fid, rc);
 605         }
 606
 607         rc = ofd_register_seq_exp(ofd);
 608         if (rc < 0) {
 609                 CERROR("%s: Can't init seq exp, rc %d\n", ofd_name(ofd), rc);
 610                 GOTO(out_fld, rc);
 611         }
 612
 613         RETURN(0);
 614
 615 out_fld:
 616         ofd_fld_fini(env, ofd);
 617 out_fid:
 618         ofd_fid_fini(env, ofd);
 619 out:
 620         return rc;
 621 }
 622
 623 /**
 624  * Initialize storage for the OFD.
 625  *
 626  * This function sets up service files for OFD. Currently, the only
 627  * service file is "health_check".
 628  *
 629  * \param[in] env       execution environment
 630  * \param[in] ofd       OFD device
 631  * \param[in] obd       OBD device (unused now)
 632  *
 633  * \retval              0 on successful setup
 634  * \retval              negative value on error
 635  */
 636 int ofd_fs_setup(const struct lu_env *env, struct ofd_device *ofd,
 637                  struct obd_device *obd)
 638 {
 639         struct ofd_thread_info  *info = ofd_info(env);
 640         struct dt_object        *fo;
 641         int                      rc = 0;
 642
 643         ENTRY;
 644
 645         rc = ofd_seqs_init(env, ofd);
 646         if (rc)
 647                 GOTO(out, rc);
 648
 649         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FS_SETUP))
 650                 GOTO(out_seqs, rc = -ENOENT);
 651
 652         lu_local_obj_fid(&info->fti_fid, OFD_HEALTH_CHECK_OID);
 653         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
 654         info->fti_attr.la_valid = LA_MODE;
 655         info->fti_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
 656         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
 657
 658         fo = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
 659                                &info->fti_dof, &info->fti_attr);
 660         if (IS_ERR(fo))
 661                 GOTO(out_seqs, rc = PTR_ERR(fo));
 662
 663         ofd->ofd_health_check_file = fo;
 664
 665         RETURN(0);
 666
 667 out_seqs:
 668         ofd_seqs_fini(env, ofd);
 669 out:
 670         return rc;
 671 }
 672
 673 /**
 674  * Cleanup service files on OFD.
 675  *
 676  * This function syncs whole OFD device and close "health check" file.
 677  *
 678  * \param[in] env       execution environment
 679  * \param[in] ofd       OFD device
 680  */
 681 void ofd_fs_cleanup(const struct lu_env *env, struct ofd_device *ofd)
 682 {
 683         int rc;
 684
 685         ENTRY;
 686
 687         ofd_seqs_fini(env, ofd);
 688
 689         rc = dt_sync(env, ofd->ofd_osd);
 690         if (rc < 0)
 691                 CWARN("%s: can't sync OFD upon cleanup: %d\n",
 692                       ofd_name(ofd), rc);
 693
 694         if (ofd->ofd_health_check_file) {
 695                 dt_object_put(env, ofd->ofd_health_check_file);
 696                 ofd->ofd_health_check_file = NULL;
 697         }
 698
 699         EXIT;
 700 }
 701