4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License version 2 for more details. A copy is
14 * included in the COPYING file that accompanied this code.
16 * You should have received a copy of the GNU General Public License
17 * along with this program; if not, write to the Free Software
18 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23 * Copyright (c) 2015, 2017, Intel Corporation.
26 * lustre/lod/lod_sub_object.c
28 * LOD sub object methods
30 * This file implements sub-object methods for LOD.
32 * LOD is Logic volume layer in the MDS stack, which will handle striping
33 * and distribute the update to different OSP/OSD. After directing the updates
34 * to one specific OSD/OSP, it also needs to do some thing before calling
35 * OSD/OSP API, for example recording updates for cross-MDT operation, get
36 * the next level transaction etc.
38 * Author: Di Wang <di.wang@intel.com>
41 #define DEBUG_SUBSYSTEM S_MDS
44 #include <obd_class.h>
45 #include <uapi/linux/lustre/lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
49 #include <lustre_fid.h>
50 #include <uapi/linux/lustre/lustre_param.h>
51 #include <md_object.h>
52 #include <lustre_linkea.h>
53 #include <lustre_log.h>
55 #include "lod_internal.h"
57 struct thandle *lod_sub_get_thandle(const struct lu_env *env,
59 const struct dt_object *sub_obj,
62 struct lod_device *lod = dt2lod_dev(th->th_dev);
63 struct top_thandle *tth;
64 struct thandle *sub_th;
65 int type = LU_SEQ_RANGE_ANY;
70 if (record_update != NULL)
71 *record_update = false;
73 if (th->th_top == NULL)
76 tth = container_of(th, struct top_thandle, tt_super);
77 tth->tt_master_sub_thandle->th_ignore_quota = th->th_ignore_quota;
79 /* local object must be mdt object, Note: during ost object
80 * creation, FID is not assigned until osp_create(),
81 * so if the FID of sub_obj is zero, it means OST object. */
82 if (!dt_object_remote(sub_obj) ||
83 fid_is_zero(lu_object_fid(&sub_obj->do_lu))) {
84 /* local MDT object */
85 if (fid_is_sane(lu_object_fid(&sub_obj->do_lu)) &&
86 tth->tt_multiple_thandle != NULL &&
87 record_update != NULL &&
89 *record_update = true;
91 RETURN(tth->tt_master_sub_thandle);
94 rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
99 /* th_complex means we need track all of updates for this
100 * transaction, include changes on OST */
101 if (type == LU_SEQ_RANGE_OST && !th->th_complex)
102 RETURN(tth->tt_master_sub_thandle);
104 sub_th = thandle_get_sub(env, th, sub_obj);
107 sub_th->th_ignore_quota = th->th_ignore_quota;
109 if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
111 *record_update = true;
117 * Declare sub-object creation.
119 * Get transaction of next layer and declare the creation of the object.
121 * \param[in] env execution environment
122 * \param[in] dt the object being created
123 * \param[in] attr the attributes of the object being created
124 * \param[in] hint the hint of the creation
125 * \param[in] dof the object format of the creation
126 * \param[th] th the transaction handle
128 * \retval 0 if the declaration succeeds
129 * \retval negative errno if the declaration fails.
131 int lod_sub_declare_create(const struct lu_env *env, struct dt_object *dt,
132 struct lu_attr *attr,
133 struct dt_allocation_hint *hint,
134 struct dt_object_format *dof, struct thandle *th)
136 struct thandle *sub_th;
139 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
141 return PTR_ERR(sub_th);
144 update_record_size(env, create, th, lu_object_fid(&dt->do_lu),
147 return dt_declare_create(env, dt, attr, hint, dof, sub_th);
153 * Get transaction of next layer, record updates if it belongs to cross-MDT
154 * operation, and create the object.
156 * \param[in] env execution environment
157 * \param[in] dt the object being created
158 * \param[in] attr the attributes of the object being created
159 * \param[in] hint the hint of the creation
160 * \param[in] dof the object format of the creation
161 * \param[th] th the transaction handle
163 * \retval 0 if the creation succeeds
164 * \retval negative errno if the creation fails.
166 int lod_sub_create(const struct lu_env *env, struct dt_object *dt,
167 struct lu_attr *attr, struct dt_allocation_hint *hint,
168 struct dt_object_format *dof, struct thandle *th)
170 struct thandle *sub_th;
175 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
177 RETURN(PTR_ERR(sub_th));
180 rc = update_record_pack(create, th,
181 lu_object_fid(&dt->do_lu),
187 rc = dt_create(env, dt, attr, hint, dof, sub_th);
193 * Declare adding reference for the sub-object
195 * Get transaction of next layer and declare the reference adding.
197 * \param[in] env execution environment
198 * \param[in] dt dt object to add reference
199 * \param[in] th transaction handle
201 * \retval 0 if the declaration succeeds.
202 * \retval negative errno if the declaration fails.
204 int lod_sub_declare_ref_add(const struct lu_env *env, struct dt_object *dt,
207 struct thandle *sub_th;
212 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
214 RETURN(PTR_ERR(sub_th));
217 update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
219 rc = dt_declare_ref_add(env, dt, sub_th);
225 * Add reference for the sub-object
227 * Get transaction of next layer, record updates if it belongs to cross-MDT
228 * operation and add reference of the object.
230 * \param[in] env execution environment
231 * \param[in] dt dt object to add reference
232 * \param[in] th transaction handle
234 * \retval 0 if it succeeds.
235 * \retval negative errno if it fails.
237 int lod_sub_ref_add(const struct lu_env *env, struct dt_object *dt,
240 struct thandle *sub_th;
245 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
247 RETURN(PTR_ERR(sub_th));
250 rc = update_record_pack(ref_add, th,
251 lu_object_fid(&dt->do_lu));
256 rc = dt_ref_add(env, dt, sub_th);
262 * Declare deleting reference for the sub-object
264 * Get transaction of next layer and declare the reference deleting.
266 * \param[in] env execution environment
267 * \param[in] dt dt object to delete reference
268 * \param[in] th transaction handle
270 * \retval 0 if the declaration succeeds.
271 * \retval negative errno if the declaration fails.
273 int lod_sub_declare_ref_del(const struct lu_env *env, struct dt_object *dt,
276 struct thandle *sub_th;
281 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
283 RETURN(PTR_ERR(sub_th));
286 update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
288 rc = dt_declare_ref_del(env, dt, sub_th);
294 * Delete reference for the sub-object
296 * Get transaction of next layer, record updates if it belongs to cross-MDT
297 * operation and delete reference of the object.
299 * \param[in] env execution environment
300 * \param[in] dt dt object to delete reference
301 * \param[in] th transaction handle
303 * \retval 0 if it succeeds.
304 * \retval negative errno if it fails.
306 int lod_sub_ref_del(const struct lu_env *env, struct dt_object *dt,
309 struct thandle *sub_th;
314 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
316 RETURN(PTR_ERR(sub_th));
319 rc = update_record_pack(ref_del, th,
320 lu_object_fid(&dt->do_lu));
325 rc = dt_ref_del(env, dt, sub_th);
331 * Declare destroying sub-object
333 * Get transaction of next layer and declare the sub-object destroy.
335 * \param[in] env execution environment
336 * \param[in] dt dt object to be destroyed
337 * \param[in] th transaction handle
339 * \retval 0 if the declaration succeeds.
340 * \retval negative errno if the declaration fails.
342 int lod_sub_declare_destroy(const struct lu_env *env, struct dt_object *dt,
345 struct thandle *sub_th;
350 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
352 RETURN(PTR_ERR(sub_th));
355 update_record_size(env, destroy, th, lu_object_fid(&dt->do_lu));
357 rc = dt_declare_destroy(env, dt, sub_th);
365 * Get transaction of next layer, record updates if it belongs to cross-MDT
366 * operation and destroy the object.
368 * \param[in] env execution environment
369 * \param[in] dt dt object to be destroyed
370 * \param[in] th transaction handle
372 * \retval 0 if the destroy succeeds.
373 * \retval negative errno if the destroy fails.
375 int lod_sub_destroy(const struct lu_env *env, struct dt_object *dt,
378 struct thandle *sub_th;
383 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
385 RETURN(PTR_ERR(sub_th));
388 rc = update_record_pack(destroy, th, lu_object_fid(&dt->do_lu));
393 rc = dt_destroy(env, dt, sub_th);
399 * Declare sub-object index insert
401 * Get transaction of next layer and declare index insert.
403 * \param[in] env execution environment
404 * \param[in] dt object for which to insert index
405 * \param[in] rec record of the index which will be inserted
406 * \param[in] key key of the index which will be inserted
407 * \param[in] th the transaction handle
409 * \retval 0 if the declaration succeeds.
410 * \retval negative errno if the declaration fails.
412 int lod_sub_declare_insert(const struct lu_env *env, struct dt_object *dt,
413 const struct dt_rec *rec,
414 const struct dt_key *key, struct thandle *th)
416 struct thandle *sub_th;
419 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
421 return PTR_ERR(sub_th);
424 update_record_size(env, index_insert, th,
425 lu_object_fid(&dt->do_lu), rec, key);
427 return dt_declare_insert(env, dt, rec, key, sub_th);
431 * Insert index of sub object
433 * Get transaction of next layer, record updates if it belongs to cross-MDT
434 * operation, and insert the index.
436 * \param[in] env execution environment
437 * \param[in] dt object for which to insert index
438 * \param[in] rec record of the index to be inserted
439 * \param[in] key key of the index to be inserted
440 * \param[in] th the transaction handle
441 * \param[in] ign whether ignore quota
443 * \retval 0 if the insertion succeeds.
444 * \retval negative errno if the insertion fails.
446 int lod_sub_insert(const struct lu_env *env, struct dt_object *dt,
447 const struct dt_rec *rec, const struct dt_key *key,
448 struct thandle *th, int ign)
450 struct thandle *sub_th;
454 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
456 return PTR_ERR(sub_th);
459 rc = update_record_pack(index_insert, th,
460 lu_object_fid(&dt->do_lu), rec, key);
465 return dt_insert(env, dt, rec, key, sub_th, ign);
469 * Declare sub-object index delete
471 * Get transaction of next layer and declare index deletion.
473 * \param[in] env execution environment
474 * \param[in] dt object for which to delete index
475 * \param[in] key key of the index which will be deleted
476 * \param[in] th the transaction handle
478 * \retval 0 if the declaration succeeds.
479 * \retval negative errno if the declaration fails.
481 int lod_sub_declare_delete(const struct lu_env *env, struct dt_object *dt,
482 const struct dt_key *key, struct thandle *th)
484 struct thandle *sub_th;
487 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
489 return PTR_ERR(sub_th);
492 update_record_size(env, index_delete, th,
493 lu_object_fid(&dt->do_lu), key);
495 return dt_declare_delete(env, dt, key, sub_th);
499 * Delete index of sub object
501 * Get transaction of next layer, record updates if it belongs to cross-MDT
502 * operation, and delete the index.
504 * \param[in] env execution environment
505 * \param[in] dt object for which to delete index
506 * \param[in] key key of the index to be deleted
507 * \param[in] th the transaction handle
509 * \retval 0 if the deletion succeeds.
510 * \retval negative errno if the deletion fails.
512 int lod_sub_delete(const struct lu_env *env, struct dt_object *dt,
513 const struct dt_key *name, struct thandle *th)
515 struct thandle *sub_th;
520 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
522 RETURN(PTR_ERR(sub_th));
525 rc = update_record_pack(index_delete, th,
526 lu_object_fid(&dt->do_lu), name);
531 rc = dt_delete(env, dt, name, sub_th);
538 * Get transaction of next layer, and declare xattr set.
540 * \param[in] env execution environment
541 * \param[in] dt object on which to set xattr
542 * \param[in] buf xattr to be set
543 * \param[in] name name of the xattr
544 * \param[in] fl flag for setting xattr
546 * \retval 0 if the declaration succeeds.
547 * \retval negative errno if the declaration fails.
549 int lod_sub_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
550 const struct lu_buf *buf, const char *name,
551 int fl, struct thandle *th)
553 struct thandle *sub_th;
558 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
560 RETURN(PTR_ERR(sub_th));
563 update_record_size(env, xattr_set, th,
564 lu_object_fid(&dt->do_lu),
567 rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
575 * Get transaction of next layer, record updates if it belongs to cross-MDT
576 * operation, and set xattr to the object.
578 * \param[in] env execution environment
579 * \param[in] dt object on which to set xattr
580 * \param[in] buf xattr to be set
581 * \param[in] name name of the xattr
582 * \param[in] fl flag for setting xattr
583 * \param[in] th transaction handle
585 * \retval 0 if the xattr setting succeeds.
586 * \retval negative errno if xattr setting fails.
588 int lod_sub_xattr_set(const struct lu_env *env, struct dt_object *dt,
589 const struct lu_buf *buf, const char *name, int fl,
592 struct thandle *sub_th;
597 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
599 RETURN(PTR_ERR(sub_th));
602 rc = update_record_pack(xattr_set, th,
603 lu_object_fid(&dt->do_lu),
609 rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
617 * Get transaction of next layer, and declare attr set.
619 * \param[in] env execution environment
620 * \param[in] dt object on which to set attr
621 * \param[in] attr attributes to be set
622 * \param[in] th transaction handle
624 * \retval 0 if the declaration succeeds.
625 * \retval negative errno if the declaration fails.
627 int lod_sub_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
628 const struct lu_attr *attr, struct thandle *th)
630 struct thandle *sub_th;
635 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
637 RETURN(PTR_ERR(sub_th));
640 update_record_size(env, attr_set, th,
641 lu_object_fid(&dt->do_lu), attr);
643 rc = dt_declare_attr_set(env, dt, attr, sub_th);
651 * Get transaction of next layer, record updates if it belongs to cross-MDT
652 * operation, and set attributes to the object.
654 * \param[in] env execution environment
655 * \param[in] dt object on which to set attr
656 * \param[in] attr attrbutes to be set
657 * \param[in] th transaction handle
659 * \retval 0 if attributes setting succeeds.
660 * \retval negative errno if the attributes setting fails.
662 int lod_sub_attr_set(const struct lu_env *env, struct dt_object *dt,
663 const struct lu_attr *attr, struct thandle *th)
666 struct thandle *sub_th;
670 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
672 RETURN(PTR_ERR(sub_th));
675 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
681 rc = dt_attr_set(env, dt, attr, sub_th);
689 * Get transaction of next layer, and declare xattr deletion.
691 * \param[in] env execution environment
692 * \param[in] dt object on which to delete xattr
693 * \param[in] name name of the xattr to be deleted
694 * \param[in] th transaction handle
696 * \retval 0 if the declaration succeeds.
697 * \retval negative errno if the declaration fails.
699 int lod_sub_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
700 const char *name, struct thandle *th)
702 struct thandle *sub_th;
707 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
709 RETURN(PTR_ERR(sub_th));
712 update_record_size(env, xattr_del, th,
713 lu_object_fid(&dt->do_lu),
716 rc = dt_declare_xattr_del(env, dt, name, sub_th);
722 * xattribute deletion
724 * Get transaction of next layer, record update if it belongs to cross-MDT
725 * operation and delete xattr.
727 * \param[in] env execution environment
728 * \param[in] dt object on which to delete xattr
729 * \param[in] name name of the xattr to be deleted
730 * \param[in] th transaction handle
732 * \retval 0 if the deletion succeeds.
733 * \retval negative errno if the deletion fails.
735 int lod_sub_xattr_del(const struct lu_env *env, struct dt_object *dt,
736 const char *name, struct thandle *th)
738 struct thandle *sub_th;
743 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
745 RETURN(PTR_ERR(sub_th));
748 rc = update_record_pack(xattr_del, th,
749 lu_object_fid(&dt->do_lu), name);
754 rc = dt_xattr_del(env, dt, name, sub_th);
760 * Declare buffer write
762 * Get transaction of next layer and declare buffer write.
764 * \param[in] env execution environment
765 * \param[in] dt object to be written
766 * \param[in] buf buffer to write which includes an embedded size field
767 * \param[in] pos offet in the object to start writing at
768 * \param[in] th transaction handle
770 * \retval 0 if the insertion succeeds.
771 * \retval negative errno if the insertion fails.
773 int lod_sub_declare_write(const struct lu_env *env, struct dt_object *dt,
774 const struct lu_buf *buf, loff_t pos,
777 struct thandle *sub_th;
782 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
784 RETURN(PTR_ERR(sub_th));
787 update_record_size(env, write, th,
788 lu_object_fid(&dt->do_lu),
791 rc = dt_declare_write(env, dt, buf, pos, sub_th);
797 * Write buffer to sub object
799 * Get transaction of next layer, records buffer write if it belongs to
800 * Cross-MDT operation, and write buffer.
802 * \param[in] env execution environment
803 * \param[in] dt object to be written
804 * \param[in] buf buffer to write which includes an embedded size field
805 * \param[in] pos offet in the object to start writing at
806 * \param[in] th transaction handle
807 * \param[in] rq enforcement for this write
809 * \retval the buffer size in bytes if it succeeds.
810 * \retval negative errno if it fails.
812 ssize_t lod_sub_write(const struct lu_env *env, struct dt_object *dt,
813 const struct lu_buf *buf, loff_t *pos,
814 struct thandle *th, int rq)
816 struct thandle *sub_th;
821 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
823 RETURN(PTR_ERR(sub_th));
826 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
832 rc = dt_write(env, dt, buf, pos, sub_th, rq);
839 * Get transaction of next layer and declare punch.
841 * \param[in] env execution environment
842 * \param[in] dt object to be written
843 * \param[in] start start offset of punch
844 * \param[in] end end offet of punch
845 * \param[in] th transaction handle
847 * \retval 0 if the insertion succeeds.
848 * \retval negative errno if the insertion fails.
850 int lod_sub_declare_punch(const struct lu_env *env, struct dt_object *dt,
851 __u64 start, __u64 end, struct thandle *th)
853 struct thandle *sub_th;
858 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
860 RETURN(PTR_ERR(sub_th));
863 update_record_size(env, punch, th,
864 lu_object_fid(&dt->do_lu),
867 rc = dt_declare_punch(env, dt, start, end, sub_th);
873 * Punch to sub object
875 * Get transaction of next layer, records buffer write if it belongs to
876 * Cross-MDT operation, and punch object.
878 * \param[in] env execution environment
879 * \param[in] dt object to be written
880 * \param[in] start start offset of punch
881 * \param[in] end end offset of punch
882 * \param[in] th transaction handle
883 * \param[in] capa capability of the write
885 * \retval the buffer size in bytes if it succeeds.
886 * \retval negative errno if it fails.
888 int lod_sub_punch(const struct lu_env *env, struct dt_object *dt,
889 __u64 start, __u64 end, struct thandle *th)
891 struct thandle *sub_th;
896 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
898 RETURN(PTR_ERR(sub_th));
901 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
907 rc = dt_punch(env, dt, start, end, sub_th);
912 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
913 struct dt_device *dt, int index)
915 struct lod_thread_info *lti = lod_env_info(env);
916 struct llog_ctxt *ctxt;
917 struct llog_handle *lgh;
918 struct llog_catid *cid = <i->lti_cid;
919 struct lu_fid *fid = <i->lti_fid;
920 struct obd_device *obd;
922 bool need_put = false;
925 lu_update_log_fid(fid, index);
927 rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
931 rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
933 CERROR("%s: can't get id from catalogs: rc = %d\n",
934 lod2obd(lod)->obd_name, rc);
938 obd = dt->dd_lu_dev.ld_obd;
939 ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
940 LASSERT(ctxt != NULL);
941 ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
942 ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
943 if (likely(logid_id(&cid->lci_logid) != 0)) {
944 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
947 /* re-create llog if it is missing */
949 logid_set_id(&cid->lci_logid, 0);
954 if (unlikely(logid_id(&cid->lci_logid) == 0)) {
955 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
958 cid->lci_logid = lgh->lgh_id;
962 LASSERT(lgh != NULL);
964 rc = llog_init_handle(env, lgh, LLOG_F_IS_CAT, NULL);
969 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
974 ctxt->loc_handle = lgh;
976 CDEBUG(D_INFO, "%s: init llog for index %d - catid "DFID":%x\n",
977 obd->obd_name, index, PFID(&cid->lci_logid.lgl_oi.oi_fid),
978 cid->lci_logid.lgl_ogen);
981 llog_cat_close(env, lgh);