Whamcloud - gitweb
LU-11997 ptlrpc: Properly swab ll_fiemap_info_key
[fs/lustre-release.git] / lustre / ofd / ofd_fs.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) 2009, 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/ofd/ofd_fs.c
33  *
34  * This file provides helper functions to handle various data stored on disk.
35  * It uses OSD API and works with any OSD.
36  *
37  * Note: this file contains also functions for sequence handling, they are
38  * placed here improperly and will be moved to the ofd_dev.c and ofd_internal.h,
39  * this comment is to be removed after that.
40  *
41  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
42  * Author: Mikhail Pershin <mike.pershin@intel.com>
43  */
44
45 #define DEBUG_SUBSYSTEM S_FILTER
46
47 #include "ofd_internal.h"
48
49 /**
50  * Restrict precreate batch count by its upper limit.
51  *
52  * The precreate batch count is a number of precreates to do in
53  * single transaction. It has upper limit - ofd_device::ofd_precreate_batch
54  * value which shouldn't be exceeded.
55  *
56  * \param[in] ofd       OFD device
57  * \param[in] batch     number of updates in the batch
58  *
59  * \retval              \a batch limited by ofd_device::ofd_precreate_batch
60  */
61 int ofd_precreate_batch(struct ofd_device *ofd, int batch)
62 {
63         int count;
64
65         spin_lock(&ofd->ofd_batch_lock);
66         count = min(ofd->ofd_precreate_batch, batch);
67         spin_unlock(&ofd->ofd_batch_lock);
68
69         return count;
70 }
71
72 /**
73  * Get ofd_seq for \a seq.
74  *
75  * Function finds appropriate structure by \a seq number and
76  * increases the reference counter of that structure.
77  *
78  * \param[in] ofd       OFD device
79  * \param[in] seq       sequence number, FID sequence number usually
80  *
81  * \retval              pointer to the requested ofd_seq structure
82  * \retval              NULL if ofd_seq is not found
83  */
84 struct ofd_seq *ofd_seq_get(struct ofd_device *ofd, u64 seq)
85 {
86         struct ofd_seq *oseq;
87
88         read_lock(&ofd->ofd_seq_list_lock);
89         list_for_each_entry(oseq, &ofd->ofd_seq_list, os_list) {
90                 if (ostid_seq(&oseq->os_oi) == seq) {
91                         atomic_inc(&oseq->os_refc);
92                         read_unlock(&ofd->ofd_seq_list_lock);
93                         return oseq;
94                 }
95         }
96         read_unlock(&ofd->ofd_seq_list_lock);
97         return NULL;
98 }
99
100 /**
101  * Drop a reference to ofd_seq.
102  *
103  * The paired function to the ofd_seq_get(). It decrease the reference counter
104  * of the ofd_seq structure and free it if that reference was last one.
105  *
106  * \param[in] env       execution environment
107  * \param[in] oseq      ofd_seq structure to put
108  */
109 void ofd_seq_put(const struct lu_env *env, struct ofd_seq *oseq)
110 {
111         if (atomic_dec_and_test(&oseq->os_refc)) {
112                 LASSERT(list_empty(&oseq->os_list));
113                 LASSERT(oseq->os_lastid_obj != NULL);
114                 dt_object_put(env, oseq->os_lastid_obj);
115                 OBD_FREE_PTR(oseq);
116         }
117 }
118
119 /**
120  * Add a new ofd_seq to the given OFD device.
121  *
122  * First it checks if there is already existent ofd_seq with the same
123  * sequence number as used by \a new_seq.
124  * If such ofd_seq is not found then the \a new_seq is added to the list
125  * of all ofd_seq structures else the \a new_seq is dropped and the found
126  * ofd_seq is returned back.
127  *
128  * \param[in] env       execution environment
129  * \param[in] ofd       OFD device
130  * \param[in] new_seq   new ofd_seq to be added
131  *
132  * \retval              ofd_seq structure
133  */
134 static struct ofd_seq *ofd_seq_add(const struct lu_env *env,
135                                    struct ofd_device *ofd,
136                                    struct ofd_seq *new_seq)
137 {
138         struct ofd_seq *os = NULL;
139
140         write_lock(&ofd->ofd_seq_list_lock);
141         list_for_each_entry(os, &ofd->ofd_seq_list, os_list) {
142                 if (ostid_seq(&os->os_oi) == ostid_seq(&new_seq->os_oi)) {
143                         atomic_inc(&os->os_refc);
144                         write_unlock(&ofd->ofd_seq_list_lock);
145                         /* The seq has not been added to the list */
146                         ofd_seq_put(env, new_seq);
147                         return os;
148                 }
149         }
150         atomic_inc(&new_seq->os_refc);
151         list_add_tail(&new_seq->os_list, &ofd->ofd_seq_list);
152         ofd->ofd_seq_count++;
153         write_unlock(&ofd->ofd_seq_list_lock);
154         return new_seq;
155 }
156
157 /**
158  * Get last object ID for the given sequence.
159  *
160  * \param[in] oseq      OFD sequence structure
161  *
162  * \retval              the last object ID for this sequence
163  */
164 u64 ofd_seq_last_oid(struct ofd_seq *oseq)
165 {
166         u64 id;
167
168         spin_lock(&oseq->os_last_oid_lock);
169         id = ostid_id(&oseq->os_oi);
170         spin_unlock(&oseq->os_last_oid_lock);
171
172         return id;
173 }
174
175 /**
176  * Set new last object ID for the given sequence.
177  *
178  * \param[in] oseq      OFD sequence
179  * \param[in] id        the new OID to set
180  */
181 void ofd_seq_last_oid_set(struct ofd_seq *oseq, u64 id)
182 {
183         spin_lock(&oseq->os_last_oid_lock);
184         if (likely(ostid_id(&oseq->os_oi) < id)) {
185                 if (ostid_set_id(&oseq->os_oi, id)) {
186                         CERROR("Bad %llu to set " DOSTID "\n",
187                                (unsigned long long)id, POSTID(&oseq->os_oi));
188                 }
189         }
190         spin_unlock(&oseq->os_last_oid_lock);
191 }
192
193 /**
194  * Update last used OID on disk for the given sequence.
195  *
196  * The last used object ID is stored persistently on disk and
197  * must be written when updated. This function writes the sequence data.
198  * The format is just an object ID of the latest used object FID.
199  * Each ID is stored in per-sequence file.
200  *
201  * \param[in] env       execution environment
202  * \param[in] ofd       OFD device
203  * \param[in] oseq      ofd_seq structure with data to write
204  *
205  * \retval              0 on successful write of data from \a oseq
206  * \retval              negative value on error
207  */
208 int ofd_seq_last_oid_write(const struct lu_env *env, struct ofd_device *ofd,
209                            struct ofd_seq *oseq)
210 {
211         struct ofd_thread_info  *info = ofd_info(env);
212         u64                      tmp;
213         struct dt_object        *obj = oseq->os_lastid_obj;
214         struct thandle          *th;
215         int                      rc;
216
217         ENTRY;
218
219         if (ofd->ofd_osd->dd_rdonly)
220                 RETURN(0);
221
222         tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
223
224         info->fti_buf.lb_buf = &tmp;
225         info->fti_buf.lb_len = sizeof(tmp);
226         info->fti_off = 0;
227
228         LASSERT(obj != NULL);
229
230         th = dt_trans_create(env, ofd->ofd_osd);
231         if (IS_ERR(th))
232                 RETURN(PTR_ERR(th));
233
234         rc = dt_declare_record_write(env, obj, &info->fti_buf,
235                                      info->fti_off, th);
236         if (rc < 0)
237                 GOTO(out, rc);
238         rc = dt_trans_start_local(env, ofd->ofd_osd, th);
239         if (rc < 0)
240                 GOTO(out, rc);
241         rc = dt_record_write(env, obj, &info->fti_buf, &info->fti_off,
242                              th);
243         if (rc < 0)
244                 GOTO(out, rc);
245
246         CDEBUG(D_INODE, "%s: write last_objid "DOSTID": rc = %d\n",
247                ofd_name(ofd), POSTID(&oseq->os_oi), rc);
248         EXIT;
249 out:
250         dt_trans_stop(env, ofd->ofd_osd, th);
251         return rc;
252 }
253
254 /**
255  * Deregister LWP items for FLDB and SEQ client on OFD.
256  *
257  * LWP is lightweight proxy - simplified connection between
258  * servers. It is used for FID Location Database (FLDB) and
259  * sequence (SEQ) client-server interactions.
260  *
261  * This function is used during server cleanup process to free
262  * LWP items that were previously set up upon OFD start.
263  *
264  * \param[in]     ofd   OFD device
265  */
266 static void ofd_deregister_seq_exp(struct ofd_device *ofd)
267 {
268         struct seq_server_site  *ss = &ofd->ofd_seq_site;
269
270         if (ss->ss_client_seq != NULL) {
271                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
272                 ss->ss_client_seq->lcs_exp = NULL;
273         }
274
275         if (ss->ss_server_fld != NULL) {
276                 lustre_deregister_lwp_item(&ss->ss_server_fld->lsf_control_exp);
277                 ss->ss_server_fld->lsf_control_exp = NULL;
278         }
279 }
280
281 /**
282  * Stop FLDB server on OFD.
283  *
284  * This function is part of OFD cleanup process.
285  *
286  * \param[in] env       execution environment
287  * \param[in] ofd       OFD device
288  *
289  */
290 static void ofd_fld_fini(const struct lu_env *env, struct ofd_device *ofd)
291 {
292         struct seq_server_site *ss = &ofd->ofd_seq_site;
293
294         if (ss != NULL && ss->ss_server_fld != NULL) {
295                 fld_server_fini(env, ss->ss_server_fld);
296                 OBD_FREE_PTR(ss->ss_server_fld);
297                 ss->ss_server_fld = NULL;
298         }
299 }
300
301 /**
302  * Free sequence structures on OFD.
303  *
304  * This function is part of OFD cleanup process, it goes through
305  * the list of ofd_seq structures stored in ofd_device structure
306  * and frees them.
307  *
308  * \param[in] env       execution environment
309  * \param[in] ofd       OFD device
310  */
311 void ofd_seqs_free(const struct lu_env *env, struct ofd_device *ofd)
312 {
313         struct ofd_seq          *oseq;
314         struct ofd_seq          *tmp;
315         struct list_head         dispose;
316
317         INIT_LIST_HEAD(&dispose);
318         write_lock(&ofd->ofd_seq_list_lock);
319         list_for_each_entry_safe(oseq, tmp, &ofd->ofd_seq_list, os_list)
320                 list_move(&oseq->os_list, &dispose);
321         write_unlock(&ofd->ofd_seq_list_lock);
322
323         while (!list_empty(&dispose)) {
324                 oseq = container_of0(dispose.next, struct ofd_seq, os_list);
325                 list_del_init(&oseq->os_list);
326                 ofd_seq_put(env, oseq);
327         }
328 }
329
330 /**
331  * Stop FLDB and SEQ services on OFD.
332  *
333  * This function is part of OFD cleanup process.
334  *
335  * \param[in] env       execution environment
336  * \param[in] ofd       OFD device
337  *
338  */
339 void ofd_seqs_fini(const struct lu_env *env, struct ofd_device *ofd)
340 {
341         int rc;
342
343         ofd_deregister_seq_exp(ofd);
344
345         rc = ofd_fid_fini(env, ofd);
346         if (rc != 0)
347                 CERROR("%s: fid fini error: rc = %d\n", ofd_name(ofd), rc);
348
349         ofd_fld_fini(env, ofd);
350
351         ofd_seqs_free(env, ofd);
352
353         LASSERT(list_empty(&ofd->ofd_seq_list));
354 }
355
356 /**
357  * Return ofd_seq structure filled with valid data.
358  *
359  * This function gets the ofd_seq by sequence number and read
360  * corresponding data from disk.
361  *
362  * \param[in] env       execution environment
363  * \param[in] ofd       OFD device
364  * \param[in] seq       sequence number
365  *
366  * \retval              ofd_seq structure filled with data
367  * \retval              ERR_PTR pointer on error
368  */
369 struct ofd_seq *ofd_seq_load(const struct lu_env *env, struct ofd_device *ofd,
370                              u64 seq)
371 {
372         struct ofd_thread_info  *info = ofd_info(env);
373         struct ofd_seq          *oseq = NULL;
374         struct dt_object        *dob;
375         u64                      lastid;
376         int                      rc;
377
378         ENTRY;
379
380         /* if seq is already initialized */
381         oseq = ofd_seq_get(ofd, seq);
382         if (oseq != NULL)
383                 RETURN(oseq);
384
385         OBD_ALLOC_PTR(oseq);
386         if (oseq == NULL)
387                 RETURN(ERR_PTR(-ENOMEM));
388
389         lu_last_id_fid(&info->fti_fid, seq, ofd->ofd_lut.lut_lsd.lsd_osd_index);
390         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
391         info->fti_attr.la_valid = LA_MODE;
392         info->fti_attr.la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
393         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
394
395         /* create object tracking per-seq last created
396          * id to be used by orphan recovery mechanism */
397         dob = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
398                                 &info->fti_dof, &info->fti_attr);
399         if (IS_ERR(dob)) {
400                 OBD_FREE_PTR(oseq);
401                 RETURN((void *)dob);
402         }
403
404         oseq->os_lastid_obj = dob;
405
406         INIT_LIST_HEAD(&oseq->os_list);
407         mutex_init(&oseq->os_create_lock);
408         spin_lock_init(&oseq->os_last_oid_lock);
409         ostid_set_seq(&oseq->os_oi, seq);
410         oseq->os_last_id_synced = 0;
411
412         atomic_set(&oseq->os_refc, 1);
413         atomic_set(&oseq->os_precreate_in_progress, 0);
414
415         rc = dt_attr_get(env, dob, &info->fti_attr);
416         if (rc)
417                 GOTO(cleanup, rc);
418
419         if (info->fti_attr.la_size == 0) {
420                 /* object is just created, initialize last id */
421                 if (OBD_FAIL_CHECK(OBD_FAIL_OFD_SET_OID))
422                         ofd_seq_last_oid_set(oseq, 0xffffff00);
423                 else
424                         ofd_seq_last_oid_set(oseq, OFD_INIT_OBJID);
425                 ofd_seq_last_oid_write(env, ofd, oseq);
426         } else if (info->fti_attr.la_size == sizeof(lastid)) {
427                 info->fti_off = 0;
428                 info->fti_buf.lb_buf = &lastid;
429                 info->fti_buf.lb_len = sizeof(lastid);
430
431                 rc = dt_record_read(env, dob, &info->fti_buf, &info->fti_off);
432                 if (rc) {
433                         CERROR("%s: can't read last_id: rc = %d\n",
434                                 ofd_name(ofd), rc);
435                         GOTO(cleanup, rc);
436                 }
437                 ofd_seq_last_oid_set(oseq, le64_to_cpu(lastid));
438         } else {
439                 CERROR("%s: corrupted size %llu LAST_ID of seq %#llx\n",
440                         ofd_name(ofd), (__u64)info->fti_attr.la_size, seq);
441                 GOTO(cleanup, rc = -EINVAL);
442         }
443
444         oseq = ofd_seq_add(env, ofd, oseq);
445         RETURN((oseq != NULL) ? oseq : ERR_PTR(-ENOENT));
446 cleanup:
447         ofd_seq_put(env, oseq);
448         return ERR_PTR(rc);
449 }
450
451 /**
452  * initialize local FLDB server.
453  *
454  * \param[in] env       execution environment
455  * \param[in] uuid      unique name for this FLDS server
456  * \param[in] ofd       OFD device
457  *
458  * \retval              0 on successful initialization
459  * \retval              negative value on error
460  */
461 static int ofd_fld_init(const struct lu_env *env, const char *uuid,
462                         struct ofd_device *ofd)
463 {
464         struct seq_server_site *ss = &ofd->ofd_seq_site;
465         int rc;
466
467         ENTRY;
468
469         OBD_ALLOC_PTR(ss->ss_server_fld);
470         if (ss->ss_server_fld == NULL)
471                 RETURN(rc = -ENOMEM);
472
473         rc = fld_server_init(env, ss->ss_server_fld, ofd->ofd_osd, uuid,
474                              LU_SEQ_RANGE_OST);
475         if (rc < 0) {
476                 OBD_FREE_PTR(ss->ss_server_fld);
477                 ss->ss_server_fld = NULL;
478                 RETURN(rc);
479         }
480         RETURN(0);
481 }
482
483 /**
484  * Update local FLDB copy from master server.
485  *
486  * This callback is called when LWP is connected to the server.
487  * It retrieves its FLDB entries from MDT0, and it only happens
488  * when upgrading the existing file system to 2.6.
489  *
490  * \param[in] data      OFD device
491  *
492  * \retval              0 on successful FLDB update
493  * \retval              negative value in case if failure
494  */
495 static int ofd_register_lwp_callback(void *data)
496 {
497         struct lu_env           *env;
498         struct ofd_device       *ofd = data;
499         struct lu_server_fld    *fld = ofd->ofd_seq_site.ss_server_fld;
500         int                     rc;
501
502         ENTRY;
503
504         if (!likely(fld->lsf_new))
505                 RETURN(0);
506
507         OBD_ALLOC_PTR(env);
508         if (env == NULL)
509                 RETURN(-ENOMEM);
510
511         rc = lu_env_init(env, LCT_DT_THREAD);
512         if (rc < 0)
513                 GOTO(out, rc);
514
515         rc = fld_update_from_controller(env, fld);
516         if (rc < 0) {
517                 CERROR("%s: cannot update controller: rc = %d\n",
518                        ofd_name(ofd), rc);
519                 GOTO(out, rc);
520         }
521         EXIT;
522 out:
523         lu_env_fini(env);
524         OBD_FREE_PTR(env);
525         return rc;
526 }
527
528 /**
529  * Get LWP exports from LWP connection for local FLDB server and SEQ client.
530  *
531  * This function is part of setup process and initialize FLDB server and SEQ
532  * client, so they may work with remote servers.
533  *
534  * \param[in] ofd       OFD device
535  *
536  * \retval              0 on successful export get
537  * \retval              negative value on error
538  */
539 static int ofd_register_seq_exp(struct ofd_device *ofd)
540 {
541         struct seq_server_site  *ss = &ofd->ofd_seq_site;
542         char                    *lwp_name = NULL;
543         int                     rc;
544
545         OBD_ALLOC(lwp_name, MAX_OBD_NAME);
546         if (lwp_name == NULL)
547                 GOTO(out_free, rc = -ENOMEM);
548
549         rc = tgt_name2lwp_name(ofd_name(ofd), lwp_name, MAX_OBD_NAME, 0);
550         if (rc != 0)
551                 GOTO(out_free, rc);
552
553         rc = lustre_register_lwp_item(lwp_name, &ss->ss_client_seq->lcs_exp,
554                                       NULL, NULL);
555         if (rc != 0)
556                 GOTO(out_free, rc);
557
558         rc = lustre_register_lwp_item(lwp_name,
559                                       &ss->ss_server_fld->lsf_control_exp,
560                                       ofd_register_lwp_callback, ofd);
561         if (rc != 0) {
562                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
563                 ss->ss_client_seq->lcs_exp = NULL;
564                 GOTO(out_free, rc);
565         }
566 out_free:
567         if (lwp_name != NULL)
568                 OBD_FREE(lwp_name, MAX_OBD_NAME);
569
570         return rc;
571 }
572
573 /**
574  * Initialize SEQ and FLD service on OFD.
575  *
576  * This is part of OFD setup process.
577  *
578  * \param[in] env       execution environment
579  * \param[in] ofd       OFD device
580  *
581  * \retval              0 on successful services initialization
582  * \retval              negative value on error
583  */
584 int ofd_seqs_init(const struct lu_env *env, struct ofd_device *ofd)
585 {
586         int rc;
587
588         rwlock_init(&ofd->ofd_seq_list_lock);
589         INIT_LIST_HEAD(&ofd->ofd_seq_list);
590         ofd->ofd_seq_count = 0;
591
592         rc = ofd_fid_init(env, ofd);
593         if (rc != 0) {
594                 CERROR("%s: fid init error: rc = %d\n", ofd_name(ofd), rc);
595                 GOTO(out, rc);
596         }
597
598         rc = ofd_fld_init(env, ofd_name(ofd), ofd);
599         if (rc < 0) {
600                 CERROR("%s: Can't init fld, rc %d\n", ofd_name(ofd), rc);
601                 GOTO(out_fid, rc);
602         }
603
604         rc = ofd_register_seq_exp(ofd);
605         if (rc < 0) {
606                 CERROR("%s: Can't init seq exp, rc %d\n", ofd_name(ofd), rc);
607                 GOTO(out_fld, rc);
608         }
609
610         RETURN(0);
611
612 out_fld:
613         ofd_fld_fini(env, ofd);
614 out_fid:
615         ofd_fid_fini(env, ofd);
616 out:
617         return rc;
618 }
619
620 /**
621  * Initialize storage for the OFD.
622  *
623  * This function sets up service files for OFD. Currently, the only
624  * service file is "health_check".
625  *
626  * \param[in] env       execution environment
627  * \param[in] ofd       OFD device
628  * \param[in] obd       OBD device (unused now)
629  *
630  * \retval              0 on successful setup
631  * \retval              negative value on error
632  */
633 int ofd_fs_setup(const struct lu_env *env, struct ofd_device *ofd,
634                  struct obd_device *obd)
635 {
636         struct ofd_thread_info  *info = ofd_info(env);
637         struct dt_object        *fo;
638         int                      rc = 0;
639
640         ENTRY;
641
642         rc = ofd_seqs_init(env, ofd);
643         if (rc)
644                 GOTO(out, rc);
645
646         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FS_SETUP))
647                 GOTO(out_seqs, rc = -ENOENT);
648
649         lu_local_obj_fid(&info->fti_fid, OFD_HEALTH_CHECK_OID);
650         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
651         info->fti_attr.la_valid = LA_MODE;
652         info->fti_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
653         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
654
655         fo = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
656                                &info->fti_dof, &info->fti_attr);
657         if (IS_ERR(fo))
658                 GOTO(out_seqs, rc = PTR_ERR(fo));
659
660         ofd->ofd_health_check_file = fo;
661
662         RETURN(0);
663
664 out_seqs:
665         ofd_seqs_fini(env, ofd);
666 out:
667         return rc;
668 }
669
670 /**
671  * Cleanup service files on OFD.
672  *
673  * This function syncs whole OFD device and close "health check" file.
674  *
675  * \param[in] env       execution environment
676  * \param[in] ofd       OFD device
677  */
678 void ofd_fs_cleanup(const struct lu_env *env, struct ofd_device *ofd)
679 {
680         int rc;
681
682         ENTRY;
683
684         ofd_seqs_fini(env, ofd);
685
686         rc = dt_sync(env, ofd->ofd_osd);
687         if (rc < 0)
688                 CWARN("%s: can't sync OFD upon cleanup: %d\n",
689                       ofd_name(ofd), rc);
690
691         if (ofd->ofd_health_check_file) {
692                 dt_object_put(env, ofd->ofd_health_check_file);
693                 ofd->ofd_health_check_file = NULL;
694         }
695
696         EXIT;
697 }
698