Whamcloud - gitweb
05eb8a49d87923107d2894bdc1659fbfb4f75a63
[fs/lustre-release.git] / lustre / lod / lod_sub_object.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,
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.
15
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
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2014, Intel Corporation.
24  */
25 /*
26  * lustre/lod/lod_sub_object.c
27  *
28  * LOD sub object methods
29  *
30  * This file implements sub-object methods for LOD.
31  *
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.
37  *
38  * Author: Di Wang <di.wang@intel.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_MDS
42
43 #include <obd.h>
44 #include <obd_class.h>
45 #include <lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
48
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>
54
55 #include "lod_internal.h"
56
57 struct thandle *lod_sub_get_thandle(const struct lu_env *env,
58                                     struct thandle *th,
59                                     const struct dt_object *sub_obj,
60                                     bool *record_update)
61 {
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;
66         __u32                   mdt_index;
67         int                     rc;
68         ENTRY;
69
70         if (record_update != NULL)
71                 *record_update = false;
72
73         if (th->th_top == NULL)
74                 RETURN(th);
75
76         tth = container_of(th, struct top_thandle, tt_super);
77
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 &&
87                     th->th_result == 0)
88                         *record_update = true;
89
90                 RETURN(tth->tt_master_sub_thandle);
91         }
92
93         rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
94                             &mdt_index, &type);
95         if (rc < 0)
96                 RETURN(ERR_PTR(rc));
97
98         if (type == LU_SEQ_RANGE_OST)
99                 RETURN(tth->tt_master_sub_thandle);
100
101         if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
102             th->th_result == 0)
103                 *record_update = true;
104
105         sub_th = thandle_get_sub(env, th, sub_obj);
106
107         RETURN(sub_th);
108 }
109
110 /**
111  * Declare sub-object creation.
112  *
113  * Get transaction of next layer and declare the creation of the object.
114  *
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
121  *
122  * \retval              0 if the declaration succeeds
123  * \retval              negative errno if the declaration fails.
124  */
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,
130                                   struct thandle *th)
131 {
132         struct thandle *sub_th;
133
134         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
135         if (IS_ERR(sub_th))
136                 return PTR_ERR(sub_th);
137
138         return dt_declare_create(env, dt, attr, hint, dof, sub_th);
139 }
140
141 /**
142  * Create sub-object.
143  *
144  * Get transaction of next layer, record updates if it belongs to cross-MDT
145  * operation, and create the object.
146  *
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
153  *
154  * \retval              0 if the creation succeeds
155  * \retval              negative errno if the creation fails.
156  */
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,
161                           struct thandle *th)
162 {
163         struct thandle     *sub_th;
164         bool               record_update;
165         int                 rc;
166         ENTRY;
167
168         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
169         if (IS_ERR(sub_th))
170                 RETURN(PTR_ERR(sub_th));
171
172         if (record_update) {
173                 rc = update_record_pack(create, th,
174                                         lu_object_fid(&dt->do_lu),
175                                         attr, hint, dof);
176                 if (rc < 0)
177                         RETURN(rc);
178         }
179
180         rc = dt_create(env, dt, attr, hint, dof, sub_th);
181
182         RETURN(rc);
183 }
184
185 /**
186  * Declare adding reference for the sub-object
187  *
188  * Get transaction of next layer and declare the reference adding.
189  *
190  * \param[in] env       execution environment
191  * \param[in] dt        dt object to add reference
192  * \param[in] th        transaction handle
193  *
194  * \retval              0 if the declaration succeeds.
195  * \retval              negative errno if the declaration fails.
196  */
197 int lod_sub_object_declare_ref_add(const struct lu_env *env,
198                                    struct dt_object *dt,
199                                    struct thandle *th)
200 {
201         struct thandle  *sub_th;
202         int             rc;
203         ENTRY;
204
205         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
206         if (IS_ERR(sub_th))
207                 RETURN(PTR_ERR(sub_th));
208
209         rc = dt_declare_ref_add(env, dt, sub_th);
210
211         RETURN(rc);
212 }
213
214 /**
215  * Add reference for the sub-object
216  *
217  * Get transaction of next layer, record updates if it belongs to cross-MDT
218  * operation and add reference of the object.
219  *
220  * \param[in] env       execution environment
221  * \param[in] dt        dt object to add reference
222  * \param[in] th        transaction handle
223  *
224  * \retval              0 if it succeeds.
225  * \retval              negative errno if it fails.
226  */
227 int lod_sub_object_ref_add(const struct lu_env *env, struct dt_object *dt,
228                            struct thandle *th)
229 {
230         struct thandle  *sub_th;
231         bool            record_update;
232         int             rc;
233         ENTRY;
234
235         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
236         if (IS_ERR(sub_th))
237                 RETURN(PTR_ERR(sub_th));
238
239         if (record_update) {
240                 rc = update_record_pack(ref_add, th,
241                                         lu_object_fid(&dt->do_lu));
242                 if (rc < 0)
243                         RETURN(rc);
244         }
245
246         rc = dt_ref_add(env, dt, sub_th);
247
248         RETURN(rc);
249 }
250
251 /**
252  * Declare deleting reference for the sub-object
253  *
254  * Get transaction of next layer and declare the reference deleting.
255  *
256  * \param[in] env       execution environment
257  * \param[in] dt        dt object to delete reference
258  * \param[in] th        transaction handle
259  *
260  * \retval              0 if the declaration succeeds.
261  * \retval              negative errno if the declaration fails.
262  */
263 int lod_sub_object_declare_ref_del(const struct lu_env *env,
264                                    struct dt_object *dt,
265                                    struct thandle *th)
266 {
267         struct thandle  *sub_th;
268         int             rc;
269         ENTRY;
270
271         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
272         if (IS_ERR(sub_th))
273                 RETURN(PTR_ERR(sub_th));
274
275         rc = dt_declare_ref_del(env, dt, sub_th);
276
277         RETURN(rc);
278 }
279
280 /**
281  * Delete reference for the sub-object
282  *
283  * Get transaction of next layer, record updates if it belongs to cross-MDT
284  * operation and delete reference of the object.
285  *
286  * \param[in] env       execution environment
287  * \param[in] dt        dt object to delete reference
288  * \param[in] th        transaction handle
289  *
290  * \retval              0 if it succeeds.
291  * \retval              negative errno if it fails.
292  */
293 int lod_sub_object_ref_del(const struct lu_env *env, struct dt_object *dt,
294                            struct thandle *th)
295 {
296         struct thandle  *sub_th;
297         bool            record_update;
298         int             rc;
299         ENTRY;
300
301         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
302         if (IS_ERR(sub_th))
303                 RETURN(PTR_ERR(sub_th));
304
305         if (record_update) {
306                 rc = update_record_pack(ref_del, th,
307                                         lu_object_fid(&dt->do_lu));
308                 if (rc < 0)
309                         RETURN(rc);
310         }
311
312         rc = dt_ref_del(env, dt, sub_th);
313
314         RETURN(rc);
315 }
316
317 /**
318  * Declare destroying sub-object
319  *
320  * Get transaction of next layer and declare the sub-object destroy.
321  *
322  * \param[in] env       execution environment
323  * \param[in] dt        dt object to be destroyed
324  * \param[in] th        transaction handle
325  *
326  * \retval              0 if the declaration succeeds.
327  * \retval              negative errno if the declaration fails.
328  */
329 int lod_sub_object_declare_destroy(const struct lu_env *env,
330                                    struct dt_object *dt,
331                                    struct thandle *th)
332 {
333         struct thandle  *sub_th;
334         int             rc;
335         ENTRY;
336
337         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
338         if (IS_ERR(sub_th))
339                 RETURN(PTR_ERR(sub_th));
340
341         rc = dt_declare_destroy(env, dt, sub_th);
342
343         RETURN(rc);
344 }
345
346 /**
347  * Destroy sub-object
348  *
349  * Get transaction of next layer, record updates if it belongs to cross-MDT
350  * operation and destroy the object.
351  *
352  * \param[in] env       execution environment
353  * \param[in] dt        dt object to be destroyed
354  * \param[in] th        transaction handle
355  *
356  * \retval              0 if the destroy succeeds.
357  * \retval              negative errno if the destroy fails.
358  */
359 int lod_sub_object_destroy(const struct lu_env *env, struct dt_object *dt,
360                            struct thandle *th)
361 {
362         struct thandle  *sub_th;
363         bool            record_update;
364         int             rc;
365         ENTRY;
366
367         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
368         if (IS_ERR(sub_th))
369                 RETURN(PTR_ERR(sub_th));
370
371         if (record_update) {
372                 rc = update_record_pack(object_destroy, th,
373                                         lu_object_fid(&dt->do_lu));
374                 if (rc < 0)
375                         RETURN(rc);
376         }
377
378         rc = dt_destroy(env, dt, sub_th);
379
380         RETURN(rc);
381 }
382
383 /**
384  * Declare sub-object index insert
385  *
386  * Get transaction of next layer and declare index insert.
387  *
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
393  *
394  * \retval              0 if the declaration succeeds.
395  * \retval              negative errno if the declaration fails.
396  */
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,
401                                   struct thandle *th)
402 {
403         struct thandle *sub_th;
404
405         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
406         if (IS_ERR(sub_th))
407                 return PTR_ERR(sub_th);
408
409         return dt_declare_insert(env, dt, rec, key, sub_th);
410 }
411
412 /**
413  * Insert index of sub object
414  *
415  * Get transaction of next layer, record updates if it belongs to cross-MDT
416  * operation, and insert the index.
417  *
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
424  *
425  * \retval              0 if the insertion succeeds.
426  * \retval              negative errno if the insertion fails.
427  */
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,
431                                 int ign)
432 {
433         struct thandle *sub_th;
434         int             rc;
435         bool            record_update;
436
437         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
438         if (IS_ERR(sub_th))
439                 return PTR_ERR(sub_th);
440
441         if (record_update) {
442                 rc = update_record_pack(index_insert, th,
443                                         lu_object_fid(&dt->do_lu), rec, key);
444                 if (rc < 0)
445                         return rc;
446         }
447
448         return dt_insert(env, dt, rec, key, sub_th, ign);
449 }
450
451 /**
452  * Declare sub-object index delete
453  *
454  * Get transaction of next layer and declare index deletion.
455  *
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
460  *
461  * \retval              0 if the declaration succeeds.
462  * \retval              negative errno if the declaration fails.
463  */
464 int lod_sub_object_declare_delete(const struct lu_env *env,
465                                   struct dt_object *dt,
466                                   const struct dt_key *key,
467                                   struct thandle *th)
468 {
469         struct thandle *sub_th;
470
471         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
472         if (IS_ERR(sub_th))
473                 return PTR_ERR(sub_th);
474
475         return dt_declare_delete(env, dt, key, sub_th);
476 }
477
478 /**
479  * Delete index of sub object
480  *
481  * Get transaction of next layer, record updates if it belongs to cross-MDT
482  * operation, and delete the index.
483  *
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
488  *
489  * \retval              0 if the deletion succeeds.
490  * \retval              negative errno if the deletion fails.
491  */
492 int lod_sub_object_delete(const struct lu_env *env, struct dt_object *dt,
493                           const struct dt_key *name, struct thandle *th)
494 {
495         struct thandle  *sub_th;
496         bool            record_update;
497         int             rc;
498         ENTRY;
499
500         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
501         if (IS_ERR(sub_th))
502                 RETURN(PTR_ERR(sub_th));
503
504         if (record_update) {
505                 rc = update_record_pack(index_delete, th,
506                                         lu_object_fid(&dt->do_lu), name);
507                 if (rc < 0)
508                         RETURN(rc);
509         }
510
511         rc = dt_delete(env, dt, name, sub_th);
512         RETURN(rc);
513 }
514
515 /**
516  * Declare xattr_set
517  *
518  * Get transaction of next layer, and declare xattr set.
519  *
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
525  *
526  * \retval              0 if the declaration succeeds.
527  * \retval              negative errno if the declaration fails.
528  */
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,
533                                      struct thandle *th)
534 {
535         struct thandle  *sub_th;
536         int             rc;
537         ENTRY;
538
539         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
540         if (IS_ERR(sub_th))
541                 RETURN(PTR_ERR(sub_th));
542
543         rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
544
545         RETURN(rc);
546 }
547
548 /**
549  * Set xattr
550  *
551  * Get transaction of next layer, record updates if it belongs to cross-MDT
552  * operation, and set xattr to the object.
553  *
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
560  *
561  * \retval              0 if the xattr setting succeeds.
562  * \retval              negative errno if xattr setting fails.
563  */
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,
566                              struct thandle *th)
567 {
568         struct thandle  *sub_th;
569         bool            record_update;
570         int             rc;
571         ENTRY;
572
573         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
574         if (IS_ERR(sub_th))
575                 RETURN(PTR_ERR(sub_th));
576
577         if (record_update) {
578                 rc = update_record_pack(xattr_set, th,
579                                         lu_object_fid(&dt->do_lu),
580                                         buf, name, fl);
581                 if (rc < 0)
582                         RETURN(rc);
583         }
584
585         rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
586
587         RETURN(rc);
588 }
589
590 /**
591  * Declare attr_set
592  *
593  * Get transaction of next layer, and declare attr set.
594  *
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
599  *
600  * \retval              0 if the declaration succeeds.
601  * \retval              negative errno if the declaration fails.
602  */
603 int lod_sub_object_declare_attr_set(const struct lu_env *env,
604                                     struct dt_object *dt,
605                                     const struct lu_attr *attr,
606                                     struct thandle *th)
607 {
608         struct thandle  *sub_th;
609         int             rc;
610         ENTRY;
611
612         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
613         if (IS_ERR(sub_th))
614                 RETURN(PTR_ERR(sub_th));
615
616         rc = dt_declare_attr_set(env, dt, attr, sub_th);
617
618         RETURN(rc);
619 }
620
621 /**
622  * attributes set
623  *
624  * Get transaction of next layer, record updates if it belongs to cross-MDT
625  * operation, and set attributes to the object.
626  *
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
631  *
632  * \retval              0 if attributes setting succeeds.
633  * \retval              negative errno if the attributes setting fails.
634  */
635 int lod_sub_object_attr_set(const struct lu_env *env,
636                             struct dt_object *dt,
637                             const struct lu_attr *attr,
638                             struct thandle *th)
639 {
640         bool               record_update;
641         struct thandle     *sub_th;
642         int                 rc;
643         ENTRY;
644
645         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
646         if (IS_ERR(sub_th))
647                 RETURN(PTR_ERR(sub_th));
648
649         if (record_update) {
650                 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
651                                         attr);
652                 if (rc < 0)
653                         RETURN(rc);
654         }
655
656         rc = dt_attr_set(env, dt, attr, sub_th);
657
658         RETURN(rc);
659 }
660
661 /**
662  * Declare xattr_del
663  *
664  * Get transaction of next layer, and declare xattr deletion.
665  *
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
670  *
671  * \retval              0 if the declaration succeeds.
672  * \retval              negative errno if the declaration fails.
673  */
674 int lod_sub_object_declare_xattr_del(const struct lu_env *env,
675                                      struct dt_object *dt,
676                                      const char *name,
677                                      struct thandle *th)
678 {
679         struct thandle  *sub_th;
680         int             rc;
681         ENTRY;
682
683         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
684         if (IS_ERR(sub_th))
685                 RETURN(PTR_ERR(sub_th));
686
687         rc = dt_declare_xattr_del(env, dt, name, sub_th);
688
689         RETURN(rc);
690 }
691
692 /**
693  * xattribute deletion
694  *
695  * Get transaction of next layer, record update if it belongs to cross-MDT
696  * operation and delete xattr.
697  *
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
702  *
703  * \retval              0 if the deletion succeeds.
704  * \retval              negative errno if the deletion fails.
705  */
706 int lod_sub_object_xattr_del(const struct lu_env *env,
707                              struct dt_object *dt,
708                              const char *name,
709                              struct thandle *th)
710 {
711         struct thandle  *sub_th;
712         bool            record_update;
713         int             rc;
714         ENTRY;
715
716         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
717         if (IS_ERR(sub_th))
718                 RETURN(PTR_ERR(sub_th));
719
720         if (record_update) {
721                 rc = update_record_pack(xattr_del, th,
722                                         lu_object_fid(&dt->do_lu), name);
723                 if (rc < 0)
724                         RETURN(rc);
725         }
726
727         rc = dt_xattr_del(env, dt, name, sub_th);
728
729         RETURN(rc);
730 }
731
732 /**
733  * Declare buffer write
734  *
735  * Get transaction of next layer and declare buffer write.
736  *
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
742  *
743  * \retval              0 if the insertion succeeds.
744  * \retval              negative errno if the insertion fails.
745  */
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,
749                                  struct thandle *th)
750 {
751         struct thandle  *sub_th;
752         int             rc;
753         ENTRY;
754
755         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
756         if (IS_ERR(sub_th))
757                 RETURN(PTR_ERR(sub_th));
758
759         rc = dt_declare_write(env, dt, buf, pos, sub_th);
760
761         RETURN(rc);
762 }
763
764 /**
765  * Write buffer to sub object
766  *
767  * Get transaction of next layer, records buffer write if it belongs to
768  * Cross-MDT operation, and write buffer.
769  *
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
776  *
777  * \retval              the buffer size in bytes if it succeeds.
778  * \retval              negative errno if it fails.
779  */
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)
783 {
784         struct thandle  *sub_th;
785         bool            record_update;
786         ssize_t         rc;
787         ENTRY;
788
789         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
790         if (IS_ERR(sub_th))
791                 RETURN(PTR_ERR(sub_th));
792
793         if (record_update) {
794                 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
795                                         buf, *pos);
796                 if (rc < 0)
797                         RETURN(rc);
798         }
799
800         rc = dt_write(env, dt, buf, pos, sub_th, rq);
801         RETURN(rc);
802 }
803
804 /**
805  * Declare punch
806  *
807  * Get transaction of next layer and declare punch.
808  *
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
814  *
815  * \retval              0 if the insertion succeeds.
816  * \retval              negative errno if the insertion fails.
817  */
818 int lod_sub_object_declare_punch(const struct lu_env *env,
819                                  struct dt_object *dt,
820                                  __u64 start, __u64 end,
821                                  struct thandle *th)
822 {
823         struct thandle  *sub_th;
824         int             rc;
825         ENTRY;
826
827         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
828         if (IS_ERR(sub_th))
829                 RETURN(PTR_ERR(sub_th));
830
831         rc = dt_declare_punch(env, dt, start, end, sub_th);
832
833         RETURN(rc);
834 }
835
836 /**
837  * Punch to sub object
838  *
839  * Get transaction of next layer, records buffer write if it belongs to
840  * Cross-MDT operation, and punch object.
841  *
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
848  *
849  * \retval              the buffer size in bytes if it succeeds.
850  * \retval              negative errno if it fails.
851  */
852 int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
853                          __u64 start, __u64 end, struct thandle *th)
854 {
855         struct thandle  *sub_th;
856         bool            record_update;
857         int             rc;
858         ENTRY;
859
860         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
861         if (IS_ERR(sub_th))
862                 RETURN(PTR_ERR(sub_th));
863
864         if (record_update) {
865                 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
866                                         start, end);
867                 if (rc < 0)
868                         RETURN(rc);
869         }
870
871         rc = dt_punch(env, dt, start, end, sub_th);
872
873         RETURN(rc);
874 }
875
876 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
877                       struct dt_device *dt, int index)
878 {
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 = &lti->lti_cid;
883         struct lu_fid           *fid = &lti->lti_fid;
884         struct obd_device       *obd;
885         int                     rc;
886         bool                    need_put = false;
887         ENTRY;
888
889         lu_update_log_fid(fid, index);
890
891         rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
892         if (rc < 0)
893                 RETURN(rc);
894
895         rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
896         if (rc != 0) {
897                 CERROR("%s: can't get id from catalogs: rc = %d\n",
898                        lod2obd(lod)->obd_name, rc);
899                 RETURN(rc);
900         }
901
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;
906
907         if (likely(logid_id(&cid->lci_logid) != 0)) {
908                 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
909                                LLOG_OPEN_EXISTS);
910
911                 /* re-create llog if it is missing */
912                 if (rc == -ENOENT)
913                         logid_set_id(&cid->lci_logid, 0);
914                 else if (rc < 0)
915                         GOTO(out_put, rc);
916         }
917
918         if (unlikely(logid_id(&cid->lci_logid) == 0)) {
919                 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
920                 if (rc < 0)
921                         GOTO(out_put, rc);
922                 cid->lci_logid = lgh->lgh_id;
923                 need_put = true;
924         }
925
926         LASSERT(lgh != NULL);
927         ctxt->loc_handle = lgh;
928
929         rc = llog_cat_init_and_process(env, lgh);
930         if (rc != 0)
931                 GOTO(out_close, rc);
932
933         if (need_put) {
934                 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
935                 if (rc != 0)
936                         GOTO(out_close, rc);
937         }
938
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);
942 out_close:
943         if (rc != 0) {
944                 llog_cat_close(env, ctxt->loc_handle);
945                 ctxt->loc_handle = NULL;
946         }
947 out_put:
948         llog_ctxt_put(ctxt);
949         RETURN(rc);
950 }
951