lustre/osp/osp_precreate.c

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