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