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) 2014, 2015, 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 <lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
49 #include <lustre_fid.h>
50 #include <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_object_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_object_declare_create(const struct lu_env *env,
130 struct dt_object *dt,
131 struct lu_attr *attr,
132 struct dt_allocation_hint *hint,
133 struct dt_object_format *dof,
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_object_create(const struct lu_env *env, struct dt_object *dt,
167 struct lu_attr *attr,
168 struct dt_allocation_hint *hint,
169 struct dt_object_format *dof,
172 struct thandle *sub_th;
177 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
179 RETURN(PTR_ERR(sub_th));
182 rc = update_record_pack(create, th,
183 lu_object_fid(&dt->do_lu),
189 rc = dt_create(env, dt, attr, hint, dof, sub_th);
195 * Declare adding reference for the sub-object
197 * Get transaction of next layer and declare the reference adding.
199 * \param[in] env execution environment
200 * \param[in] dt dt object to add reference
201 * \param[in] th transaction handle
203 * \retval 0 if the declaration succeeds.
204 * \retval negative errno if the declaration fails.
206 int lod_sub_object_declare_ref_add(const struct lu_env *env,
207 struct dt_object *dt,
210 struct thandle *sub_th;
215 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
217 RETURN(PTR_ERR(sub_th));
220 update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
222 rc = dt_declare_ref_add(env, dt, sub_th);
228 * Add reference for the sub-object
230 * Get transaction of next layer, record updates if it belongs to cross-MDT
231 * operation and add reference of the object.
233 * \param[in] env execution environment
234 * \param[in] dt dt object to add reference
235 * \param[in] th transaction handle
237 * \retval 0 if it succeeds.
238 * \retval negative errno if it fails.
240 int lod_sub_object_ref_add(const struct lu_env *env, struct dt_object *dt,
243 struct thandle *sub_th;
248 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
250 RETURN(PTR_ERR(sub_th));
253 rc = update_record_pack(ref_add, th,
254 lu_object_fid(&dt->do_lu));
259 rc = dt_ref_add(env, dt, sub_th);
265 * Declare deleting reference for the sub-object
267 * Get transaction of next layer and declare the reference deleting.
269 * \param[in] env execution environment
270 * \param[in] dt dt object to delete reference
271 * \param[in] th transaction handle
273 * \retval 0 if the declaration succeeds.
274 * \retval negative errno if the declaration fails.
276 int lod_sub_object_declare_ref_del(const struct lu_env *env,
277 struct dt_object *dt,
280 struct thandle *sub_th;
285 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
287 RETURN(PTR_ERR(sub_th));
290 update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
292 rc = dt_declare_ref_del(env, dt, sub_th);
298 * Delete reference for the sub-object
300 * Get transaction of next layer, record updates if it belongs to cross-MDT
301 * operation and delete reference of the object.
303 * \param[in] env execution environment
304 * \param[in] dt dt object to delete reference
305 * \param[in] th transaction handle
307 * \retval 0 if it succeeds.
308 * \retval negative errno if it fails.
310 int lod_sub_object_ref_del(const struct lu_env *env, struct dt_object *dt,
313 struct thandle *sub_th;
318 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
320 RETURN(PTR_ERR(sub_th));
323 rc = update_record_pack(ref_del, th,
324 lu_object_fid(&dt->do_lu));
329 rc = dt_ref_del(env, dt, sub_th);
335 * Declare destroying sub-object
337 * Get transaction of next layer and declare the sub-object destroy.
339 * \param[in] env execution environment
340 * \param[in] dt dt object to be destroyed
341 * \param[in] th transaction handle
343 * \retval 0 if the declaration succeeds.
344 * \retval negative errno if the declaration fails.
346 int lod_sub_object_declare_destroy(const struct lu_env *env,
347 struct dt_object *dt,
350 struct thandle *sub_th;
355 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
357 RETURN(PTR_ERR(sub_th));
360 update_record_size(env, object_destroy, th,
361 lu_object_fid(&dt->do_lu));
363 rc = dt_declare_destroy(env, dt, sub_th);
371 * Get transaction of next layer, record updates if it belongs to cross-MDT
372 * operation and destroy the object.
374 * \param[in] env execution environment
375 * \param[in] dt dt object to be destroyed
376 * \param[in] th transaction handle
378 * \retval 0 if the destroy succeeds.
379 * \retval negative errno if the destroy fails.
381 int lod_sub_object_destroy(const struct lu_env *env, struct dt_object *dt,
384 struct thandle *sub_th;
389 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
391 RETURN(PTR_ERR(sub_th));
394 rc = update_record_pack(object_destroy, th,
395 lu_object_fid(&dt->do_lu));
400 rc = dt_destroy(env, dt, sub_th);
406 * Declare sub-object index insert
408 * Get transaction of next layer and declare index insert.
410 * \param[in] env execution environment
411 * \param[in] dt object for which to insert index
412 * \param[in] rec record of the index which will be inserted
413 * \param[in] key key of the index which will be inserted
414 * \param[in] th the transaction handle
416 * \retval 0 if the declaration succeeds.
417 * \retval negative errno if the declaration fails.
419 int lod_sub_object_declare_insert(const struct lu_env *env,
420 struct dt_object *dt,
421 const struct dt_rec *rec,
422 const struct dt_key *key,
425 struct thandle *sub_th;
428 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
430 return PTR_ERR(sub_th);
433 update_record_size(env, index_insert, th,
434 lu_object_fid(&dt->do_lu), rec, key);
436 return dt_declare_insert(env, dt, rec, key, sub_th);
440 * Insert index of sub object
442 * Get transaction of next layer, record updates if it belongs to cross-MDT
443 * operation, and insert the index.
445 * \param[in] env execution environment
446 * \param[in] dt object for which to insert index
447 * \param[in] rec record of the index to be inserted
448 * \param[in] key key of the index to be inserted
449 * \param[in] th the transaction handle
450 * \param[in] ign whether ignore quota
452 * \retval 0 if the insertion succeeds.
453 * \retval negative errno if the insertion fails.
455 int lod_sub_object_index_insert(const struct lu_env *env, struct dt_object *dt,
456 const struct dt_rec *rec,
457 const struct dt_key *key, struct thandle *th,
460 struct thandle *sub_th;
464 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
466 return PTR_ERR(sub_th);
469 rc = update_record_pack(index_insert, th,
470 lu_object_fid(&dt->do_lu), rec, key);
475 return dt_insert(env, dt, rec, key, sub_th, ign);
479 * Declare sub-object index delete
481 * Get transaction of next layer and declare index deletion.
483 * \param[in] env execution environment
484 * \param[in] dt object for which to delete index
485 * \param[in] key key of the index which will be deleted
486 * \param[in] th the transaction handle
488 * \retval 0 if the declaration succeeds.
489 * \retval negative errno if the declaration fails.
491 int lod_sub_object_declare_delete(const struct lu_env *env,
492 struct dt_object *dt,
493 const struct dt_key *key,
496 struct thandle *sub_th;
499 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
501 return PTR_ERR(sub_th);
504 update_record_size(env, index_delete, th,
505 lu_object_fid(&dt->do_lu), key);
507 return dt_declare_delete(env, dt, key, sub_th);
511 * Delete index of sub object
513 * Get transaction of next layer, record updates if it belongs to cross-MDT
514 * operation, and delete the index.
516 * \param[in] env execution environment
517 * \param[in] dt object for which to delete index
518 * \param[in] key key of the index to be deleted
519 * \param[in] th the transaction handle
521 * \retval 0 if the deletion succeeds.
522 * \retval negative errno if the deletion fails.
524 int lod_sub_object_delete(const struct lu_env *env, struct dt_object *dt,
525 const struct dt_key *name, struct thandle *th)
527 struct thandle *sub_th;
532 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
534 RETURN(PTR_ERR(sub_th));
537 rc = update_record_pack(index_delete, th,
538 lu_object_fid(&dt->do_lu), name);
543 rc = dt_delete(env, dt, name, sub_th);
550 * Get transaction of next layer, and declare xattr set.
552 * \param[in] env execution environment
553 * \param[in] dt object on which to set xattr
554 * \param[in] buf xattr to be set
555 * \param[in] name name of the xattr
556 * \param[in] fl flag for setting xattr
558 * \retval 0 if the declaration succeeds.
559 * \retval negative errno if the declaration fails.
561 int lod_sub_object_declare_xattr_set(const struct lu_env *env,
562 struct dt_object *dt,
563 const struct lu_buf *buf,
564 const char *name, int fl,
567 struct thandle *sub_th;
572 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
574 RETURN(PTR_ERR(sub_th));
577 update_record_size(env, xattr_set, th,
578 lu_object_fid(&dt->do_lu),
581 rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
589 * Get transaction of next layer, record updates if it belongs to cross-MDT
590 * operation, and set xattr to the object.
592 * \param[in] env execution environment
593 * \param[in] dt object on which to set xattr
594 * \param[in] buf xattr to be set
595 * \param[in] name name of the xattr
596 * \param[in] fl flag for setting xattr
597 * \param[in] th transaction handle
599 * \retval 0 if the xattr setting succeeds.
600 * \retval negative errno if xattr setting fails.
602 int lod_sub_object_xattr_set(const struct lu_env *env, struct dt_object *dt,
603 const struct lu_buf *buf, const char *name, int fl,
606 struct thandle *sub_th;
611 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
613 RETURN(PTR_ERR(sub_th));
616 rc = update_record_pack(xattr_set, th,
617 lu_object_fid(&dt->do_lu),
623 rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
631 * Get transaction of next layer, and declare attr set.
633 * \param[in] env execution environment
634 * \param[in] dt object on which to set attr
635 * \param[in] attr attributes to be set
636 * \param[in] th transaction handle
638 * \retval 0 if the declaration succeeds.
639 * \retval negative errno if the declaration fails.
641 int lod_sub_object_declare_attr_set(const struct lu_env *env,
642 struct dt_object *dt,
643 const struct lu_attr *attr,
646 struct thandle *sub_th;
651 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
653 RETURN(PTR_ERR(sub_th));
656 update_record_size(env, attr_set, th,
657 lu_object_fid(&dt->do_lu), attr);
659 rc = dt_declare_attr_set(env, dt, attr, sub_th);
667 * Get transaction of next layer, record updates if it belongs to cross-MDT
668 * operation, and set attributes to the object.
670 * \param[in] env execution environment
671 * \param[in] dt object on which to set attr
672 * \param[in] attr attrbutes to be set
673 * \param[in] th transaction handle
675 * \retval 0 if attributes setting succeeds.
676 * \retval negative errno if the attributes setting fails.
678 int lod_sub_object_attr_set(const struct lu_env *env,
679 struct dt_object *dt,
680 const struct lu_attr *attr,
684 struct thandle *sub_th;
688 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
690 RETURN(PTR_ERR(sub_th));
693 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
699 rc = dt_attr_set(env, dt, attr, sub_th);
707 * Get transaction of next layer, and declare xattr deletion.
709 * \param[in] env execution environment
710 * \param[in] dt object on which to delete xattr
711 * \param[in] name name of the xattr to be deleted
712 * \param[in] th transaction handle
714 * \retval 0 if the declaration succeeds.
715 * \retval negative errno if the declaration fails.
717 int lod_sub_object_declare_xattr_del(const struct lu_env *env,
718 struct dt_object *dt,
722 struct thandle *sub_th;
727 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
729 RETURN(PTR_ERR(sub_th));
732 update_record_size(env, xattr_del, th,
733 lu_object_fid(&dt->do_lu),
736 rc = dt_declare_xattr_del(env, dt, name, sub_th);
742 * xattribute deletion
744 * Get transaction of next layer, record update if it belongs to cross-MDT
745 * operation and delete xattr.
747 * \param[in] env execution environment
748 * \param[in] dt object on which to delete xattr
749 * \param[in] name name of the xattr to be deleted
750 * \param[in] th transaction handle
752 * \retval 0 if the deletion succeeds.
753 * \retval negative errno if the deletion fails.
755 int lod_sub_object_xattr_del(const struct lu_env *env,
756 struct dt_object *dt,
760 struct thandle *sub_th;
765 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
767 RETURN(PTR_ERR(sub_th));
770 rc = update_record_pack(xattr_del, th,
771 lu_object_fid(&dt->do_lu), name);
776 rc = dt_xattr_del(env, dt, name, sub_th);
782 * Declare buffer write
784 * Get transaction of next layer and declare buffer write.
786 * \param[in] env execution environment
787 * \param[in] dt object to be written
788 * \param[in] buf buffer to write which includes an embedded size field
789 * \param[in] pos offet in the object to start writing at
790 * \param[in] th transaction handle
792 * \retval 0 if the insertion succeeds.
793 * \retval negative errno if the insertion fails.
795 int lod_sub_object_declare_write(const struct lu_env *env,
796 struct dt_object *dt,
797 const struct lu_buf *buf, loff_t pos,
800 struct thandle *sub_th;
805 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
807 RETURN(PTR_ERR(sub_th));
810 update_record_size(env, write, th,
811 lu_object_fid(&dt->do_lu),
814 rc = dt_declare_write(env, dt, buf, pos, sub_th);
820 * Write buffer to sub object
822 * Get transaction of next layer, records buffer write if it belongs to
823 * Cross-MDT operation, and write buffer.
825 * \param[in] env execution environment
826 * \param[in] dt object to be written
827 * \param[in] buf buffer to write which includes an embedded size field
828 * \param[in] pos offet in the object to start writing at
829 * \param[in] th transaction handle
830 * \param[in] rq enforcement for this write
832 * \retval the buffer size in bytes if it succeeds.
833 * \retval negative errno if it fails.
835 ssize_t lod_sub_object_write(const struct lu_env *env, struct dt_object *dt,
836 const struct lu_buf *buf, loff_t *pos,
837 struct thandle *th, int rq)
839 struct thandle *sub_th;
844 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
846 RETURN(PTR_ERR(sub_th));
849 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
855 rc = dt_write(env, dt, buf, pos, sub_th, rq);
862 * Get transaction of next layer and declare punch.
864 * \param[in] env execution environment
865 * \param[in] dt object to be written
866 * \param[in] start start offset of punch
867 * \param[in] end end offet of punch
868 * \param[in] th transaction handle
870 * \retval 0 if the insertion succeeds.
871 * \retval negative errno if the insertion fails.
873 int lod_sub_object_declare_punch(const struct lu_env *env,
874 struct dt_object *dt,
875 __u64 start, __u64 end,
878 struct thandle *sub_th;
883 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
885 RETURN(PTR_ERR(sub_th));
888 update_record_size(env, punch, th,
889 lu_object_fid(&dt->do_lu),
892 rc = dt_declare_punch(env, dt, start, end, sub_th);
898 * Punch to sub object
900 * Get transaction of next layer, records buffer write if it belongs to
901 * Cross-MDT operation, and punch object.
903 * \param[in] env execution environment
904 * \param[in] dt object to be written
905 * \param[in] start start offset of punch
906 * \param[in] end end offset of punch
907 * \param[in] th transaction handle
908 * \param[in] capa capability of the write
910 * \retval the buffer size in bytes if it succeeds.
911 * \retval negative errno if it fails.
913 int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
914 __u64 start, __u64 end, struct thandle *th)
916 struct thandle *sub_th;
921 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
923 RETURN(PTR_ERR(sub_th));
926 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
932 rc = dt_punch(env, dt, start, end, sub_th);
937 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
938 struct dt_device *dt, int index)
940 struct lod_thread_info *lti = lod_env_info(env);
941 struct llog_ctxt *ctxt;
942 struct llog_handle *lgh;
943 struct llog_catid *cid = <i->lti_cid;
944 struct lu_fid *fid = <i->lti_fid;
945 struct obd_device *obd;
947 bool need_put = false;
950 lu_update_log_fid(fid, index);
952 rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
956 rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
958 CERROR("%s: can't get id from catalogs: rc = %d\n",
959 lod2obd(lod)->obd_name, rc);
963 obd = dt->dd_lu_dev.ld_obd;
964 ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
965 LASSERT(ctxt != NULL);
966 ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
967 ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
968 if (likely(logid_id(&cid->lci_logid) != 0)) {
969 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
972 /* re-create llog if it is missing */
974 logid_set_id(&cid->lci_logid, 0);
979 if (unlikely(logid_id(&cid->lci_logid) == 0)) {
980 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
983 cid->lci_logid = lgh->lgh_id;
987 LASSERT(lgh != NULL);
989 rc = llog_cat_init_and_process(env, lgh);
994 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
999 ctxt->loc_handle = lgh;
1001 CDEBUG(D_INFO, "%s: Init llog for %d - catid "DOSTID":%x\n",
1002 obd->obd_name, index, POSTID(&cid->lci_logid.lgl_oi),
1003 cid->lci_logid.lgl_ogen);
1006 llog_cat_close(env, lgh);
1008 llog_ctxt_put(ctxt);