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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Intel, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/osp/osp_sync.c
  37  *
  38  * Lustre OST Proxy Device
  39  *
  40  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.pershin@intel.com>
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include "osp_internal.h"
  50
  51 /*
  52  * there are two specific states to take care about:
  53  *
  54  * = import is disconnected =
  55  *
  56  * = import is inactive =
  57  *   in this case osp_declare_object_create() returns an error
  58  *
  59  */
  60
  61 /*
  62  * statfs
  63  */
  64 static inline int osp_statfs_need_update(struct osp_device *d)
  65 {
  66         return !cfs_time_before(cfs_time_current(),
  67                                 d->opd_statfs_fresh_till);
  68 }
  69
  70 static void osp_statfs_timer_cb(unsigned long _d)
  71 {
  72         struct osp_device *d = (struct osp_device *) _d;
  73
  74         LASSERT(d);
  75         cfs_waitq_signal(&d->opd_pre_waitq);
  76 }
  77
  78 static int osp_statfs_interpret(const struct lu_env *env,
  79                                 struct ptlrpc_request *req,
  80                                 union ptlrpc_async_args *aa, int rc)
  81 {
  82         struct obd_import       *imp = req->rq_import;
  83         struct obd_statfs       *msfs;
  84         struct osp_device       *d;
  85
  86         ENTRY;
  87
  88         aa = ptlrpc_req_async_args(req);
  89         d = aa->pointer_arg[0];
  90         LASSERT(d);
  91
  92         if (rc != 0)
  93                 GOTO(out, rc);
  94
  95         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
  96         if (msfs == NULL)
  97                 GOTO(out, rc = -EPROTO);
  98
  99         d->opd_statfs = *msfs;
 100
 101         osp_pre_update_status(d, rc);
 102
 103         /* schedule next update */
 104         d->opd_statfs_fresh_till = cfs_time_shift(d->opd_statfs_maxage);
 105         cfs_timer_arm(&d->opd_statfs_timer, d->opd_statfs_fresh_till);
 106         d->opd_statfs_update_in_progress = 0;
 107
 108         CDEBUG(D_CACHE, "updated statfs %p\n", d);
 109
 110         RETURN(0);
 111 out:
 112         /* couldn't update statfs, try again as soon as possible */
 113         cfs_waitq_signal(&d->opd_pre_waitq);
 114         if (req->rq_import_generation == imp->imp_generation)
 115                 CERROR("%s: couldn't update statfs: rc = %d\n",
 116                        d->opd_obd->obd_name, rc);
 117         RETURN(rc);
 118 }
 119
 120 static int osp_statfs_update(struct osp_device *d)
 121 {
 122         struct ptlrpc_request   *req;
 123         struct obd_import       *imp;
 124         union ptlrpc_async_args *aa;
 125         int                      rc;
 126
 127         ENTRY;
 128
 129         CDEBUG(D_CACHE, "going to update statfs\n");
 130
 131         imp = d->opd_obd->u.cli.cl_import;
 132         LASSERT(imp);
 133
 134         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
 135         if (req == NULL)
 136                 RETURN(-ENOMEM);
 137
 138         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
 139         if (rc) {
 140                 ptlrpc_request_free(req);
 141                 RETURN(rc);
 142         }
 143         ptlrpc_request_set_replen(req);
 144         req->rq_request_portal = OST_CREATE_PORTAL;
 145         ptlrpc_at_set_req_timeout(req);
 146
 147         req->rq_interpret_reply = (ptlrpc_interpterer_t)osp_statfs_interpret;
 148         aa = ptlrpc_req_async_args(req);
 149         aa->pointer_arg[0] = d;
 150
 151         /*
 152          * no updates till reply
 153          */
 154         cfs_timer_disarm(&d->opd_statfs_timer);
 155         d->opd_statfs_fresh_till = cfs_time_shift(obd_timeout * 1000);
 156         d->opd_statfs_update_in_progress = 1;
 157
 158         ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 159
 160         RETURN(0);
 161 }
 162
 163 /*
 164  * XXX: there might be a case where removed object(s) do not add free
 165  * space (empty object). if the number of such deletions is high, then
 166  * we can start to update statfs too often - a rpc storm
 167  * TODO: some throttling is needed
 168  */
 169 void osp_statfs_need_now(struct osp_device *d)
 170 {
 171         if (!d->opd_statfs_update_in_progress) {
 172                 /*
 173                  * if current status is -ENOSPC (lack of free space on OST)
 174                  * then we should poll OST immediately once object destroy
 175                  * is replied
 176                  */
 177                 d->opd_statfs_fresh_till = cfs_time_shift(-1);
 178                 cfs_timer_disarm(&d->opd_statfs_timer);
 179                 cfs_waitq_signal(&d->opd_pre_waitq);
 180         }
 181 }
 182
 183
 184 /*
 185  * OSP tries to maintain pool of available objects so that calls to create
 186  * objects don't block most of time
 187  *
 188  * each time OSP gets connected to OST, we should start from precreation cleanup
 189  */
 190 static inline int osp_precreate_running(struct osp_device *d)
 191 {
 192         return !!(d->opd_pre_thread.t_flags & SVC_RUNNING);
 193 }
 194
 195 static inline int osp_precreate_stopped(struct osp_device *d)
 196 {
 197         return !!(d->opd_pre_thread.t_flags & SVC_STOPPED);
 198 }
 199
 200 static inline int osp_precreate_near_empty_nolock(struct osp_device *d)
 201 {
 202         int window = d->opd_pre_last_created - d->opd_pre_used_id;
 203
 204         /* don't consider new precreation till OST is healty and
 205          * has free space */
 206         return ((window - d->opd_pre_reserved < d->opd_pre_grow_count / 2) &&
 207                 (d->opd_pre_status == 0));
 208 }
 209
 210 static inline int osp_precreate_near_empty(struct osp_device *d)
 211 {
 212         int rc;
 213
 214         /* XXX: do we really need locking here? */
 215         cfs_spin_lock(&d->opd_pre_lock);
 216         rc = osp_precreate_near_empty_nolock(d);
 217         cfs_spin_unlock(&d->opd_pre_lock);
 218         return rc;
 219 }
 220
 221 static int osp_precreate_send(struct osp_device *d)
 222 {
 223         struct ptlrpc_request   *req;
 224         struct obd_import       *imp;
 225         struct ost_body         *body;
 226         int                      rc, grow, diff;
 227
 228         ENTRY;
 229
 230         /* don't precreate new objects till OST healthy and has free space */
 231         if (unlikely(d->opd_pre_status)) {
 232                 CDEBUG(D_INFO, "%s: don't send new precreate: rc = %d\n",
 233                        d->opd_obd->obd_name, d->opd_pre_status);
 234                 RETURN(0);
 235         }
 236
 237         /*
 238          * if not connection/initialization is compeleted, ignore
 239          */
 240         imp = d->opd_obd->u.cli.cl_import;
 241         LASSERT(imp);
 242
 243         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 244         if (req == NULL)
 245                 RETURN(-ENOMEM);
 246         req->rq_request_portal = OST_CREATE_PORTAL;
 247         /* we should not resend create request - anyway we will have delorphan
 248          * and kill these objects */
 249         req->rq_no_delay = req->rq_no_resend = 1;
 250
 251         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 252         if (rc) {
 253                 ptlrpc_request_free(req);
 254                 RETURN(rc);
 255         }
 256
 257         cfs_spin_lock(&d->opd_pre_lock);
 258         if (d->opd_pre_grow_count > d->opd_pre_max_grow_count / 2)
 259                 d->opd_pre_grow_count = d->opd_pre_max_grow_count / 2;
 260         grow = d->opd_pre_grow_count;
 261         cfs_spin_unlock(&d->opd_pre_lock);
 262
 263         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 264         LASSERT(body);
 265         body->oa.o_id = d->opd_pre_last_created + grow;
 266         body->oa.o_seq = FID_SEQ_OST_MDT0; /* XXX: support for CMD? */
 267         body->oa.o_valid = OBD_MD_FLGROUP;
 268
 269         ptlrpc_request_set_replen(req);
 270
 271         rc = ptlrpc_queue_wait(req);
 272         if (rc) {
 273                 CERROR("%s: can't precreate: rc = %d\n",
 274                        d->opd_obd->obd_name, rc);
 275                 GOTO(out_req, rc);
 276         }
 277         LASSERT(req->rq_transno == 0);
 278
 279         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 280         if (body == NULL)
 281                 GOTO(out_req, rc = -EPROTO);
 282
 283         CDEBUG(D_HA, "new last_created %lu\n", (unsigned long) body->oa.o_id);
 284         LASSERT(body->oa.o_id > d->opd_pre_used_id);
 285
 286         diff = body->oa.o_id - d->opd_pre_last_created;
 287
 288         cfs_spin_lock(&d->opd_pre_lock);
 289         if (diff < grow) {
 290                 /* the OST has not managed to create all the
 291                  * objects we asked for */
 292                 d->opd_pre_grow_count = max(diff, OST_MIN_PRECREATE);
 293                 d->opd_pre_grow_slow = 1;
 294         } else {
 295                 /* the OST is able to keep up with the work,
 296                  * we could consider increasing grow_count
 297                  * next time if needed */
 298                 d->opd_pre_grow_slow = 0;
 299         }
 300         d->opd_pre_last_created = body->oa.o_id;
 301         cfs_spin_unlock(&d->opd_pre_lock);
 302         CDEBUG(D_OTHER, "current precreated pool: %llu-%llu\n",
 303                d->opd_pre_used_id, d->opd_pre_last_created);
 304
 305 out_req:
 306         /* now we can wakeup all users awaiting for objects */
 307         osp_pre_update_status(d, rc);
 308         cfs_waitq_signal(&d->opd_pre_user_waitq);
 309
 310         ptlrpc_req_finished(req);
 311         RETURN(rc);
 312 }
 313
 314
 315 static int osp_get_lastid_from_ost(struct osp_device *d)
 316 {
 317         struct ptlrpc_request   *req;
 318         struct obd_import       *imp;
 319         obd_id                  *reply;
 320         char                    *tmp;
 321         int                      rc;
 322
 323         imp = d->opd_obd->u.cli.cl_import;
 324         LASSERT(imp);
 325
 326         req = ptlrpc_request_alloc(imp, &RQF_OST_GET_INFO_LAST_ID);
 327         if (req == NULL)
 328                 RETURN(-ENOMEM);
 329
 330         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
 331                              RCL_CLIENT, sizeof(KEY_LAST_ID));
 332         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
 333         if (rc) {
 334                 ptlrpc_request_free(req);
 335                 RETURN(rc);
 336         }
 337
 338         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
 339         memcpy(tmp, KEY_LAST_ID, sizeof(KEY_LAST_ID));
 340
 341         req->rq_no_delay = req->rq_no_resend = 1;
 342         ptlrpc_request_set_replen(req);
 343         rc = ptlrpc_queue_wait(req);
 344         if (rc) {
 345                 /* bad-bad OST.. let sysadm sort this out */
 346                 ptlrpc_set_import_active(imp, 0);
 347                 GOTO(out, rc);
 348         }
 349
 350         reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
 351         if (reply == NULL)
 352                 GOTO(out, rc = -EPROTO);
 353
 354         d->opd_last_used_id = *reply;
 355         CDEBUG(D_HA, "%s: got last_id "LPU64" from OST\n",
 356                d->opd_obd->obd_name, d->opd_last_used_id);
 357
 358 out:
 359         ptlrpc_req_finished(req);
 360         RETURN(rc);
 361
 362 }
 363
 364 /**
 365  * asks OST to clean precreate orphans
 366  * and gets next id for new objects
 367  */
 368 static int osp_precreate_cleanup_orphans(struct osp_device *d)
 369 {
 370         struct ptlrpc_request   *req = NULL;
 371         struct obd_import       *imp;
 372         struct ost_body         *body;
 373         int                      rc;
 374
 375         ENTRY;
 376
 377         LASSERT(d->opd_recovery_completed);
 378         LASSERT(d->opd_pre_reserved == 0);
 379
 380         CDEBUG(D_HA, "%s: going to cleanup orphans since "LPU64"\n",
 381                 d->opd_obd->obd_name, d->opd_last_used_id);
 382
 383         if (d->opd_last_used_id < 2) {
 384                 /* lastid looks strange... ask OST */
 385                 rc = osp_get_lastid_from_ost(d);
 386                 if (rc)
 387                         GOTO(out, rc);
 388         }
 389
 390         imp = d->opd_obd->u.cli.cl_import;
 391         LASSERT(imp);
 392
 393         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 394         if (req == NULL)
 395                 GOTO(out, rc = -ENOMEM);
 396
 397         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 398         if (rc) {
 399                 ptlrpc_request_free(req);
 400                 GOTO(out, rc);
 401         }
 402
 403         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 404         if (body == NULL)
 405                 GOTO(out, rc = -EPROTO);
 406
 407         body->oa.o_flags = OBD_FL_DELORPHAN;
 408         body->oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
 409         body->oa.o_seq = FID_SEQ_OST_MDT0;
 410
 411         body->oa.o_id = d->opd_last_used_id;
 412
 413         ptlrpc_request_set_replen(req);
 414
 415         /* Don't resend the delorphan req */
 416         req->rq_no_resend = req->rq_no_delay = 1;
 417
 418         rc = ptlrpc_queue_wait(req);
 419         if (rc)
 420                 GOTO(out, rc);
 421
 422         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 423         if (body == NULL)
 424                 GOTO(out, rc = -EPROTO);
 425
 426         /*
 427          * OST provides us with id new pool starts from in body->oa.o_id
 428          */
 429         cfs_spin_lock(&d->opd_pre_lock);
 430         if (le64_to_cpu(d->opd_last_used_id) > body->oa.o_id) {
 431                 d->opd_pre_grow_count = OST_MIN_PRECREATE +
 432                                         le64_to_cpu(d->opd_last_used_id) -
 433                                         body->oa.o_id;
 434                 d->opd_pre_last_created = le64_to_cpu(d->opd_last_used_id) + 1;
 435         } else {
 436                 d->opd_pre_grow_count = OST_MIN_PRECREATE;
 437                 d->opd_pre_last_created = body->oa.o_id + 1;
 438         }
 439         d->opd_pre_used_id = d->opd_pre_last_created - 1;
 440         d->opd_pre_grow_slow = 0;
 441         cfs_spin_unlock(&d->opd_pre_lock);
 442
 443         CDEBUG(D_HA, "Got last_id "LPU64" from OST, last_used is "LPU64
 444                ", next "LPU64"\n", body->oa.o_id,
 445                le64_to_cpu(d->opd_last_used_id), d->opd_pre_used_id);
 446
 447 out:
 448         if (req)
 449                 ptlrpc_req_finished(req);
 450
 451         RETURN(rc);
 452 }
 453
 454 /*
 455  * the function updates current precreation status used: functional or not
 456  *
 457  * rc is a last code from the transport, rc == 0 meaning transport works
 458  * well and users of lod can use objects from this OSP
 459  *
 460  * the status depends on current usage of OST
 461  */
 462 void osp_pre_update_status(struct osp_device *d, int rc)
 463 {
 464         struct obd_statfs       *msfs = &d->opd_statfs;
 465         int                      old = d->opd_pre_status;
 466         __u64                    used;
 467
 468         d->opd_pre_status = rc;
 469         if (rc)
 470                 goto out;
 471
 472         /* Add a bit of hysteresis so this flag isn't continually flapping,
 473          * and ensure that new files don't get extremely fragmented due to
 474          * only a small amount of available space in the filesystem.
 475          * We want to set the NOSPC flag when there is less than ~0.1% free
 476          * and clear it when there is at least ~0.2% free space, so:
 477          *                   avail < ~0.1% max          max = avail + used
 478          *            1025 * avail < avail + used       used = blocks - free
 479          *            1024 * avail < used
 480          *            1024 * avail < blocks - free
 481          *                   avail < ((blocks - free) >> 10)
 482          *
 483          * On very large disk, say 16TB 0.1% will be 16 GB. We don't want to
 484          * lose that amount of space so in those cases we report no space left
 485          * if their is less than 1 GB left.                             */
 486         if (likely(msfs->os_type)) {
 487                 used = min_t(__u64, (msfs->os_blocks - msfs->os_bfree) >> 10,
 488                                     1 << 30);
 489                 if ((msfs->os_ffree < 32) || (msfs->os_bavail < used)) {
 490                         d->opd_pre_status = -ENOSPC;
 491                         if (old != -ENOSPC)
 492                                 CDEBUG(D_INFO, "%s: status: "LPU64" blocks, "
 493                                        LPU64" free, "LPU64" used, "LPU64" "
 494                                        "avail -> %d: rc = %d\n",
 495                                        d->opd_obd->obd_name, msfs->os_blocks,
 496                                        msfs->os_bfree, used, msfs->os_bavail,
 497                                        d->opd_pre_status, rc);
 498                         CDEBUG(D_INFO,
 499                                "non-commited changes: %lu, in progress: %u\n",
 500                                d->opd_syn_changes, d->opd_syn_rpc_in_progress);
 501                 } else if (old == -ENOSPC) {
 502                         d->opd_pre_status = 0;
 503                         d->opd_pre_grow_slow = 0;
 504                         d->opd_pre_grow_count = OST_MIN_PRECREATE;
 505                         cfs_waitq_signal(&d->opd_pre_waitq);
 506                         CDEBUG(D_INFO, "%s: no space: "LPU64" blocks, "LPU64
 507                                " free, "LPU64" used, "LPU64" avail -> %d: "
 508                                "rc = %d\n", d->opd_obd->obd_name,
 509                                msfs->os_blocks, msfs->os_bfree, used,
 510                                msfs->os_bavail, d->opd_pre_status, rc);
 511                 }
 512         }
 513
 514 out:
 515         cfs_waitq_signal(&d->opd_pre_user_waitq);
 516 }
 517
 518 static int osp_precreate_thread(void *_arg)
 519 {
 520         struct osp_device       *d = _arg;
 521         struct ptlrpc_thread    *thread = &d->opd_pre_thread;
 522         struct l_wait_info       lwi = { 0 };
 523         char                     pname[16];
 524         int                      rc;
 525
 526         ENTRY;
 527
 528         sprintf(pname, "osp-pre-%u\n", d->opd_index);
 529         cfs_daemonize(pname);
 530
 531         cfs_spin_lock(&d->opd_pre_lock);
 532         thread->t_flags = SVC_RUNNING;
 533         cfs_spin_unlock(&d->opd_pre_lock);
 534         cfs_waitq_signal(&thread->t_ctl_waitq);
 535
 536         while (osp_precreate_running(d)) {
 537                 /*
 538                  * need to be connected to OST
 539                  */
 540                 while (osp_precreate_running(d)) {
 541                         l_wait_event(d->opd_pre_waitq,
 542                                      !osp_precreate_running(d) ||
 543                                      d->opd_new_connection, &lwi);
 544
 545                         if (!osp_precreate_running(d))
 546                                 break;
 547
 548                         if (!d->opd_new_connection)
 549                                 continue;
 550
 551                         /* got connected */
 552                         d->opd_new_connection = 0;
 553                         d->opd_got_disconnected = 0;
 554                         break;
 555                 }
 556
 557                 osp_statfs_update(d);
 558
 559                 /*
 560                  * wait for local recovery to finish, so we can cleanup orphans
 561                  * orphans are all objects since "last used" (assigned), but
 562                  * there might be objects reserved and in some cases they won't
 563                  * be used. we can't cleanup them till we're sure they won't be
 564                  * used. so we block new reservations and wait till all reserved
 565                  * objects either user or released.
 566                  */
 567                 l_wait_event(d->opd_pre_waitq, (!d->opd_pre_reserved &&
 568                                                 d->opd_recovery_completed) ||
 569                              !osp_precreate_running(d) ||
 570                              d->opd_got_disconnected, &lwi);
 571
 572                 if (osp_precreate_running(d) && !d->opd_got_disconnected) {
 573                         rc = osp_precreate_cleanup_orphans(d);
 574                         if (rc) {
 575                                 CERROR("%s: cannot cleanup orphans: rc = %d\n",
 576                                        d->opd_obd->obd_name,  rc);
 577                                 /* we can't proceed from here, OST seem to
 578                                  * be in a bad shape, better to wait for
 579                                  * a new instance of the server and repeat
 580                                  * from the beginning. notify possible waiters
 581                                  * this OSP isn't quite functional yet */
 582                                 osp_pre_update_status(d, rc);
 583                                 cfs_waitq_signal(&d->opd_pre_user_waitq);
 584                                 l_wait_event(d->opd_pre_waitq,
 585                                              !osp_precreate_running(d) ||
 586                                              d->opd_new_connection, &lwi);
 587                                 continue;
 588
 589                         }
 590                 }
 591
 592                 /*
 593                  * connected, can handle precreates now
 594                  */
 595                 while (osp_precreate_running(d)) {
 596                         l_wait_event(d->opd_pre_waitq,
 597                                      !osp_precreate_running(d) ||
 598                                      osp_precreate_near_empty(d) ||
 599                                      osp_statfs_need_update(d) ||
 600                                      d->opd_got_disconnected, &lwi);
 601
 602                         if (!osp_precreate_running(d))
 603                                 break;
 604
 605                         /* something happened to the connection
 606                          * have to start from the beginning */
 607                         if (d->opd_got_disconnected)
 608                                 break;
 609
 610                         if (osp_statfs_need_update(d))
 611                                 osp_statfs_update(d);
 612
 613                         if (osp_precreate_near_empty(d)) {
 614                                 rc = osp_precreate_send(d);
 615                                 /* osp_precreate_send() sets opd_pre_status
 616                                  * in case of error, that prevent the using of
 617                                  * failed device. */
 618                                 if (rc != 0 && rc != -ENOSPC &&
 619                                     rc != -ETIMEDOUT && rc != -ENOTCONN)
 620                                         CERROR("%s: cannot precreate objects:"
 621                                                " rc = %d\n",
 622                                                d->opd_obd->obd_name, rc);
 623                         }
 624                 }
 625         }
 626
 627         thread->t_flags = SVC_STOPPED;
 628         cfs_waitq_signal(&thread->t_ctl_waitq);
 629
 630         RETURN(0);
 631 }
 632
 633 static int osp_precreate_ready_condition(struct osp_device *d)
 634 {
 635         __u64 next;
 636
 637         /* ready if got enough precreated objects */
 638         /* we need to wait for others (opd_pre_reserved) and our object (+1) */
 639         next = d->opd_pre_used_id + d->opd_pre_reserved + 1;
 640         if (next <= d->opd_pre_last_created)
 641                 return 1;
 642
 643         /* ready if OST reported no space and no destoys in progress */
 644         if (d->opd_syn_changes + d->opd_syn_rpc_in_progress == 0 &&
 645             d->opd_pre_status != 0)
 646                 return 1;
 647
 648         return 0;
 649 }
 650
 651 static int osp_precreate_timeout_condition(void *data)
 652 {
 653         struct osp_device *d = data;
 654
 655         LCONSOLE_WARN("%s: slow creates, last="LPU64", next="LPU64", "
 656                       "reserved="LPU64", syn_changes=%lu, "
 657                       "syn_rpc_in_progress=%d, status=%d\n",
 658                       d->opd_obd->obd_name, d->opd_pre_last_created,
 659                       d->opd_pre_used_id, d->opd_pre_reserved,
 660                       d->opd_syn_changes, d->opd_syn_rpc_in_progress,
 661                       d->opd_pre_status);
 662
 663         return 0;
 664 }
 665
 666 /*
 667  * called to reserve object in the pool
 668  * return codes:
 669  *  ENOSPC - no space on corresponded OST
 670  *  EAGAIN - precreation is in progress, try later
 671  *  EIO    - no access to OST
 672  */
 673 int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d)
 674 {
 675         struct l_wait_info       lwi;
 676         cfs_time_t               expire = cfs_time_shift(obd_timeout);
 677         int                      precreated, rc;
 678         int                      count = 0;
 679
 680         ENTRY;
 681
 682         LASSERT(d->opd_pre_last_created >= d->opd_pre_used_id);
 683
 684         lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
 685                           osp_precreate_timeout_condition, d);
 686
 687         /*
 688          * wait till:
 689          *  - preallocation is done
 690          *  - no free space expected soon
 691          *  - can't connect to OST for too long (obd_timeout)
 692          */
 693         while ((rc = d->opd_pre_status) == 0 || rc == -ENOSPC ||
 694                 rc == -ENODEV) {
 695                 if (unlikely(rc == -ENODEV)) {
 696                         if (cfs_time_aftereq(cfs_time_current(), expire))
 697                                 break;
 698                 }
 699
 700 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 90, 0)
 701                 /*
 702                  * to address Andreas's concern on possible busy-loop
 703                  * between this thread and osp_precreate_send()
 704                  */
 705                 if (unlikely(count++ == 1000)) {
 706                         osp_precreate_timeout_condition(d);
 707                         LBUG();
 708                 }
 709 #endif
 710
 711                 /*
 712                  * increase number of precreations
 713                  */
 714                 if (d->opd_pre_grow_count < d->opd_pre_max_grow_count &&
 715                     d->opd_pre_grow_slow == 0 &&
 716                     (d->opd_pre_last_created - d->opd_pre_used_id <=
 717                      d->opd_pre_grow_count / 4 + 1)) {
 718                         cfs_spin_lock(&d->opd_pre_lock);
 719                         d->opd_pre_grow_slow = 1;
 720                         d->opd_pre_grow_count *= 2;
 721                         cfs_spin_unlock(&d->opd_pre_lock);
 722                 }
 723
 724                 /*
 725                  * we never use the last object in the window
 726                  */
 727                 cfs_spin_lock(&d->opd_pre_lock);
 728                 precreated = d->opd_pre_last_created - d->opd_pre_used_id;
 729                 if (precreated > d->opd_pre_reserved) {
 730                         d->opd_pre_reserved++;
 731                         cfs_spin_unlock(&d->opd_pre_lock);
 732                         rc = 0;
 733
 734                         /* XXX: don't wake up if precreation is in progress */
 735                         if (osp_precreate_near_empty_nolock(d))
 736                                 cfs_waitq_signal(&d->opd_pre_waitq);
 737
 738                         break;
 739                 }
 740                 cfs_spin_unlock(&d->opd_pre_lock);
 741
 742                 /*
 743                  * all precreated objects have been used and no-space
 744                  * status leave us no chance to succeed very soon
 745                  * but if there is destroy in progress, then we should
 746                  * wait till that is done - some space might be released
 747                  */
 748                 if (unlikely(rc == -ENOSPC)) {
 749                         if (d->opd_syn_changes) {
 750                                 /* force local commit to release space */
 751                                 dt_commit_async(env, d->opd_storage);
 752                         }
 753                         if (d->opd_syn_rpc_in_progress) {
 754                                 /* just wait till destroys are done */
 755                                 /* see l_wait_even() few lines below */
 756                         }
 757                         if (d->opd_syn_changes +
 758                             d->opd_syn_rpc_in_progress == 0) {
 759                                 /* no hope for free space */
 760                                 break;
 761                         }
 762                 }
 763
 764                 /* XXX: don't wake up if precreation is in progress */
 765                 cfs_waitq_signal(&d->opd_pre_waitq);
 766
 767                 l_wait_event(d->opd_pre_user_waitq,
 768                              osp_precreate_ready_condition(d), &lwi);
 769         }
 770
 771         RETURN(rc);
 772 }
 773
 774 /*
 775  * this function relies on reservation made before
 776  */
 777 __u64 osp_precreate_get_id(struct osp_device *d)
 778 {
 779         obd_id objid;
 780
 781         /* grab next id from the pool */
 782         cfs_spin_lock(&d->opd_pre_lock);
 783         LASSERT(d->opd_pre_used_id < d->opd_pre_last_created);
 784         objid = ++d->opd_pre_used_id;
 785         d->opd_pre_reserved--;
 786         /*
 787          * last_used_id must be changed along with getting new id otherwise
 788          * we might miscalculate gap causing object loss or leak
 789          */
 790         osp_update_last_id(d, objid);
 791         cfs_spin_unlock(&d->opd_pre_lock);
 792
 793         /*
 794          * probably main thread suspended orphan cleanup till
 795          * all reservations are released, see comment in
 796          * osp_precreate_thread() just before orphan cleanup
 797          */
 798         if (unlikely(d->opd_pre_reserved == 0 && d->opd_pre_status))
 799                 cfs_waitq_signal(&d->opd_pre_waitq);
 800
 801         return objid;
 802 }
 803
 804 /*
 805  *
 806  */
 807 int osp_object_truncate(const struct lu_env *env, struct dt_object *dt,
 808                         __u64 size)
 809 {
 810         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 811         struct ptlrpc_request   *req = NULL;
 812         struct obd_import       *imp;
 813         struct ost_body         *body;
 814         struct obdo             *oa = NULL;
 815         int                      rc;
 816
 817         ENTRY;
 818
 819         imp = d->opd_obd->u.cli.cl_import;
 820         LASSERT(imp);
 821
 822         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
 823         if (req == NULL)
 824                 RETURN(-ENOMEM);
 825
 826         /* XXX: capa support? */
 827         /* osc_set_capa_size(req, &RMF_CAPA1, capa); */
 828         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 829         if (rc) {
 830                 ptlrpc_request_free(req);
 831                 RETURN(rc);
 832         }
 833
 834         /*
 835          * XXX: decide how do we do here with resend
 836          * if we don't resend, then client may see wrong file size
 837          * if we do resend, then MDS thread can get stuck for quite long
 838          */
 839         req->rq_no_resend = req->rq_no_delay = 1;
 840
 841         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 842         ptlrpc_at_set_req_timeout(req);
 843
 844         OBD_ALLOC_PTR(oa);
 845         if (oa == NULL)
 846                 GOTO(out, rc = -ENOMEM);
 847
 848         rc = fid_ostid_pack(lu_object_fid(&dt->do_lu), &oa->o_oi);
 849         LASSERT(rc == 0);
 850         oa->o_size = size;
 851         oa->o_blocks = OBD_OBJECT_EOF;
 852         oa->o_valid = OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 853                       OBD_MD_FLID | OBD_MD_FLGROUP;
 854
 855         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 856         LASSERT(body);
 857         lustre_set_wire_obdo(&body->oa, oa);
 858
 859         /* XXX: capa support? */
 860         /* osc_pack_capa(req, body, capa); */
 861
 862         ptlrpc_request_set_replen(req);
 863
 864         rc = ptlrpc_queue_wait(req);
 865         if (rc)
 866                 CERROR("can't punch object: %d\n", rc);
 867 out:
 868         ptlrpc_req_finished(req);
 869         if (oa)
 870                 OBD_FREE_PTR(oa);
 871         RETURN(rc);
 872 }
 873
 874 int osp_init_precreate(struct osp_device *d)
 875 {
 876         struct l_wait_info       lwi = { 0 };
 877         int                      rc;
 878
 879         ENTRY;
 880
 881         /* initially precreation isn't ready */
 882         d->opd_pre_status = -EAGAIN;
 883         d->opd_pre_used_id = 0;
 884         d->opd_pre_last_created = 0;
 885         d->opd_pre_reserved = 0;
 886         d->opd_got_disconnected = 1;
 887         d->opd_pre_grow_slow = 0;
 888         d->opd_pre_grow_count = OST_MIN_PRECREATE;
 889         d->opd_pre_min_grow_count = OST_MIN_PRECREATE;
 890         d->opd_pre_max_grow_count = OST_MAX_PRECREATE;
 891
 892         cfs_spin_lock_init(&d->opd_pre_lock);
 893         cfs_waitq_init(&d->opd_pre_waitq);
 894         cfs_waitq_init(&d->opd_pre_user_waitq);
 895         cfs_waitq_init(&d->opd_pre_thread.t_ctl_waitq);
 896
 897         /*
 898          * Initialize statfs-related things
 899          */
 900         d->opd_statfs_maxage = 5; /* default update interval */
 901         d->opd_statfs_fresh_till = cfs_time_shift(-1000);
 902         CDEBUG(D_OTHER, "current %llu, fresh till %llu\n",
 903                (unsigned long long)cfs_time_current(),
 904                (unsigned long long)d->opd_statfs_fresh_till);
 905         cfs_timer_init(&d->opd_statfs_timer, osp_statfs_timer_cb, d);
 906
 907         /*
 908          * start thread handling precreation and statfs updates
 909          */
 910         rc = cfs_create_thread(osp_precreate_thread, d, 0);
 911         if (rc < 0) {
 912                 CERROR("can't start precreate thread %d\n", rc);
 913                 RETURN(rc);
 914         }
 915
 916         l_wait_event(d->opd_pre_thread.t_ctl_waitq,
 917                      osp_precreate_running(d) || osp_precreate_stopped(d),
 918                      &lwi);
 919
 920         RETURN(0);
 921 }
 922
 923 void osp_precreate_fini(struct osp_device *d)
 924 {
 925         struct ptlrpc_thread *thread = &d->opd_pre_thread;
 926
 927         ENTRY;
 928
 929         cfs_timer_disarm(&d->opd_statfs_timer);
 930
 931         thread->t_flags = SVC_STOPPING;
 932         cfs_waitq_signal(&d->opd_pre_waitq);
 933
 934         cfs_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);
 935
 936         EXIT;
 937 }
 938