Whamcloud - gitweb
7f328eeeb667113a5bfb5cf4f912b7c738b9f8f6
[fs/lustre-release.git] / lustre / ofd / ofd_objects.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, 2014, 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_objects.c
33  *
34  * This file contains OSD API methods related to OBD Filter Device (OFD)
35  * object operations.
36  *
37  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
38  * Author: Mikhail Pershin <mike.pershin@intel.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_FILTER
42
43 #include <dt_object.h>
44 #include <lustre/lustre_idl.h>
45 #include <lustre_lfsck.h>
46
47 #include "ofd_internal.h"
48
49 /**
50  * Get object version from disk and check it.
51  *
52  * This function checks object version from disk with
53  * ofd_thread_info::fti_pre_version filled from incoming RPC. This is part of
54  * VBR (Version-Based Recovery) and ensures that object has the same version
55  * upon replay as it has during original modification.
56  *
57  * \param[in]  info     execution thread OFD private data
58  * \param[in]  fo       OFD object
59  *
60  * \retval              0 if version matches
61  * \retval              -EOVERFLOW on version mismatch
62  */
63 static int ofd_version_get_check(struct ofd_thread_info *info,
64                                  struct ofd_object *fo)
65 {
66         dt_obj_version_t curr_version;
67
68         LASSERT(ofd_object_exists(fo));
69
70         if (info->fti_exp == NULL)
71                 RETURN(0);
72
73         curr_version = dt_version_get(info->fti_env, ofd_object_child(fo));
74         if ((__s64)curr_version == -EOPNOTSUPP)
75                 RETURN(0);
76         /* VBR: version is checked always because costs nothing */
77         if (info->fti_pre_version != 0 &&
78             info->fti_pre_version != curr_version) {
79                 CDEBUG(D_INODE, "Version mismatch %#llx != %#llx\n",
80                        info->fti_pre_version, curr_version);
81                 spin_lock(&info->fti_exp->exp_lock);
82                 info->fti_exp->exp_vbr_failed = 1;
83                 spin_unlock(&info->fti_exp->exp_lock);
84                 RETURN (-EOVERFLOW);
85         }
86         info->fti_pre_version = curr_version;
87         RETURN(0);
88 }
89
90 /**
91  * Get OFD object by FID.
92  *
93  * This function finds OFD slice of compound object with the given FID.
94  *
95  * \param[in] env       execution environment
96  * \param[in] ofd       OFD device
97  * \param[in] fid       FID of the object
98  *
99  * \retval              pointer to the found ofd_object
100  * \retval              ERR_PTR(errno) in case of error
101  */
102 struct ofd_object *ofd_object_find(const struct lu_env *env,
103                                    struct ofd_device *ofd,
104                                    const struct lu_fid *fid)
105 {
106         struct ofd_object *fo;
107         struct lu_object  *o;
108
109         ENTRY;
110
111         o = lu_object_find(env, &ofd->ofd_dt_dev.dd_lu_dev, fid, NULL);
112         if (likely(!IS_ERR(o)))
113                 fo = ofd_obj(o);
114         else
115                 fo = ERR_CAST(o); /* return error */
116
117         RETURN(fo);
118 }
119
120 /**
121  * Get FID of parent MDT object.
122  *
123  * This function reads extended attribute XATTR_NAME_FID of OFD object which
124  * contains the MDT parent object FID and saves it in ofd_object::ofo_ff.
125  *
126  * The filter_fid::ff_parent::f_ver field currently holds
127  * the OST-object index in the parent MDT-object's layout EA,
128  * not the actual FID::f_ver of the parent. We therefore access
129  * it via the macro f_stripe_idx.
130  *
131  * \param[in] env       execution environment
132  * \param[in] fo        OFD object
133  *
134  * \retval              0 if successful
135  * \retval              -ENODATA if there is no such xattr
136  * \retval              negative value on error
137  */
138 int ofd_object_ff_load(const struct lu_env *env, struct ofd_object *fo)
139 {
140         struct ofd_thread_info *info = ofd_info(env);
141         struct filter_fid *ff = &fo->ofo_ff;
142         struct lu_buf *buf = &info->fti_buf;
143         int rc = 0;
144
145         if (fid_is_sane(&ff->ff_parent))
146                 return 0;
147
148         buf->lb_buf = ff;
149         buf->lb_len = sizeof(*ff);
150         rc = dt_xattr_get(env, ofd_object_child(fo), buf, XATTR_NAME_FID);
151         if (rc < 0)
152                 return rc;
153
154         if (unlikely(rc < sizeof(struct lu_fid))) {
155                 fid_zero(&ff->ff_parent);
156
157                 return -ENODATA;
158         }
159
160         filter_fid_le_to_cpu(ff, ff, rc);
161
162         return 0;
163 }
164
165 /**
166  * Precreate the given number \a nr of objects in the given sequence \a oseq.
167  *
168  * This function precreates new OST objects in the given sequence.
169  * The precreation starts from \a id and creates \a nr objects sequentially.
170  *
171  * Notes:
172  * This function may create fewer objects than requested.
173  *
174  * We mark object SUID+SGID to flag it for accepting UID+GID from client on
175  * first write. Currently the permission bits on the OST are never used,
176  * so this is OK.
177  *
178  * Initialize a/c/m time so any client timestamp will always be newer and
179  * update the inode. The ctime = 0 case is also handled specially in
180  * osd_inode_setattr(). See LU-221, LU-1042 for details.
181  *
182  * \param[in] env       execution environment
183  * \param[in] ofd       OFD device
184  * \param[in] id        object ID to start precreation from
185  * \param[in] oseq      object sequence
186  * \param[in] nr        number of objects to precreate
187  * \param[in] sync      synchronous precreation flag
188  *
189  * \retval              0 if successful
190  * \retval              negative value on error
191  */
192 int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
193                           u64 id, struct ofd_seq *oseq, int nr, int sync)
194 {
195         struct ofd_thread_info  *info = ofd_info(env);
196         struct ofd_object       *fo = NULL;
197         struct dt_object        *next;
198         struct thandle          *th;
199         struct ofd_object       **batch;
200         struct lu_fid           *fid = &info->fti_fid;
201         u64                     tmp;
202         int                     rc;
203         int                     rc2;
204         int                     i;
205         int                     objects = 0;
206         int                     nr_saved = nr;
207
208         ENTRY;
209
210         /* Don't create objects beyond the valid range for this SEQ */
211         if (unlikely(fid_seq_is_mdt0(ostid_seq(&oseq->os_oi)) &&
212                      (id + nr) >= IDIF_MAX_OID)) {
213                 CERROR("%s:"DOSTID" hit the IDIF_MAX_OID (1<<48)!\n",
214                        ofd_name(ofd), id, ostid_seq(&oseq->os_oi));
215                 RETURN(rc = -ENOSPC);
216         } else if (unlikely(!fid_seq_is_mdt0(ostid_seq(&oseq->os_oi)) &&
217                             (id + nr) >= OBIF_MAX_OID)) {
218                 CERROR("%s:"DOSTID" hit the OBIF_MAX_OID (1<<32)!\n",
219                        ofd_name(ofd), id, ostid_seq(&oseq->os_oi));
220                 RETURN(rc = -ENOSPC);
221         }
222
223         OBD_ALLOC(batch, nr_saved * sizeof(struct ofd_object *));
224         if (batch == NULL)
225                 RETURN(-ENOMEM);
226
227         info->fti_attr.la_valid = LA_TYPE | LA_MODE;
228         info->fti_attr.la_mode = S_IFREG | S_ISUID | S_ISGID | 0666;
229         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
230
231         info->fti_attr.la_valid |= LA_ATIME | LA_MTIME | LA_CTIME;
232         info->fti_attr.la_atime = 0;
233         info->fti_attr.la_mtime = 0;
234         info->fti_attr.la_ctime = 0;
235
236         LASSERT(id != 0);
237
238         /* prepare objects */
239         *fid = *lu_object_fid(&oseq->os_lastid_obj->do_lu);
240         for (i = 0; i < nr; i++) {
241                 rc = fid_set_id(fid, id + i);
242                 if (rc != 0) {
243                         if (i == 0)
244                                 GOTO(out, rc);
245
246                         nr = i;
247                         break;
248                 }
249
250                 fo = ofd_object_find(env, ofd, fid);
251                 if (IS_ERR(fo)) {
252                         if (i == 0)
253                                 GOTO(out, rc = PTR_ERR(fo));
254
255                         nr = i;
256                         break;
257                 }
258
259                 ofd_write_lock(env, fo);
260                 batch[i] = fo;
261         }
262         info->fti_buf.lb_buf = &tmp;
263         info->fti_buf.lb_len = sizeof(tmp);
264         info->fti_off = 0;
265
266         th = ofd_trans_create(env, ofd);
267         if (IS_ERR(th))
268                 GOTO(out, rc = PTR_ERR(th));
269
270         th->th_sync |= sync;
271
272         rc = dt_declare_record_write(env, oseq->os_lastid_obj, &info->fti_buf,
273                                      info->fti_off, th);
274         if (rc)
275                 GOTO(trans_stop, rc);
276
277         for (i = 0; i < nr; i++) {
278                 fo = batch[i];
279                 LASSERT(fo);
280
281                 if (unlikely(ofd_object_exists(fo))) {
282                         /* object may exist being re-created by write replay */
283                         CDEBUG(D_INODE, "object %#llx/%#llx exists: "
284                                DFID"\n", ostid_seq(&oseq->os_oi), id,
285                                PFID(lu_object_fid(&fo->ofo_obj.do_lu)));
286                         continue;
287                 }
288
289                 next = ofd_object_child(fo);
290                 LASSERT(next != NULL);
291
292                 rc = dt_declare_create(env, next, &info->fti_attr, NULL,
293                                        &info->fti_dof, th);
294                 if (rc < 0) {
295                         if (i == 0)
296                                 GOTO(trans_stop, rc);
297
298                         nr = i;
299                         break;
300                 }
301         }
302
303         rc = dt_trans_start_local(env, ofd->ofd_osd, th);
304         if (rc)
305                 GOTO(trans_stop, rc);
306
307         CDEBUG(D_OTHER, "%s: create new object "DFID" nr %d\n",
308                ofd_name(ofd), PFID(fid), nr);
309
310          /* When the LFSCK scanning the whole device to verify the LAST_ID file
311           * consistency, it will load the last_id into RAM firstly, and compare
312           * the last_id with each OST-object's ID. If the later one is larger,
313           * then it will regard the LAST_ID file crashed. But during the LFSCK
314           * scanning, the OFD may continue to create new OST-objects. Those new
315           * created OST-objects will have larger IDs than the LFSCK known ones.
316           * So from the LFSCK view, it needs to re-load the last_id from disk
317           * file, and if the latest last_id is still smaller than the object's
318           * ID, then the LAST_ID file is real crashed.
319           *
320           * To make above mechanism to work, before OFD pre-create OST-objects,
321           * it needs to update the LAST_ID file firstly, otherwise, the LFSCK
322           * may cannot get latest last_id although new OST-object created. */
323         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_SKIP_LASTID)) {
324                 tmp = cpu_to_le64(id + nr - 1);
325                 dt_write_lock(env, oseq->os_lastid_obj, 0);
326                 rc = dt_record_write(env, oseq->os_lastid_obj,
327                                      &info->fti_buf, &info->fti_off, th);
328                 dt_write_unlock(env, oseq->os_lastid_obj);
329                 if (rc != 0)
330                         GOTO(trans_stop, rc);
331         }
332
333         for (i = 0; i < nr; i++) {
334                 fo = batch[i];
335                 LASSERT(fo);
336
337                 /* Only the new created objects need to be recorded. */
338                 if (ofd->ofd_osd->dd_record_fid_accessed) {
339                         struct lfsck_req_local *lrl = &ofd_info(env)->fti_lrl;
340
341                         lfsck_pack_rfa(lrl, lu_object_fid(&fo->ofo_obj.do_lu),
342                                        LEL_FID_ACCESSED, LFSCK_TYPE_LAYOUT);
343                         lfsck_in_notify_local(env, ofd->ofd_osd, lrl, NULL);
344                 }
345
346                 if (likely(!ofd_object_exists(fo) &&
347                            !OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING))) {
348                         next = ofd_object_child(fo);
349                         LASSERT(next != NULL);
350
351                         rc = dt_create(env, next, &info->fti_attr, NULL,
352                                        &info->fti_dof, th);
353                         if (rc < 0) {
354                                 if (i == 0)
355                                         GOTO(trans_stop, rc);
356
357                                 rc = 0;
358                                 break;
359                         }
360                         LASSERT(ofd_object_exists(fo));
361                 }
362                 ofd_seq_last_oid_set(oseq, id + i);
363         }
364
365         objects = i;
366         /* NOT all the wanted objects have been created,
367          * set the LAST_ID as the real created. */
368         if (unlikely(objects < nr)) {
369                 int rc1;
370
371                 info->fti_off = 0;
372                 tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
373                 dt_write_lock(env, oseq->os_lastid_obj, 0);
374                 rc1 = dt_record_write(env, oseq->os_lastid_obj,
375                                       &info->fti_buf, &info->fti_off, th);
376                 dt_write_unlock(env, oseq->os_lastid_obj);
377                 if (rc1 != 0)
378                         CERROR("%s: fail to reset the LAST_ID for seq (%#llx"
379                                ") from %llu to %llu\n", ofd_name(ofd),
380                                ostid_seq(&oseq->os_oi), id + nr - 1,
381                                ofd_seq_last_oid(oseq));
382         }
383
384 trans_stop:
385         rc2 = ofd_trans_stop(env, ofd, th, rc);
386         if (rc2)
387                 CERROR("%s: failed to stop transaction: rc = %d\n",
388                        ofd_name(ofd), rc2);
389         if (!rc)
390                 rc = rc2;
391 out:
392         for (i = 0; i < nr_saved; i++) {
393                 fo = batch[i];
394                 if (fo) {
395                         ofd_write_unlock(env, fo);
396                         ofd_object_put(env, fo);
397                 }
398         }
399         OBD_FREE(batch, nr_saved * sizeof(struct ofd_object *));
400
401         CDEBUG((objects == 0 && rc == 0) ? D_ERROR : D_OTHER,
402                "created %d/%d objects: %d\n", objects, nr_saved, rc);
403
404         LASSERT(ergo(objects == 0, rc < 0));
405         RETURN(objects > 0 ? objects : rc);
406 }
407
408 /**
409  * Fix the OFD object ownership.
410  *
411  * If the object still has SUID+SGID bits set, meaning that it was precreated
412  * by the MDT before it was assigned to any file, (see ofd_precreate_objects())
413  * then we will accept the UID/GID/PROJID if sent by the client for initializing
414  * the ownership of this object.  We only allow this to happen once (so clear
415  * these bits) and later only allow setattr.
416  *
417  * \param[in] env        execution environment
418  * \param[in] fo         OFD object
419  * \param[in] la         object attributes
420  * \param[in] is_setattr was this function called from setattr or not
421  *
422  * \retval              0 if successful
423  * \retval              negative value on error
424  */
425 int ofd_attr_handle_id(const struct lu_env *env, struct ofd_object *fo,
426                          struct lu_attr *la, int is_setattr)
427 {
428         struct ofd_thread_info  *info = ofd_info(env);
429         struct lu_attr          *ln = &info->fti_attr2;
430         __u32                    mask = 0;
431         int                      rc;
432
433         ENTRY;
434
435         if (!(la->la_valid & LA_UID) && !(la->la_valid & LA_GID) &&
436             !(la->la_valid & LA_PROJID))
437                 RETURN(0);
438
439         rc = dt_attr_get(env, ofd_object_child(fo), ln);
440         if (rc != 0)
441                 RETURN(rc);
442
443         LASSERT(ln->la_valid & LA_MODE);
444
445         /*
446          * Only allow setattr to change UID/GID/PROJID, if
447          * SUID+SGID is not set which means this is not
448          * initialization of this objects.
449          */
450         if (!is_setattr) {
451                 if (!(ln->la_mode & S_ISUID))
452                         la->la_valid &= ~(LA_UID | LA_PROJID);
453                 if (!(ln->la_mode & S_ISGID))
454                         la->la_valid &= ~(LA_GID | LA_PROJID);
455         }
456
457         /* Initialize ownership of this object, clear SUID+SGID bits*/
458         if ((la->la_valid & LA_UID) && (ln->la_mode & S_ISUID))
459                 mask |= S_ISUID;
460         if ((la->la_valid & LA_GID) && (ln->la_mode & S_ISGID))
461                 mask |= S_ISGID;
462         if (mask != 0) {
463                 if (!(la->la_valid & LA_MODE) || !is_setattr) {
464                         la->la_mode = ln->la_mode;
465                         la->la_valid |= LA_MODE;
466                 }
467                 la->la_mode &= ~mask;
468         }
469
470         RETURN(0);
471 }
472
473 /**
474  * Set OFD object attributes.
475  *
476  * This function sets OFD object attributes taken from incoming request.
477  * It sets not only regular attributes but also XATTR_NAME_FID extended
478  * attribute if needed. The "fid" xattr allows the object's MDT parent inode
479  * to be found and verified by LFSCK and other tools in case of inconsistency.
480  *
481  * \param[in] env       execution environment
482  * \param[in] fo        OFD object
483  * \param[in] la        object attributes
484  * \param[in] ff        filter_fid structure, contains additional attributes
485  *
486  * \retval              0 if successful
487  * \retval              negative value on error
488  */
489 int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
490                  struct lu_attr *la, struct filter_fid *ff)
491 {
492         struct ofd_thread_info  *info = ofd_info(env);
493         struct ofd_device       *ofd = ofd_obj2dev(fo);
494         struct thandle          *th;
495         struct ofd_mod_data     *fmd;
496         int                     ff_needed = 0;
497         int                     rc;
498         int                     rc2;
499         ENTRY;
500
501         ofd_write_lock(env, fo);
502         if (!ofd_object_exists(fo))
503                 GOTO(unlock, rc = -ENOENT);
504
505         if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME)) {
506                 fmd = ofd_fmd_get(info->fti_exp, &fo->ofo_header.loh_fid);
507                 if (fmd && fmd->fmd_mactime_xid < info->fti_xid)
508                         fmd->fmd_mactime_xid = info->fti_xid;
509                 ofd_fmd_put(info->fti_exp, fmd);
510         }
511
512         /* VBR: version recovery check */
513         rc = ofd_version_get_check(info, fo);
514         if (rc)
515                 GOTO(unlock, rc);
516
517         rc = ofd_attr_handle_id(env, fo, la, 1 /* is_setattr */);
518         if (rc != 0)
519                 GOTO(unlock, rc);
520
521         if (ff != NULL) {
522                 rc = ofd_object_ff_load(env, fo);
523                 if (rc == -ENODATA)
524                         ff_needed = 1;
525                 else if (rc < 0)
526                         GOTO(unlock, rc);
527         }
528
529         th = ofd_trans_create(env, ofd);
530         if (IS_ERR(th))
531                 GOTO(unlock, rc = PTR_ERR(th));
532
533         rc = dt_declare_attr_set(env, ofd_object_child(fo), la, th);
534         if (rc)
535                 GOTO(stop, rc);
536
537         if (ff_needed) {
538                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
539                         ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
540                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
541                         le32_add_cpu(&ff->ff_parent.f_oid, -1);
542
543                 info->fti_buf.lb_buf = ff;
544                 info->fti_buf.lb_len = sizeof(*ff);
545                 rc = dt_declare_xattr_set(env, ofd_object_child(fo),
546                                           &info->fti_buf, XATTR_NAME_FID, 0,
547                                           th);
548                 if (rc)
549                         GOTO(stop, rc);
550         }
551
552         rc = ofd_trans_start(env, ofd, la->la_valid & LA_SIZE ? fo : NULL, th);
553         if (rc)
554                 GOTO(stop, rc);
555
556         rc = dt_attr_set(env, ofd_object_child(fo), la, th);
557         if (rc)
558                 GOTO(stop, rc);
559
560         if (ff_needed) {
561                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
562                         GOTO(stop, rc);
563
564                 rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
565                                   XATTR_NAME_FID, 0, th);
566                 if (!rc)
567                         filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
568         }
569
570         GOTO(stop, rc);
571
572 stop:
573         rc2 = ofd_trans_stop(env, ofd, th, rc);
574         if (rc2)
575                 CERROR("%s: failed to stop transaction: rc = %d\n",
576                        ofd_name(ofd), rc2);
577         if (!rc)
578                 rc = rc2;
579
580 unlock:
581         ofd_write_unlock(env, fo);
582
583         return rc;
584 }
585
586 /**
587  * Truncate/punch OFD object.
588  *
589  * This function frees all of the allocated object's space from the \a start
590  * offset to the \a end offset. For truncate() operations the \a end offset
591  * is OBD_OBJECT_EOF. The functionality to punch holes in an object via
592  * fallocate(FALLOC_FL_PUNCH_HOLE) is not yet implemented (see LU-3606).
593  *
594  * \param[in] env       execution environment
595  * \param[in] fo        OFD object
596  * \param[in] start     start offset to punch from
597  * \param[in] end       end of punch
598  * \param[in] la        object attributes
599  * \param[in] ff        filter_fid structure
600  * \param[in] oa        obdo struct from incoming request
601  *
602  * \retval              0 if successful
603  * \retval              negative value on error
604  */
605 int ofd_object_punch(const struct lu_env *env, struct ofd_object *fo,
606                      __u64 start, __u64 end, struct lu_attr *la,
607                      struct filter_fid *ff, struct obdo *oa)
608 {
609         struct ofd_thread_info  *info = ofd_info(env);
610         struct ofd_device       *ofd = ofd_obj2dev(fo);
611         struct ofd_mod_data     *fmd;
612         struct dt_object        *dob = ofd_object_child(fo);
613         struct thandle          *th;
614         int                     ff_needed = 0;
615         int                     rc;
616         int                     rc2;
617
618         ENTRY;
619
620         /* we support truncate, not punch yet */
621         LASSERT(end == OBD_OBJECT_EOF);
622
623         ofd_write_lock(env, fo);
624         fmd = ofd_fmd_get(info->fti_exp, &fo->ofo_header.loh_fid);
625         if (fmd && fmd->fmd_mactime_xid < info->fti_xid)
626                 fmd->fmd_mactime_xid = info->fti_xid;
627         ofd_fmd_put(info->fti_exp, fmd);
628
629         if (!ofd_object_exists(fo))
630                 GOTO(unlock, rc = -ENOENT);
631
632         if (ofd->ofd_lfsck_verify_pfid && oa->o_valid & OBD_MD_FLFID) {
633                 rc = ofd_verify_ff(env, fo, oa);
634                 if (rc != 0)
635                         GOTO(unlock, rc);
636         }
637
638         /* VBR: version recovery check */
639         rc = ofd_version_get_check(info, fo);
640         if (rc)
641                 GOTO(unlock, rc);
642
643         rc = ofd_attr_handle_id(env, fo, la, 0 /* !is_setattr */);
644         if (rc != 0)
645                 GOTO(unlock, rc);
646
647         if (ff != NULL) {
648                 rc = ofd_object_ff_load(env, fo);
649                 if (rc == -ENODATA)
650                         ff_needed = 1;
651                 else if (rc < 0)
652                         GOTO(unlock, rc);
653         }
654
655         th = ofd_trans_create(env, ofd);
656         if (IS_ERR(th))
657                 GOTO(unlock, rc = PTR_ERR(th));
658
659         rc = dt_declare_attr_set(env, dob, la, th);
660         if (rc)
661                 GOTO(stop, rc);
662
663         rc = dt_declare_punch(env, dob, start, OBD_OBJECT_EOF, th);
664         if (rc)
665                 GOTO(stop, rc);
666
667         if (ff_needed) {
668                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
669                         ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
670                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
671                         le32_add_cpu(&ff->ff_parent.f_oid, -1);
672
673                 info->fti_buf.lb_buf = ff;
674                 info->fti_buf.lb_len = sizeof(*ff);
675                 rc = dt_declare_xattr_set(env, ofd_object_child(fo),
676                                           &info->fti_buf, XATTR_NAME_FID, 0,
677                                           th);
678                 if (rc)
679                         GOTO(stop, rc);
680         }
681
682         rc = ofd_trans_start(env, ofd, fo, th);
683         if (rc)
684                 GOTO(stop, rc);
685
686         rc = dt_punch(env, dob, start, OBD_OBJECT_EOF, th);
687         if (rc)
688                 GOTO(stop, rc);
689
690         rc = dt_attr_set(env, dob, la, th);
691         if (rc)
692                 GOTO(stop, rc);
693
694         if (ff_needed) {
695                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
696                         GOTO(stop, rc);
697
698                 rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
699                                   XATTR_NAME_FID, 0, th);
700                 if (!rc)
701                         filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
702         }
703
704         GOTO(stop, rc);
705
706 stop:
707         rc2 = ofd_trans_stop(env, ofd, th, rc);
708         if (rc2 != 0)
709                 CERROR("%s: failed to stop transaction: rc = %d\n",
710                        ofd_name(ofd), rc2);
711         if (!rc)
712                 rc = rc2;
713 unlock:
714         ofd_write_unlock(env, fo);
715
716         return rc;
717 }
718
719 /**
720  * Destroy OFD object.
721  *
722  * This function destroys OFD object. If object wasn't used at all (orphan)
723  * then local transaction is used, which means the transaction data is not
724  * returned back in reply.
725  *
726  * \param[in] env       execution environment
727  * \param[in] fo        OFD object
728  * \param[in] orphan    flag to indicate that object is orphaned
729  *
730  * \retval              0 if successful
731  * \retval              negative value on error
732  */
733 int ofd_destroy(const struct lu_env *env, struct ofd_object *fo,
734                        int orphan)
735 {
736         struct ofd_device       *ofd = ofd_obj2dev(fo);
737         struct thandle          *th;
738         int                     rc = 0;
739         int                     rc2;
740
741         ENTRY;
742
743         ofd_write_lock(env, fo);
744         if (!ofd_object_exists(fo))
745                 GOTO(unlock, rc = -ENOENT);
746
747         th = ofd_trans_create(env, ofd);
748         if (IS_ERR(th))
749                 GOTO(unlock, rc = PTR_ERR(th));
750
751         rc = dt_declare_ref_del(env, ofd_object_child(fo), th);
752         if (rc < 0)
753                 GOTO(stop, rc);
754
755         rc = dt_declare_destroy(env, ofd_object_child(fo), th);
756         if (rc < 0)
757                 GOTO(stop, rc);
758
759         if (orphan)
760                 rc = dt_trans_start_local(env, ofd->ofd_osd, th);
761         else
762                 rc = ofd_trans_start(env, ofd, NULL, th);
763         if (rc)
764                 GOTO(stop, rc);
765
766         ofd_fmd_drop(ofd_info(env)->fti_exp, &fo->ofo_header.loh_fid);
767
768         dt_ref_del(env, ofd_object_child(fo), th);
769         dt_destroy(env, ofd_object_child(fo), th);
770 stop:
771         rc2 = ofd_trans_stop(env, ofd, th, rc);
772         if (rc2)
773                 CERROR("%s failed to stop transaction: %d\n",
774                        ofd_name(ofd), rc2);
775         if (!rc)
776                 rc = rc2;
777 unlock:
778         ofd_write_unlock(env, fo);
779         RETURN(rc);
780 }
781
782 /**
783  * Get OFD object attributes.
784  *
785  * This function gets OFD object regular attributes. It is used to serve
786  * incoming request as well as for local OFD purposes.
787  *
788  * \param[in] env       execution environment
789  * \param[in] fo        OFD object
790  * \param[in] la        object attributes
791  *
792  * \retval              0 if successful
793  * \retval              negative value on error
794  */
795 int ofd_attr_get(const struct lu_env *env, struct ofd_object *fo,
796                  struct lu_attr *la)
797 {
798         int rc = 0;
799
800         ENTRY;
801
802         if (ofd_object_exists(fo)) {
803                 rc = dt_attr_get(env, ofd_object_child(fo), la);
804         } else {
805                 rc = -ENOENT;
806         }
807         RETURN(rc);
808 }