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);
78 /* local object must be mdt object, Note: during ost object
79 * creation, FID is not assigned until osp_create(),
80 * so if the FID of sub_obj is zero, it means OST object. */
81 if (!dt_object_remote(sub_obj) ||
82 fid_is_zero(lu_object_fid(&sub_obj->do_lu))) {
83 /* local MDT object */
84 if (fid_is_sane(lu_object_fid(&sub_obj->do_lu)) &&
85 tth->tt_multiple_thandle != NULL &&
86 record_update != NULL &&
88 *record_update = true;
90 RETURN(tth->tt_master_sub_thandle);
93 rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
98 /* th_complex means we need track all of updates for this
99 * transaction, include changes on OST */
100 if (type == LU_SEQ_RANGE_OST && !th->th_complex)
101 RETURN(tth->tt_master_sub_thandle);
103 sub_th = thandle_get_sub(env, th, sub_obj);
107 if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
109 *record_update = true;
115 * Declare sub-object creation.
117 * Get transaction of next layer and declare the creation of the object.
119 * \param[in] env execution environment
120 * \param[in] dt the object being created
121 * \param[in] attr the attributes of the object being created
122 * \param[in] hint the hint of the creation
123 * \param[in] dof the object format of the creation
124 * \param[th] th the transaction handle
126 * \retval 0 if the declaration succeeds
127 * \retval negative errno if the declaration fails.
129 int lod_sub_declare_create(const struct lu_env *env, struct dt_object *dt,
130 struct lu_attr *attr,
131 struct dt_allocation_hint *hint,
132 struct dt_object_format *dof, struct thandle *th)
134 struct thandle *sub_th;
137 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
139 return PTR_ERR(sub_th);
142 update_record_size(env, create, th, lu_object_fid(&dt->do_lu),
145 return dt_declare_create(env, dt, attr, hint, dof, sub_th);
151 * Get transaction of next layer, record updates if it belongs to cross-MDT
152 * operation, and create the object.
154 * \param[in] env execution environment
155 * \param[in] dt the object being created
156 * \param[in] attr the attributes of the object being created
157 * \param[in] hint the hint of the creation
158 * \param[in] dof the object format of the creation
159 * \param[th] th the transaction handle
161 * \retval 0 if the creation succeeds
162 * \retval negative errno if the creation fails.
164 int lod_sub_create(const struct lu_env *env, struct dt_object *dt,
165 struct lu_attr *attr, struct dt_allocation_hint *hint,
166 struct dt_object_format *dof, struct thandle *th)
168 struct thandle *sub_th;
173 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
175 RETURN(PTR_ERR(sub_th));
178 rc = update_record_pack(create, th,
179 lu_object_fid(&dt->do_lu),
185 rc = dt_create(env, dt, attr, hint, dof, sub_th);
191 * Declare adding reference for the sub-object
193 * Get transaction of next layer and declare the reference adding.
195 * \param[in] env execution environment
196 * \param[in] dt dt object to add reference
197 * \param[in] th transaction handle
199 * \retval 0 if the declaration succeeds.
200 * \retval negative errno if the declaration fails.
202 int lod_sub_declare_ref_add(const struct lu_env *env, struct dt_object *dt,
205 struct thandle *sub_th;
210 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
212 RETURN(PTR_ERR(sub_th));
215 update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
217 rc = dt_declare_ref_add(env, dt, sub_th);
223 * Add reference for the sub-object
225 * Get transaction of next layer, record updates if it belongs to cross-MDT
226 * operation and add reference of the object.
228 * \param[in] env execution environment
229 * \param[in] dt dt object to add reference
230 * \param[in] th transaction handle
232 * \retval 0 if it succeeds.
233 * \retval negative errno if it fails.
235 int lod_sub_ref_add(const struct lu_env *env, struct dt_object *dt,
238 struct thandle *sub_th;
243 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
245 RETURN(PTR_ERR(sub_th));
248 rc = update_record_pack(ref_add, th,
249 lu_object_fid(&dt->do_lu));
254 rc = dt_ref_add(env, dt, sub_th);
260 * Declare deleting reference for the sub-object
262 * Get transaction of next layer and declare the reference deleting.
264 * \param[in] env execution environment
265 * \param[in] dt dt object to delete reference
266 * \param[in] th transaction handle
268 * \retval 0 if the declaration succeeds.
269 * \retval negative errno if the declaration fails.
271 int lod_sub_declare_ref_del(const struct lu_env *env, struct dt_object *dt,
274 struct thandle *sub_th;
279 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
281 RETURN(PTR_ERR(sub_th));
284 update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
286 rc = dt_declare_ref_del(env, dt, sub_th);
292 * Delete reference for the sub-object
294 * Get transaction of next layer, record updates if it belongs to cross-MDT
295 * operation and delete reference of the object.
297 * \param[in] env execution environment
298 * \param[in] dt dt object to delete reference
299 * \param[in] th transaction handle
301 * \retval 0 if it succeeds.
302 * \retval negative errno if it fails.
304 int lod_sub_ref_del(const struct lu_env *env, struct dt_object *dt,
307 struct thandle *sub_th;
312 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
314 RETURN(PTR_ERR(sub_th));
317 rc = update_record_pack(ref_del, th,
318 lu_object_fid(&dt->do_lu));
323 rc = dt_ref_del(env, dt, sub_th);
329 * Declare destroying sub-object
331 * Get transaction of next layer and declare the sub-object destroy.
333 * \param[in] env execution environment
334 * \param[in] dt dt object to be destroyed
335 * \param[in] th transaction handle
337 * \retval 0 if the declaration succeeds.
338 * \retval negative errno if the declaration fails.
340 int lod_sub_declare_destroy(const struct lu_env *env, struct dt_object *dt,
343 struct thandle *sub_th;
348 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
350 RETURN(PTR_ERR(sub_th));
353 update_record_size(env, destroy, th, lu_object_fid(&dt->do_lu));
355 rc = dt_declare_destroy(env, dt, sub_th);
363 * Get transaction of next layer, record updates if it belongs to cross-MDT
364 * operation and destroy the object.
366 * \param[in] env execution environment
367 * \param[in] dt dt object to be destroyed
368 * \param[in] th transaction handle
370 * \retval 0 if the destroy succeeds.
371 * \retval negative errno if the destroy fails.
373 int lod_sub_destroy(const struct lu_env *env, struct dt_object *dt,
376 struct thandle *sub_th;
381 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
383 RETURN(PTR_ERR(sub_th));
386 rc = update_record_pack(destroy, th, lu_object_fid(&dt->do_lu));
391 rc = dt_destroy(env, dt, sub_th);
397 * Declare sub-object index insert
399 * Get transaction of next layer and declare index insert.
401 * \param[in] env execution environment
402 * \param[in] dt object for which to insert index
403 * \param[in] rec record of the index which will be inserted
404 * \param[in] key key of the index which will be inserted
405 * \param[in] th the transaction handle
407 * \retval 0 if the declaration succeeds.
408 * \retval negative errno if the declaration fails.
410 int lod_sub_declare_insert(const struct lu_env *env, struct dt_object *dt,
411 const struct dt_rec *rec,
412 const struct dt_key *key, struct thandle *th)
414 struct thandle *sub_th;
417 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
419 return PTR_ERR(sub_th);
422 update_record_size(env, index_insert, th,
423 lu_object_fid(&dt->do_lu), rec, key);
425 return dt_declare_insert(env, dt, rec, key, sub_th);
429 * Insert index of sub object
431 * Get transaction of next layer, record updates if it belongs to cross-MDT
432 * operation, and insert the index.
434 * \param[in] env execution environment
435 * \param[in] dt object for which to insert index
436 * \param[in] rec record of the index to be inserted
437 * \param[in] key key of the index to be inserted
438 * \param[in] th the transaction handle
439 * \param[in] ign whether ignore quota
441 * \retval 0 if the insertion succeeds.
442 * \retval negative errno if the insertion fails.
444 int lod_sub_insert(const struct lu_env *env, struct dt_object *dt,
445 const struct dt_rec *rec, const struct dt_key *key,
446 struct thandle *th, int ign)
448 struct thandle *sub_th;
452 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
454 return PTR_ERR(sub_th);
457 rc = update_record_pack(index_insert, th,
458 lu_object_fid(&dt->do_lu), rec, key);
463 return dt_insert(env, dt, rec, key, sub_th, ign);
467 * Declare sub-object index delete
469 * Get transaction of next layer and declare index deletion.
471 * \param[in] env execution environment
472 * \param[in] dt object for which to delete index
473 * \param[in] key key of the index which will be deleted
474 * \param[in] th the transaction handle
476 * \retval 0 if the declaration succeeds.
477 * \retval negative errno if the declaration fails.
479 int lod_sub_declare_delete(const struct lu_env *env, struct dt_object *dt,
480 const struct dt_key *key, struct thandle *th)
482 struct thandle *sub_th;
485 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
487 return PTR_ERR(sub_th);
490 update_record_size(env, index_delete, th,
491 lu_object_fid(&dt->do_lu), key);
493 return dt_declare_delete(env, dt, key, sub_th);
497 * Delete index of sub object
499 * Get transaction of next layer, record updates if it belongs to cross-MDT
500 * operation, and delete the index.
502 * \param[in] env execution environment
503 * \param[in] dt object for which to delete index
504 * \param[in] key key of the index to be deleted
505 * \param[in] th the transaction handle
507 * \retval 0 if the deletion succeeds.
508 * \retval negative errno if the deletion fails.
510 int lod_sub_delete(const struct lu_env *env, struct dt_object *dt,
511 const struct dt_key *name, struct thandle *th)
513 struct thandle *sub_th;
518 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
520 RETURN(PTR_ERR(sub_th));
523 rc = update_record_pack(index_delete, th,
524 lu_object_fid(&dt->do_lu), name);
529 rc = dt_delete(env, dt, name, sub_th);
536 * Get transaction of next layer, and declare xattr set.
538 * \param[in] env execution environment
539 * \param[in] dt object on which to set xattr
540 * \param[in] buf xattr to be set
541 * \param[in] name name of the xattr
542 * \param[in] fl flag for setting xattr
544 * \retval 0 if the declaration succeeds.
545 * \retval negative errno if the declaration fails.
547 int lod_sub_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
548 const struct lu_buf *buf, const char *name,
549 int fl, struct thandle *th)
551 struct thandle *sub_th;
556 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
558 RETURN(PTR_ERR(sub_th));
561 update_record_size(env, xattr_set, th,
562 lu_object_fid(&dt->do_lu),
565 rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
573 * Get transaction of next layer, record updates if it belongs to cross-MDT
574 * operation, and set xattr to the object.
576 * \param[in] env execution environment
577 * \param[in] dt object on which to set xattr
578 * \param[in] buf xattr to be set
579 * \param[in] name name of the xattr
580 * \param[in] fl flag for setting xattr
581 * \param[in] th transaction handle
583 * \retval 0 if the xattr setting succeeds.
584 * \retval negative errno if xattr setting fails.
586 int lod_sub_xattr_set(const struct lu_env *env, struct dt_object *dt,
587 const struct lu_buf *buf, const char *name, int fl,
590 struct thandle *sub_th;
595 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
597 RETURN(PTR_ERR(sub_th));
600 rc = update_record_pack(xattr_set, th,
601 lu_object_fid(&dt->do_lu),
607 rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
615 * Get transaction of next layer, and declare attr set.
617 * \param[in] env execution environment
618 * \param[in] dt object on which to set attr
619 * \param[in] attr attributes to be set
620 * \param[in] th transaction handle
622 * \retval 0 if the declaration succeeds.
623 * \retval negative errno if the declaration fails.
625 int lod_sub_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
626 const struct lu_attr *attr, struct thandle *th)
628 struct thandle *sub_th;
633 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
635 RETURN(PTR_ERR(sub_th));
638 update_record_size(env, attr_set, th,
639 lu_object_fid(&dt->do_lu), attr);
641 rc = dt_declare_attr_set(env, dt, attr, sub_th);
649 * Get transaction of next layer, record updates if it belongs to cross-MDT
650 * operation, and set attributes to the object.
652 * \param[in] env execution environment
653 * \param[in] dt object on which to set attr
654 * \param[in] attr attrbutes to be set
655 * \param[in] th transaction handle
657 * \retval 0 if attributes setting succeeds.
658 * \retval negative errno if the attributes setting fails.
660 int lod_sub_attr_set(const struct lu_env *env, struct dt_object *dt,
661 const struct lu_attr *attr, struct thandle *th)
664 struct thandle *sub_th;
668 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
670 RETURN(PTR_ERR(sub_th));
673 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
679 rc = dt_attr_set(env, dt, attr, sub_th);
687 * Get transaction of next layer, and declare xattr deletion.
689 * \param[in] env execution environment
690 * \param[in] dt object on which to delete xattr
691 * \param[in] name name of the xattr to be deleted
692 * \param[in] th transaction handle
694 * \retval 0 if the declaration succeeds.
695 * \retval negative errno if the declaration fails.
697 int lod_sub_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
698 const char *name, struct thandle *th)
700 struct thandle *sub_th;
705 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
707 RETURN(PTR_ERR(sub_th));
710 update_record_size(env, xattr_del, th,
711 lu_object_fid(&dt->do_lu),
714 rc = dt_declare_xattr_del(env, dt, name, sub_th);
720 * xattribute deletion
722 * Get transaction of next layer, record update if it belongs to cross-MDT
723 * operation and delete xattr.
725 * \param[in] env execution environment
726 * \param[in] dt object on which to delete xattr
727 * \param[in] name name of the xattr to be deleted
728 * \param[in] th transaction handle
730 * \retval 0 if the deletion succeeds.
731 * \retval negative errno if the deletion fails.
733 int lod_sub_xattr_del(const struct lu_env *env, struct dt_object *dt,
734 const char *name, struct thandle *th)
736 struct thandle *sub_th;
741 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
743 RETURN(PTR_ERR(sub_th));
746 rc = update_record_pack(xattr_del, th,
747 lu_object_fid(&dt->do_lu), name);
752 rc = dt_xattr_del(env, dt, name, sub_th);
758 * Declare buffer write
760 * Get transaction of next layer and declare buffer write.
762 * \param[in] env execution environment
763 * \param[in] dt object to be written
764 * \param[in] buf buffer to write which includes an embedded size field
765 * \param[in] pos offet in the object to start writing at
766 * \param[in] th transaction handle
768 * \retval 0 if the insertion succeeds.
769 * \retval negative errno if the insertion fails.
771 int lod_sub_declare_write(const struct lu_env *env, struct dt_object *dt,
772 const struct lu_buf *buf, loff_t pos,
775 struct thandle *sub_th;
780 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
782 RETURN(PTR_ERR(sub_th));
785 update_record_size(env, write, th,
786 lu_object_fid(&dt->do_lu),
789 rc = dt_declare_write(env, dt, buf, pos, sub_th);
795 * Write buffer to sub object
797 * Get transaction of next layer, records buffer write if it belongs to
798 * Cross-MDT operation, and write buffer.
800 * \param[in] env execution environment
801 * \param[in] dt object to be written
802 * \param[in] buf buffer to write which includes an embedded size field
803 * \param[in] pos offet in the object to start writing at
804 * \param[in] th transaction handle
805 * \param[in] rq enforcement for this write
807 * \retval the buffer size in bytes if it succeeds.
808 * \retval negative errno if it fails.
810 ssize_t lod_sub_write(const struct lu_env *env, struct dt_object *dt,
811 const struct lu_buf *buf, loff_t *pos,
812 struct thandle *th, int rq)
814 struct thandle *sub_th;
819 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
821 RETURN(PTR_ERR(sub_th));
824 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
830 rc = dt_write(env, dt, buf, pos, sub_th, rq);
837 * Get transaction of next layer and declare punch.
839 * \param[in] env execution environment
840 * \param[in] dt object to be written
841 * \param[in] start start offset of punch
842 * \param[in] end end offet of punch
843 * \param[in] th transaction handle
845 * \retval 0 if the insertion succeeds.
846 * \retval negative errno if the insertion fails.
848 int lod_sub_declare_punch(const struct lu_env *env, struct dt_object *dt,
849 __u64 start, __u64 end, struct thandle *th)
851 struct thandle *sub_th;
856 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
858 RETURN(PTR_ERR(sub_th));
861 update_record_size(env, punch, th,
862 lu_object_fid(&dt->do_lu),
865 rc = dt_declare_punch(env, dt, start, end, sub_th);
871 * Punch to sub object
873 * Get transaction of next layer, records buffer write if it belongs to
874 * Cross-MDT operation, and punch object.
876 * \param[in] env execution environment
877 * \param[in] dt object to be written
878 * \param[in] start start offset of punch
879 * \param[in] end end offset of punch
880 * \param[in] th transaction handle
881 * \param[in] capa capability of the write
883 * \retval the buffer size in bytes if it succeeds.
884 * \retval negative errno if it fails.
886 int lod_sub_punch(const struct lu_env *env, struct dt_object *dt,
887 __u64 start, __u64 end, struct thandle *th)
889 struct thandle *sub_th;
894 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
896 RETURN(PTR_ERR(sub_th));
899 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
905 rc = dt_punch(env, dt, start, end, sub_th);
910 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
911 struct dt_device *dt, int index)
913 struct lod_thread_info *lti = lod_env_info(env);
914 struct llog_ctxt *ctxt;
915 struct llog_handle *lgh;
916 struct llog_catid *cid = <i->lti_cid;
917 struct lu_fid *fid = <i->lti_fid;
918 struct obd_device *obd;
920 bool need_put = false;
923 lu_update_log_fid(fid, index);
925 rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
929 rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
931 CERROR("%s: can't get id from catalogs: rc = %d\n",
932 lod2obd(lod)->obd_name, rc);
936 obd = dt->dd_lu_dev.ld_obd;
937 ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
938 LASSERT(ctxt != NULL);
939 ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
940 ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
941 if (likely(logid_id(&cid->lci_logid) != 0)) {
942 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
945 /* re-create llog if it is missing */
947 logid_set_id(&cid->lci_logid, 0);
952 if (unlikely(logid_id(&cid->lci_logid) == 0)) {
953 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
956 cid->lci_logid = lgh->lgh_id;
960 LASSERT(lgh != NULL);
962 rc = llog_init_handle(env, lgh, LLOG_F_IS_CAT, NULL);
967 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
972 ctxt->loc_handle = lgh;
974 CDEBUG(D_INFO, "%s: init llog for index %d - catid "DFID":%x\n",
975 obd->obd_name, index, PFID(&cid->lci_logid.lgl_oi.oi_fid),
976 cid->lci_logid.lgl_ogen);
979 llog_cat_close(env, lgh);