lustre/osp/osp_precreate.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/osp/osp_precreate.c
  32  *
  33  * Lustre OST Proxy Device
  34  *
  35  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  36  * Author: Mikhail Pershin <mike.pershin@intel.com>
  37  * Author: Di Wang <di.wang@intel.com>
  38  */
  39
  40 #define DEBUG_SUBSYSTEM S_MDS
  41
  42 #include <linux/kthread.h>
  43
  44 #include <lustre_obdo.h>
  45
  46 #include "osp_internal.h"
  47
  48 /*
  49  * there are two specific states to take care about:
  50  *
  51  * = import is disconnected =
  52  *
  53  * = import is inactive =
  54  *   in this case osp_declare_create() returns an error
  55  *
  56  */
  57
  58 /**
  59  * Check whether statfs data is expired
  60  *
  61  * OSP device caches statfs data for the target, the function checks
  62  * whether the data is expired or not.
  63  *
  64  * \param[in] d         OSP device
  65  *
  66  * \retval              0 - not expired, 1 - expired
  67  */
  68 static inline int osp_statfs_need_update(struct osp_device *d)
  69 {
  70         return !ktime_before(ktime_get(), d->opd_statfs_fresh_till);
  71 }
  72
  73 /*
  74  * OSP tries to maintain pool of available objects so that calls to create
  75  * objects don't block most of time
  76  *
  77  * each time OSP gets connected to OST, we should start from precreation cleanup
  78  */
  79 static void osp_statfs_timer_cb(cfs_timer_cb_arg_t data)
  80 {
  81         struct osp_device *d = cfs_from_timer(d, data, opd_statfs_timer);
  82
  83         LASSERT(d);
  84         /* invalidate statfs data so osp_precreate_thread() can refresh */
  85         d->opd_statfs_fresh_till = ktime_sub_ns(ktime_get(), NSEC_PER_SEC);
  86         if (d->opd_pre_task)
  87                 wake_up(&d->opd_pre_waitq);
  88 }
  89
  90 static void osp_pre_update_msfs(struct osp_device *d, struct obd_statfs *msfs);
  91
  92 /*
  93  * The function updates current precreation status if broken, and
  94  * updates that cached statfs state if functional, then wakes up waiters.
  95  * We don't clear opd_pre_status directly here, but rather leave this
  96  * to osp_pre_update_msfs() to do if everything is OK so that we don't
  97  * have a race to clear opd_pre_status and then set it to -ENOSPC again.
  98  *
  99  * \param[in] d         OSP device
 100  * \param[in] msfs      statfs data
 101  * \param[in] rc        new precreate status for device \a d
 102  */
 103 static void osp_pre_update_status_msfs(struct osp_device *d,
 104                                        struct obd_statfs *msfs, int rc)
 105 {
 106         CDEBUG(D_INFO, "%s: Updating status = %d\n", d->opd_obd->obd_name, rc);
 107         if (rc)
 108                 d->opd_pre_status = rc;
 109         else
 110                 osp_pre_update_msfs(d, msfs);
 111
 112         wake_up_all(&d->opd_pre_user_waitq);
 113 }
 114
 115 /* Pass in the old statfs data in case the limits have changed */
 116 void osp_pre_update_status(struct osp_device *d, int rc)
 117 {
 118         osp_pre_update_status_msfs(d, &d->opd_statfs, rc);
 119 }
 120
 121
 122 /**
 123  * RPC interpret callback for OST_STATFS RPC
 124  *
 125  * An interpretation callback called by ptlrpc for OST_STATFS RPC when it is
 126  * replied by the target. It's used to maintain statfs cache for the target.
 127  * The function fills data from the reply if successful and schedules another
 128  * update.
 129  *
 130  * \param[in] env       LU environment provided by the caller
 131  * \param[in] req       RPC replied
 132  * \param[in] aa        callback data
 133  * \param[in] rc        RPC result
 134  *
 135  * \retval 0            on success
 136  * \retval negative     negated errno on error
 137  */
 138 static int osp_statfs_interpret(const struct lu_env *env,
 139                                 struct ptlrpc_request *req, void *args, int rc)
 140 {
 141         union ptlrpc_async_args *aa = args;
 142         struct obd_import *imp = req->rq_import;
 143         struct obd_statfs *msfs;
 144         struct obd_statfs *sfs;
 145         struct osp_device *d;
 146         u64 maxage_ns;
 147
 148         ENTRY;
 149
 150         aa = ptlrpc_req_async_args(aa, req);
 151         d = aa->pointer_arg[0];
 152         LASSERT(d);
 153
 154         if (rc != 0)
 155                 GOTO(out, rc);
 156
 157         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
 158         if (msfs == NULL)
 159                 GOTO(out, rc = -EPROTO);
 160
 161         if (d->opd_pre)
 162                 osp_pre_update_status_msfs(d, msfs, 0);
 163         else
 164                 osp_pre_update_msfs(d, msfs);
 165
 166         /* schedule next update */
 167         maxage_ns = d->opd_statfs_maxage * NSEC_PER_SEC;
 168         d->opd_statfs_fresh_till = ktime_add_ns(ktime_get(), maxage_ns);
 169         mod_timer(&d->opd_statfs_timer,
 170                   jiffies + cfs_time_seconds(d->opd_statfs_maxage));
 171         d->opd_statfs_update_in_progress = 0;
 172
 173         sfs = &d->opd_statfs;
 174         CDEBUG(D_CACHE,
 175                "%s (%p): %llu blocks, %llu free, %llu avail, %u bsize, %u reserved mb low, %u reserved mb high, %u reserved ino low, %u reserved ino high, %llu files, %llu free files %#x\n",
 176                d->opd_obd->obd_name, d, sfs->os_blocks, sfs->os_bfree,
 177                sfs->os_bavail, sfs->os_bsize, d->opd_reserved_mb_low,
 178                d->opd_reserved_mb_high, d->opd_reserved_ino_low,
 179                d->opd_reserved_ino_high, sfs->os_files, sfs->os_ffree,
 180                sfs->os_state);
 181
 182         RETURN(0);
 183 out:
 184         /* couldn't update statfs, try again with a small delay */
 185         d->opd_statfs_fresh_till = ktime_add_ns(ktime_get(), 10 * NSEC_PER_SEC);
 186         d->opd_statfs_update_in_progress = 0;
 187         if (d->opd_pre && d->opd_pre_task)
 188                 wake_up(&d->opd_pre_waitq);
 189
 190         if (req->rq_import_generation == imp->imp_generation)
 191                 CDEBUG(D_CACHE, "%s: couldn't update statfs: rc = %d\n",
 192                        d->opd_obd->obd_name, rc);
 193         RETURN(rc);
 194 }
 195
 196 /**
 197  * Send OST_STATFS RPC
 198  *
 199  * Sends OST_STATFS RPC to refresh cached statfs data for the target.
 200  * Also disables scheduled updates as times OSP may need to refresh
 201  * statfs data before expiration. The function doesn't block, instead
 202  * an interpretation callback osp_statfs_interpret() is used.
 203  *
 204  * \param[in] d         OSP device
 205  */
 206 static int osp_statfs_update(const struct lu_env *env, struct osp_device *d)
 207 {
 208         u64 expire = obd_timeout * 1000 * NSEC_PER_SEC;
 209         struct ptlrpc_request   *req;
 210         struct obd_import       *imp;
 211         union ptlrpc_async_args *aa;
 212         int rc;
 213
 214         ENTRY;
 215
 216         CDEBUG(D_CACHE, "going to update statfs\n");
 217
 218         imp = d->opd_obd->u.cli.cl_import;
 219         LASSERT(imp);
 220
 221         req = ptlrpc_request_alloc(imp,
 222                            d->opd_pre ? &RQF_OST_STATFS : &RQF_MDS_STATFS);
 223         if (req == NULL)
 224                 RETURN(-ENOMEM);
 225
 226         rc = ptlrpc_request_pack(req,
 227                          d->opd_pre ? LUSTRE_OST_VERSION : LUSTRE_MDS_VERSION,
 228                          d->opd_pre ? OST_STATFS : MDS_STATFS);
 229         if (rc) {
 230                 ptlrpc_request_free(req);
 231                 RETURN(rc);
 232         }
 233         ptlrpc_request_set_replen(req);
 234         if (d->opd_pre)
 235                 req->rq_request_portal = OST_CREATE_PORTAL;
 236         ptlrpc_at_set_req_timeout(req);
 237
 238         req->rq_interpret_reply = osp_statfs_interpret;
 239         aa = ptlrpc_req_async_args(aa, req);
 240         aa->pointer_arg[0] = d;
 241
 242         /*
 243          * no updates till reply
 244          */
 245         timer_delete(&d->opd_statfs_timer);
 246         d->opd_statfs_fresh_till = ktime_add_ns(ktime_get(), expire);
 247         d->opd_statfs_update_in_progress = 1;
 248
 249         ptlrpcd_add_req(req);
 250
 251         /* we still want to sync changes if no new changes are coming */
 252         if (ktime_before(ktime_get(), d->opd_sync_next_commit_cb))
 253                 GOTO(out, rc);
 254
 255         if (atomic_read(&d->opd_sync_changes)) {
 256                 struct thandle *th;
 257
 258                 th = dt_trans_create(env, d->opd_storage);
 259                 if (IS_ERR(th)) {
 260                         CERROR("%s: can't sync\n", d->opd_obd->obd_name);
 261                         GOTO(out, rc);
 262                 }
 263                 rc = dt_trans_start_local(env, d->opd_storage, th);
 264                 if (rc == 0) {
 265                         CDEBUG(D_OTHER, "%s: sync forced, %d changes\n",
 266                                d->opd_obd->obd_name,
 267                                atomic_read(&d->opd_sync_changes));
 268                         osp_sync_add_commit_cb_1s(env, d, th);
 269                 }
 270                 dt_trans_stop(env, d->opd_storage, th);
 271         }
 272
 273 out:
 274         RETURN(0);
 275 }
 276
 277 /**
 278  * Schedule an immediate update for statfs data
 279  *
 280  * If cached statfs data claim no free space, but OSP has got a request to
 281  * destroy an object (so release some space probably), then we may need to
 282  * refresh cached statfs data sooner than planned. The function checks there
 283  * is no statfs update going and schedules immediate update if so.
 284  * XXX: there might be a case where removed object(s) do not add free space (empty
 285  * object). If the number of such deletions is high, then we can start to update
 286  * statfs too often causing a RPC storm. some throttling is needed...
 287  *
 288  * \param[in] d         OSP device where statfs data needs to be refreshed
 289  */
 290 void osp_statfs_need_now(struct osp_device *d)
 291 {
 292         if (!d->opd_statfs_update_in_progress) {
 293                 /*
 294                  * if current status is -ENOSPC (lack of free space on OST)
 295                  * then we should poll OST immediately once object destroy
 296                  * is replied
 297                  */
 298                 d->opd_statfs_fresh_till = ktime_sub_ns(ktime_get(), NSEC_PER_SEC);
 299                 timer_delete(&d->opd_statfs_timer);
 300                 wake_up(&d->opd_pre_waitq);
 301         }
 302 }
 303
 304 /**
 305  * Return number of precreated objects
 306  *
 307  * A simple helper to calculate the number of precreated objects on the device.
 308  *
 309  * \param[in] env       LU environment provided by the caller
 310  * \param[in] osp       OSP device
 311  *
 312  * \retval              the number of the precreated objects
 313  */
 314 static inline int osp_objs_precreated(const struct lu_env *env,
 315                                       struct osp_device *osp)
 316 {
 317         return osp_fid_diff(&osp->opd_pre_last_created_fid,
 318                             &osp->opd_pre_used_fid);
 319 }
 320
 321 /**
 322  * Check pool of precreated objects is nearly empty
 323  *
 324  * We should not wait till the pool of the precreated objects is exhausted,
 325  * because then there will be a long period of OSP being unavailable for the
 326  * new creations due to lenghty precreate RPC. Instead we ask for another
 327  * precreation ahead and hopefully have it ready before the current pool is
 328  * empty. Notice this function relies on an external locking.
 329  *
 330  * \param[in] env       LU environment provided by the caller
 331  * \param[in] d         OSP device
 332  *
 333  * \retval              0 - current pool is good enough, 1 - time to precreate
 334  */
 335 static inline int osp_precreate_near_empty_nolock(const struct lu_env *env,
 336                                                   struct osp_device *d)
 337 {
 338         int window = osp_objs_precreated(env, d);
 339
 340         /* don't consider new precreation till OST is healty and
 341          * has free space */
 342         return ((window - d->opd_pre_reserved < d->opd_pre_create_count / 2 ||
 343                  d->opd_force_creation) && (d->opd_pre_status == 0));
 344 }
 345
 346 /**
 347  * Check pool of precreated objects
 348  *
 349  * This is protected version of osp_precreate_near_empty_nolock(), check that
 350  * for the details.
 351  *
 352  * \param[in] env       LU environment provided by the caller
 353  * \param[in] d         OSP device
 354  *
 355  * \retval              0 - current pool is good enough, 1 - time to precreate
 356  */
 357 static inline int osp_precreate_near_empty(const struct lu_env *env,
 358                                            struct osp_device *d)
 359 {
 360         int rc;
 361
 362         if (d->opd_pre == NULL)
 363                 return 0;
 364
 365         /* XXX: do we really need locking here? */
 366         spin_lock(&d->opd_pre_lock);
 367         rc = osp_precreate_near_empty_nolock(env, d);
 368         spin_unlock(&d->opd_pre_lock);
 369         return rc;
 370 }
 371
 372 /**
 373  * Write FID into into last_oid/last_seq file
 374  *
 375  * The function stores the sequence and the in-sequence id into two dedicated
 376  * files. The sync argument can be used to request synchronous commit, so the
 377  * function won't return until the updates are committed.
 378  *
 379  * \param[in] env       LU environment provided by the caller
 380  * \param[in] osp       OSP device
 381  * \param[in] fid       fid where sequence/id is taken
 382  * \param[in] sync      update mode: 0 - asynchronously, 1 - synchronously
 383  *
 384  * \retval 0            on success
 385  * \retval negative     negated errno on error
 386  **/
 387 int osp_write_last_oid_seq_files(struct lu_env *env, struct osp_device *osp,
 388                                  struct lu_fid *fid, int sync)
 389 {
 390         struct osp_thread_info  *oti = osp_env_info(env);
 391         struct lu_buf      *lb_oid = &oti->osi_lb;
 392         struct lu_buf      *lb_oseq = &oti->osi_lb2;
 393         loff_t             oid_off;
 394         u64                oid;
 395         loff_t             oseq_off;
 396         struct thandle    *th;
 397         int                   rc;
 398         ENTRY;
 399
 400         if (osp->opd_storage->dd_rdonly)
 401                 RETURN(0);
 402
 403         /* Note: through f_oid is only 32 bits, it will also write 64 bits
 404          * for oid to keep compatibility with the previous version. */
 405         oid = fid->f_oid;
 406         osp_objid_buf_prep(lb_oid, &oid_off,
 407                            &oid, osp->opd_index);
 408
 409         osp_objseq_buf_prep(lb_oseq, &oseq_off,
 410                             &fid->f_seq, osp->opd_index);
 411
 412         th = dt_trans_create(env, osp->opd_storage);
 413         if (IS_ERR(th))
 414                 RETURN(PTR_ERR(th));
 415
 416         th->th_sync |= sync;
 417         rc = dt_declare_record_write(env, osp->opd_last_used_oid_file,
 418                                      lb_oid, oid_off, th);
 419         if (rc != 0)
 420                 GOTO(out, rc);
 421
 422         rc = dt_declare_record_write(env, osp->opd_last_used_seq_file,
 423                                      lb_oseq, oseq_off, th);
 424         if (rc != 0)
 425                 GOTO(out, rc);
 426
 427         rc = dt_trans_start_local(env, osp->opd_storage, th);
 428         if (rc != 0)
 429                 GOTO(out, rc);
 430
 431         rc = dt_record_write(env, osp->opd_last_used_oid_file, lb_oid,
 432                              &oid_off, th);
 433         if (rc != 0) {
 434                 CERROR("%s: can not write to last seq file: rc = %d\n",
 435                         osp->opd_obd->obd_name, rc);
 436                 GOTO(out, rc);
 437         }
 438         rc = dt_record_write(env, osp->opd_last_used_seq_file, lb_oseq,
 439                              &oseq_off, th);
 440         if (rc) {
 441                 CERROR("%s: can not write to last seq file: rc = %d\n",
 442                         osp->opd_obd->obd_name, rc);
 443                 GOTO(out, rc);
 444         }
 445 out:
 446         dt_trans_stop(env, osp->opd_storage, th);
 447         RETURN(rc);
 448 }
 449
 450 /**
 451  * Switch to another sequence
 452  *
 453  * When a current sequence has no available IDs left, OSP has to switch to
 454  * another new sequence. OSP requests it using the regular FLDB protocol
 455  * and stores synchronously before that is used in precreate. This is needed
 456  * to basically have the sequences referenced (not orphaned), otherwise it's
 457  * possible that OST has some objects precreated and the clients have data
 458  * written to it, but after MDT failover nobody refers those objects and OSP
 459  * has no idea that the sequence need cleanup to be done.
 460  * While this is very expensive operation, it's supposed to happen infrequently
 461  * because sequence has LUSTRE_DATA_SEQ_MAX_WIDTH=32M objects by default.
 462  *
 463  * \param[in] env       LU environment provided by the caller
 464  * \param[in] osp       OSP device
 465  *
 466  * \retval 0            on success
 467  * \retval negative     negated errno on error
 468  */
 469 static int osp_precreate_rollover_new_seq(struct lu_env *env,
 470                                           struct osp_device *osp)
 471 {
 472         struct lu_fid   *fid = &osp_env_info(env)->osi_fid;
 473         struct lu_fid   *last_fid = &osp->opd_last_used_fid;
 474         int             rc;
 475         ENTRY;
 476
 477         rc = seq_client_get_seq(env, osp->opd_obd->u.cli.cl_seq, &fid->f_seq);
 478         if (rc != 0) {
 479                 CERROR("%s: alloc fid error: rc = %d\n",
 480                        osp->opd_obd->obd_name, rc);
 481                 RETURN(rc);
 482         }
 483
 484         fid->f_oid = 1;
 485         fid->f_ver = 0;
 486         LASSERTF(fid_seq(fid) != fid_seq(last_fid),
 487                  "fid "DFID", last_fid "DFID"\n", PFID(fid),
 488                  PFID(last_fid));
 489
 490         rc = osp_write_last_oid_seq_files(env, osp, fid, 1);
 491         if (rc != 0) {
 492                 CERROR("%s: Can not update oid/seq file: rc = %d\n",
 493                        osp->opd_obd->obd_name, rc);
 494                 RETURN(rc);
 495         }
 496
 497         LCONSOLE(D_INFO, "%s: update sequence from %#llx to %#llx\n",
 498                  osp->opd_obd->obd_name, fid_seq(last_fid),
 499                  fid_seq(fid));
 500         /* Update last_xxx to the new seq */
 501         spin_lock(&osp->opd_pre_lock);
 502         osp->opd_last_used_fid = *fid;
 503         osp_fid_to_obdid(fid, &osp->opd_last_id);
 504         osp->opd_gap_start_fid = *fid;
 505         osp->opd_pre_used_fid = *fid;
 506         osp->opd_pre_last_created_fid = *fid;
 507         spin_unlock(&osp->opd_pre_lock);
 508
 509         RETURN(rc);
 510 }
 511
 512 /**
 513  * Find IDs available in current sequence
 514  *
 515  * The function calculates the highest possible ID and the number of IDs
 516  * available in the current sequence OSP is using. The number is limited
 517  * artifically by the caller (grow param) and the number of IDs available
 518  * in the sequence by nature. The function doesn't require an external
 519  * locking.
 520  *
 521  * \param[in] env       LU environment provided by the caller
 522  * \param[in] osp       OSP device
 523  * \param[in] fid       FID the caller wants to start with
 524  * \param[in] grow      how many the caller wants
 525  * \param[out] fid      the highest calculated FID
 526  * \param[out] grow     the number of available IDs calculated
 527  *
 528  * \retval              0 on success, 1 - the sequence is empty
 529  */
 530 static int osp_precreate_fids(const struct lu_env *env, struct osp_device *osp,
 531                               struct lu_fid *fid, int *grow)
 532 {
 533         struct osp_thread_info *osi = osp_env_info(env);
 534         __u64 seq_width = osp->opd_pre_seq_width;
 535         __u64 end;
 536         int i = 0;
 537
 538         if (fid_is_idif(fid)) {
 539                 struct lu_fid   *last_fid;
 540                 struct ost_id   *oi = &osi->osi_oi;
 541                 int rc;
 542
 543                 spin_lock(&osp->opd_pre_lock);
 544                 last_fid = &osp->opd_pre_last_created_fid;
 545                 fid_to_ostid(last_fid, oi);
 546                 end = min(ostid_id(oi) + *grow, min(IDIF_MAX_OID, seq_width));
 547                 *grow = end - ostid_id(oi);
 548                 rc = ostid_set_id(oi, ostid_id(oi) + *grow);
 549                 spin_unlock(&osp->opd_pre_lock);
 550
 551                 if (*grow == 0 || rc)
 552                         return 1;
 553
 554                 ostid_to_fid(fid, oi, osp->opd_index);
 555                 return 0;
 556         }
 557
 558         spin_lock(&osp->opd_pre_lock);
 559         *fid = osp->opd_pre_last_created_fid;
 560         end = fid->f_oid;
 561         end = min((end + *grow), min(OBIF_MAX_OID, seq_width));
 562         *grow = end - fid->f_oid;
 563         fid->f_oid += end - fid->f_oid;
 564         spin_unlock(&osp->opd_pre_lock);
 565
 566         CDEBUG(D_INFO, "Expect %d, actual %d ["DFID" -- "DFID"]\n",
 567                *grow, i, PFID(fid), PFID(&osp->opd_pre_last_created_fid));
 568
 569         return *grow > 0 ? 0 : 1;
 570 }
 571
 572 /**
 573  * Prepare and send precreate RPC
 574  *
 575  * The function finds how many objects should be precreated.  Then allocates,
 576  * prepares and schedules precreate RPC synchronously. Upon reply the function
 577  * wakes up the threads waiting for the new objects on this target. If the
 578  * target wasn't able to create all the objects requested, then the next
 579  * precreate will be asking for fewer objects (i.e. slow precreate down).
 580  *
 581  * \param[in] env       LU environment provided by the caller
 582  * \param[in] d         OSP device
 583  *
 584  * \retval 0            on success
 585  * \retval negative     negated errno on error
 586  **/
 587 static int osp_precreate_send(const struct lu_env *env, struct osp_device *d)
 588 {
 589         struct osp_thread_info  *oti = osp_env_info(env);
 590         struct ptlrpc_request   *req;
 591         struct obd_import       *imp;
 592         struct ost_body         *body;
 593         int                      rc, grow, diff;
 594         struct lu_fid           *fid = &oti->osi_fid;
 595         ENTRY;
 596
 597         /* don't precreate new objects till OST healthy and has free space */
 598         if (unlikely(d->opd_pre_status)) {
 599                 CDEBUG(D_INFO, "%s: don't send new precreate: rc = %d\n",
 600                        d->opd_obd->obd_name, d->opd_pre_status);
 601                 RETURN(0);
 602         }
 603
 604         /*
 605          * if not connection/initialization is compeleted, ignore
 606          */
 607         imp = d->opd_obd->u.cli.cl_import;
 608         LASSERT(imp);
 609
 610         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 611         if (req == NULL)
 612                 RETURN(-ENOMEM);
 613         req->rq_request_portal = OST_CREATE_PORTAL;
 614
 615         /* Delorphan happens only with a first MDT-OST connect. resend/replay
 616          * handles objects creation on reconnects, no need to do delorhpan
 617          * in this case.
 618          */
 619
 620         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 621         if (rc) {
 622                 ptlrpc_request_free(req);
 623                 RETURN(rc);
 624         }
 625
 626         spin_lock(&d->opd_pre_lock);
 627         if (d->opd_force_creation)
 628                 d->opd_pre_create_count = OST_MIN_PRECREATE;
 629         else if (d->opd_pre_create_count > d->opd_pre_max_create_count / 2)
 630                 d->opd_pre_create_count = d->opd_pre_max_create_count / 2;
 631         grow = d->opd_pre_create_count;
 632         spin_unlock(&d->opd_pre_lock);
 633
 634         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 635         LASSERT(body);
 636
 637         *fid = d->opd_pre_last_created_fid;
 638         rc = osp_precreate_fids(env, d, fid, &grow);
 639         if (rc == 1)
 640                 /* Current seq has been used up*/
 641                 GOTO(out_req, rc = -ENOSPC);
 642
 643         if (!osp_is_fid_client(d)) {
 644                 /* Non-FID client will always send seq 0 because of
 645                  * compatiblity */
 646                 LASSERTF(fid_is_idif(fid), "Invalid fid "DFID"\n", PFID(fid));
 647                 fid->f_seq = 0;
 648         }
 649
 650         fid_to_ostid(fid, &body->oa.o_oi);
 651         body->oa.o_valid = OBD_MD_FLGROUP;
 652
 653         ptlrpc_request_set_replen(req);
 654
 655         if (OBD_FAIL_CHECK(OBD_FAIL_OSP_FAKE_PRECREATE))
 656                 GOTO(ready, rc = 0);
 657
 658         rc = ptlrpc_queue_wait(req);
 659         if (rc) {
 660                 CERROR("%s: can't precreate: rc = %d\n", d->opd_obd->obd_name,
 661                        rc);
 662                 if (req->rq_net_err)
 663                         /* have osp_precreate_reserve() to wait for repeat */
 664                         rc = -ENOTCONN;
 665                 GOTO(out_req, rc);
 666         }
 667
 668         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 669         if (body == NULL)
 670                 GOTO(out_req, rc = -EPROTO);
 671
 672         ostid_to_fid(fid, &body->oa.o_oi, d->opd_index);
 673
 674 ready:
 675         if (osp_fid_diff(fid, &d->opd_pre_used_fid) <= 0) {
 676                 CERROR("%s: precreate fid "DFID" <= local used fid "DFID
 677                        ": rc = %d\n", d->opd_obd->obd_name,
 678                        PFID(fid), PFID(&d->opd_pre_used_fid), -ESTALE);
 679                 GOTO(out_req, rc = -ESTALE);
 680         }
 681
 682         diff = osp_fid_diff(fid, &d->opd_pre_last_created_fid);
 683
 684         spin_lock(&d->opd_pre_lock);
 685         if (diff < grow) {
 686                 /* the OST has not managed to create all the
 687                  * objects we asked for */
 688                 d->opd_pre_create_count = max(diff, OST_MIN_PRECREATE);
 689                 d->opd_pre_create_slow = 1;
 690         } else {
 691                 /* the OST is able to keep up with the work,
 692                  * we could consider increasing create_count
 693                  * next time if needed */
 694                 d->opd_pre_create_slow = 0;
 695         }
 696
 697         if ((body->oa.o_valid & OBD_MD_FLSIZE) && body->oa.o_size)
 698                 d->opd_pre_seq_width = body->oa.o_size;
 699
 700         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 701         fid_to_ostid(fid, &body->oa.o_oi);
 702
 703         d->opd_pre_last_created_fid = *fid;
 704         d->opd_force_creation = false;
 705         spin_unlock(&d->opd_pre_lock);
 706
 707         CDEBUG(D_HA, "%s: current precreated pool: "DFID"-"DFID"\n",
 708                d->opd_obd->obd_name, PFID(&d->opd_pre_used_fid),
 709                PFID(&d->opd_pre_last_created_fid));
 710 out_req:
 711         /* now we can wakeup all users awaiting for objects */
 712         osp_pre_update_status(d, rc);
 713
 714         /* pause to let osp_precreate_reserve to go first */
 715         CFS_FAIL_TIMEOUT(OBD_FAIL_OSP_PRECREATE_PAUSE, 2);
 716
 717         ptlrpc_req_finished(req);
 718         RETURN(rc);
 719 }
 720
 721 /**
 722  * Get last precreated object from target (OST)
 723  *
 724  * Sends synchronous RPC to the target (OST) to learn the last precreated
 725  * object. This later is used to remove all unused objects (cleanup orphan
 726  * procedure). Also, the next object after one we got will be used as a
 727  * starting point for the new precreates.
 728  *
 729  * \param[in] env       LU environment provided by the caller
 730  * \param[in] d         OSP device
 731  * \param[in] update    update or not update last used fid
 732  *
 733  * \retval 0            on success
 734  * \retval negative     negated errno on error
 735  **/
 736 static int osp_get_lastfid_from_ost(const struct lu_env *env,
 737                                     struct osp_device *d, bool update)
 738 {
 739         struct ptlrpc_request   *req = NULL;
 740         struct obd_import       *imp;
 741         struct lu_fid           *last_fid;
 742         char                    *tmp;
 743         int                     rc;
 744         ENTRY;
 745
 746         imp = d->opd_obd->u.cli.cl_import;
 747         LASSERT(imp);
 748
 749         req = ptlrpc_request_alloc(imp, &RQF_OST_GET_INFO_LAST_FID);
 750         if (req == NULL)
 751                 RETURN(-ENOMEM);
 752
 753         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_KEY, RCL_CLIENT,
 754                              sizeof(KEY_LAST_FID));
 755
 756         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
 757         if (rc) {
 758                 ptlrpc_request_free(req);
 759                 RETURN(rc);
 760         }
 761
 762         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_KEY);
 763         memcpy(tmp, KEY_LAST_FID, sizeof(KEY_LAST_FID));
 764
 765         req->rq_no_delay = req->rq_no_resend = 1;
 766         last_fid = req_capsule_client_get(&req->rq_pill, &RMF_FID);
 767         fid_cpu_to_le(last_fid, &d->opd_last_used_fid);
 768
 769         ptlrpc_request_set_replen(req);
 770
 771         rc = ptlrpc_queue_wait(req);
 772         if (rc) {
 773                 /* -EFAULT means reading LAST_FID failed (see ofd_get_info_hld),
 774                  * let sysadm sort this * out.
 775                  */
 776                 if (rc == -EFAULT)
 777                         ptlrpc_set_import_active(imp, 0);
 778                 GOTO(out, rc);
 779         }
 780
 781         last_fid = req_capsule_server_get(&req->rq_pill, &RMF_FID);
 782         if (last_fid == NULL) {
 783                 CERROR("%s: Got last_fid failed.\n", d->opd_obd->obd_name);
 784                 GOTO(out, rc = -EPROTO);
 785         }
 786
 787         if (!fid_is_sane(last_fid)) {
 788                 CERROR("%s: Got insane last_fid "DFID"\n",
 789                        d->opd_obd->obd_name, PFID(last_fid));
 790                 GOTO(out, rc = -EPROTO);
 791         }
 792
 793         /* Only update the last used fid, if the OST has objects for
 794          * this sequence, i.e. fid_oid > 0 */
 795         if (fid_oid(last_fid) > 0 && update)
 796                 d->opd_last_used_fid = *last_fid;
 797
 798         if (fid_seq(last_fid) == fid_seq(&d->opd_last_used_fid)) {
 799                 if (fid_oid(last_fid) == 0 ||
 800                     (fid_seq_is_norm(fid_seq(last_fid)) &&
 801                      fid_oid(last_fid) == LUSTRE_FID_INIT_OID)) {
 802                         /* reformatted OST, it requires creation request
 803                          * to recreate objects
 804                          */
 805                         spin_lock(&d->opd_pre_lock);
 806                         d->opd_force_creation = true;
 807                         spin_unlock(&d->opd_pre_lock);
 808                 }
 809         }
 810         CDEBUG(D_HA, "%s: Got last_fid "DFID"\n", d->opd_obd->obd_name,
 811                PFID(last_fid));
 812
 813 out:
 814         ptlrpc_req_finished(req);
 815         RETURN(rc);
 816 }
 817
 818 /**
 819  * Cleanup orphans on OST
 820  *
 821  * This function is called in a contex of a dedicated thread handling
 822  * all the precreation suff. The function waits till local recovery
 823  * is complete, then identify all the unreferenced objects (orphans)
 824  * using the highest ID referenced by a local and the highest object
 825  * precreated by the target. The found range is a subject to removal
 826  * using specially flagged RPC. During this process OSP is marked
 827  * unavailable for new objects.
 828  *
 829  * \param[in] env       LU environment provided by the caller
 830  * \param[in] d         OSP device
 831  *
 832  * \retval 0            on success
 833  * \retval negative     negated errno on error
 834  */
 835 static int osp_precreate_cleanup_orphans(struct lu_env *env,
 836                                          struct osp_device *d)
 837 {
 838         struct osp_thread_info  *osi = osp_env_info(env);
 839         struct lu_fid           *last_fid = &osi->osi_fid;
 840         struct ptlrpc_request   *req = NULL;
 841         struct obd_import       *imp;
 842         struct ost_body         *body;
 843         int                      update_status = 0;
 844         int                      rc;
 845         int                      diff;
 846
 847         ENTRY;
 848
 849         /*
 850          * Do cleanup orphans only with a first connection, after that
 851          * all precreate requests uses resend/replay flags to support OST
 852          * failover/reconnect.
 853          */
 854         if (d->opd_cleanup_orphans_done) {
 855                 rc = osp_get_lastfid_from_ost(env, d, false);
 856                 RETURN(0);
 857         }
 858         /*
 859          * wait for local recovery to finish, so we can cleanup orphans
 860          * orphans are all objects since "last used" (assigned), but
 861          * there might be objects reserved and in some cases they won't
 862          * be used. we can't cleanup them till we're sure they won't be
 863          * used. also can't we allow new reservations because they may
 864          * end up getting orphans being cleaned up below. so we block
 865          * new reservations and wait till all reserved objects either
 866          * user or released.
 867          */
 868         spin_lock(&d->opd_pre_lock);
 869         d->opd_pre_recovering = 1;
 870         spin_unlock(&d->opd_pre_lock);
 871         /*
 872          * The locking above makes sure the opd_pre_reserved check below will
 873          * catch all osp_precreate_reserve() calls who find
 874          * "!opd_pre_recovering".
 875          */
 876         wait_event_idle(d->opd_pre_waitq,
 877                         (!d->opd_pre_reserved && d->opd_recovery_completed) ||
 878                         !d->opd_pre_task || d->opd_got_disconnected);
 879         if (!d->opd_pre_task || d->opd_got_disconnected)
 880                 GOTO(out, rc = -EAGAIN);
 881
 882         CDEBUG(D_HA, "%s: going to cleanup orphans since "DFID"\n",
 883                d->opd_obd->obd_name, PFID(&d->opd_last_used_fid));
 884
 885         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_DELAY_DELORPHAN, cfs_fail_val);
 886
 887         *last_fid = d->opd_last_used_fid;
 888         /* The OSP should already get the valid seq now */
 889         LASSERT(!fid_is_zero(last_fid));
 890         if (fid_oid(&d->opd_last_used_fid) < 2 ||
 891             OBD_FAIL_CHECK(OBD_FAIL_OSP_GET_LAST_FID)) {
 892                 /* lastfid looks strange... ask OST */
 893                 rc = osp_get_lastfid_from_ost(env, d, true);
 894                 if (rc)
 895                         GOTO(out, rc);
 896         }
 897
 898         imp = d->opd_obd->u.cli.cl_import;
 899         LASSERT(imp);
 900
 901         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 902         if (req == NULL)
 903                 GOTO(out, rc = -ENOMEM);
 904
 905         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 906         if (rc) {
 907                 ptlrpc_request_free(req);
 908                 req = NULL;
 909                 GOTO(out, rc);
 910         }
 911
 912         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 913         if (body == NULL)
 914                 GOTO(out, rc = -EPROTO);
 915
 916         body->oa.o_flags = OBD_FL_DELORPHAN;
 917         body->oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
 918
 919         fid_to_ostid(&d->opd_last_used_fid, &body->oa.o_oi);
 920
 921         ptlrpc_request_set_replen(req);
 922
 923         /* Don't resend the delorphan req */
 924         req->rq_no_resend = req->rq_no_delay = 1;
 925
 926         rc = ptlrpc_queue_wait(req);
 927         if (rc) {
 928                 update_status = 1;
 929                 GOTO(out, rc);
 930         }
 931
 932         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 933         if (body == NULL)
 934                 GOTO(out, rc = -EPROTO);
 935
 936         /*
 937          * OST provides us with id new pool starts from in body->oa.o_id
 938          */
 939         ostid_to_fid(last_fid, &body->oa.o_oi, d->opd_index);
 940
 941         spin_lock(&d->opd_pre_lock);
 942         diff = osp_fid_diff(&d->opd_last_used_fid, last_fid);
 943         if (diff > 0) {
 944                 d->opd_pre_create_count = OST_MIN_PRECREATE + diff;
 945                 d->opd_pre_last_created_fid = d->opd_last_used_fid;
 946         } else {
 947                 d->opd_pre_create_count = OST_MIN_PRECREATE;
 948                 d->opd_pre_last_created_fid = *last_fid;
 949         }
 950         /*
 951          * This empties the pre-creation pool and effectively blocks any new
 952          * reservations.
 953          */
 954         LASSERTF(fid_oid(&d->opd_pre_last_created_fid) <= IDIF_MAX_OID,
 955                  "%s: last_created_fid "DFID" > %llu\n",
 956                  d->opd_obd->obd_name, PFID(&d->opd_pre_last_created_fid),
 957                  IDIF_MAX_OID);
 958         d->opd_pre_used_fid = d->opd_pre_last_created_fid;
 959         d->opd_pre_create_slow = 0;
 960         if ((body->oa.o_valid & OBD_MD_FLSIZE) && body->oa.o_size)
 961                 d->opd_pre_seq_width = body->oa.o_size;
 962         spin_unlock(&d->opd_pre_lock);
 963
 964         CDEBUG(D_HA, "%s: Got last_id "DFID" from OST, last_created "DFID
 965                "last_used is "DFID"\n", d->opd_obd->obd_name, PFID(last_fid),
 966                PFID(&d->opd_pre_last_created_fid), PFID(&d->opd_last_used_fid));
 967 out:
 968         if (req)
 969                 ptlrpc_req_finished(req);
 970
 971
 972         /*
 973          * If rc is zero, the pre-creation window should have been emptied.
 974          * Since waking up the herd would be useless without pre-created
 975          * objects, we defer the signal to osp_precreate_send() in that case.
 976          */
 977         if (rc != 0) {
 978                 if (update_status) {
 979                         CERROR("%s: cannot cleanup orphans: rc = %d\n",
 980                                d->opd_obd->obd_name, rc);
 981                         /* we can't proceed from here, OST seem to
 982                          * be in a bad shape, better to wait for
 983                          * a new instance of the server and repeat
 984                          * from the beginning. notify possible waiters
 985                          * this OSP isn't quite functional yet */
 986                         osp_pre_update_status(d, rc);
 987                 } else {
 988                         wake_up_all(&d->opd_pre_user_waitq);
 989                 }
 990         } else {
 991                 spin_lock(&d->opd_pre_lock);
 992                 d->opd_pre_recovering = 0;
 993                 spin_unlock(&d->opd_pre_lock);
 994                 d->opd_cleanup_orphans_done = true;
 995         }
 996
 997         RETURN(rc);
 998 }
 999
