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