Whamcloud - gitweb
LU-13922 osd-ldiskfs: no need to add OI cache in readdir
[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, 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_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_lfsck.h>
45
46 #include "ofd_internal.h"
47
48 /**
49  * Get object version from disk and check it.
50  *
51  * This function checks object version from disk with
52  * ofd_thread_info::fti_pre_version filled from incoming RPC. This is part of
53  * VBR (Version-Based Recovery) and ensures that object has the same version
54  * upon replay as it has during original modification.
55  *
56  * \param[in]  info     execution thread OFD private data
57  * \param[in]  fo       OFD object
58  *
59  * \retval              0 if version matches
60  * \retval              -EOVERFLOW on version mismatch
61  */
62 static int ofd_version_get_check(struct ofd_thread_info *info,
63                                  struct ofd_object *fo)
64 {
65         dt_obj_version_t curr_version;
66
67         LASSERT(ofd_object_exists(fo));
68
69         if (info->fti_exp == NULL)
70                 RETURN(0);
71
72         curr_version = dt_version_get(info->fti_env, ofd_object_child(fo));
73         if ((__s64)curr_version == -EOPNOTSUPP)
74                 RETURN(0);
75         /* VBR: version is checked always because costs nothing */
76         if (info->fti_pre_version != 0 &&
77             info->fti_pre_version != curr_version) {
78                 CDEBUG(D_INODE, "Version mismatch %#llx != %#llx\n",
79                        info->fti_pre_version, curr_version);
80                 spin_lock(&info->fti_exp->exp_lock);
81                 info->fti_exp->exp_vbr_failed = 1;
82                 spin_unlock(&info->fti_exp->exp_lock);
83                 RETURN (-EOVERFLOW);
84         }
85         info->fti_pre_version = curr_version;
86         RETURN(0);
87 }
88
89 /**
90  * Get OFD object by FID.
91  *
92  * This function finds OFD slice of compound object with the given FID.
93  *
94  * \param[in] env       execution environment
95  * \param[in] ofd       OFD device
96  * \param[in] fid       FID of the object
97  *
98  * \retval              pointer to the found ofd_object
99  * \retval              ERR_PTR(errno) in case of error
100  */
101 struct ofd_object *ofd_object_find(const struct lu_env *env,
102                                    struct ofd_device *ofd,
103                                    const struct lu_fid *fid)
104 {
105         struct ofd_object *fo;
106         struct lu_object  *o;
107
108         ENTRY;
109
110         o = lu_object_find(env, &ofd->ofd_dt_dev.dd_lu_dev, fid, NULL);
111         if (likely(!IS_ERR(o)))
112                 fo = ofd_obj(o);
113         else
114                 fo = ERR_CAST(o); /* return error */
115
116         RETURN(fo);
117 }
118
119 /**
120  * Get FID of parent MDT object.
121  *
122  * This function reads extended attribute XATTR_NAME_FID of OFD object which
123  * contains the MDT parent object FID and saves it in ofd_object::ofo_ff.
124  *
125  * The filter_fid::ff_parent::f_ver field currently holds
126  * the OST-object index in the parent MDT-object's layout EA,
127  * not the actual FID::f_ver of the parent. We therefore access
128  * it via the macro f_stripe_idx.
129  *
130  * \param[in] env       execution environment
131  * \param[in] fo        OFD object
132  *
133  * \retval              0 if successful
134  * \retval              -ENODATA if there is no such xattr
135  * \retval              negative value on error
136  */
137 int ofd_object_ff_load(const struct lu_env *env, struct ofd_object *fo)
138 {
139         struct ofd_thread_info *info = ofd_info(env);
140         struct filter_fid *ff = &fo->ofo_ff;
141         struct lu_buf *buf = &info->fti_buf;
142         int rc = 0;
143
144         if (fid_is_sane(&ff->ff_parent))
145                 return 0;
146
147         buf->lb_buf = ff;
148         buf->lb_len = sizeof(*ff);
149         rc = dt_xattr_get(env, ofd_object_child(fo), buf, XATTR_NAME_FID);
150         if (rc < 0)
151                 return rc;
152
153         if (unlikely(rc < sizeof(struct lu_fid))) {
154                 fid_zero(&ff->ff_parent);
155                 return -EINVAL;
156         }
157
158         filter_fid_le_to_cpu(ff, ff, rc);
159
160         return 0;
161 }
162
163 struct ofd_precreate_cb {
164         struct dt_txn_commit_cb  opc_cb;
165         struct ofd_seq          *opc_oseq;
166         int                      opc_objects;
167 };
168
169 static void ofd_cb_precreate(struct lu_env *env, struct thandle *th,
170                              struct dt_txn_commit_cb *cb, int err)
171 {
172         struct ofd_precreate_cb *opc;
173         struct ofd_seq *oseq;
174
175         opc = container_of(cb, struct ofd_precreate_cb, opc_cb);
176         oseq = opc->opc_oseq;
177
178         CDEBUG(D_OTHER, "Sub %d from %d for "DFID", th_sync %d\n",
179                opc->opc_objects, atomic_read(&oseq->os_precreate_in_progress),
180                PFID(&oseq->os_oi.oi_fid), th->th_sync);
181         atomic_sub(opc->opc_objects, &oseq->os_precreate_in_progress);
182         ofd_seq_put(env, opc->opc_oseq);
183         OBD_FREE_PTR(opc);
184 }
185
186 static int ofd_precreate_cb_add(const struct lu_env *env, struct thandle *th,
187                                 struct ofd_seq *oseq, int objects)
188 {
189         struct ofd_precreate_cb *opc;
190         struct dt_txn_commit_cb *dcb;
191         int precreate, rc;
192
193         OBD_ALLOC_PTR(opc);
194         if (!opc)
195                 return -ENOMEM;
196
197         precreate = atomic_read(&oseq->os_precreate_in_progress);
198         atomic_inc(&oseq->os_refc);
199         opc->opc_oseq = oseq;
200         opc->opc_objects = objects;
201         CDEBUG(D_OTHER, "Add %d to %d for "DFID", th_sync %d\n",
202                opc->opc_objects, precreate,
203                PFID(&oseq->os_oi.oi_fid), th->th_sync);
204
205         if ((precreate + objects) >= (5 * OST_MAX_PRECREATE))
206                 th->th_sync = 1;
207
208         dcb = &opc->opc_cb;
209         dcb->dcb_func = ofd_cb_precreate;
210         INIT_LIST_HEAD(&dcb->dcb_linkage);
211         strlcpy(dcb->dcb_name, "ofd_cb_precreate", sizeof(dcb->dcb_name));
212
213         rc = dt_trans_cb_add(th, dcb);
214         if (rc) {
215                 ofd_seq_put(env, oseq);
216                 OBD_FREE_PTR(opc);
217                 return rc;
218         }
219
220         atomic_add(objects, &oseq->os_precreate_in_progress);
221
222         return 0;
223 }
224
225 /**
226  * Precreate the given number \a nr of objects in the given sequence \a oseq.
227  *
228  * This function precreates new OST objects in the given sequence.
229  * The precreation starts from \a id and creates \a nr objects sequentially.
230  *
231  * Notes:
232  * This function may create fewer objects than requested.
233  *
234  * We mark object SUID+SGID to flag it for accepting UID+GID from client on
235  * first write. Currently the permission bits on the OST are never used,
236  * so this is OK.
237  *
238  * Initialize a/c/m time so any client timestamp will always be newer and
239  * update the inode. The ctime = 0 case is also handled specially in
240  * osd_inode_setattr(). See LU-221, LU-1042 for details.
241  *
242  * \param[in] env       execution environment
243  * \param[in] ofd       OFD device
244  * \param[in] id        object ID to start precreation from
245  * \param[in] oseq      object sequence
246  * \param[in] nr        number of objects to precreate
247  * \param[in] sync      synchronous precreation flag
248  *
249  * \retval              0 if successful
250  * \retval              negative value on error
251  */
252 int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
253                           u64 id, struct ofd_seq *oseq, int nr, int sync)
254 {
255         struct ofd_thread_info  *info = ofd_info(env);
256         struct ofd_object       *fo = NULL;
257         struct dt_object        *next;
258         struct thandle          *th;
259         struct ofd_object       **batch;
260         struct lu_fid           *fid = &info->fti_fid;
261         u64                     tmp;
262         int                     rc;
263         int                     rc2;
264         int                     i;
265         int                     objects = 0;
266         int                     nr_saved = nr;
267
268         ENTRY;
269
270         /* Don't create objects beyond the valid range for this SEQ */
271         if (unlikely(fid_seq_is_mdt0(ostid_seq(&oseq->os_oi)) &&
272                      (id + nr) > IDIF_MAX_OID)) {
273                 CERROR("%s:"DOSTID" hit the IDIF_MAX_OID (1<<48)!\n",
274                        ofd_name(ofd), id, ostid_seq(&oseq->os_oi));
275                 RETURN(rc = -ENOSPC);
276         } else if (unlikely(!fid_seq_is_mdt0(ostid_seq(&oseq->os_oi)) &&
277                             (id + nr) > OBIF_MAX_OID)) {
278                 CERROR("%s:"DOSTID" hit the OBIF_MAX_OID (1<<32)!\n",
279                        ofd_name(ofd), id, ostid_seq(&oseq->os_oi));
280                 RETURN(rc = -ENOSPC);
281         }
282
283         OBD_ALLOC_PTR_ARRAY(batch, nr_saved);
284         if (batch == NULL)
285                 RETURN(-ENOMEM);
286
287         info->fti_attr.la_valid = LA_TYPE | LA_MODE;
288         info->fti_attr.la_mode = S_IFREG | S_ISUID | S_ISGID | S_ISVTX | 0666;
289         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
290
291         info->fti_attr.la_valid |= LA_ATIME | LA_MTIME | LA_CTIME;
292         info->fti_attr.la_atime = 0;
293         info->fti_attr.la_mtime = 0;
294         info->fti_attr.la_ctime = 0;
295
296         LASSERT(id != 0);
297
298         /* prepare objects */
299         *fid = *lu_object_fid(&oseq->os_lastid_obj->do_lu);
300         for (i = 0; i < nr; i++) {
301                 rc = fid_set_id(fid, id + i);
302                 if (rc != 0) {
303                         if (i == 0)
304                                 GOTO(out, rc);
305
306                         nr = i;
307                         break;
308                 }
309
310                 fo = ofd_object_find(env, ofd, fid);
311                 if (IS_ERR(fo)) {
312                         if (i == 0)
313                                 GOTO(out, rc = PTR_ERR(fo));
314
315                         nr = i;
316                         break;
317                 }
318
319                 batch[i] = fo;
320         }
321         info->fti_buf.lb_buf = &tmp;
322         info->fti_buf.lb_len = sizeof(tmp);
323         info->fti_off = 0;
324
325         th = ofd_trans_create(env, ofd);
326         if (IS_ERR(th))
327                 GOTO(out, rc = PTR_ERR(th));
328
329         th->th_sync |= sync;
330
331         rc = dt_declare_record_write(env, oseq->os_lastid_obj, &info->fti_buf,
332                                      info->fti_off, th);
333         if (rc)
334                 GOTO(trans_stop, rc);
335
336         for (i = 0; i < nr; i++) {
337                 fo = batch[i];
338                 LASSERT(fo);
339
340                 if (unlikely(ofd_object_exists(fo))) {
341                         /* object may exist being re-created by write replay */
342                         CDEBUG(D_INODE, "object %#llx/%#llx exists: "
343                                DFID"\n", ostid_seq(&oseq->os_oi), id,
344                                PFID(lu_object_fid(&fo->ofo_obj.do_lu)));
345                         continue;
346                 }
347
348                 next = ofd_object_child(fo);
349                 LASSERT(next != NULL);
350
351                 rc = dt_declare_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                         nr = i;
358                         break;
359                 }
360         }
361
362         rc = dt_trans_start_local(env, ofd->ofd_osd, th);
363         if (rc)
364                 GOTO(trans_stop, rc);
365
366         CDEBUG(D_OTHER, "%s: create new object "DFID" nr %d\n",
367                ofd_name(ofd), PFID(fid), nr);
368
369          /* When the LFSCK scanning the whole device to verify the LAST_ID file
370           * consistency, it will load the last_id into RAM firstly, and compare
371           * the last_id with each OST-object's ID. If the later one is larger,
372           * then it will regard the LAST_ID file crashed. But during the LFSCK
373           * scanning, the OFD may continue to create new OST-objects. Those new
374           * created OST-objects will have larger IDs than the LFSCK known ones.
375           * So from the LFSCK view, it needs to re-load the last_id from disk
376           * file, and if the latest last_id is still smaller than the object's
377           * ID, then the LAST_ID file is real crashed.
378           *
379           * To make above mechanism to work, before OFD pre-create OST-objects,
380           * it needs to update the LAST_ID file firstly, otherwise, the LFSCK
381           * may cannot get latest last_id although new OST-object created. */
382         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_SKIP_LASTID)) {
383                 tmp = cpu_to_le64(id + nr - 1);
384                 dt_write_lock(env, oseq->os_lastid_obj, DT_LASTID);
385                 rc = dt_record_write(env, oseq->os_lastid_obj,
386                                      &info->fti_buf, &info->fti_off, th);
387                 dt_write_unlock(env, oseq->os_lastid_obj);
388                 if (rc != 0)
389                         GOTO(trans_stop, rc);
390         }
391
392         for (i = 0; i < nr; i++) {
393                 fo = batch[i];
394                 LASSERT(fo);
395
396                 ofd_write_lock(env, fo);
397
398                 /* Only the new created objects need to be recorded. */
399                 if (ofd->ofd_osd->dd_record_fid_accessed) {
400                         struct lfsck_req_local *lrl = &ofd_info(env)->fti_lrl;
401
402                         lfsck_pack_rfa(lrl, lu_object_fid(&fo->ofo_obj.do_lu),
403                                        LEL_FID_ACCESSED, LFSCK_TYPE_LAYOUT);
404                         lfsck_in_notify_local(env, ofd->ofd_osd, lrl, NULL);
405                 }
406
407                 if (likely(!ofd_object_exists(fo) &&
408                            !OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING))) {
409                         next = ofd_object_child(fo);
410                         LASSERT(next != NULL);
411
412                         rc = dt_create(env, next, &info->fti_attr, NULL,
413                                        &info->fti_dof, th);
414                         ofd_write_unlock(env, fo);
415                         if (rc < 0) {
416                                 if (i == 0)
417                                         GOTO(trans_stop, rc);
418
419                                 rc = 0;
420                                 break;
421                         }
422                         LASSERT(ofd_object_exists(fo));
423                 } else {
424                         ofd_write_unlock(env, fo);
425                 }
426
427                 ofd_seq_last_oid_set(oseq, id + i);
428         }
429
430         objects = i;
431         /* NOT all the wanted objects have been created,
432          * set the LAST_ID as the real created. */
433         if (unlikely(objects < nr)) {
434                 int rc1;
435
436                 info->fti_off = 0;
437                 tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
438                 dt_write_lock(env, oseq->os_lastid_obj, DT_LASTID);
439                 rc1 = dt_record_write(env, oseq->os_lastid_obj,
440                                       &info->fti_buf, &info->fti_off, th);
441                 dt_write_unlock(env, oseq->os_lastid_obj);
442                 if (rc1 != 0)
443                         CERROR("%s: fail to reset the LAST_ID for seq (%#llx"
444                                ") from %llu to %llu\n", ofd_name(ofd),
445                                ostid_seq(&oseq->os_oi), id + nr - 1,
446                                ofd_seq_last_oid(oseq));
447         }
448
449         if (objects)
450                 ofd_precreate_cb_add(env, th, oseq, objects);
451 trans_stop:
452         rc2 = ofd_trans_stop(env, ofd, th, rc);
453         if (rc2)
454                 CERROR("%s: failed to stop transaction: rc = %d\n",
455                        ofd_name(ofd), rc2);
456         if (!rc)
457                 rc = rc2;
458 out:
459         for (i = 0; i < nr_saved; i++) {
460                 fo = batch[i];
461                 if (!fo)
462                         continue;
463                 ofd_object_put(env, fo);
464         }
465         OBD_FREE_PTR_ARRAY(batch, nr_saved);
466
467         CDEBUG((objects == 0 && rc == 0) ? D_ERROR : D_OTHER,
468                "created %d/%d objects: %d\n", objects, nr_saved, rc);
469
470         LASSERT(ergo(objects == 0, rc < 0));
471         RETURN(objects > 0 ? objects : rc);
472 }
473
474 /**
475  * Fix the OFD object ownership.
476  *
477  * If the object still has SUID+SGID bits set, meaning that it was precreated
478  * by the MDT before it was assigned to any file, (see ofd_precreate_objects())
479  * then we will accept the UID/GID/PROJID if sent by the client for initializing
480  * the ownership of this object.  We only allow this to happen once (so clear
481  * these bits) and later only allow setattr.
482  *
483  * \param[in] env        execution environment
484  * \param[in] fo         OFD object
485  * \param[in] la         object attributes
486  * \param[in] is_setattr was this function called from setattr or not
487  *
488  * \retval              0 if successful
489  * \retval              negative value on error
490  */
491 int ofd_attr_handle_id(const struct lu_env *env, struct ofd_object *fo,
492                          struct lu_attr *la, int is_setattr)
493 {
494         struct ofd_thread_info  *info = ofd_info(env);
495         struct lu_attr          *ln = &info->fti_attr2;
496         __u32                    mask = 0;
497         int                      rc;
498
499         ENTRY;
500
501         if (!(la->la_valid & LA_UID) && !(la->la_valid & LA_GID) &&
502             !(la->la_valid & LA_PROJID))
503                 RETURN(0);
504
505         rc = dt_attr_get(env, ofd_object_child(fo), ln);
506         if (rc != 0)
507                 RETURN(rc);
508
509         LASSERT(ln->la_valid & LA_MODE);
510
511         /*
512          * Only allow setattr to change UID/GID/PROJID, if
513          * SUID+SGID is not set which means this is not
514          * initialization of this objects.
515          */
516         if (!is_setattr) {
517                 if (!(ln->la_mode & S_ISUID))
518                         la->la_valid &= ~LA_UID;
519                 if (!(ln->la_mode & S_ISGID))
520                         la->la_valid &= ~LA_GID;
521                 if (!(ln->la_mode & S_ISVTX))
522                         la->la_valid &= ~LA_PROJID;
523         }
524
525         /* Initialize ownership of this object, clear SUID+SGID bits*/
526         if ((la->la_valid & LA_UID) && (ln->la_mode & S_ISUID))
527                 mask |= S_ISUID;
528         if ((la->la_valid & LA_GID) && (ln->la_mode & S_ISGID))
529                 mask |= S_ISGID;
530         if ((la->la_valid & LA_PROJID) && (ln->la_mode & S_ISVTX))
531                 mask |= S_ISVTX;
532         if (mask != 0) {
533                 if (!(la->la_valid & LA_MODE) || !is_setattr) {
534                         la->la_mode = ln->la_mode;
535                         la->la_valid |= LA_MODE;
536                 }
537                 la->la_mode &= ~mask;
538         }
539
540         RETURN(0);
541 }
542
543 /**
544  * Check if it needs to update filter_fid by the value of @oa.
545  *
546  * \param[in] env       env
547  * \param[in] fo        ofd object
548  * \param[in] oa        obdo from client or MDT
549  * \param[out] ff       if filter_fid needs updating, this field is used to
550  *                      return the new buffer
551  *
552  * \retval < 0          error occurred
553  * \retval 0            doesn't need to update filter_fid
554  * \retval FL_XATTR_{CREATE,REPLACE}    flag for xattr update
555  */
556 int ofd_object_ff_update(const struct lu_env *env, struct ofd_object *fo,
557                          const struct obdo *oa, struct filter_fid *ff)
558 {
559         int rc = 0;
560         ENTRY;
561
562         if (!(oa->o_valid &
563               (OBD_MD_FLFID | OBD_MD_FLOSTLAYOUT | OBD_MD_LAYOUT_VERSION)))
564                 RETURN(0);
565
566         rc = ofd_object_ff_load(env, fo);
567         if (rc < 0 && rc != -ENODATA)
568                 RETURN(rc);
569
570         LASSERT(ff != &fo->ofo_ff);
571         if (rc == -ENODATA) {
572                 rc = LU_XATTR_CREATE;
573                 memset(ff, 0, sizeof(*ff));
574         } else {
575                 rc = LU_XATTR_REPLACE;
576                 memcpy(ff, &fo->ofo_ff, sizeof(*ff));
577         }
578
579         if (oa->o_valid & OBD_MD_FLFID) {
580                 /* packing fid and converting it to LE for storing into EA.
581                  * Here ->o_stripe_idx should be filled by LOV and rest of
582                  * fields - by client. */
583                 ff->ff_parent.f_seq = oa->o_parent_seq;
584                 ff->ff_parent.f_oid = oa->o_parent_oid;
585                 /* XXX: we are ignoring o_parent_ver here, since this should
586                  *      be the same for all objects in this fileset. */
587                 ff->ff_parent.f_ver = oa->o_stripe_idx;
588         }
589         if (oa->o_valid & OBD_MD_FLOSTLAYOUT)
590                 ff->ff_layout = oa->o_layout;
591
592         if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
593                 CDEBUG(D_INODE, DFID": OST("DFID") layout version %u -> %u\n",
594                        PFID(&fo->ofo_ff.ff_parent),
595                        PFID(lu_object_fid(&fo->ofo_obj.do_lu)),
596                        ff->ff_layout_version, oa->o_layout_version);
597
598                 /* only the MDS has the authority to update layout version */
599                 if (!(exp_connect_flags(ofd_info(env)->fti_exp) &
600                       OBD_CONNECT_MDS)) {
601                         CERROR(DFID": update layout version from client\n",
602                                PFID(&fo->ofo_ff.ff_parent));
603
604                         RETURN(-EPERM);
605                 }
606
607                 if (ff->ff_layout_version & LU_LAYOUT_RESYNC) {
608                         /* this opens a new era of writing */
609                         ff->ff_layout_version = 0;
610                         ff->ff_range = 0;
611                 }
612
613                 /* it's not allowed to change it to a smaller value */
614                 if (oa->o_layout_version < ff->ff_layout_version)
615                         RETURN(-EINVAL);
616
617                 if (ff->ff_layout_version == 0 ||
618                     oa->o_layout_version & LU_LAYOUT_RESYNC) {
619                         /* if LU_LAYOUT_RESYNC is set, it closes the era of
620                          * writing. Only mirror I/O can write this object. */
621                         ff->ff_layout_version = oa->o_layout_version;
622                         ff->ff_range = 0;
623                 } else if (oa->o_layout_version > ff->ff_layout_version) {
624                         ff->ff_range = max_t(__u32, ff->ff_range,
625                                              oa->o_layout_version -
626                                              ff->ff_layout_version);
627                 }
628         }
629
630         if (memcmp(ff, &fo->ofo_ff, sizeof(*ff)))
631                 filter_fid_cpu_to_le(ff, ff, sizeof(*ff));
632         else /* no change */
633                 rc = 0;
634
635         RETURN(rc);
636 }
637
638 /**
639  * Set OFD object attributes.
640  *
641  * This function sets OFD object attributes taken from incoming request.
642  * It sets not only regular attributes but also XATTR_NAME_FID extended
643  * attribute if needed. The "fid" xattr allows the object's MDT parent inode
644  * to be found and verified by LFSCK and other tools in case of inconsistency.
645  *
646  * \param[in] env       execution environment
647  * \param[in] fo        OFD object
648  * \param[in] la        object attributes
649  * \param[in] oa        obdo carries fid, ost_layout, layout version
650  *
651  * \retval              0 if successful
652  * \retval              negative value on error
653  */
654 int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
655                  struct lu_attr *la, struct obdo *oa)
656 {
657         struct ofd_thread_info *info = ofd_info(env);
658         struct ofd_device *ofd = ofd_obj2dev(fo);
659         struct filter_fid *ff = &info->fti_mds_fid;
660         struct thandle *th;
661         int fl, rc, rc2;
662
663         ENTRY;
664
665         if (!ofd_object_exists(fo))
666                 GOTO(out, rc = -ENOENT);
667
668         /* VBR: version recovery check */
669         rc = ofd_version_get_check(info, fo);
670         if (rc)
671                 GOTO(out, rc);
672
673         rc = ofd_attr_handle_id(env, fo, la, 1 /* is_setattr */);
674         if (rc != 0)
675                 GOTO(out, rc);
676
677         th = ofd_trans_create(env, ofd);
678         if (IS_ERR(th))
679                 GOTO(out, rc = PTR_ERR(th));
680
681         rc = dt_declare_attr_set(env, ofd_object_child(fo), la, th);
682         if (rc)
683                 GOTO(stop, rc);
684
685         info->fti_buf.lb_buf = ff;
686         info->fti_buf.lb_len = sizeof(*ff);
687         rc = dt_declare_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
688                                   XATTR_NAME_FID, 0, th);
689         if (rc)
690                 GOTO(stop, rc);
691
692         rc = ofd_trans_start(env, ofd, la->la_valid & LA_SIZE ? fo : NULL, th);
693         if (rc)
694                 GOTO(stop, rc);
695
696         ofd_write_lock(env, fo);
697         if (!ofd_object_exists(fo))
698                 GOTO(unlock, rc = -ENOENT);
699
700         /* serialize vs ofd_commitrw_write() */
701         if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME))
702                 tgt_fmd_update(info->fti_exp, &fo->ofo_header.loh_fid,
703                                info->fti_xid);
704
705         rc = dt_attr_set(env, ofd_object_child(fo), la, th);
706         if (rc)
707                 GOTO(unlock, rc);
708
709         fl = ofd_object_ff_update(env, fo, oa, ff);
710         if (fl < 0)
711                 GOTO(unlock, rc = fl);
712
713         if (fl) {
714                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
715                         ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
716                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
717                         le32_add_cpu(&ff->ff_parent.f_oid, -1);
718                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
719                         GOTO(unlock, rc);
720
721                 info->fti_buf.lb_buf = ff;
722                 info->fti_buf.lb_len = sizeof(*ff);
723                 rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
724                                   XATTR_NAME_FID, fl, th);
725                 if (!rc)
726                         filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
727         }
728
729         GOTO(unlock, rc);
730
731 unlock:
732         ofd_write_unlock(env, fo);
733 stop:
734         rc2 = ofd_trans_stop(env, ofd, th, rc);
735         if (rc2)
736                 CERROR("%s: failed to stop transaction: rc = %d\n",
737                        ofd_name(ofd), rc2);
738         if (!rc)
739                 rc = rc2;
740 out:
741         return rc;
742 }
743
744 /**
745  * Fallocate(Preallocate) space for OFD object.
746  *
747  * This function allocates space for the object from the \a start
748  * offset to the \a end offset.
749  *
750  * \param[in] env       execution environment
751  * \param[in] fo        OFD object
752  * \param[in] start     start offset to allocate from
753  * \param[in] end       end of allocate
754  * \param[in] mode      fallocate mode
755  * \param[in] la        object attributes
756  * \param[in] ff        filter_fid structure
757  *
758  * \retval              0 if successful
759  * \retval              negative value on error
760  */
761 int ofd_object_fallocate(const struct lu_env *env, struct ofd_object *fo,
762                         __u64 start, __u64 end, int mode, struct lu_attr *la,
763                         struct obdo *oa)
764 {
765         struct ofd_thread_info *info = ofd_info(env);
766         struct ofd_device *ofd = ofd_obj2dev(fo);
767         struct dt_object *dob = ofd_object_child(fo);
768         struct thandle *th;
769         struct filter_fid *ff = &info->fti_mds_fid;
770         bool ff_needed = false;
771         int rc;
772
773         ENTRY;
774
775         ofd_write_lock(env, fo);
776         if (!ofd_object_exists(fo))
777                 GOTO(unlock, rc = -ENOENT);
778
779         /* VBR: version recovery check */
780         rc = ofd_version_get_check(info, fo);
781         if (rc != 0)
782                 GOTO(unlock, rc);
783
784         if (ff != NULL) {
785                 rc = ofd_object_ff_load(env, fo);
786                 if (rc == -ENODATA)
787                         ff_needed = true;
788                 else if (rc < 0)
789                         GOTO(unlock, rc);
790         }
791
792         th = ofd_trans_create(env, ofd);
793         if (IS_ERR(th))
794                 GOTO(unlock, rc = PTR_ERR(th));
795
796         rc = dt_declare_attr_set(env, dob, la, th);
797         if (rc)
798                 GOTO(stop, rc);
799
800         rc = dt_declare_falloc(env, dob, th);
801         if (rc)
802                 GOTO(stop, rc);
803
804         rc = ofd_trans_start(env, ofd, fo, th);
805         if (rc)
806                 GOTO(stop, rc);
807
808         rc = dt_falloc(env, dob, start, end, mode, th);
809         if (rc)
810                 GOTO(stop, rc);
811
812         rc = dt_attr_set(env, dob, la, th);
813         if (rc)
814                 GOTO(stop, rc);
815
816         if (ff_needed) {
817                 rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
818                                   XATTR_NAME_FID, 0, th);
819                 if (!rc)
820                         filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
821         }
822 stop:
823         ofd_trans_stop(env, ofd, th, rc);
824 unlock:
825         ofd_write_unlock(env, fo);
826         RETURN(rc);
827 }
828
829 /**
830  * Truncate/punch OFD object.
831  *
832  * This function frees all of the allocated object's space from the \a start
833  * offset to the \a end offset. For truncate() operations the \a end offset
834  * is OBD_OBJECT_EOF. The functionality to punch holes in an object via
835  * fallocate(FALLOC_FL_PUNCH_HOLE) is not yet implemented (see LU-3606).
836  *
837  * \param[in] env       execution environment
838  * \param[in] fo        OFD object
839  * \param[in] start     start offset to punch from
840  * \param[in] end       end of punch
841  * \param[in] la        object attributes
842  * \param[in] oa        obdo struct from incoming request
843  *
844  * \retval              0 if successful
845  * \retval              negative value on error
846  */
847 int ofd_object_punch(const struct lu_env *env, struct ofd_object *fo,
848                      __u64 start, __u64 end, struct lu_attr *la,
849                      struct obdo *oa)
850 {
851         struct ofd_thread_info *info = ofd_info(env);
852         struct ofd_device *ofd = ofd_obj2dev(fo);
853         struct dt_object *dob = ofd_object_child(fo);
854         struct filter_fid *ff = &info->fti_mds_fid;
855         struct thandle *th;
856         int fl, rc, rc2;
857
858         ENTRY;
859
860         /* we support truncate, not punch yet */
861         LASSERT(end == OBD_OBJECT_EOF);
862
863         if (!ofd_object_exists(fo))
864                 GOTO(out, rc = -ENOENT);
865
866         if (ofd->ofd_lfsck_verify_pfid && oa->o_valid & OBD_MD_FLFID) {
867                 rc = ofd_verify_ff(env, fo, oa);
868                 if (rc != 0)
869                         GOTO(out, rc);
870         }
871
872         /* VBR: version recovery check */
873         rc = ofd_version_get_check(info, fo);
874         if (rc)
875                 GOTO(out, rc);
876
877         rc = ofd_attr_handle_id(env, fo, la, 0 /* !is_setattr */);
878         if (rc != 0)
879                 GOTO(out, rc);
880
881         th = ofd_trans_create(env, ofd);
882         if (IS_ERR(th))
883                 GOTO(out, rc = PTR_ERR(th));
884
885         rc = dt_declare_attr_set(env, dob, la, th);
886         if (rc)
887                 GOTO(stop, rc);
888
889         rc = dt_declare_punch(env, dob, start, OBD_OBJECT_EOF, th);
890         if (rc)
891                 GOTO(stop, rc);
892
893         info->fti_buf.lb_buf = ff;
894         info->fti_buf.lb_len = sizeof(*ff);
895         rc = dt_declare_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
896                                   XATTR_NAME_FID, 0, th);
897         if (rc)
898                 GOTO(stop, rc);
899
900         rc = ofd_trans_start(env, ofd, fo, th);
901         if (rc)
902                 GOTO(stop, rc);
903
904         ofd_write_lock(env, fo);
905
906         if (la->la_valid & (LA_ATIME | LA_MTIME | LA_CTIME))
907                 tgt_fmd_update(info->fti_exp, &fo->ofo_header.loh_fid,
908                                info->fti_xid);
909
910         if (!ofd_object_exists(fo))
911                 GOTO(unlock, rc = -ENOENT);
912
913         /* need to verify layout version */
914         if (oa->o_valid & OBD_MD_LAYOUT_VERSION) {
915                 rc = ofd_verify_layout_version(env, fo, oa);
916                 if (rc)
917                         GOTO(unlock, rc);
918
919                 oa->o_valid &= ~OBD_MD_LAYOUT_VERSION;
920         }
921
922         if (oa->o_valid & OBD_MD_FLFLAGS && oa->o_flags & LUSTRE_ENCRYPT_FL) {
923                 /* punch must be aware we are dealing with an encrypted file */
924                 struct lu_attr la = {
925                         .la_valid = LA_FLAGS,
926                         .la_flags = LUSTRE_ENCRYPT_FL,
927                 };
928
929                 rc = dt_attr_set(env, dob, &la, th);
930                 if (rc)
931                         GOTO(unlock, rc);
932         }
933         rc = dt_punch(env, dob, start, OBD_OBJECT_EOF, th);
934         if (rc)
935                 GOTO(unlock, rc);
936
937         fl = ofd_object_ff_update(env, fo, oa, ff);
938         if (fl < 0)
939                 GOTO(unlock, rc = fl);
940
941         rc = dt_attr_set(env, dob, la, th);
942         if (rc)
943                 GOTO(unlock, rc);
944
945         if (fl) {
946                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR1))
947                         ff->ff_parent.f_oid = cpu_to_le32(1UL << 31);
948                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_UNMATCHED_PAIR2))
949                         le32_add_cpu(&ff->ff_parent.f_oid, -1);
950                 else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NOPFID))
951                         GOTO(unlock, rc);
952
953                 info->fti_buf.lb_buf = ff;
954                 info->fti_buf.lb_len = sizeof(*ff);
955                 rc = dt_xattr_set(env, ofd_object_child(fo), &info->fti_buf,
956                                   XATTR_NAME_FID, fl, th);
957                 if (!rc)
958                         filter_fid_le_to_cpu(&fo->ofo_ff, ff, sizeof(*ff));
959         }
960
961         GOTO(unlock, rc);
962
963 unlock:
964         ofd_write_unlock(env, fo);
965 stop:
966         rc2 = ofd_trans_stop(env, ofd, th, rc);
967         if (rc2 != 0)
968                 CERROR("%s: failed to stop transaction: rc = %d\n",
969                        ofd_name(ofd), rc2);
970         if (!rc)
971                 rc = rc2;
972 out:
973         return rc;
974 }
975
976 /**
977  * Destroy OFD object.
978  *
979  * This function destroys OFD object. If object wasn't used at all (orphan)
980  * then local transaction is used, which means the transaction data is not
981  * returned back in reply.
982  *
983  * \param[in] env       execution environment
984  * \param[in] fo        OFD object
985  * \param[in] orphan    flag to indicate that object is orphaned
986  *
987  * \retval              0 if successful
988  * \retval              negative value on error
989  */
990 int ofd_destroy(const struct lu_env *env, struct ofd_object *fo,
991                        int orphan)
992 {
993         struct ofd_device       *ofd = ofd_obj2dev(fo);
994         struct thandle          *th;
995         int                     rc = 0;
996         int                     rc2;
997
998         ENTRY;
999
1000         if (!ofd_object_exists(fo))
1001                 GOTO(out, rc = -ENOENT);
1002
1003         th = ofd_trans_create(env, ofd);
1004         if (IS_ERR(th))
1005                 GOTO(out, rc = PTR_ERR(th));
1006
1007         rc = dt_declare_ref_del(env, ofd_object_child(fo), th);
1008         if (rc < 0)
1009                 GOTO(stop, rc);
1010
1011         rc = dt_declare_destroy(env, ofd_object_child(fo), th);
1012         if (rc < 0)
1013                 GOTO(stop, rc);
1014
1015         if (orphan)
1016                 rc = dt_trans_start_local(env, ofd->ofd_osd, th);
1017         else
1018                 rc = ofd_trans_start(env, ofd, NULL, th);
1019         if (rc)
1020                 GOTO(stop, rc);
1021
1022         ofd_write_lock(env, fo);
1023         if (!ofd_object_exists(fo))
1024                 GOTO(stop, rc = -ENOENT);
1025
1026         tgt_fmd_drop(ofd_info(env)->fti_exp, &fo->ofo_header.loh_fid);
1027
1028         dt_ref_del(env, ofd_object_child(fo), th);
1029         dt_destroy(env, ofd_object_child(fo), th);
1030         ofd_write_unlock(env, fo);
1031
1032 stop:
1033         rc2 = ofd_trans_stop(env, ofd, th, rc);
1034         if (rc2)
1035                 CERROR("%s failed to stop transaction: %d\n",
1036                        ofd_name(ofd), rc2);
1037         if (!rc)
1038                 rc = rc2;
1039 out:
1040         RETURN(rc);
1041 }
1042
1043 /**
1044  * Get OFD object attributes.
1045  *
1046  * This function gets OFD object regular attributes. It is used to serve
1047  * incoming request as well as for local OFD purposes.
1048  *
1049  * \param[in] env       execution environment
1050  * \param[in] fo        OFD object
1051  * \param[in] la        object attributes
1052  *
1053  * \retval              0 if successful
1054  * \retval              negative value on error
1055  */
1056 int ofd_attr_get(const struct lu_env *env, struct ofd_object *fo,
1057                  struct lu_attr *la)
1058 {
1059         int rc = 0;
1060
1061         ENTRY;
1062
1063         if (ofd_object_exists(fo)) {
1064                 rc = dt_attr_get(env, ofd_object_child(fo), la);
1065         } else {
1066                 rc = -ENOENT;
1067         }
1068         RETURN(rc);
1069 }