1000 /**
1001  * Update precreate status using statfs data
1002  *
1003  * The function decides whether this OSP should be used for new objects.
1004  * IOW, whether this OST is used up or has some free space. Cached statfs
1005  * data is used to make this decision. If the latest result of statfs
1006  * request (rc argument) is not success, then just mark OSP unavailable
1007  * right away.
1008  *
1009  * The new statfs data is passed in \a msfs and needs to be stored into
1010  * opd_statfs, but only after the various flags in os_state are set, so
1011  * that the new statfs data is not visible without appropriate flags set.
1012  * As such, there is no need to clear the flags here, since this is called
1013  * with new statfs data, and they should not be cleared if sent from OST.
1014  *
1015  * Add a bit of hysteresis so this flag isn't continually flapping, and
1016  * ensure that new files don't get extremely fragmented due to only a
1017  * small amount of available space in the filesystem.  We want to set
1018  * the ENOSPC/ENOINO flags unconditionally when there is less than the
1019  * reserved size free, and still copy them from the old state when there
1020  * is less than 2*reserved size free space or inodes.
1021  *
1022  * \param[in] d         OSP device
1023  * \param[in] msfs      statfs data
1024  */
1025 static void osp_pre_update_msfs(struct osp_device *d, struct obd_statfs *msfs)
1026 {
1027         u32 old_state = d->opd_statfs.os_state;
1028         u64 available_mb;
1029
1030         /* statfs structure not initialized yet */
1031         if (unlikely(!msfs->os_type))
1032                 return;
1033
1034         /* if the low and high watermarks have not been initialized yet */
1035         if (unlikely(d->opd_reserved_mb_high == 0 &&
1036                      d->opd_reserved_mb_low == 0)) {
1037                 /* Use ~0.1% by default to disable object allocation,
1038                  * and ~0.2% to enable, size in MB, set both watermark
1039                  */
1040                 spin_lock(&d->opd_pre_lock);
1041                 if (d->opd_reserved_mb_high == 0 &&
1042                     d->opd_reserved_mb_low == 0) {
1043                         d->opd_reserved_mb_low = ((msfs->os_bsize >> 10) *
1044                                                   msfs->os_blocks) >> 20;
1045                         if (d->opd_reserved_mb_low < 1)
1046                                 d->opd_reserved_mb_low = 1;
1047                         d->opd_reserved_mb_high =
1048                                 (d->opd_reserved_mb_low << 1) + 1;
1049                 }
1050                 spin_unlock(&d->opd_pre_lock);
1051         }
1052
1053         if (unlikely(d->opd_reserved_ino_high == 0 &&
1054                      d->opd_reserved_ino_low == 0)) {
1055                 /* Use ~0.0001% by default to disallow distributed transactions,
1056                  * and ~0.0002% to allow, set both watermark
1057                  */
1058                 spin_lock(&d->opd_pre_lock);
1059                 if (d->opd_reserved_ino_high == 0 &&
1060                     d->opd_reserved_ino_low == 0) {
1061                         d->opd_reserved_ino_low = msfs->os_ffree >> 20;
1062                         if (d->opd_reserved_ino_low < 32)
1063                                 d->opd_reserved_ino_low = 32;
1064                         d->opd_reserved_ino_high =
1065                                 (d->opd_reserved_ino_low << 1) + 1;
1066                 }
1067                 spin_unlock(&d->opd_pre_lock);
1068         }
1069
1070         available_mb = (msfs->os_bavail * (msfs->os_bsize >> 10)) >> 10;
1071         if (msfs->os_ffree < d->opd_reserved_ino_low)
1072                 msfs->os_state |= OS_STATFS_ENOINO;
1073         else if (msfs->os_ffree <= d->opd_reserved_ino_high)
1074                 msfs->os_state |= old_state & OS_STATFS_ENOINO;
1075         /* else don't clear flags in new msfs->os_state sent from OST */
1076
1077         if (available_mb < d->opd_reserved_mb_low)
1078                 msfs->os_state |= OS_STATFS_ENOSPC;
1079         else if (available_mb <= d->opd_reserved_mb_high)
1080                 msfs->os_state |= old_state & OS_STATFS_ENOSPC;
1081         /* else don't clear flags in new msfs->os_state sent from OST */
1082
1083         CDEBUG(D_INFO,
1084                "%s: blocks=%llu free=%llu avail=%llu avail_mb=%llu hwm_mb=%u files=%llu ffree=%llu state=%x: rc = %d\n",
1085                d->opd_obd->obd_name, msfs->os_blocks, msfs->os_bfree,
1086                msfs->os_bavail, available_mb, d->opd_reserved_mb_high,
1087                msfs->os_files, msfs->os_ffree, msfs->os_state,
1088                d->opd_pre ? d->opd_pre_status : 0);
1089
1090         if (!d->opd_pre)
1091                 goto update;
1092
1093         if (msfs->os_state & (OS_STATFS_ENOINO | OS_STATFS_ENOSPC)) {
1094                 d->opd_pre_status = -ENOSPC;
1095                 if (!(old_state & (OS_STATFS_ENOINO | OS_STATFS_ENOSPC)))
1096                         CDEBUG(D_INFO, "%s: full: state=%x: rc = %x\n",
1097                                d->opd_obd->obd_name, msfs->os_state,
1098                                d->opd_pre_status);
1099                 CDEBUG(D_INFO, "uncommitted changes=%u in_progress=%u\n",
1100                        atomic_read(&d->opd_sync_changes),
1101                        atomic_read(&d->opd_sync_rpcs_in_progress));
1102         } else if (old_state & (OS_STATFS_ENOINO | OS_STATFS_ENOSPC)) {
1103                 d->opd_pre_status = 0;
1104                 spin_lock(&d->opd_pre_lock);
1105                 d->opd_pre_create_slow = 0;
1106                 d->opd_pre_create_count = OST_MIN_PRECREATE;
1107                 spin_unlock(&d->opd_pre_lock);
1108                 wake_up(&d->opd_pre_waitq);
1109
1110                 CDEBUG(D_INFO,
1111                        "%s: available: state=%x: rc = %d\n",
1112                        d->opd_obd->obd_name, msfs->os_state,
1113                        d->opd_pre_status);
1114         } else {
1115                 /* we only get here if rc == 0 in the caller */
1116                 d->opd_pre_status = 0;
1117         }
1118
1119         /* Object precreation skipped on OST if manually disabled */
1120         if (d->opd_pre_max_create_count == 0)
1121                 msfs->os_state |= OS_STATFS_NOPRECREATE;
1122         /* else don't clear flags in new msfs->os_state sent from OST */
1123
1124 update:
1125         /* copy only new statfs state to make it visible to MDS threads */
1126         if (&d->opd_statfs != msfs)
1127                 d->opd_statfs = *msfs;
1128 }
1129
1130 /**
1131  * Initialize FID for precreation
1132  *
1133  * For a just created new target, a new sequence should be taken.
1134  * The function checks there is no IDIF in use (if the target was
1135  * added with the older version of Lustre), then requests a new
1136  * sequence from FLDB using the regular protocol. Then this new
1137  * sequence is stored on a persisten storage synchronously to prevent
1138  * possible object leakage (for the detail see the description for
1139  * osp_precreate_rollover_new_seq()).
1140  *
1141  * \param[in] osp       OSP device
1142  *
1143  * \retval 0            on success
1144  * \retval negative     negated errno on error
1145  */
1146 int osp_init_pre_fid(struct lu_env *env, struct osp_device *osp)
1147 {
1148         struct osp_thread_info  *osi;
1149         struct lu_client_seq    *cli_seq;
1150         struct lu_fid           *last_fid;
1151         int                     rc;
1152         ENTRY;
1153
1154         LASSERT(osp->opd_pre != NULL);
1155
1156         /* Let's check if the current last_seq/fid is valid,
1157          * otherwise request new sequence from the controller */
1158         if (osp_is_fid_client(osp) && osp->opd_group != 0) {
1159                 /* Non-MDT0 can only use normal sequence for
1160                  * OST objects */
1161                 if (fid_is_norm(&osp->opd_last_used_fid))
1162                         RETURN(0);
1163         } else {
1164                 /* Initially MDT0 will start with IDIF, after
1165                  * that it will request new sequence from the
1166                  * controller */
1167                 if (fid_is_idif(&osp->opd_last_used_fid) ||
1168                     fid_is_norm(&osp->opd_last_used_fid))
1169                         RETURN(0);
1170         }
1171
1172         if (!fid_is_zero(&osp->opd_last_used_fid))
1173                 CWARN("%s: invalid last used fid "DFID
1174                       ", try to get new sequence.\n",
1175                       osp->opd_obd->obd_name,
1176                       PFID(&osp->opd_last_used_fid));
1177
1178         osi = osp_env_info(env);
1179         last_fid = &osi->osi_fid;
1180         fid_zero(last_fid);
1181         /* For a freshed fs, it will allocate a new sequence first */
1182         if (osp_is_fid_client(osp) && osp->opd_group != 0) {
1183                 cli_seq = osp->opd_obd->u.cli.cl_seq;
1184                 rc = seq_client_get_seq(env, cli_seq, &last_fid->f_seq);
1185                 if (rc != 0) {
1186                         CERROR("%s: alloc fid error: rc = %d\n",
1187                                osp->opd_obd->obd_name, rc);
1188                         GOTO(out, rc);
1189                 }
1190         } else {
1191                 last_fid->f_seq = fid_idif_seq(0, osp->opd_index);
1192         }
1193         last_fid->f_oid = 1;
1194         last_fid->f_ver = 0;
1195
1196         spin_lock(&osp->opd_pre_lock);
1197         osp->opd_last_used_fid = *last_fid;
1198         osp->opd_pre_used_fid = *last_fid;
1199         osp->opd_pre_last_created_fid = *last_fid;
1200         spin_unlock(&osp->opd_pre_lock);
1201         rc = osp_write_last_oid_seq_files(env, osp, last_fid, 1);
1202         if (rc != 0) {
1203                 CERROR("%s: write fid error: rc = %d\n",
1204                        osp->opd_obd->obd_name, rc);
1205                 GOTO(out, rc);
1206         }
1207 out:
1208         RETURN(rc);
1209 }
1210
1211 struct opt_args {
1212         struct osp_device       *opta_dev;
1213         struct lu_env           opta_env;
1214         struct completion       *opta_started;
1215 };
1216 /**
1217  * The core of precreate functionality
1218  *
1219  * The function implements the main precreation loop. Basically it
1220  * involves connecting to the target, precerate FID initialization,
1221  * identifying and removing orphans, then serving precreation. As
1222  * part of the latter, the thread is responsible for statfs data
1223  * updates. The precreation is mostly driven by another threads
1224  * asking for new OST objects - those askers wake the thread when
1225  * the number of precreated objects reach low watermark.
1226  * After a disconnect, the sequence above repeats. This is keep going
1227  * until the thread is requested to stop.
1228  *
1229  * \param[in] _arg      private data the thread (OSP device to handle)
1230  *
1231  * \retval 0            on success
1232  * \retval negative     negated errno on error
1233  */
1234 static int osp_precreate_thread(void *_args)
1235 {
1236         struct opt_args         *args = _args;
1237         struct osp_device       *d = args->opta_dev;
1238         struct lu_env           *env = &args->opta_env;
1239         int                      rc;
1240
1241         ENTRY;
1242
1243         complete(args->opta_started);
1244
1245         /* wait for connection from the layers above */
1246         wait_event_idle(d->opd_pre_waitq,
1247                         kthread_should_stop() ||
1248                         d->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
1249
1250         while (!kthread_should_stop()) {
1251                 /*
1252                  * need to be connected to OST
1253                  */
1254                 while (!kthread_should_stop()) {
1255                         if ((d->opd_pre == NULL || d->opd_pre_recovering) &&
1256                             d->opd_imp_connected &&
1257                             !d->opd_got_disconnected)
1258                                 break;
1259                         wait_event_idle(d->opd_pre_waitq,
1260                                         kthread_should_stop() ||
1261                                         d->opd_new_connection);
1262
1263                         if (!d->opd_new_connection)
1264                                 continue;
1265
1266                         OBD_FAIL_TIMEOUT(OBD_FAIL_OSP_CON_EVENT_DELAY,
1267                                          cfs_fail_val);
1268                         d->opd_new_connection = 0;
1269                         d->opd_got_disconnected = 0;
1270                         break;
1271                 }
1272
1273                 if (kthread_should_stop())
1274                         break;
1275
1276                 if (d->opd_pre) {
1277                         LASSERT(d->opd_obd->u.cli.cl_seq != NULL);
1278                         LASSERT(d->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
1279
1280                         /* Init fid for osp_precreate if necessary */
1281                         rc = osp_init_pre_fid(env, d);
1282                         if (rc != 0) {
1283                                 class_export_put(d->opd_exp);
1284                                 d->opd_obd->u.cli.cl_seq->lcs_exp = NULL;
1285                                 CERROR("%s: init pre fid error: rc = %d\n",
1286                                                 d->opd_obd->obd_name, rc);
1287                                 continue;
1288                         }
1289                 }
1290
1291                 if (osp_statfs_update(env, d)) {
1292                         if (wait_event_idle_timeout(d->opd_pre_waitq,
1293                                                     kthread_should_stop(),
1294                                                     cfs_time_seconds(5)) == 0)
1295                                 l_wait_event_abortable(
1296                                         d->opd_pre_waitq,
1297                                         kthread_should_stop());
1298                         continue;
1299                 }
1300
1301                 if (d->opd_pre) {
1302                         /*
1303                          * Clean up orphans or recreate missing objects.
1304                          */
1305                         rc = osp_precreate_cleanup_orphans(env, d);
1306                         if (rc != 0) {
1307                                 schedule_timeout_interruptible(cfs_time_seconds(1));
1308                                 continue;
1309                         }
1310                 }
1311
1312                 /*
1313                  * connected, can handle precreates now
1314                  */
1315                 while (!kthread_should_stop()) {
1316                         wait_event_idle(d->opd_pre_waitq,
1317                                         kthread_should_stop() ||
1318                                         (osp_precreate_near_empty(env, d) &&
1319                                          !(osp_precreate_end_seq(env, d) &&
1320                                            osp_objs_precreated(env, d) != 0)) ||
1321                                         osp_statfs_need_update(d) ||
1322                                         d->opd_got_disconnected);
1323
1324                         if (kthread_should_stop())
1325                                 break;
1326
1327                         /* something happened to the connection
1328                          * have to start from the beginning */
1329                         if (d->opd_got_disconnected)
1330                                 break;
1331
1332                         if (osp_statfs_need_update(d))
1333                                 if (osp_statfs_update(env, d))
1334                                         break;
1335
1336                         if (d->opd_pre == NULL)
1337                                 continue;
1338
1339                         if (OBD_FAIL_CHECK(OBD_FAIL_OSP_GET_LAST_FID)) {
1340                                 d->opd_pre_recovering = 1;
1341                                 break;
1342                         }
1343
1344                         /* To avoid handling different seq in precreate/orphan
1345                          * cleanup, it will hold precreate until current seq is
1346                          * used up. */
1347                         if (unlikely(osp_precreate_end_seq(env, d))) {
1348                                 if (osp_objs_precreated(env, d) == 0) {
1349                                         rc = osp_precreate_rollover_new_seq(env, d);
1350                                         if (rc)
1351                                                 continue;
1352                                 } else {
1353                                         continue;
1354                                 }
1355                         }
1356
1357                         if (osp_precreate_near_empty(env, d)) {
1358                                 rc = osp_precreate_send(env, d);
1359                                 /* osp_precreate_send() sets opd_pre_status
1360                                  * in case of error, that prevent the using of
1361                                  * failed device. */
1362                                 if (rc < 0 && rc != -ENOSPC &&
1363                                     rc != -ETIMEDOUT && rc != -ENOTCONN)
1364                                         CERROR("%s: cannot precreate objects:"
1365                                                " rc = %d\n",
1366                                                d->opd_obd->obd_name, rc);
1367                         }
1368                 }
1369         }
1370
1371         lu_env_fini(env);
1372         OBD_FREE_PTR(args);
1373
1374         RETURN(0);
1375 }
1376
1377 /**
1378  * Check when to stop to wait for precreate objects.
1379  *
1380  * The caller wanting a new OST object can't wait undefinitely. The
1381  * function checks for few conditions including available new OST
1382  * objects, disconnected OST, lack of space with no pending destroys,
1383  * etc. IOW, it checks whether the current OSP state is good to keep
1384  * waiting or it's better to give up.
1385  *
1386  * \param[in] env       LU environment provided by the caller
1387  * \param[in] d         OSP device
1388  *
1389  * \retval              0 - keep waiting, 1 - no luck
1390  */
1391 static int osp_precreate_ready_condition(const struct lu_env *env,
1392                                          struct osp_device *d)
1393 {
1394         /* Bail out I/O fails to OST */
1395         if (d->opd_pre_status != 0 &&
1396             d->opd_pre_status != -EAGAIN &&
1397             d->opd_pre_status != -ENODEV &&
1398             d->opd_pre_status != -ENOTCONN &&
1399             d->opd_pre_status != -ENOSPC) {
1400                 /* DEBUG LU-3230 */
1401                 if (d->opd_pre_status != -EIO)
1402                         CERROR("%s: precreate failed opd_pre_status %d\n",
1403                                d->opd_obd->obd_name, d->opd_pre_status);
1404                 return 1;
1405         }
1406
1407         if (d->opd_pre_recovering || d->opd_force_creation)
1408                 return 0;
1409
1410         /* ready if got enough precreated objects */
1411         /* we need to wait for others (opd_pre_reserved) and our object (+1) */
1412         if (d->opd_pre_reserved + 1 < osp_objs_precreated(env, d))
1413                 return 1;
1414
1415         /* ready if OST reported no space and no destroys in progress */
1416         if (atomic_read(&d->opd_sync_changes) +
1417             atomic_read(&d->opd_sync_rpcs_in_progress) == 0 &&
1418             d->opd_pre_status == -ENOSPC)
1419                 return 1;
1420
1421         return 0;
1422 }
1423
1424 /**
1425  * Reserve object in precreate pool
1426  *
1427  * When the caller wants to create a new object on this target (target
1428  * represented by the given OSP), it should declare this intention using
1429  * a regular ->dt_declare_create() OSD API method. Then OSP will be trying
1430  * to reserve an object in the existing precreated pool or wait up to
1431  * obd_timeout for the available object to appear in the pool (a dedicated
1432  * thread will be doing real precreation in background). The object can be
1433  * consumed later with osp_precreate_get_fid() or be released with call to
1434  * lu_object_put(). Notice the function doesn't reserve a specific ID, just
1435  * some ID. The actual ID assignment happen in osp_precreate_get_fid().
1436  * If the space on the target is short and there is a pending object destroy,
1437  * then the function forces local commit to speedup space release (see
1438  * osp_sync.c for the details).
1439  *
1440  * \param[in] env       LU environment provided by the caller
1441  * \param[in] d         OSP device
1442  *
1443  * \retval              0 on success
1444  * \retval              -ENOSPC when no space on OST
1445  * \retval              -EAGAIN try later, slow precreation in progress
1446  * \retval              -EIO when no access to OST
1447  */
1448 int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d,
1449                           bool can_block)
1450 {
1451         time64_t expire = ktime_get_seconds() + obd_timeout;
1452         int precreated, rc, synced = 0;
1453
1454         ENTRY;
1455
1456         LASSERTF(osp_objs_precreated(env, d) >= 0, "Last created FID "DFID
1457                  "Next FID "DFID"\n", PFID(&d->opd_pre_last_created_fid),
1458                  PFID(&d->opd_pre_used_fid));
1459
1460         /* opd_pre_max_create_count 0 to not use specified OST. */
1461         if (d->opd_pre_max_create_count == 0)
1462                 RETURN(-ENOBUFS);
1463
1464         /*
1465          * wait till:
1466          *  - preallocation is done
1467          *  - no free space expected soon
1468          *  - can't connect to OST for too long (obd_timeout)
1469          *  - OST can allocate fid sequence.
1470          */
1471         while ((rc = d->opd_pre_status) == 0 || rc == -ENOSPC ||
1472                 rc == -ENODEV || rc == -EAGAIN || rc == -ENOTCONN) {
1473
1474                 /*
1475                  * increase number of precreations
1476                  */
1477                 precreated = osp_objs_precreated(env, d);
1478                 if (d->opd_pre_create_count < d->opd_pre_max_create_count &&
1479                     d->opd_pre_create_slow == 0 &&
1480                     precreated <= (d->opd_pre_create_count / 4 + 1)) {
1481                         spin_lock(&d->opd_pre_lock);
1482                         d->opd_pre_create_slow = 1;
1483                         d->opd_pre_create_count *= 2;
1484                         spin_unlock(&d->opd_pre_lock);
1485                 }
1486
1487                 spin_lock(&d->opd_pre_lock);
1488                 precreated = osp_objs_precreated(env, d);
1489                 if (!d->opd_pre_recovering && !d->opd_force_creation) {
1490                         if (precreated > d->opd_pre_reserved) {
1491                                 d->opd_pre_reserved++;
1492                                 spin_unlock(&d->opd_pre_lock);
1493                                 rc = 0;
1494
1495                                 /*
1496                                  * XXX: don't wake up if precreation
1497                                  * is in progress
1498                                  */
1499                                 if (osp_precreate_near_empty_nolock(env, d) &&
1500                                    !osp_precreate_end_seq_nolock(env, d))
1501                                         wake_up(&d->opd_pre_waitq);
1502
1503                                 break;
1504                         } else if (unlikely(precreated &&
1505                                    osp_precreate_end_seq_nolock(env, d))) {
1506                                 /*
1507                                  * precreate pool is reaching the end of the
1508                                  * current seq, and doesn't have enough objects
1509                                  */
1510                                 rc = -ENOSPC;
1511                                 spin_unlock(&d->opd_pre_lock);
1512                                 break;
1513                         }
1514                 }
1515                 spin_unlock(&d->opd_pre_lock);
1516
1517                 /*
1518                  * all precreated objects have been used and no-space
1519                  * status leave us no chance to succeed very soon
1520                  * but if there is destroy in progress, then we should
1521                  * wait till that is done - some space might be released
1522                  */
1523                 if (unlikely(rc == -ENOSPC)) {
1524                         if (atomic_read(&d->opd_sync_changes) && synced == 0) {
1525                                 /* force local commit to release space */
1526                                 dt_commit_async(env, d->opd_storage);
1527                                 osp_sync_check_for_work(d);
1528                                 synced = 1;
1529                         }
1530                         if (atomic_read(&d->opd_sync_rpcs_in_progress)) {
1531                                 /* just wait till destroys are done
1532                                  * see wait_event_idle_timeout() below
1533                                  */
1534                         }
1535                         if (atomic_read(&d->opd_sync_changes) +
1536                             atomic_read(&d->opd_sync_rpcs_in_progress) == 0) {
1537                                 /* no hope for free space */
1538                                 break;
1539                         }
1540                 }
1541
1542                 /* XXX: don't wake up if precreation is in progress */
1543                 wake_up(&d->opd_pre_waitq);
1544
1545                 if (ktime_get_seconds() >= expire) {
1546                         rc = -ETIMEDOUT;
1547                         break;
1548                 }
1549
1550                 if (!can_block) {
1551                         LASSERT(d->opd_pre);
1552                         rc = -ENOBUFS;
1553                         break;
1554                 }
1555
1556                 CDEBUG(D_INFO, "%s: Sleeping on objects\n",
1557                        d->opd_obd->obd_name);
1558                 if (wait_event_idle_timeout(
1559                             d->opd_pre_user_waitq,
1560                             osp_precreate_ready_condition(env, d),
1561                             cfs_time_seconds(obd_timeout)) == 0) {
1562                         CDEBUG(D_HA,
1563                                "%s: slow creates, last="DFID", next="DFID", "
1564                                "reserved=%llu, sync_changes=%u, "
1565                                "sync_rpcs_in_progress=%d, status=%d\n",
1566                                d->opd_obd->obd_name,
1567                                PFID(&d->opd_pre_last_created_fid),
1568                                PFID(&d->opd_pre_used_fid), d->opd_pre_reserved,
1569                                atomic_read(&d->opd_sync_changes),
1570                                atomic_read(&d->opd_sync_rpcs_in_progress),
1571                                d->opd_pre_status);
1572                 } else {
1573                         CDEBUG(D_INFO, "%s: Waked up, status=%d\n",
1574                                d->opd_obd->obd_name, d->opd_pre_status);
1575                 }
1576         }
1577
1578         RETURN(rc);
1579 }
1580
1581 /**
1582  * Get a FID from precreation pool
1583  *
1584  * The function is a companion for osp_precreate_reserve() - it assigns
1585  * a specific FID from the precreate. The function should be called only
1586  * if the call to osp_precreate_reserve() was successful. The function
1587  * updates a local storage to remember the highest object ID referenced
1588  * by the node in the given sequence.
1589  *
1590  * A very importan details: this is supposed to be called once the
1591  * transaction is started, so on-disk update will be atomic with the
1592  * data (like LOVEA) refering this object. Then the object won't be leaked:
1593  * either it's referenced by the committed transaction or it's a subject
1594  * to the orphan cleanup procedure.
1595  *
1596  * \param[in] env       LU environment provided by the caller
1597  * \param[in] d         OSP device
1598  * \param[out] fid      generated FID
1599  *
1600  * \retval 0            on success
1601  * \retval negative     negated errno on error
1602  */
1603 int osp_precreate_get_fid(const struct lu_env *env, struct osp_device *d,
1604                           struct lu_fid *fid)
1605 {
1606         struct lu_fid *pre_used_fid = &d->opd_pre_used_fid;
1607
1608         /* grab next id from the pool */
1609         spin_lock(&d->opd_pre_lock);
1610
1611         LASSERTF(osp_fid_diff(&d->opd_pre_used_fid,
1612                              &d->opd_pre_last_created_fid) < 0,
1613                  "next fid "DFID" > last created fid "DFID"\n",
1614                  PFID(&d->opd_pre_used_fid),
1615                  PFID(&d->opd_pre_last_created_fid));
1616
1617         /* Non-IDIF FIDs shouldn't get here with OID == OBIF_MAX_OID. For IDIF,
1618          * f_oid wraps and "f_seq" (holding high 16 bits of ID) needs increment
1619          */
1620         if (fid_is_idif(pre_used_fid) &&
1621             unlikely(fid_oid(pre_used_fid) == OBIF_MAX_OID)) {
1622                 struct ost_id oi;
1623                 __u32 idx = fid_idif_ost_idx(pre_used_fid);
1624
1625                 fid_to_ostid(pre_used_fid, &oi);
1626                 oi.oi.oi_id++;
1627                 ostid_to_fid(pre_used_fid, &oi, idx);
1628         } else {
1629                 pre_used_fid->f_oid++;
1630         }
1631
1632         memcpy(fid, pre_used_fid, sizeof(*fid));
1633         d->opd_pre_reserved--;
1634         /*
1635          * last_used_id must be changed along with getting new id otherwise
1636          * we might miscalculate gap causing object loss or leak
1637          */
1638         osp_update_last_fid(d, fid);
1639         spin_unlock(&d->opd_pre_lock);
1640
1641         /*
1642          * probably main thread suspended orphan cleanup till
1643          * all reservations are released, see comment in
1644          * osp_precreate_thread() just before orphan cleanup
1645          */
1646         if (unlikely(d->opd_pre_reserved == 0 &&
1647                      (d->opd_pre_recovering || d->opd_pre_status)))
1648                 wake_up(&d->opd_pre_waitq);
1649
1650         return 0;
1651 }
1652
1653 /*
1654  * Set size regular attribute on an object
1655  *
1656  * When a striping is created late, it's possible that size is already
1657  * initialized on the file. Then the new striping should inherit size
1658  * from the file. The function sets size on the object using the regular
1659  * protocol (OST_PUNCH).
1660  * XXX: should be re-implemented using OUT ?
1661  *
1662  * \param[in] env       LU environment provided by the caller
1663  * \param[in] dt        object
1664  * \param[in] size      size to set.
1665  *
1666  * \retval 0            on success
1667  * \retval negative     negated errno on error
1668  */
1669 int osp_object_truncate(const struct lu_env *env, struct dt_object *dt,
1670                         __u64 size)
1671 {
1672         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1673         struct ptlrpc_request   *req = NULL;
1674         struct obd_import       *imp;
1675         struct ost_body         *body;
1676         struct obdo             *oa = NULL;
1677         int                      rc;
1678
1679         ENTRY;
1680
1681         imp = d->opd_obd->u.cli.cl_import;
1682         LASSERT(imp);
1683
1684         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
1685         if (req == NULL)
1686                 RETURN(-ENOMEM);
1687
1688         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
1689         if (rc) {
1690                 ptlrpc_request_free(req);
1691                 RETURN(rc);
1692         }
1693
1694         /*
1695          * XXX: decide how do we do here with resend
1696          * if we don't resend, then client may see wrong file size
1697          * if we do resend, then MDS thread can get stuck for quite long
1698          * and if we don't resend, then client will also get -EAGAIN !!
1699          * (see LU-7975 and sanity/test_27F use cases)
1700          * but let's decide not to resend/delay this truncate request to OST
1701          * and allow Client to decide to resend, in a less agressive way from
1702          * after_reply(), by returning -EINPROGRESS instead of
1703          * -EAGAIN/-EAGAIN upon return from ptlrpc_queue_wait() at the
1704          * end of this routine
1705          */
1706         req->rq_no_resend = req->rq_no_delay = 1;
1707
1708         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1709         ptlrpc_at_set_req_timeout(req);
1710
1711         OBD_ALLOC_PTR(oa);
1712         if (oa == NULL)
1713                 GOTO(out, rc = -ENOMEM);
1714
1715         rc = fid_to_ostid(lu_object_fid(&dt->do_lu), &oa->o_oi);
1716         LASSERT(rc == 0);
1717         oa->o_size = size;
1718         oa->o_blocks = OBD_OBJECT_EOF;
1719         oa->o_valid = OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
1720                       OBD_MD_FLID | OBD_MD_FLGROUP;
1721
1722         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1723         LASSERT(body);
1724         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
1725
1726         /* XXX: capa support? */
1727         /* osc_pack_capa(req, body, capa); */
1728
1729         ptlrpc_request_set_replen(req);
1730
1731         rc = ptlrpc_queue_wait(req);
1732         if (rc) {
1733                 /* -EAGAIN/-EWOULDBLOCK means OST is unreachable at the moment
1734                  * since we have decided not to resend/delay, but this could
1735                  * lead to wrong size to be seen at Client side and even process
1736                  * trying to open to exit/fail if not itself handling -EAGAIN.
1737                  * So it should be better to return -EINPROGRESS instead and
1738                  * leave the decision to resend at Client side in after_reply()
1739                  */
1740                 if (rc == -EAGAIN) {
1741                         rc = -EINPROGRESS;
1742                         CDEBUG(D_HA, "returning -EINPROGRESS instead of "
1743                                "-EWOULDBLOCK/-EAGAIN to allow Client to "
1744                                "resend\n");
1745                 } else {
1746                         CERROR("can't punch object: %d\n", rc);
1747                 }
1748         }
1749 out:
1750         ptlrpc_req_finished(req);
1751         if (oa)
1752                 OBD_FREE_PTR(oa);
1753         RETURN(rc);
1754 }
1755
1756 /**
1757  * Initialize precreation functionality of OSP
1758  *
1759  * Prepares all the internal structures and starts the precreate thread
1760  *
1761  * \param[in] d         OSP device
1762  *
1763  * \retval 0            on success
1764  * \retval negative     negated errno on error
1765  */
1766 int osp_init_precreate(struct osp_device *d)
1767 {
1768         ENTRY;
1769
1770         OBD_ALLOC_PTR(d->opd_pre);
1771         if (d->opd_pre == NULL)
1772                 RETURN(-ENOMEM);
1773
1774         /* initially precreation isn't ready */
1775         init_waitqueue_head(&d->opd_pre_user_waitq);
1776         d->opd_pre_status = -EAGAIN;
1777         fid_zero(&d->opd_pre_used_fid);
1778         d->opd_pre_used_fid.f_oid = 1;
1779         fid_zero(&d->opd_pre_last_created_fid);
1780         d->opd_pre_last_created_fid.f_oid = 1;
1781         d->opd_last_id = 0;
1782         d->opd_pre_reserved = 0;
1783         d->opd_pre_seq_width = LUSTRE_DATA_SEQ_MAX_WIDTH;
1784         d->opd_got_disconnected = 1;
1785         d->opd_pre_create_slow = 0;
1786         d->opd_pre_create_count = OST_MIN_PRECREATE;
1787         d->opd_pre_min_create_count = OST_MIN_PRECREATE;
1788         d->opd_pre_max_create_count = OST_MAX_PRECREATE;
1789         d->opd_reserved_mb_high = 0;
1790         d->opd_reserved_mb_low = 0;
1791         d->opd_cleanup_orphans_done = false;
1792         d->opd_force_creation = false;
1793
1794         RETURN(0);
1795 }
1796
1797 /**
1798  * Finish precreate functionality of OSP
1799  *
1800  *
1801  * Asks all the activity (the thread, update timer) to stop, then
1802  * wait till that is done.
1803  *
1804  * \param[in] d         OSP device
1805  */
1806 void osp_precreate_fini(struct osp_device *d)
1807 {
1808         ENTRY;
1809
1810         if (d->opd_pre == NULL)
1811                 RETURN_EXIT;
1812
1813         OBD_FREE_PTR(d->opd_pre);
1814         d->opd_pre = NULL;
1815
1816         EXIT;
1817 }
1818
1819 int osp_init_statfs(struct osp_device *d)
1820 {
1821         struct task_struct      *task;
1822         struct opt_args         *args;
1823         DECLARE_COMPLETION_ONSTACK(started);
1824         int                     rc;
1825
1826         ENTRY;
1827
1828         spin_lock_init(&d->opd_pre_lock);
1829         init_waitqueue_head(&d->opd_pre_waitq);
1830
1831         /*
1832          * Initialize statfs-related things
1833          */
1834         d->opd_statfs_maxage = 5; /* defaultupdate interval */
1835         d->opd_statfs_fresh_till = ktime_sub_ns(ktime_get(),
1836                                                 1000 * NSEC_PER_SEC);
1837         CDEBUG(D_OTHER, "current %lldns, fresh till %lldns\n",
1838                ktime_get_ns(),
1839                ktime_to_ns(d->opd_statfs_fresh_till));
1840         cfs_timer_setup(&d->opd_statfs_timer, osp_statfs_timer_cb,
1841                         (unsigned long)d, 0);
1842
1843         if (d->opd_storage->dd_rdonly)
1844                 RETURN(0);
1845
1846         OBD_ALLOC_PTR(args);
1847         if (!args)
1848                 RETURN(0);
1849         args->opta_dev = d;
1850         args->opta_started = &started;
1851         rc = lu_env_init(&args->opta_env,
1852                          d->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
1853         if (rc) {
1854                 CERROR("%s: init env error: rc = %d\n", d->opd_obd->obd_name,
1855                        rc);
1856                 OBD_FREE_PTR(args);
1857                 RETURN(0);
1858         }
1859
1860         /*
1861          * start thread handling precreation and statfs updates
1862          */
1863         task = kthread_create(osp_precreate_thread, args,
1864                               "osp-pre-%u-%u", d->opd_index, d->opd_group);
1865         if (IS_ERR(task)) {
1866                 CERROR("can't start precreate thread %ld\n", PTR_ERR(task));
1867                 lu_env_fini(&args->opta_env);
1868                 OBD_FREE_PTR(args);
1869                 RETURN(PTR_ERR(task));
1870         }
1871         d->opd_pre_task = task;
1872         wake_up_process(task);
1873         wait_for_completion(&started);
1874
1875         RETURN(0);
1876 }
1877
1878 void osp_statfs_fini(struct osp_device *d)
1879 {
1880         struct task_struct *task = d->opd_pre_task;
1881         ENTRY;
1882
1883         timer_delete(&d->opd_statfs_timer);
1884
1885         d->opd_pre_task = NULL;
1886         if (task)
1887                 kthread_stop(task);
1888
1889         EXIT;
1890 }