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, 2016, 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         /* Let's check if the current last_seq/fid is valid,
1042          * otherwise request new sequence from the controller */
1043         if (osp_is_fid_client(osp) && osp->opd_group != 0) {
1044                 /* Non-MDT0 can only use normal sequence for
1045                  * OST objects */
1046                 if (fid_is_norm(&osp->opd_last_used_fid))
1047                         RETURN(0);
1048         } else {
1049                 /* Initially MDT0 will start with IDIF, after
1050                  * that it will request new sequence from the
1051                  * controller */
1052                 if (fid_is_idif(&osp->opd_last_used_fid) ||
1053                     fid_is_norm(&osp->opd_last_used_fid))
1054                         RETURN(0);
1055         }
1056
1057         if (!fid_is_zero(&osp->opd_last_used_fid))
1058                 CWARN("%s: invalid last used fid "DFID
1059                       ", try to get new sequence.\n",
1060                       osp->opd_obd->obd_name,
1061                       PFID(&osp->opd_last_used_fid));
1062
1063         rc = lu_env_init(&env, osp->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
1064         if (rc) {
1065                 CERROR("%s: init env error: rc = %d\n",
1066                        osp->opd_obd->obd_name, rc);
1067                 RETURN(rc);
1068         }
1069
1070         osi = osp_env_info(&env);
1071         last_fid = &osi->osi_fid;
1072         fid_zero(last_fid);
1073         /* For a freshed fs, it will allocate a new sequence first */
1074         if (osp_is_fid_client(osp) && osp->opd_group != 0) {
1075                 cli_seq = osp->opd_obd->u.cli.cl_seq;
1076                 rc = seq_client_get_seq(&env, cli_seq, &last_fid->f_seq);
1077                 if (rc != 0) {
1078                         CERROR("%s: alloc fid error: rc = %d\n",
1079                                osp->opd_obd->obd_name, rc);
1080                         GOTO(out, rc);
1081                 }
1082         } else {
1083                 last_fid->f_seq = fid_idif_seq(0, osp->opd_index);
1084         }
1085         last_fid->f_oid = 1;
1086         last_fid->f_ver = 0;
1087
1088         spin_lock(&osp->opd_pre_lock);
1089         osp->opd_last_used_fid = *last_fid;
1090         osp->opd_pre_used_fid = *last_fid;
1091         osp->opd_pre_last_created_fid = *last_fid;
1092         spin_unlock(&osp->opd_pre_lock);
1093         rc = osp_write_last_oid_seq_files(&env, osp, last_fid, 1);
1094         if (rc != 0) {
1095                 CERROR("%s: write fid error: rc = %d\n",
1096                        osp->opd_obd->obd_name, rc);
1097                 GOTO(out, rc);
1098         }
1099 out:
1100         lu_env_fini(&env);
1101         RETURN(rc);
1102 }
1103
1104 /**
1105  * The core of precreate functionality
1106  *
1107  * The function implements the main precreation loop. Basically it
1108  * involves connecting to the target, precerate FID initialization,
1109  * identifying and removing orphans, then serving precreation. As
1110  * part of the latter, the thread is responsible for statfs data
1111  * updates. The precreation is mostly driven by another threads
1112  * asking for new OST objects - those askers wake the thread when
1113  * the number of precreated objects reach low watermark.
1114  * After a disconnect, the sequence above repeats. This is keep going
1115  * until the thread is requested to stop.
1116  *
1117  * \param[in] _arg      private data the thread (OSP device to handle)
1118  *
1119  * \retval 0            on success
1120  * \retval negative     negated errno on error
1121  */
1122 static int osp_precreate_thread(void *_arg)
1123 {
1124         struct osp_device       *d = _arg;
1125         struct ptlrpc_thread    *thread = &d->opd_pre_thread;
1126         struct l_wait_info       lwi = { 0 };
1127         struct l_wait_info       lwi2 = LWI_TIMEOUT(cfs_time_seconds(5),
1128                                                     back_to_sleep, NULL);
1129         struct lu_env            env;
1130         int                      rc;
1131
1132         ENTRY;
1133
1134         rc = lu_env_init(&env, d->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
1135         if (rc) {
1136                 CERROR("%s: init env error: rc = %d\n", d->opd_obd->obd_name,
1137                        rc);
1138                 RETURN(rc);
1139         }
1140
1141         spin_lock(&d->opd_pre_lock);
1142         thread->t_flags = SVC_RUNNING;
1143         spin_unlock(&d->opd_pre_lock);
1144         wake_up(&thread->t_ctl_waitq);
1145
1146         while (osp_precreate_running(d)) {
1147                 /*
1148                  * need to be connected to OST
1149                  */
1150                 while (osp_precreate_running(d)) {
1151                         l_wait_event(d->opd_pre_waitq,
1152                                      !osp_precreate_running(d) ||
1153                                      d->opd_new_connection,
1154                                      &lwi);
1155
1156                         if (!d->opd_new_connection)
1157                                 continue;
1158
1159                         d->opd_new_connection = 0;
1160                         d->opd_got_disconnected = 0;
1161                         break;
1162                 }
1163
1164                 if (!osp_precreate_running(d))
1165                         break;
1166
1167                 LASSERT(d->opd_obd->u.cli.cl_seq != NULL);
1168                 /* Sigh, fid client is not ready yet */
1169                 if (d->opd_obd->u.cli.cl_seq->lcs_exp == NULL)
1170                         continue;
1171
1172                 /* Init fid for osp_precreate if necessary */
1173                 rc = osp_init_pre_fid(d);
1174                 if (rc != 0) {
1175                         class_export_put(d->opd_exp);
1176                         d->opd_obd->u.cli.cl_seq->lcs_exp = NULL;
1177                         CERROR("%s: init pre fid error: rc = %d\n",
1178                                d->opd_obd->obd_name, rc);
1179                         continue;
1180                 }
1181
1182                 if (osp_statfs_update(d)) {
1183                         l_wait_event(d->opd_pre_waitq,
1184                                      !osp_precreate_running(d), &lwi2);
1185                         continue;
1186                 }
1187
1188                 /*
1189                  * Clean up orphans or recreate missing objects.
1190                  */
1191                 rc = osp_precreate_cleanup_orphans(&env, d);
1192                 if (rc != 0)
1193                         continue;
1194                 /*
1195                  * connected, can handle precreates now
1196                  */
1197                 while (osp_precreate_running(d)) {
1198                         l_wait_event(d->opd_pre_waitq,
1199                                      !osp_precreate_running(d) ||
1200                                      osp_precreate_near_empty(&env, d) ||
1201                                      osp_statfs_need_update(d) ||
1202                                      d->opd_got_disconnected, &lwi);
1203
1204                         if (!osp_precreate_running(d))
1205                                 break;
1206
1207                         /* something happened to the connection
1208                          * have to start from the beginning */
1209                         if (d->opd_got_disconnected)
1210                                 break;
1211
1212                         if (osp_statfs_need_update(d))
1213                                 if (osp_statfs_update(d))
1214                                         break;
1215
1216                         /* To avoid handling different seq in precreate/orphan
1217                          * cleanup, it will hold precreate until current seq is
1218                          * used up. */
1219                         if (unlikely(osp_precreate_end_seq(&env, d) &&
1220                             !osp_create_end_seq(&env, d)))
1221                                 continue;
1222
1223                         if (unlikely(osp_precreate_end_seq(&env, d) &&
1224                                      osp_create_end_seq(&env, d))) {
1225                                 LCONSOLE_INFO("%s:%#llx is used up."
1226                                               " Update to new seq\n",
1227                                               d->opd_obd->obd_name,
1228                                          fid_seq(&d->opd_pre_last_created_fid));
1229                                 rc = osp_precreate_rollover_new_seq(&env, d);
1230                                 if (rc)
1231                                         continue;
1232                         }
1233
1234                         if (osp_precreate_near_empty(&env, d)) {
1235                                 rc = osp_precreate_send(&env, d);
1236                                 /* osp_precreate_send() sets opd_pre_status
1237                                  * in case of error, that prevent the using of
1238                                  * failed device. */
1239                                 if (rc < 0 && rc != -ENOSPC &&
1240                                     rc != -ETIMEDOUT && rc != -ENOTCONN)
1241                                         CERROR("%s: cannot precreate objects:"
1242                                                " rc = %d\n",
1243                                                d->opd_obd->obd_name, rc);
1244                         }
1245                 }
1246         }
1247
1248         thread->t_flags = SVC_STOPPED;
1249         lu_env_fini(&env);
1250         wake_up(&thread->t_ctl_waitq);
1251
1252         RETURN(0);
1253 }
1254
1255 /**
1256  * Check when to stop to wait for precreate objects.
1257  *
1258  * The caller wanting a new OST object can't wait undefinitely. The
1259  * function checks for few conditions including available new OST
1260  * objects, disconnected OST, lack of space with no pending destroys,
1261  * etc. IOW, it checks whether the current OSP state is good to keep
1262  * waiting or it's better to give up.
1263  *
1264  * \param[in] env       LU environment provided by the caller
1265  * \param[in] d         OSP device
1266  *
1267  * \retval              0 - keep waiting, 1 - no luck
1268  */
1269 static int osp_precreate_ready_condition(const struct lu_env *env,
1270                                          struct osp_device *d)
1271 {
1272         if (d->opd_pre_recovering)
1273                 return 0;
1274
1275         /* ready if got enough precreated objects */
1276         /* we need to wait for others (opd_pre_reserved) and our object (+1) */
1277         if (d->opd_pre_reserved + 1 < osp_objs_precreated(env, d))
1278                 return 1;
1279
1280         /* ready if OST reported no space and no destroys in progress */
1281         if (atomic_read(&d->opd_syn_changes) +
1282             atomic_read(&d->opd_syn_rpc_in_progress) == 0 &&
1283             d->opd_pre_status == -ENOSPC)
1284                 return 1;
1285
1286         /* Bail out I/O fails to OST */
1287         if (d->opd_pre_status != 0 &&
1288             d->opd_pre_status != -EAGAIN &&
1289             d->opd_pre_status != -ENODEV &&
1290             d->opd_pre_status != -ENOTCONN &&
1291             d->opd_pre_status != -ENOSPC) {
1292                 /* DEBUG LU-3230 */
1293                 if (d->opd_pre_status != -EIO)
1294                         CERROR("%s: precreate failed opd_pre_status %d\n",
1295                                d->opd_obd->obd_name, d->opd_pre_status);
1296                 return 1;
1297         }
1298
1299         return 0;
1300 }
1301
1302 static int osp_precreate_timeout_condition(void *data)
1303 {
1304         struct osp_device *d = data;
1305
1306         CDEBUG(D_HA, "%s: slow creates, last="DFID", next="DFID", "
1307               "reserved=%llu, syn_changes=%u, "
1308               "syn_rpc_in_progress=%d, status=%d\n",
1309               d->opd_obd->obd_name, PFID(&d->opd_pre_last_created_fid),
1310               PFID(&d->opd_pre_used_fid), d->opd_pre_reserved,
1311               atomic_read(&d->opd_syn_changes),
1312               atomic_read(&d->opd_syn_rpc_in_progress),
1313               d->opd_pre_status);
1314
1315         return 1;
1316 }
1317
1318 /**
1319  * Reserve object in precreate pool
1320  *
1321  * When the caller wants to create a new object on this target (target
1322  * represented by the given OSP), it should declare this intention using
1323  * a regular ->dt_declare_create() OSD API method. Then OSP will be trying
1324  * to reserve an object in the existing precreated pool or wait up to
1325  * obd_timeout for the available object to appear in the pool (a dedicated
1326  * thread will be doing real precreation in background). The object can be
1327  * consumed later with osp_precreate_get_fid() or be released with call to
1328  * lu_object_put(). Notice the function doesn't reserve a specific ID, just
1329  * some ID. The actual ID assignment happen in osp_precreate_get_fid().
1330  * If the space on the target is short and there is a pending object destroy,
1331  * then the function forces local commit to speedup space release (see
1332  * osp_sync.c for the details).
1333  *
1334  * \param[in] env       LU environment provided by the caller
1335  * \param[in] d         OSP device
1336  *
1337  * \retval              0 on success
1338  * \retval              -ENOSPC when no space on OST
1339  * \retval              -EAGAIN try later, slow precreation in progress
1340  * \retval              -EIO when no access to OST
1341  */
1342 int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d)
1343 {
1344         struct l_wait_info       lwi;
1345         cfs_time_t               expire = cfs_time_shift(obd_timeout);
1346         int                      precreated, rc;
1347
1348         ENTRY;
1349
1350         LASSERTF(osp_objs_precreated(env, d) >= 0, "Last created FID "DFID
1351                  "Next FID "DFID"\n", PFID(&d->opd_pre_last_created_fid),
1352                  PFID(&d->opd_pre_used_fid));
1353
1354         /* opd_pre_max_create_count 0 to not use specified OST. */
1355         if (d->opd_pre_max_create_count == 0)
1356                 RETURN(-ENOBUFS);
1357
1358         /*
1359          * wait till:
1360          *  - preallocation is done
1361          *  - no free space expected soon
1362          *  - can't connect to OST for too long (obd_timeout)
1363          *  - OST can allocate fid sequence.
1364          */
1365         while ((rc = d->opd_pre_status) == 0 || rc == -ENOSPC ||
1366                 rc == -ENODEV || rc == -EAGAIN || rc == -ENOTCONN) {
1367
1368                 /*
1369                  * increase number of precreations
1370                  */
1371                 precreated = osp_objs_precreated(env, d);
1372                 if (d->opd_pre_create_count < d->opd_pre_max_create_count &&
1373                     d->opd_pre_create_slow == 0 &&
1374                     precreated <= (d->opd_pre_create_count / 4 + 1)) {
1375                         spin_lock(&d->opd_pre_lock);
1376                         d->opd_pre_create_slow = 1;
1377                         d->opd_pre_create_count *= 2;
1378                         spin_unlock(&d->opd_pre_lock);
1379                 }
1380
1381                 spin_lock(&d->opd_pre_lock);
1382                 precreated = osp_objs_precreated(env, d);
1383                 if (precreated > d->opd_pre_reserved &&
1384                     !d->opd_pre_recovering) {
1385                         d->opd_pre_reserved++;
1386                         spin_unlock(&d->opd_pre_lock);
1387                         rc = 0;
1388
1389                         /* XXX: don't wake up if precreation is in progress */
1390                         if (osp_precreate_near_empty_nolock(env, d) &&
1391                            !osp_precreate_end_seq_nolock(env, d))
1392                                 wake_up(&d->opd_pre_waitq);
1393
1394                         break;
1395                 }
1396                 spin_unlock(&d->opd_pre_lock);
1397
1398                 /*
1399                  * all precreated objects have been used and no-space
1400                  * status leave us no chance to succeed very soon
1401                  * but if there is destroy in progress, then we should
1402                  * wait till that is done - some space might be released
1403                  */
1404                 if (unlikely(rc == -ENOSPC)) {
1405                         if (atomic_read(&d->opd_syn_changes)) {
1406                                 /* force local commit to release space */
1407                                 dt_commit_async(env, d->opd_storage);
1408                         }
1409                         if (atomic_read(&d->opd_syn_rpc_in_progress)) {
1410                                 /* just wait till destroys are done */
1411                                 /* see l_wait_even() few lines below */
1412                         }
1413                         if (atomic_read(&d->opd_syn_changes) +
1414                             atomic_read(&d->opd_syn_rpc_in_progress) == 0) {
1415                                 /* no hope for free space */
1416                                 break;
1417                         }
1418                 }
1419
1420                 /* XXX: don't wake up if precreation is in progress */
1421                 wake_up(&d->opd_pre_waitq);
1422
1423                 lwi = LWI_TIMEOUT(expire - cfs_time_current(),
1424                                 osp_precreate_timeout_condition, d);
1425                 if (cfs_time_aftereq(cfs_time_current(), expire)) {
1426                         rc = -ETIMEDOUT;
1427                         break;
1428                 }
1429
1430                 l_wait_event(d->opd_pre_user_waitq,
1431                              osp_precreate_ready_condition(env, d), &lwi);
1432         }
1433
1434         RETURN(rc);
1435 }
1436
1437 /**
1438  * Get a FID from precreation pool
1439  *
1440  * The function is a companion for osp_precreate_reserve() - it assigns
1441  * a specific FID from the precreate. The function should be called only
1442  * if the call to osp_precreate_reserve() was successful. The function
1443  * updates a local storage to remember the highest object ID referenced
1444  * by the node in the given sequence.
1445  *
1446  * A very importan details: this is supposed to be called once the
1447  * transaction is started, so on-disk update will be atomic with the
1448  * data (like LOVEA) refering this object. Then the object won't be leaked:
1449  * either it's referenced by the committed transaction or it's a subject
1450  * to the orphan cleanup procedure.
1451  *
1452  * \param[in] env       LU environment provided by the caller
1453  * \param[in] d         OSP device
1454  * \param[out] fid      generated FID
1455  *
1456  * \retval 0            on success
1457  * \retval negative     negated errno on error
1458  */
1459 int osp_precreate_get_fid(const struct lu_env *env, struct osp_device *d,
1460                           struct lu_fid *fid)
1461 {
1462         /* grab next id from the pool */
1463         spin_lock(&d->opd_pre_lock);
1464
1465         LASSERTF(osp_fid_diff(&d->opd_pre_used_fid,
1466                              &d->opd_pre_last_created_fid) < 0,
1467                  "next fid "DFID" last created fid "DFID"\n",
1468                  PFID(&d->opd_pre_used_fid),
1469                  PFID(&d->opd_pre_last_created_fid));
1470
1471         d->opd_pre_used_fid.f_oid++;
1472         memcpy(fid, &d->opd_pre_used_fid, sizeof(*fid));
1473         d->opd_pre_reserved--;
1474         /*
1475          * last_used_id must be changed along with getting new id otherwise
1476          * we might miscalculate gap causing object loss or leak
1477          */
1478         osp_update_last_fid(d, fid);
1479         spin_unlock(&d->opd_pre_lock);
1480
1481         /*
1482          * probably main thread suspended orphan cleanup till
1483          * all reservations are released, see comment in
1484          * osp_precreate_thread() just before orphan cleanup
1485          */
1486         if (unlikely(d->opd_pre_reserved == 0 && d->opd_pre_status))
1487                 wake_up(&d->opd_pre_waitq);
1488
1489         return 0;
1490 }
1491
1492 /*
1493  * Set size regular attribute on an object
1494  *
1495  * When a striping is created late, it's possible that size is already
1496  * initialized on the file. Then the new striping should inherit size
1497  * from the file. The function sets size on the object using the regular
1498  * protocol (OST_PUNCH).
1499  * XXX: should be re-implemented using OUT ?
1500  *
1501  * \param[in] env       LU environment provided by the caller
1502  * \param[in] dt        object
1503  * \param[in] size      size to set.
1504  *
1505  * \retval 0            on success
1506  * \retval negative     negated errno on error
1507  */
1508 int osp_object_truncate(const struct lu_env *env, struct dt_object *dt,
1509                         __u64 size)
1510 {
1511         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1512         struct ptlrpc_request   *req = NULL;
1513         struct obd_import       *imp;
1514         struct ost_body         *body;
1515         struct obdo             *oa = NULL;
1516         int                      rc;
1517
1518         ENTRY;
1519
1520         imp = d->opd_obd->u.cli.cl_import;
1521         LASSERT(imp);
1522
1523         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
1524         if (req == NULL)
1525                 RETURN(-ENOMEM);
1526
1527         /* XXX: capa support? */
1528         /* osc_set_capa_size(req, &RMF_CAPA1, capa); */
1529         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
1530         if (rc) {
1531                 ptlrpc_request_free(req);
1532                 RETURN(rc);
1533         }
1534
1535         /*
1536          * XXX: decide how do we do here with resend
1537          * if we don't resend, then client may see wrong file size
1538          * if we do resend, then MDS thread can get stuck for quite long
1539          * and if we don't resend, then client will also get -EWOULDBLOCK !!
1540          * (see LU-7975 and sanity/test_27F use cases)
1541          * but let's decide not to resend/delay this truncate request to OST
1542          * and allow Client to decide to resend, in a less agressive way from
1543          * after_reply(), by returning -EINPROGRESS instead of
1544          * -EAGAIN/-EWOULDBLOCK upon return from ptlrpc_queue_wait() at the
1545          * end of this routine
1546          */
1547         req->rq_no_resend = req->rq_no_delay = 1;
1548
1549         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1550         ptlrpc_at_set_req_timeout(req);
1551
1552         OBD_ALLOC_PTR(oa);
1553         if (oa == NULL)
1554                 GOTO(out, rc = -ENOMEM);
1555
1556         rc = fid_to_ostid(lu_object_fid(&dt->do_lu), &oa->o_oi);
1557         LASSERT(rc == 0);
1558         oa->o_size = size;
1559         oa->o_blocks = OBD_OBJECT_EOF;
1560         oa->o_valid = OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
1561                       OBD_MD_FLID | OBD_MD_FLGROUP;
1562
1563         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1564         LASSERT(body);
1565         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
1566
1567         /* XXX: capa support? */
1568         /* osc_pack_capa(req, body, capa); */
1569
1570         ptlrpc_request_set_replen(req);
1571
1572         rc = ptlrpc_queue_wait(req);
1573         if (rc) {
1574                 /* -EWOULDBLOCK/-EAGAIN means OST is unreachable at the moment
1575                  * since we have decided not to resend/delay, but this could
1576                  * lead to wrong size to be seen at Client side and even process
1577                  * trying to open to exit/fail if not itself handling -EAGAIN.
1578                  * So it should be better to return -EINPROGRESS instead and
1579                  * leave the decision to resend at Client side in after_reply()
1580                  */
1581                 if (rc == -EWOULDBLOCK) {
1582                         rc = -EINPROGRESS;
1583                         CDEBUG(D_HA, "returning -EINPROGRESS instead of "
1584                                "-EWOULDBLOCK/-EAGAIN to allow Client to "
1585                                "resend\n");
1586                 } else {
1587                         CERROR("can't punch object: %d\n", rc);
1588                 }
1589         }
1590 out:
1591         ptlrpc_req_finished(req);
1592         if (oa)
1593                 OBD_FREE_PTR(oa);
1594         RETURN(rc);
1595 }
1596
1597 /**
1598  * Initialize precreation functionality of OSP
1599  *
1600  * Prepares all the internal structures and starts the precreate thread
1601  *
1602  * \param[in] d         OSP device
1603  *
1604  * \retval 0            on success
1605  * \retval negative     negated errno on error
1606  */
1607 int osp_init_precreate(struct osp_device *d)
1608 {
1609         struct l_wait_info       lwi = { 0 };
1610         struct task_struct              *task;
1611
1612         ENTRY;
1613
1614         OBD_ALLOC_PTR(d->opd_pre);
1615         if (d->opd_pre == NULL)
1616                 RETURN(-ENOMEM);
1617
1618         /* initially precreation isn't ready */
1619         d->opd_pre_status = -EAGAIN;
1620         fid_zero(&d->opd_pre_used_fid);
1621         d->opd_pre_used_fid.f_oid = 1;
1622         fid_zero(&d->opd_pre_last_created_fid);
1623         d->opd_pre_last_created_fid.f_oid = 1;
1624         d->opd_pre_reserved = 0;
1625         d->opd_got_disconnected = 1;
1626         d->opd_pre_create_slow = 0;
1627         d->opd_pre_create_count = OST_MIN_PRECREATE;
1628         d->opd_pre_min_create_count = OST_MIN_PRECREATE;
1629         d->opd_pre_max_create_count = OST_MAX_PRECREATE;
1630         d->opd_reserved_mb_high = 0;
1631         d->opd_reserved_mb_low = 0;
1632
1633         spin_lock_init(&d->opd_pre_lock);
1634         init_waitqueue_head(&d->opd_pre_waitq);
1635         init_waitqueue_head(&d->opd_pre_user_waitq);
1636         init_waitqueue_head(&d->opd_pre_thread.t_ctl_waitq);
1637
1638         /*
1639          * Initialize statfs-related things
1640          */
1641         d->opd_statfs_maxage = 5; /* default update interval */
1642         d->opd_statfs_fresh_till = cfs_time_shift(-1000);
1643         CDEBUG(D_OTHER, "current %llu, fresh till %llu\n",
1644                (unsigned long long)cfs_time_current(),
1645                (unsigned long long)d->opd_statfs_fresh_till);
1646         cfs_timer_init(&d->opd_statfs_timer, osp_statfs_timer_cb, d);
1647
1648         /*
1649          * start thread handling precreation and statfs updates
1650          */
1651         task = kthread_run(osp_precreate_thread, d,
1652                            "osp-pre-%u-%u", d->opd_index, d->opd_group);
1653         if (IS_ERR(task)) {
1654                 CERROR("can't start precreate thread %ld\n", PTR_ERR(task));
1655                 RETURN(PTR_ERR(task));
1656         }
1657
1658         l_wait_event(d->opd_pre_thread.t_ctl_waitq,
1659                      osp_precreate_running(d) || osp_precreate_stopped(d),
1660                      &lwi);
1661
1662         RETURN(0);
1663 }
1664
1665 /**
1666  * Finish precreate functionality of OSP
1667  *
1668  *
1669  * Asks all the activity (the thread, update timer) to stop, then
1670  * wait till that is done.
1671  *
1672  * \param[in] d         OSP device
1673  */
1674 void osp_precreate_fini(struct osp_device *d)
1675 {
1676         struct ptlrpc_thread *thread;
1677
1678         ENTRY;
1679
1680         cfs_timer_disarm(&d->opd_statfs_timer);
1681
1682         if (d->opd_pre == NULL)
1683                 RETURN_EXIT;
1684
1685         thread = &d->opd_pre_thread;
1686
1687         thread->t_flags = SVC_STOPPING;
1688         wake_up(&d->opd_pre_waitq);
1689
1690         wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);
1691
1692         OBD_FREE_PTR(d->opd_pre);
1693         d->opd_pre = NULL;
1694
1695         EXIT;
1696 }
1697