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, 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 if (type == LU_SEQ_RANGE_OST)
99 RETURN(tth->tt_master_sub_thandle);
101 if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
103 *record_update = true;
105 sub_th = thandle_get_sub(env, th, sub_obj);
111 * Declare sub-object creation.
113 * Get transaction of next layer and declare the creation of the object.
115 * \param[in] env execution environment
116 * \param[in] dt the object being created
117 * \param[in] attr the attributes of the object being created
118 * \param[in] hint the hint of the creation
119 * \param[in] dof the object format of the creation
120 * \param[th] th the transaction handle
122 * \retval 0 if the declaration succeeds
123 * \retval negative errno if the declaration fails.
125 int lod_sub_object_declare_create(const struct lu_env *env,
126 struct dt_object *dt,
127 struct lu_attr *attr,
128 struct dt_allocation_hint *hint,
129 struct dt_object_format *dof,
132 struct thandle *sub_th;
134 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
136 return PTR_ERR(sub_th);
138 return dt_declare_create(env, dt, attr, hint, dof, sub_th);
144 * Get transaction of next layer, record updates if it belongs to cross-MDT
145 * operation, and create the object.
147 * \param[in] env execution environment
148 * \param[in] dt the object being created
149 * \param[in] attr the attributes of the object being created
150 * \param[in] hint the hint of the creation
151 * \param[in] dof the object format of the creation
152 * \param[th] th the transaction handle
154 * \retval 0 if the creation succeeds
155 * \retval negative errno if the creation fails.
157 int lod_sub_object_create(const struct lu_env *env, struct dt_object *dt,
158 struct lu_attr *attr,
159 struct dt_allocation_hint *hint,
160 struct dt_object_format *dof,
163 struct thandle *sub_th;
168 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
170 RETURN(PTR_ERR(sub_th));
173 rc = update_record_pack(create, th,
174 lu_object_fid(&dt->do_lu),
180 rc = dt_create(env, dt, attr, hint, dof, sub_th);
186 * Declare adding reference for the sub-object
188 * Get transaction of next layer and declare the reference adding.
190 * \param[in] env execution environment
191 * \param[in] dt dt object to add reference
192 * \param[in] th transaction handle
194 * \retval 0 if the declaration succeeds.
195 * \retval negative errno if the declaration fails.
197 int lod_sub_object_declare_ref_add(const struct lu_env *env,
198 struct dt_object *dt,
201 struct thandle *sub_th;
205 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
207 RETURN(PTR_ERR(sub_th));
209 rc = dt_declare_ref_add(env, dt, sub_th);
215 * Add reference for the sub-object
217 * Get transaction of next layer, record updates if it belongs to cross-MDT
218 * operation and add reference of the object.
220 * \param[in] env execution environment
221 * \param[in] dt dt object to add reference
222 * \param[in] th transaction handle
224 * \retval 0 if it succeeds.
225 * \retval negative errno if it fails.
227 int lod_sub_object_ref_add(const struct lu_env *env, struct dt_object *dt,
230 struct thandle *sub_th;
235 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
237 RETURN(PTR_ERR(sub_th));
240 rc = update_record_pack(ref_add, th,
241 lu_object_fid(&dt->do_lu));
246 rc = dt_ref_add(env, dt, sub_th);
252 * Declare deleting reference for the sub-object
254 * Get transaction of next layer and declare the reference deleting.
256 * \param[in] env execution environment
257 * \param[in] dt dt object to delete reference
258 * \param[in] th transaction handle
260 * \retval 0 if the declaration succeeds.
261 * \retval negative errno if the declaration fails.
263 int lod_sub_object_declare_ref_del(const struct lu_env *env,
264 struct dt_object *dt,
267 struct thandle *sub_th;
271 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
273 RETURN(PTR_ERR(sub_th));
275 rc = dt_declare_ref_del(env, dt, sub_th);
281 * Delete reference for the sub-object
283 * Get transaction of next layer, record updates if it belongs to cross-MDT
284 * operation and delete reference of the object.
286 * \param[in] env execution environment
287 * \param[in] dt dt object to delete reference
288 * \param[in] th transaction handle
290 * \retval 0 if it succeeds.
291 * \retval negative errno if it fails.
293 int lod_sub_object_ref_del(const struct lu_env *env, struct dt_object *dt,
296 struct thandle *sub_th;
301 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
303 RETURN(PTR_ERR(sub_th));
306 rc = update_record_pack(ref_del, th,
307 lu_object_fid(&dt->do_lu));
312 rc = dt_ref_del(env, dt, sub_th);
318 * Declare destroying sub-object
320 * Get transaction of next layer and declare the sub-object destroy.
322 * \param[in] env execution environment
323 * \param[in] dt dt object to be destroyed
324 * \param[in] th transaction handle
326 * \retval 0 if the declaration succeeds.
327 * \retval negative errno if the declaration fails.
329 int lod_sub_object_declare_destroy(const struct lu_env *env,
330 struct dt_object *dt,
333 struct thandle *sub_th;
337 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
339 RETURN(PTR_ERR(sub_th));
341 rc = dt_declare_destroy(env, dt, sub_th);
349 * Get transaction of next layer, record updates if it belongs to cross-MDT
350 * operation and destroy the object.
352 * \param[in] env execution environment
353 * \param[in] dt dt object to be destroyed
354 * \param[in] th transaction handle
356 * \retval 0 if the destroy succeeds.
357 * \retval negative errno if the destroy fails.
359 int lod_sub_object_destroy(const struct lu_env *env, struct dt_object *dt,
362 struct thandle *sub_th;
367 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
369 RETURN(PTR_ERR(sub_th));
372 rc = update_record_pack(object_destroy, th,
373 lu_object_fid(&dt->do_lu));
378 rc = dt_destroy(env, dt, sub_th);
384 * Declare sub-object index insert
386 * Get transaction of next layer and declare index insert.
388 * \param[in] env execution environment
389 * \param[in] dt object for which to insert index
390 * \param[in] rec record of the index which will be inserted
391 * \param[in] key key of the index which will be inserted
392 * \param[in] th the transaction handle
394 * \retval 0 if the declaration succeeds.
395 * \retval negative errno if the declaration fails.
397 int lod_sub_object_declare_insert(const struct lu_env *env,
398 struct dt_object *dt,
399 const struct dt_rec *rec,
400 const struct dt_key *key,
403 struct thandle *sub_th;
405 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
407 return PTR_ERR(sub_th);
409 return dt_declare_insert(env, dt, rec, key, sub_th);
413 * Insert index of sub object
415 * Get transaction of next layer, record updates if it belongs to cross-MDT
416 * operation, and insert the index.
418 * \param[in] env execution environment
419 * \param[in] dt object for which to insert index
420 * \param[in] rec record of the index to be inserted
421 * \param[in] key key of the index to be inserted
422 * \param[in] th the transaction handle
423 * \param[in] ign whether ignore quota
425 * \retval 0 if the insertion succeeds.
426 * \retval negative errno if the insertion fails.
428 int lod_sub_object_index_insert(const struct lu_env *env, struct dt_object *dt,
429 const struct dt_rec *rec,
430 const struct dt_key *key, struct thandle *th,
433 struct thandle *sub_th;
437 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
439 return PTR_ERR(sub_th);
442 rc = update_record_pack(index_insert, th,
443 lu_object_fid(&dt->do_lu), rec, key);
448 return dt_insert(env, dt, rec, key, sub_th, ign);
452 * Declare sub-object index delete
454 * Get transaction of next layer and declare index deletion.
456 * \param[in] env execution environment
457 * \param[in] dt object for which to delete index
458 * \param[in] key key of the index which will be deleted
459 * \param[in] th the transaction handle
461 * \retval 0 if the declaration succeeds.
462 * \retval negative errno if the declaration fails.
464 int lod_sub_object_declare_delete(const struct lu_env *env,
465 struct dt_object *dt,
466 const struct dt_key *key,
469 struct thandle *sub_th;
471 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
473 return PTR_ERR(sub_th);
475 return dt_declare_delete(env, dt, key, sub_th);
479 * Delete index of sub object
481 * Get transaction of next layer, record updates if it belongs to cross-MDT
482 * operation, and delete the index.
484 * \param[in] env execution environment
485 * \param[in] dt object for which to delete index
486 * \param[in] key key of the index to be deleted
487 * \param[in] th the transaction handle
489 * \retval 0 if the deletion succeeds.
490 * \retval negative errno if the deletion fails.
492 int lod_sub_object_delete(const struct lu_env *env, struct dt_object *dt,
493 const struct dt_key *name, struct thandle *th)
495 struct thandle *sub_th;
500 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
502 RETURN(PTR_ERR(sub_th));
505 rc = update_record_pack(index_delete, th,
506 lu_object_fid(&dt->do_lu), name);
511 rc = dt_delete(env, dt, name, sub_th);
518 * Get transaction of next layer, and declare xattr set.
520 * \param[in] env execution environment
521 * \param[in] dt object on which to set xattr
522 * \param[in] buf xattr to be set
523 * \param[in] name name of the xattr
524 * \param[in] fl flag for setting xattr
526 * \retval 0 if the declaration succeeds.
527 * \retval negative errno if the declaration fails.
529 int lod_sub_object_declare_xattr_set(const struct lu_env *env,
530 struct dt_object *dt,
531 const struct lu_buf *buf,
532 const char *name, int fl,
535 struct thandle *sub_th;
539 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
541 RETURN(PTR_ERR(sub_th));
543 rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
551 * Get transaction of next layer, record updates if it belongs to cross-MDT
552 * operation, and set xattr to the object.
554 * \param[in] env execution environment
555 * \param[in] dt object on which to set xattr
556 * \param[in] buf xattr to be set
557 * \param[in] name name of the xattr
558 * \param[in] fl flag for setting xattr
559 * \param[in] th transaction handle
561 * \retval 0 if the xattr setting succeeds.
562 * \retval negative errno if xattr setting fails.
564 int lod_sub_object_xattr_set(const struct lu_env *env, struct dt_object *dt,
565 const struct lu_buf *buf, const char *name, int fl,
568 struct thandle *sub_th;
573 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
575 RETURN(PTR_ERR(sub_th));
578 rc = update_record_pack(xattr_set, th,
579 lu_object_fid(&dt->do_lu),
585 rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
593 * Get transaction of next layer, and declare attr set.
595 * \param[in] env execution environment
596 * \param[in] dt object on which to set attr
597 * \param[in] attr attributes to be set
598 * \param[in] th transaction handle
600 * \retval 0 if the declaration succeeds.
601 * \retval negative errno if the declaration fails.
603 int lod_sub_object_declare_attr_set(const struct lu_env *env,
604 struct dt_object *dt,
605 const struct lu_attr *attr,
608 struct thandle *sub_th;
612 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
614 RETURN(PTR_ERR(sub_th));
616 rc = dt_declare_attr_set(env, dt, attr, sub_th);
624 * Get transaction of next layer, record updates if it belongs to cross-MDT
625 * operation, and set attributes to the object.
627 * \param[in] env execution environment
628 * \param[in] dt object on which to set attr
629 * \param[in] attr attrbutes to be set
630 * \param[in] th transaction handle
632 * \retval 0 if attributes setting succeeds.
633 * \retval negative errno if the attributes setting fails.
635 int lod_sub_object_attr_set(const struct lu_env *env,
636 struct dt_object *dt,
637 const struct lu_attr *attr,
641 struct thandle *sub_th;
645 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
647 RETURN(PTR_ERR(sub_th));
650 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
656 rc = dt_attr_set(env, dt, attr, sub_th);
664 * Get transaction of next layer, and declare xattr deletion.
666 * \param[in] env execution environment
667 * \param[in] dt object on which to delete xattr
668 * \param[in] name name of the xattr to be deleted
669 * \param[in] th transaction handle
671 * \retval 0 if the declaration succeeds.
672 * \retval negative errno if the declaration fails.
674 int lod_sub_object_declare_xattr_del(const struct lu_env *env,
675 struct dt_object *dt,
679 struct thandle *sub_th;
683 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
685 RETURN(PTR_ERR(sub_th));
687 rc = dt_declare_xattr_del(env, dt, name, sub_th);
693 * xattribute deletion
695 * Get transaction of next layer, record update if it belongs to cross-MDT
696 * operation and delete xattr.
698 * \param[in] env execution environment
699 * \param[in] dt object on which to delete xattr
700 * \param[in] name name of the xattr to be deleted
701 * \param[in] th transaction handle
703 * \retval 0 if the deletion succeeds.
704 * \retval negative errno if the deletion fails.
706 int lod_sub_object_xattr_del(const struct lu_env *env,
707 struct dt_object *dt,
711 struct thandle *sub_th;
716 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
718 RETURN(PTR_ERR(sub_th));
721 rc = update_record_pack(xattr_del, th,
722 lu_object_fid(&dt->do_lu), name);
727 rc = dt_xattr_del(env, dt, name, sub_th);
733 * Declare buffer write
735 * Get transaction of next layer and declare buffer write.
737 * \param[in] env execution environment
738 * \param[in] dt object to be written
739 * \param[in] buf buffer to write which includes an embedded size field
740 * \param[in] pos offet in the object to start writing at
741 * \param[in] th transaction handle
743 * \retval 0 if the insertion succeeds.
744 * \retval negative errno if the insertion fails.
746 int lod_sub_object_declare_write(const struct lu_env *env,
747 struct dt_object *dt,
748 const struct lu_buf *buf, loff_t pos,
751 struct thandle *sub_th;
755 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
757 RETURN(PTR_ERR(sub_th));
759 rc = dt_declare_write(env, dt, buf, pos, sub_th);
765 * Write buffer to sub object
767 * Get transaction of next layer, records buffer write if it belongs to
768 * Cross-MDT operation, and write buffer.
770 * \param[in] env execution environment
771 * \param[in] dt object to be written
772 * \param[in] buf buffer to write which includes an embedded size field
773 * \param[in] pos offet in the object to start writing at
774 * \param[in] th transaction handle
775 * \param[in] rq enforcement for this write
777 * \retval the buffer size in bytes if it succeeds.
778 * \retval negative errno if it fails.
780 ssize_t lod_sub_object_write(const struct lu_env *env, struct dt_object *dt,
781 const struct lu_buf *buf, loff_t *pos,
782 struct thandle *th, int rq)
784 struct thandle *sub_th;
789 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
791 RETURN(PTR_ERR(sub_th));
794 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
800 rc = dt_write(env, dt, buf, pos, sub_th, rq);
807 * Get transaction of next layer and declare punch.
809 * \param[in] env execution environment
810 * \param[in] dt object to be written
811 * \param[in] start start offset of punch
812 * \param[in] end end offet of punch
813 * \param[in] th transaction handle
815 * \retval 0 if the insertion succeeds.
816 * \retval negative errno if the insertion fails.
818 int lod_sub_object_declare_punch(const struct lu_env *env,
819 struct dt_object *dt,
820 __u64 start, __u64 end,
823 struct thandle *sub_th;
827 sub_th = lod_sub_get_thandle(env, th, dt, NULL);
829 RETURN(PTR_ERR(sub_th));
831 rc = dt_declare_punch(env, dt, start, end, sub_th);
837 * Punch to sub object
839 * Get transaction of next layer, records buffer write if it belongs to
840 * Cross-MDT operation, and punch object.
842 * \param[in] env execution environment
843 * \param[in] dt object to be written
844 * \param[in] start start offset of punch
845 * \param[in] end end offset of punch
846 * \param[in] th transaction handle
847 * \param[in] capa capability of the write
849 * \retval the buffer size in bytes if it succeeds.
850 * \retval negative errno if it fails.
852 int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
853 __u64 start, __u64 end, struct thandle *th)
855 struct thandle *sub_th;
860 sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
862 RETURN(PTR_ERR(sub_th));
865 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
871 rc = dt_punch(env, dt, start, end, sub_th);
876 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
877 struct dt_device *dt, int index)
879 struct lod_thread_info *lti = lod_env_info(env);
880 struct llog_ctxt *ctxt;
881 struct llog_handle *lgh;
882 struct llog_catid *cid = <i->lti_cid;
883 struct lu_fid *fid = <i->lti_fid;
884 struct obd_device *obd;
886 bool need_put = false;
889 lu_update_log_fid(fid, index);
891 rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
895 rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
897 CERROR("%s: can't get id from catalogs: rc = %d\n",
898 lod2obd(lod)->obd_name, rc);
902 obd = dt->dd_lu_dev.ld_obd;
903 ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
904 LASSERT(ctxt != NULL);
905 ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
907 if (likely(logid_id(&cid->lci_logid) != 0)) {
908 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
911 /* re-create llog if it is missing */
913 logid_set_id(&cid->lci_logid, 0);
918 if (unlikely(logid_id(&cid->lci_logid) == 0)) {
919 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
922 cid->lci_logid = lgh->lgh_id;
926 LASSERT(lgh != NULL);
927 ctxt->loc_handle = lgh;
929 rc = llog_cat_init_and_process(env, lgh);
934 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
939 CDEBUG(D_INFO, "%s: Init llog for %d - catid "DOSTID":%x\n",
940 obd->obd_name, index, POSTID(&cid->lci_logid.lgl_oi),
941 cid->lci_logid.lgl_ogen);
944 llog_cat_close(env, ctxt->loc_handle);
945 ctxt->loc_handle = NULL;