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