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         ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 152
 153         cfs_timer_disarm(&d->opd_statfs_timer);
 154
 155         /*
 156          * no updates till reply
 157          */
 158         d->opd_statfs_fresh_till = cfs_time_shift(obd_timeout * 1000);
 159         d->opd_statfs_update_in_progress = 1;
 160
 161         RETURN(0);
 162 }
 163
 164 /*
 165  * XXX: there might be a case where removed object(s) do not add free
 166  * space (empty object). if the number of such deletions is high, then
 167  * we can start to update statfs too often - a rpc storm
 168  * TODO: some throttling is needed
 169  */
 170 void osp_statfs_need_now(struct osp_device *d)
 171 {
 172         if (!d->opd_statfs_update_in_progress) {
 173                 /*
 174                  * if current status is -ENOSPC (lack of free space on OST)
 175                  * then we should poll OST immediately once object destroy
 176                  * is replied
 177                  */
 178                 d->opd_statfs_fresh_till = cfs_time_shift(-1);
 179                 cfs_timer_disarm(&d->opd_statfs_timer);
 180                 cfs_waitq_signal(&d->opd_pre_waitq);
 181         }
 182 }
 183
 184
 185 /*
 186  * OSP tries to maintain pool of available objects so that calls to create
 187  * objects don't block most of time
 188  *
 189  * each time OSP gets connected to OST, we should start from precreation cleanup
 190  */
 191 static inline int osp_precreate_running(struct osp_device *d)
 192 {
 193         return !!(d->opd_pre_thread.t_flags & SVC_RUNNING);
 194 }
 195
 196 static inline int osp_precreate_stopped(struct osp_device *d)
 197 {
 198         return !!(d->opd_pre_thread.t_flags & SVC_STOPPED);
 199 }
 200
 201 static inline int osp_precreate_near_empty_nolock(struct osp_device *d)
 202 {
 203         int window = d->opd_pre_last_created - d->opd_pre_next;
 204
 205         /* don't consider new precreation till OST is healty and
 206          * has free space */
 207         return ((window - d->opd_pre_reserved < d->opd_pre_grow_count / 2) &&
 208                 (d->opd_pre_status == 0));
 209 }
 210
 211 static inline int osp_precreate_near_empty(struct osp_device *d)
 212 {
 213         int rc;
 214
 215         /* XXX: do we really need locking here? */
 216         cfs_spin_lock(&d->opd_pre_lock);
 217         rc = osp_precreate_near_empty_nolock(d);
 218         cfs_spin_unlock(&d->opd_pre_lock);
 219         return rc;
 220 }
 221
 222 static int osp_precreate_send(struct osp_device *d)
 223 {
 224         struct ptlrpc_request   *req;
 225         struct obd_import       *imp;
 226         struct ost_body         *body;
 227         int                      rc, grow, diff;
 228
 229         ENTRY;
 230
 231         /* don't precreate new objects till OST healthy and has free space */
 232         if (unlikely(d->opd_pre_status)) {
 233                 CDEBUG(D_INFO, "%s: don't send new precreate: rc = %d\n",
 234                        d->opd_obd->obd_name, d->opd_pre_status);
 235                 RETURN(0);
 236         }
 237
 238         /*
 239          * if not connection/initialization is compeleted, ignore
 240          */
 241         imp = d->opd_obd->u.cli.cl_import;
 242         LASSERT(imp);
 243
 244         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 245         if (req == NULL)
 246                 RETURN(-ENOMEM);
 247         req->rq_request_portal = OST_CREATE_PORTAL;
 248         /* we should not resend create request - anyway we will have delorphan
 249          * and kill these objects */
 250         req->rq_no_delay = req->rq_no_resend = 1;
 251
 252         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 253         if (rc) {
 254                 ptlrpc_request_free(req);
 255                 RETURN(rc);
 256         }
 257
 258         cfs_spin_lock(&d->opd_pre_lock);
 259         if (d->opd_pre_grow_count > d->opd_pre_max_grow_count / 2)
 260                 d->opd_pre_grow_count = d->opd_pre_max_grow_count / 2;
 261         grow = d->opd_pre_grow_count;
 262         cfs_spin_unlock(&d->opd_pre_lock);
 263
 264         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 265         LASSERT(body);
 266         body->oa.o_id = d->opd_pre_last_created + grow;
 267         body->oa.o_seq = FID_SEQ_OST_MDT0; /* XXX: support for CMD? */
 268         body->oa.o_valid = OBD_MD_FLGROUP;
 269
 270         ptlrpc_request_set_replen(req);
 271
 272         rc = ptlrpc_queue_wait(req);
 273         if (rc) {
 274                 CERROR("%s: can't precreate: rc = %d\n",
 275                        d->opd_obd->obd_name, rc);
 276                 GOTO(out_req, rc);
 277         }
 278         LASSERT(req->rq_transno == 0);
 279
 280         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 281         if (body == NULL)
 282                 GOTO(out_req, rc = -EPROTO);
 283
 284         CDEBUG(D_HA, "new last_created %lu\n", (unsigned long) body->oa.o_id);
 285         LASSERT(body->oa.o_id > d->opd_pre_next);
 286
 287         diff = body->oa.o_id - d->opd_pre_last_created;
 288
 289         cfs_spin_lock(&d->opd_pre_lock);
 290         if (diff < grow) {
 291                 /* the OST has not managed to create all the
 292                  * objects we asked for */
 293                 d->opd_pre_grow_count = max(diff, OST_MIN_PRECREATE);
 294                 d->opd_pre_grow_slow = 1;
 295         } else {
 296                 /* the OST is able to keep up with the work,
 297                  * we could consider increasing grow_count
 298                  * next time if needed */
 299                 d->opd_pre_grow_slow = 0;
 300         }
 301         d->opd_pre_last_created = body->oa.o_id;
 302         cfs_spin_unlock(&d->opd_pre_lock);
 303         CDEBUG(D_OTHER, "current precreated pool: %llu-%llu\n",
 304                d->opd_pre_next, d->opd_pre_last_created);
 305
 306 out_req:
 307         /* now we can wakeup all users awaiting for objects */
 308         osp_pre_update_status(d, rc);
 309         cfs_waitq_signal(&d->opd_pre_user_waitq);
 310
 311         ptlrpc_req_finished(req);
 312         RETURN(rc);
 313 }
 314
 315 /**
 316  * asks OST to clean precreate orphans
 317  * and gets next id for new objects
 318  */
 319 static int osp_precreate_cleanup_orphans(struct osp_device *d)
 320 {
 321         struct ptlrpc_request   *req = NULL;
 322         struct obd_import       *imp;
 323         struct ost_body         *body;
 324         int                      rc;
 325
 326         ENTRY;
 327
 328         LASSERT(d->opd_recovery_completed);
 329         LASSERT(d->opd_pre_reserved == 0);
 330
 331         imp = d->opd_obd->u.cli.cl_import;
 332         LASSERT(imp);
 333
 334         req = ptlrpc_request_alloc(imp, &RQF_OST_CREATE);
 335         if (req == NULL)
 336                 RETURN(-ENOMEM);
 337
 338         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 339         if (rc) {
 340                 ptlrpc_request_free(req);
 341                 RETURN(rc);
 342         }
 343
 344         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 345         if (body == NULL)
 346                 GOTO(out_req, rc = -EPROTO);
 347
 348         body->oa.o_flags = OBD_FL_DELORPHAN;
 349         body->oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
 350         body->oa.o_seq = FID_SEQ_OST_MDT0;
 351
 352         /* remove from NEXT after used one */
 353         body->oa.o_id = d->opd_last_used_id + 1;
 354
 355         ptlrpc_request_set_replen(req);
 356
 357         /* Don't resend the delorphan req */
 358         req->rq_no_resend = req->rq_no_delay = 1;
 359
 360         rc = ptlrpc_queue_wait(req);
 361         if (rc) {
 362                 ptlrpc_set_import_active(imp, 0);
 363                 GOTO(out_req, rc);
 364         }
 365
 366         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 367         if (body == NULL)
 368                 GOTO(out_req, rc = -EPROTO);
 369
 370         /*
 371          * OST provides us with id new pool starts from in body->oa.o_id
 372          */
 373         cfs_spin_lock(&d->opd_pre_lock);
 374         if (le64_to_cpu(d->opd_last_used_id) > body->oa.o_id) {
 375                 d->opd_pre_grow_count = OST_MIN_PRECREATE +
 376                                         le64_to_cpu(d->opd_last_used_id) -
 377                                         body->oa.o_id;
 378                 d->opd_pre_last_created = le64_to_cpu(d->opd_last_used_id) + 1;
 379         } else {
 380                 d->opd_pre_grow_count = OST_MIN_PRECREATE;
 381                 d->opd_pre_last_created = body->oa.o_id + 1;
 382         }
 383         d->opd_pre_next = d->opd_pre_last_created;
 384         d->opd_pre_grow_slow = 0;
 385         cfs_spin_unlock(&d->opd_pre_lock);
 386
 387         /* now we can wakeup all users awaiting for objects */
 388         osp_pre_update_status(d, rc);
 389         cfs_waitq_signal(&d->opd_pre_user_waitq);
 390
 391         CDEBUG(D_HA, "Got last_id "LPU64" from OST, last_used is "LPU64
 392                ", next "LPU64"\n", body->oa.o_id,
 393                le64_to_cpu(d->opd_last_used_id), d->opd_pre_next);
 394
 395 out_req:
 396         ptlrpc_req_finished(req);
 397         RETURN(rc);
 398 }
 399
 400 /*
 401  * the function updates current precreation status used: functional or not
 402  *
 403  * rc is a last code from the transport, rc == 0 meaning transport works
 404  * well and users of lod can use objects from this OSP
 405  *
 406  * the status depends on current usage of OST
 407  */
 408 void osp_pre_update_status(struct osp_device *d, int rc)
 409 {
 410         struct obd_statfs       *msfs = &d->opd_statfs;
 411         int                      old = d->opd_pre_status;
 412         __u64                    used;
 413
 414         d->opd_pre_status = rc;
 415         if (rc)
 416                 goto out;
 417
 418         if (likely(msfs->os_type)) {
 419                 used = min_t(__u64, (msfs->os_blocks - msfs->os_bfree) >> 10,
 420                                     1 << 30);
 421                 if ((msfs->os_ffree < 32) || (msfs->os_bavail < used)) {
 422                         d->opd_pre_status = -ENOSPC;
 423                         if (old != -ENOSPC)
 424                                 CDEBUG(D_INFO, "%s: status: "LPU64" blocks, "
 425                                        LPU64" free, "LPU64" used, "LPU64" "
 426                                        "avail -> %d: rc = %d\n",
 427                                        d->opd_obd->obd_name, msfs->os_blocks,
 428                                        msfs->os_bfree, used, msfs->os_bavail,
 429                                        d->opd_pre_status, rc);
 430                         CDEBUG(D_INFO,
 431                                "non-commited changes: %lu, in progress: %u\n",
 432                                d->opd_syn_changes, d->opd_syn_rpc_in_progress);
 433                 } else if (old == -ENOSPC) {
 434                         d->opd_pre_status = 0;
 435                         d->opd_pre_grow_slow = 0;
 436                         d->opd_pre_grow_count = OST_MIN_PRECREATE;
 437                         cfs_waitq_signal(&d->opd_pre_waitq);
 438                         CDEBUG(D_INFO, "%s: no space: "LPU64" blocks, "LPU64
 439                                " free, "LPU64" used, "LPU64" avail -> %d: "
 440                                "rc = %d\n", d->opd_obd->obd_name,
 441                                msfs->os_blocks, msfs->os_bfree, used,
 442                                msfs->os_bavail, d->opd_pre_status, rc);
 443                 }
 444         }
 445
 446 out:
 447         cfs_waitq_signal(&d->opd_pre_user_waitq);
 448 }
 449
 450 static int osp_precreate_thread(void *_arg)
 451 {
 452         struct osp_device       *d = _arg;
 453         struct ptlrpc_thread    *thread = &d->opd_pre_thread;
 454         struct l_wait_info       lwi = { 0 };
 455         char                     pname[16];
 456         int                      rc;
 457
 458         ENTRY;
 459
 460         sprintf(pname, "osp-pre-%u\n", d->opd_index);
 461         cfs_daemonize(pname);
 462
 463         cfs_spin_lock(&d->opd_pre_lock);
 464         thread->t_flags = SVC_RUNNING;
 465         cfs_spin_unlock(&d->opd_pre_lock);
 466         cfs_waitq_signal(&thread->t_ctl_waitq);
 467
 468         while (osp_precreate_running(d)) {
 469                 /*
 470                  * need to be connected to OST
 471                  */
 472                 while (osp_precreate_running(d)) {
 473                         l_wait_event(d->opd_pre_waitq,
 474                                      !osp_precreate_running(d) ||
 475                                      d->opd_new_connection, &lwi);
 476
 477                         if (!osp_precreate_running(d))
 478                                 break;
 479
 480                         if (!d->opd_new_connection)
 481                                 continue;
 482
 483                         /* got connected */
 484                         d->opd_new_connection = 0;
 485                         d->opd_got_disconnected = 0;
 486                         break;
 487                 }
 488
 489                 osp_statfs_update(d);
 490
 491                 /*
 492                  * wait for local recovery to finish, so we can cleanup orphans
 493                  * orphans are all objects since "last used" (assigned), but
 494                  * there might be objects reserved and in some cases they won't
 495                  * be used. we can't cleanup them till we're sure they won't be
 496                  * used. so we block new reservations and wait till all reserved
 497                  * objects either user or released.
 498                  */
 499                 l_wait_event(d->opd_pre_waitq, (!d->opd_pre_reserved &&
 500                                                 d->opd_recovery_completed) ||
 501                              !osp_precreate_running(d) ||
 502                              d->opd_got_disconnected, &lwi);
 503
 504                 if (osp_precreate_running(d) && !d->opd_got_disconnected) {
 505                         rc = osp_precreate_cleanup_orphans(d);
 506                         if (rc) {
 507                                 CERROR("%s: cannot cleanup orphans: rc = %d\n",
 508                                        d->opd_obd->obd_name,  rc);
 509                         }
 510                 }
 511
 512                 /*
 513                  * connected, can handle precreates now
 514                  */
 515                 while (osp_precreate_running(d)) {
 516                         l_wait_event(d->opd_pre_waitq,
 517                                      !osp_precreate_running(d) ||
 518                                      osp_precreate_near_empty(d) ||
 519                                      osp_statfs_need_update(d) ||
 520                                      d->opd_got_disconnected, &lwi);
 521
 522                         if (!osp_precreate_running(d))
 523                                 break;
 524
 525                         /* something happened to the connection
 526                          * have to start from the beginning */
 527                         if (d->opd_got_disconnected)
 528                                 break;
 529
 530                         if (osp_statfs_need_update(d))
 531                                 osp_statfs_update(d);
 532
 533                         if (osp_precreate_near_empty(d)) {
 534                                 rc = osp_precreate_send(d);
 535                                 /* osp_precreate_send() sets opd_pre_status
 536                                  * in case of error, that prevent the using of
 537                                  * failed device. */
 538                                 if (rc != 0 && rc != -ENOSPC &&
 539                                     rc != -ETIMEDOUT && rc != -ENOTCONN)
 540                                         CERROR("%s: cannot precreate objects:"
 541                                                " rc = %d\n",
 542                                                d->opd_obd->obd_name, rc);
 543                         }
 544                 }
 545         }
 546
 547         thread->t_flags = SVC_STOPPED;
 548         cfs_waitq_signal(&thread->t_ctl_waitq);
 549
 550         RETURN(0);
 551 }
 552
 553 static int osp_precreate_ready_condition(struct osp_device *d)
 554 {
 555         /* ready if got enough precreated objects */
 556         if (d->opd_pre_next + d->opd_pre_reserved < d->opd_pre_last_created)
 557                 return 1;
 558
 559         /* ready if OST reported no space and no destoys in progress */
 560         if (d->opd_syn_changes + d->opd_syn_rpc_in_progress == 0 &&
 561             d->opd_pre_status != 0)
 562                 return 1;
 563
 564         return 0;
 565 }
 566
 567 static int osp_precreate_timeout_condition(void *data)
 568 {
 569         struct osp_device *d = data;
 570
 571         LCONSOLE_WARN("%s: slow creates, last="LPU64", next="LPU64", "
 572                       "reserved="LPU64", syn_changes=%lu, "
 573                       "syn_rpc_in_progress=%d, status=%d\n",
 574                       d->opd_obd->obd_name, d->opd_pre_last_created,
 575                       d->opd_pre_next, d->opd_pre_reserved,
 576                       d->opd_syn_changes, d->opd_syn_rpc_in_progress,
 577                       d->opd_pre_status);
 578
 579         return 0;
 580 }
 581
 582 /*
 583  * called to reserve object in the pool
 584  * return codes:
 585  *  ENOSPC - no space on corresponded OST
 586  *  EAGAIN - precreation is in progress, try later
 587  *  EIO    - no access to OST
 588  */
 589 int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d)
 590 {
 591         struct l_wait_info       lwi;
 592         cfs_time_t               expire = cfs_time_shift(obd_timeout);
 593         int                      precreated, rc;
 594
 595         ENTRY;
 596
 597         LASSERT(d->opd_pre_last_created >= d->opd_pre_next);
 598
 599         lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
 600                           osp_precreate_timeout_condition, d);
 601
 602         /*
 603          * wait till:
 604          *  - preallocation is done
 605          *  - no free space expected soon
 606          *  - can't connect to OST for too long (obd_timeout)
 607          */
 608         while ((rc = d->opd_pre_status) == 0 || rc == -ENOSPC ||
 609                 rc == -ENODEV) {
 610                 if (unlikely(rc == -ENODEV)) {
 611                         if (cfs_time_aftereq(cfs_time_current(), expire))
 612                                 break;
 613                 }
 614
 615                 /*
 616                  * increase number of precreations
 617                  */
 618                 if (d->opd_pre_grow_count < d->opd_pre_max_grow_count &&
 619                     d->opd_pre_grow_slow == 0 &&
 620                     (d->opd_pre_last_created - d->opd_pre_next <=
 621                      d->opd_pre_grow_count / 4 + 1)) {
 622                         cfs_spin_lock(&d->opd_pre_lock);
 623                         d->opd_pre_grow_slow = 1;
 624                         d->opd_pre_grow_count *= 2;
 625                         cfs_spin_unlock(&d->opd_pre_lock);
 626                 }
 627
 628                 /*
 629                  * we never use the last object in the window
 630                  */
 631                 cfs_spin_lock(&d->opd_pre_lock);
 632                 precreated = d->opd_pre_last_created - d->opd_pre_next;
 633                 if (precreated > d->opd_pre_reserved) {
 634                         d->opd_pre_reserved++;
 635                         cfs_spin_unlock(&d->opd_pre_lock);
 636                         rc = 0;
 637
 638                         /* XXX: don't wake up if precreation is in progress */
 639                         if (osp_precreate_near_empty_nolock(d))
 640                                 cfs_waitq_signal(&d->opd_pre_waitq);
 641
 642                         break;
 643                 }
 644                 cfs_spin_unlock(&d->opd_pre_lock);
 645
 646                 /*
 647                  * all precreated objects have been used and no-space
 648                  * status leave us no chance to succeed very soon
 649                  * but if there is destroy in progress, then we should
 650                  * wait till that is done - some space might be released
 651                  */
 652                 if (unlikely(rc == -ENOSPC)) {
 653                         if (d->opd_syn_changes) {
 654                                 /* force local commit to release space */
 655                                 dt_commit_async(env, d->opd_storage);
 656                         }
 657                         if (d->opd_syn_rpc_in_progress) {
 658                                 /* just wait till destroys are done */
 659                                 /* see l_wait_even() few lines below */
 660                         }
 661                         if (d->opd_syn_changes +
 662                             d->opd_syn_rpc_in_progress == 0) {
 663                                 /* no hope for free space */
 664                                 break;
 665                         }
 666                 }
 667
 668                 /* XXX: don't wake up if precreation is in progress */
 669                 cfs_waitq_signal(&d->opd_pre_waitq);
 670
 671                 l_wait_event(d->opd_pre_user_waitq,
 672                              osp_precreate_ready_condition(d), &lwi);
 673         }
 674
 675         RETURN(rc);
 676 }
 677
 678 /*
 679  * this function relies on reservation made before
 680  */
 681 __u64 osp_precreate_get_id(struct osp_device *d)
 682 {
 683         obd_id objid;
 684
 685         /* grab next id from the pool */
 686         cfs_spin_lock(&d->opd_pre_lock);
 687         LASSERT(d->opd_pre_next <= d->opd_pre_last_created);
 688         objid = d->opd_pre_next++;
 689         d->opd_pre_reserved--;
 690         /*
 691          * last_used_id must be changed along with getting new id otherwise
 692          * we might miscalculate gap causing object loss or leak
 693          */
 694         osp_update_last_id(d, objid);
 695         cfs_spin_unlock(&d->opd_pre_lock);
 696
 697         /*
 698          * probably main thread suspended orphan cleanup till
 699          * all reservations are released, see comment in
 700          * osp_precreate_thread() just before orphan cleanup
 701          */
 702         if (unlikely(d->opd_pre_reserved == 0 && d->opd_pre_status))
 703                 cfs_waitq_signal(&d->opd_pre_waitq);
 704
 705         return objid;
 706 }
 707
 708 /*
 709  *
 710  */
 711 int osp_object_truncate(const struct lu_env *env, struct dt_object *dt,
 712                         __u64 size)
 713 {
 714         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 715         struct ptlrpc_request   *req = NULL;
 716         struct obd_import       *imp;
 717         struct ost_body         *body;
 718         struct obdo             *oa = NULL;
 719         int                      rc;
 720
 721         ENTRY;
 722
 723         imp = d->opd_obd->u.cli.cl_import;
 724         LASSERT(imp);
 725
 726         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
 727         if (req == NULL)
 728                 RETURN(-ENOMEM);
 729
 730         /* XXX: capa support? */
 731         /* osc_set_capa_size(req, &RMF_CAPA1, capa); */
 732         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 733         if (rc) {
 734                 ptlrpc_request_free(req);
 735                 RETURN(rc);
 736         }
 737
 738         /*
 739          * XXX: decide how do we do here with resend
 740          * if we don't resend, then client may see wrong file size
 741          * if we do resend, then MDS thread can get stuck for quite long
 742          */
 743         req->rq_no_resend = req->rq_no_delay = 1;
 744
 745         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 746         ptlrpc_at_set_req_timeout(req);
 747
 748         OBD_ALLOC_PTR(oa);
 749         if (oa == NULL)
 750                 GOTO(out, rc = -ENOMEM);
 751
 752         rc = fid_ostid_pack(lu_object_fid(&dt->do_lu), &oa->o_oi);
 753         LASSERT(rc == 0);
 754         oa->o_size = size;
 755         oa->o_blocks = OBD_OBJECT_EOF;
 756         oa->o_valid = OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 757                       OBD_MD_FLID | OBD_MD_FLGROUP;
 758
 759         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 760         LASSERT(body);
 761         lustre_set_wire_obdo(&body->oa, oa);
 762
 763         /* XXX: capa support? */
 764         /* osc_pack_capa(req, body, capa); */
 765
 766         ptlrpc_request_set_replen(req);
 767
 768         rc = ptlrpc_queue_wait(req);
 769         if (rc)
 770                 CERROR("can't punch object: %d\n", rc);
 771 out:
 772         ptlrpc_req_finished(req);
 773         if (oa)
 774                 OBD_FREE_PTR(oa);
 775         RETURN(rc);
 776 }
 777
 778 int osp_init_precreate(struct osp_device *d)
 779 {
 780         struct l_wait_info       lwi = { 0 };
 781         int                      rc;
 782
 783         ENTRY;
 784
 785         /* initially precreation isn't ready */
 786         d->opd_pre_status = -EAGAIN;
 787         d->opd_pre_next = 1;
 788         d->opd_pre_last_created = 1;
 789         d->opd_pre_reserved = 0;
 790         d->opd_got_disconnected = 1;
 791         d->opd_pre_grow_slow = 0;
 792         d->opd_pre_grow_count = OST_MIN_PRECREATE;
 793         d->opd_pre_min_grow_count = OST_MIN_PRECREATE;
 794         d->opd_pre_max_grow_count = OST_MAX_PRECREATE;
 795
 796         cfs_spin_lock_init(&d->opd_pre_lock);
 797         cfs_waitq_init(&d->opd_pre_waitq);
 798         cfs_waitq_init(&d->opd_pre_user_waitq);
 799         cfs_waitq_init(&d->opd_pre_thread.t_ctl_waitq);
 800
 801         /*
 802          * Initialize statfs-related things
 803          */
 804         d->opd_statfs_maxage = 5; /* default update interval */
 805         d->opd_statfs_fresh_till = cfs_time_shift(-1000);
 806         CDEBUG(D_OTHER, "current %llu, fresh till %llu\n",
 807                (unsigned long long)cfs_time_current(),
 808                (unsigned long long)d->opd_statfs_fresh_till);
 809         cfs_timer_init(&d->opd_statfs_timer, osp_statfs_timer_cb, d);
 810
 811         /*
 812          * start thread handling precreation and statfs updates
 813          */
 814         rc = cfs_create_thread(osp_precreate_thread, d, 0);
 815         if (rc < 0) {
 816                 CERROR("can't start precreate thread %d\n", rc);
 817                 RETURN(rc);
 818         }
 819
 820         l_wait_event(d->opd_pre_thread.t_ctl_waitq,
 821                      osp_precreate_running(d) || osp_precreate_stopped(d),
 822                      &lwi);
 823
 824         RETURN(0);
 825 }
 826
 827 void osp_precreate_fini(struct osp_device *d)
 828 {
 829         struct ptlrpc_thread *thread = &d->opd_pre_thread;
 830
 831         ENTRY;
 832
 833         cfs_timer_disarm(&d->opd_statfs_timer);
 834
 835         thread->t_flags = SVC_STOPPING;
 836         cfs_waitq_signal(&d->opd_pre_waitq);
 837
 838         cfs_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);
 839
 840         EXIT;
 841 }
 842