Whamcloud - gitweb
a2bda2495dfff376a6f8e0ea13a93d4258911726
[fs/lustre-release.git] / lustre / lov / lov_ea.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/lov/lov_ea.c
37  *
38  * Author: Wang Di <wangdi@clusterfs.com>
39  */
40
41 #ifndef EXPORT_SYMTAB
42 # define EXPORT_SYMTAB
43 #endif
44 #define DEBUG_SUBSYSTEM S_LOV
45
46 #ifdef __KERNEL__
47 #include <asm/div64.h>
48 #include <libcfs/libcfs.h>
49 #else
50 #include <liblustre.h>
51 #endif
52
53 #include <obd_class.h>
54 #include <obd_lov.h>
55 #include <lustre/lustre_idl.h>
56 #include <lustre_log.h>
57
58 #include "lov_internal.h"
59
60 struct lovea_unpack_args {
61         struct lov_stripe_md *lsm;
62         int                   cursor;
63 };
64
65 static int lsm_lmm_verify_common(struct lov_mds_md *lmm, int lmm_bytes,
66                                  int stripe_count)
67 {
68
69         if (stripe_count == 0 || stripe_count > LOV_V1_INSANE_STRIPE_COUNT) {
70                 CERROR("bad stripe count %d\n", stripe_count);
71                 lov_dump_lmm(D_WARNING, lmm);
72                 return -EINVAL;
73         }
74
75         if (lmm->lmm_object_id == 0) {
76                 CERROR("zero object id\n");
77                 lov_dump_lmm(D_WARNING, lmm);
78                 return -EINVAL;
79         }
80
81         if (lmm->lmm_pattern != cpu_to_le32(LOV_PATTERN_RAID0)) {
82                 CERROR("bad striping pattern\n");
83                 lov_dump_lmm(D_WARNING, lmm);
84                 return -EINVAL;
85         }
86
87         if (lmm->lmm_stripe_size == 0 ||
88             (stripe_count != -1 &&
89              (__u64)le32_to_cpu(lmm->lmm_stripe_size)*stripe_count >
90              0xffffffff)) {
91                 CERROR("bad stripe size %u\n",
92                        le32_to_cpu(lmm->lmm_stripe_size));
93                 lov_dump_lmm(D_WARNING, lmm);
94                 return -EINVAL;
95         }
96         return 0;
97 }
98
99 struct lov_stripe_md *lsm_alloc_plain(int stripe_count, int *size)
100 {
101         struct lov_stripe_md *lsm;
102         int i, oinfo_ptrs_size;
103         struct lov_oinfo *loi;
104
105         LASSERT(stripe_count > 0);
106
107         oinfo_ptrs_size = sizeof(struct lov_oinfo *) * stripe_count;
108         *size = sizeof(struct lov_stripe_md) + oinfo_ptrs_size;
109
110         OBD_ALLOC(lsm, *size);
111         if (!lsm)
112                 return NULL;;
113
114         for (i = 0; i < stripe_count; i++) {
115                 OBD_SLAB_ALLOC_PTR_GFP(loi, lov_oinfo_slab, CFS_ALLOC_IO);
116                 if (loi == NULL)
117                         goto err;
118                 lsm->lsm_oinfo[i] = loi;
119         }
120         lsm->lsm_stripe_count = stripe_count;
121         lsm->lsm_pool_name[0] = '\0';
122         return lsm;
123
124 err:
125         while (--i >= 0)
126                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab, sizeof(*loi));
127         OBD_FREE(lsm, *size);
128         return NULL;
129 }
130
131 void lsm_free_plain(struct lov_stripe_md *lsm)
132 {
133         int stripe_count = lsm->lsm_stripe_count;
134         int i;
135
136         for (i = 0; i < stripe_count; i++)
137                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab,
138                               sizeof(struct lov_oinfo));
139         OBD_FREE(lsm, sizeof(struct lov_stripe_md) +
140                  stripe_count * sizeof(struct lov_oinfo *));
141 }
142
143 static void lsm_unpackmd_common(struct lov_stripe_md *lsm,
144                                 struct lov_mds_md *lmm)
145 {
146         /*
147          * This supposes lov_mds_md_v1/v3 first fields are
148          * are the same
149          */
150         lsm->lsm_object_id = le64_to_cpu(lmm->lmm_object_id);
151         lsm->lsm_object_gr = le64_to_cpu(lmm->lmm_object_gr);
152         lsm->lsm_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
153         lsm->lsm_pattern = le32_to_cpu(lmm->lmm_pattern);
154         lsm->lsm_pool_name[0] = '\0';
155 }
156
157 static void
158 lsm_stripe_by_index_plain(struct lov_stripe_md *lsm, int *stripeno,
159                            obd_off *lov_off, unsigned long *swidth)
160 {
161         if (swidth)
162                 *swidth = (unsigned long)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
163 }
164
165 static void
166 lsm_stripe_by_offset_plain(struct lov_stripe_md *lsm, int *stripeno,
167                            obd_off *lov_off, unsigned long *swidth)
168 {
169         if (swidth)
170                 *swidth = (unsigned long)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
171 }
172
173 static obd_off
174 lsm_stripe_offset_by_index_plain(struct lov_stripe_md *lsm,
175                                   int stripe_index)
176 {
177         return 0;
178 }
179
180 static obd_off
181 lsm_stripe_offset_by_offset_plain(struct lov_stripe_md *lsm,
182                                   obd_off lov_off)
183 {
184         return 0;
185 }
186
187 static int
188 lsm_stripe_index_by_offset_plain(struct lov_stripe_md *lsm,
189                                   obd_off lov_off)
190 {
191         return 0;
192 }
193
194 static int lsm_revalidate_plain(struct lov_stripe_md *lsm,
195                                 struct obd_device *obd)
196 {
197         return 0;
198 }
199
200 static int lsm_destroy_plain(struct lov_stripe_md *lsm, struct obdo *oa,
201                              struct obd_export *md_exp)
202 {
203         return 0;
204 }
205
206 static int lsm_lmm_verify_v1(struct lov_mds_md_v1 *lmm, int lmm_bytes,
207                              int *stripe_count)
208 {
209         if (lmm_bytes < sizeof(*lmm)) {
210                 CERROR("lov_mds_md_v1 too small: %d, need at least %d\n",
211                        lmm_bytes, (int)sizeof(*lmm));
212                 return -EINVAL;
213         }
214
215         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
216
217         if (lmm_bytes < lov_mds_md_size(*stripe_count, LOV_MAGIC_V1)) {
218                 CERROR("LOV EA V1 too small: %d, need %d\n",
219                        lmm_bytes, lov_mds_md_size(*stripe_count, LOV_MAGIC_V1));
220                 lov_dump_lmm_v1(D_WARNING, lmm);
221                 return -EINVAL;
222         }
223
224         return lsm_lmm_verify_common(lmm, lmm_bytes, *stripe_count);
225 }
226
227 int lsm_unpackmd_v1(struct lov_obd *lov, struct lov_stripe_md *lsm,
228                     struct lov_mds_md_v1 *lmm)
229 {
230         struct lov_oinfo *loi;
231         int i;
232
233         lsm_unpackmd_common(lsm, lmm);
234
235         for (i = 0; i < lsm->lsm_stripe_count; i++) {
236                 /* XXX LOV STACKING call down to osc_unpackmd() */
237                 loi = lsm->lsm_oinfo[i];
238                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
239                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
240                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
241                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
242                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
243                         CERROR("OST index %d more than OST count %d\n",
244                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
245                         lov_dump_lmm_v1(D_WARNING, lmm);
246                         return -EINVAL;
247                 }
248                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
249                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
250                         lov_dump_lmm_v1(D_WARNING, lmm);
251                         return -EINVAL;
252                 }
253         }
254
255         return 0;
256 }
257
258 const struct lsm_operations lsm_v1_ops = {
259         .lsm_free            = lsm_free_plain,
260         .lsm_destroy         = lsm_destroy_plain,
261         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
262         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
263         .lsm_revalidate         = lsm_revalidate_plain,
264         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
265         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
266         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
267         .lsm_lmm_verify         = lsm_lmm_verify_v1,
268         .lsm_unpackmd           = lsm_unpackmd_v1,
269 };
270
271 struct lov_extent *lovea_off2le(struct lov_stripe_md *lsm, obd_off lov_off)
272 {
273         struct lov_array_info *lai;
274         struct lov_extent *le;
275         int i = 0;
276
277         LASSERT(lsm->lsm_array != NULL);
278         lai = lsm->lsm_array;
279         LASSERT(lai->lai_ext_count > 1);
280
281         for (le = lai->lai_ext_array, i = 0;
282              i < lai->lai_ext_count && le->le_start + le->le_len <= lov_off
283              && le->le_len != -1;
284              i ++, le ++) {
285                ; /* empty loop */
286         }
287
288         CDEBUG(D_INFO, "off "LPU64" idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
289                lov_off, i, le->le_start, le->le_len, le->le_loi_idx,
290                le->le_stripe_count);
291
292         RETURN(le);
293 }
294
295 struct lov_extent *lovea_idx2le(struct lov_stripe_md *lsm, int stripe_no)
296 {
297         struct lov_extent *le;
298         struct lov_array_info *lai;
299         int i, stripe_index;
300
301         LASSERT(lsm->lsm_array != NULL);
302         LASSERT(stripe_no >= 0 && stripe_no <= lsm->lsm_stripe_count);
303         lai = lsm->lsm_array;
304         LASSERT(lai->lai_ext_count > 1);
305
306         for (le = lai->lai_ext_array, i = 0, stripe_index = le->le_stripe_count;
307              i < lai->lai_ext_count && stripe_index <= stripe_no &&
308              le->le_len != -1; i ++, le ++,
309              stripe_index += le->le_stripe_count) {
310                 ; /* empty loop */
311         }
312
313         CDEBUG(D_INFO, "stripe %d idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
314                stripe_no, i, le->le_start, le->le_len, le->le_loi_idx,
315                le->le_stripe_count);
316         RETURN(le);
317 }
318
319 static void lovea_free_array_info(struct lov_stripe_md *lsm)
320 {
321         if (!lsm || !lsm->lsm_array)
322                 return;
323
324         if (lsm->lsm_array->lai_ext_array)
325                 OBD_FREE(lsm->lsm_array->lai_ext_array,
326                          lsm->lsm_array->lai_ext_count *
327                          sizeof(struct lov_extent));
328
329         OBD_FREE_PTR(lsm->lsm_array);
330 }
331
332 static void lsm_free_join(struct lov_stripe_md *lsm)
333 {
334         lovea_free_array_info(lsm);
335         lsm_free_plain(lsm);
336 }
337
338 static void
339 lsm_stripe_by_index_join(struct lov_stripe_md *lsm, int *stripeno,
340                            obd_off *lov_off, unsigned long *swidth)
341 {
342         struct lov_extent *le;
343
344         LASSERT(stripeno != NULL);
345
346         le = lovea_idx2le(lsm, *stripeno);
347
348         LASSERT(le != NULL && le->le_stripe_count != 0);
349
350         *stripeno -= le->le_loi_idx;
351
352         if (swidth)
353                 *swidth = (unsigned long)lsm->lsm_stripe_size * le->le_stripe_count;
354
355         if (lov_off) {
356                 struct lov_extent *lov_le = lovea_off2le(lsm, *lov_off);
357                 if (lov_le == le) {
358                         *lov_off = (*lov_off > le->le_start) ?
359                                    (*lov_off - le->le_start) : 0;
360                 } else {
361                         *lov_off = (*lov_off > le->le_start) ?
362                                    le->le_len : 0;
363                         LASSERT(*lov_off != -1);
364                 }
365         }
366 }
367
368 static void
369 lsm_stripe_by_offset_join(struct lov_stripe_md *lsm, int *stripeno,
370                            obd_off *lov_off, unsigned long *swidth)
371 {
372         struct lov_extent *le;
373
374         LASSERT(lov_off != NULL);
375
376         le = lovea_off2le(lsm, *lov_off);
377
378         LASSERT(le != NULL && le->le_stripe_count != 0);
379
380         *lov_off = (*lov_off > le->le_start) ? (*lov_off - le->le_start) : 0;
381
382         if (stripeno)
383                 *stripeno -= le->le_loi_idx;
384
385         if (swidth)
386                 *swidth = (unsigned long)lsm->lsm_stripe_size * le->le_stripe_count;
387 }
388
389 static obd_off
390 lsm_stripe_offset_by_index_join(struct lov_stripe_md *lsm,
391                                  int stripe_index)
392 {
393         struct lov_extent *le;
394
395         le = lovea_idx2le(lsm, stripe_index);
396
397         return le ? le->le_start : 0;
398 }
399
400 static obd_off
401 lsm_stripe_offset_by_offset_join(struct lov_stripe_md *lsm,
402                                  obd_off lov_off)
403 {
404         struct lov_extent *le;
405
406         le = lovea_off2le(lsm, lov_off);
407
408         return le ? le->le_start : 0;
409 }
410
411 static int
412 lsm_stripe_index_by_offset_join(struct lov_stripe_md *lsm,
413                                  obd_off lov_off)
414 {
415         struct lov_extent *le = NULL;
416
417         le = lovea_off2le(lsm, lov_off);
418
419         return le ? le->le_loi_idx : 0;
420 }
421
422 static int lovea_unpack_array(struct llog_handle *handle,
423                               struct llog_rec_hdr *rec, void *data)
424 {
425         struct lovea_unpack_args *args = (struct lovea_unpack_args *)data;
426         struct llog_array_rec *la_rec = (struct llog_array_rec*)rec;
427         struct mds_extent_desc *med = &la_rec->lmr_med;
428         struct lov_stripe_md *lsm = args->lsm;
429         int cursor = args->cursor++;
430         struct lov_mds_md *lmm;
431         struct lov_array_info *lai;
432         struct lov_oinfo * loi;
433         int i, loi_index;
434         ENTRY;
435
436         /* sanity check */
437         LASSERT(lsm->lsm_stripe_count != 0);
438         lmm = &med->med_lmm;
439         LASSERT(lsm->lsm_array != NULL);
440
441         lai = lsm->lsm_array;
442
443         if (cursor == 0) {
444                lai->lai_ext_array[cursor].le_loi_idx = 0;
445         } else {
446                int next_loi_index = lai->lai_ext_array[cursor - 1].le_loi_idx +
447                                  lai->lai_ext_array[cursor - 1].le_stripe_count;
448                lai->lai_ext_array[cursor].le_loi_idx = next_loi_index;
449         }
450         /* insert extent desc into lsm extent array  */
451         lai->lai_ext_array[cursor].le_start = le64_to_cpu(med->med_start);
452         lai->lai_ext_array[cursor].le_len   = le64_to_cpu(med->med_len);
453         lai->lai_ext_array[cursor].le_stripe_count =
454                                    le32_to_cpu(lmm->lmm_stripe_count);
455
456         /* unpack extent's lmm to lov_oinfo array */
457         loi_index = lai->lai_ext_array[cursor].le_loi_idx;
458         CDEBUG(D_INFO, "lovea upackmd cursor %d, loi_index %d extent "
459                         LPU64":"LPU64"\n", cursor, loi_index, med->med_start,
460                         med->med_len);
461
462         for (i = 0; i < le32_to_cpu(lmm->lmm_stripe_count); i ++, loi_index++) {
463                 /* XXX LOV STACKING call down to osc_unpackmd() */
464                 loi = lsm->lsm_oinfo[loi_index];
465                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
466                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
467                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
468                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
469         }
470
471         RETURN(0);
472 }
473
474 static int lsm_revalidate_join(struct lov_stripe_md *lsm,
475                                struct obd_device *obd)
476 {
477         struct llog_handle *llh;
478         struct llog_ctxt *ctxt;
479         struct lovea_unpack_args args;
480         int rc, rc2;
481         ENTRY;
482
483         LASSERT(lsm->lsm_array != NULL);
484
485         /*Revalidate lsm might be called from client or MDS server.
486          *So the ctxt might be in different position
487          */
488         ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
489         if (!ctxt)
490                 ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
491
492         LASSERT(ctxt);
493
494         if (lsm->lsm_array && lsm->lsm_array->lai_ext_array)
495                 GOTO(release_ctxt, rc = 0);
496
497         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
498                lsm->lsm_array->lai_array_id.lgl_oid,
499                lsm->lsm_array->lai_array_id.lgl_ogr);
500         OBD_ALLOC(lsm->lsm_array->lai_ext_array,lsm->lsm_array->lai_ext_count *
501                                                 sizeof (struct lov_extent));
502         if (!lsm->lsm_array->lai_ext_array)
503                 GOTO(release_ctxt, rc = -ENOMEM);
504
505         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
506                lsm->lsm_array->lai_array_id.lgl_oid,
507                lsm->lsm_array->lai_array_id.lgl_ogr);
508
509         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id, NULL);
510         if (rc)
511                 GOTO(out, rc);
512
513         args.lsm = lsm;
514         args.cursor = 0;
515         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
516         if (rc == 0)
517                 rc = llog_process(llh, lovea_unpack_array, &args, NULL);
518         rc2 = llog_close(llh);
519         if (rc == 0)
520                 rc = rc2;
521 out:
522         if (rc)
523                 lovea_free_array_info(lsm);
524 release_ctxt:
525         llog_ctxt_put(ctxt);
526         RETURN(rc);
527 }
528
529 int lsm_destroy_join(struct lov_stripe_md *lsm, struct obdo *oa,
530                       struct obd_export *md_exp)
531 {
532         struct llog_ctxt *ctxt;
533         struct llog_handle *llh;
534         int rc = 0;
535         ENTRY;
536
537         LASSERT(md_exp != NULL);
538         /*for those orphan inode, we should keep array id*/
539         if (!(oa->o_valid & OBD_MD_FLCOOKIE))
540                 RETURN(rc);
541
542         ctxt = llog_get_context(md_exp->exp_obd, LLOG_LOVEA_REPL_CTXT);
543         if (!ctxt)
544                 RETURN(-EINVAL);
545
546         LASSERT(lsm->lsm_array != NULL);
547         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id,
548                          NULL);
549         if (rc)
550                 GOTO(out, rc);
551
552         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
553         if (rc == 0) {
554                 rc = llog_destroy(llh);
555         }
556         llog_free_handle(llh);
557 out:
558         llog_ctxt_put(ctxt);
559         RETURN(rc);
560 }
561
562 static int lsm_lmm_verify_join(struct lov_mds_md *lmm, int lmm_bytes,
563                                int *stripe_count)
564 {
565         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join *)lmm;
566
567         if (lmm_bytes < sizeof(*lmmj)) {
568                 CERROR("lov_mds_md too small: %d, need at least %d\n",
569                        lmm_bytes, (int)sizeof(*lmmj));
570                 return -EINVAL;
571         }
572
573         if (lmmj->lmmj_array_id.lgl_oid == 0) {
574                 CERROR("zero array object id\n");
575                 return -EINVAL;
576         }
577
578         *stripe_count = le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count);
579
580         return lsm_lmm_verify_common(&lmmj->lmmj_md, lmm_bytes, *stripe_count);
581 }
582
583 static int lovea_init_array_info(struct lov_stripe_md *lsm,
584                                  struct llog_logid *logid,
585                                  __u32 extent_count)
586 {
587         struct lov_array_info *lai;
588         ENTRY;
589
590         OBD_ALLOC_PTR(lai);
591         if (!lai)
592                 RETURN(-ENOMEM);
593
594         lai->lai_array_id.lgl_oid = le64_to_cpu(logid->lgl_oid);
595         lai->lai_array_id.lgl_ogr = le64_to_cpu(logid->lgl_ogr);
596         lai->lai_array_id.lgl_ogen = le32_to_cpu(logid->lgl_ogen);
597         lai->lai_ext_count = le32_to_cpu(extent_count);
598         lsm->lsm_array = lai;
599         RETURN(0);
600 }
601
602 static int lsm_unpackmd_join(struct lov_obd *lov, struct lov_stripe_md *lsm,
603                       struct lov_mds_md *lmm)
604 {
605         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join*)lmm;
606         int    rc;
607         ENTRY;
608
609         lsm_unpackmd_common(lsm, &lmmj->lmmj_md);
610
611         rc = lovea_init_array_info(lsm, &lmmj->lmmj_array_id,
612                                    lmmj->lmmj_extent_count);
613         if (rc) {
614                 CERROR("Init joined lsm id"LPU64" arrary error %d",
615                         lsm->lsm_object_id, rc);
616                 GOTO(out, rc);
617         }
618 out:
619         RETURN(rc);
620 }
621
622 const struct lsm_operations lsm_join_ops = {
623         .lsm_free             = lsm_free_join,
624         .lsm_destroy          = lsm_destroy_join,
625         .lsm_stripe_by_index  = lsm_stripe_by_index_join,
626         .lsm_stripe_by_offset = lsm_stripe_by_offset_join,
627         .lsm_revalidate       = lsm_revalidate_join,
628         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_join,
629         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_join,
630         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_join,
631         .lsm_lmm_verify         = lsm_lmm_verify_join,
632         .lsm_unpackmd           = lsm_unpackmd_join,
633 };
634
635
636 static int lsm_lmm_verify_v3(struct lov_mds_md *lmmv1, int lmm_bytes,
637                              int *stripe_count)
638 {
639         struct lov_mds_md_v3 *lmm;
640
641         lmm = (struct lov_mds_md_v3 *)lmmv1;
642
643         if (lmm_bytes < sizeof(*lmm)) {
644                 CERROR("lov_mds_md_v3 too small: %d, need at least %d\n",
645                        lmm_bytes, (int)sizeof(*lmm));
646                 return -EINVAL;
647         }
648
649         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
650
651         if (lmm_bytes < lov_mds_md_size(*stripe_count, LOV_MAGIC_V3)) {
652                 CERROR("LOV EA V3 too small: %d, need %d\n",
653                        lmm_bytes, lov_mds_md_size(*stripe_count, LOV_MAGIC_V3));
654                 lov_dump_lmm_v3(D_WARNING, lmm);
655                 return -EINVAL;
656         }
657
658         return lsm_lmm_verify_common((struct lov_mds_md_v1 *)lmm, lmm_bytes,
659                                      *stripe_count);
660 }
661
662 int lsm_unpackmd_v3(struct lov_obd *lov, struct lov_stripe_md *lsm,
663                     struct lov_mds_md *lmmv1)
664 {
665         struct lov_mds_md_v3 *lmm;
666         struct lov_oinfo *loi;
667         int i;
668
669         lmm = (struct lov_mds_md_v3 *)lmmv1;
670
671         lsm_unpackmd_common(lsm, (struct lov_mds_md_v1 *)lmm);
672         strncpy(lsm->lsm_pool_name, lmm->lmm_pool_name, LOV_MAXPOOLNAME);
673
674         for (i = 0; i < lsm->lsm_stripe_count; i++) {
675                 /* XXX LOV STACKING call down to osc_unpackmd() */
676                 loi = lsm->lsm_oinfo[i];
677                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
678                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
679                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
680                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
681                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
682                         CERROR("OST index %d more than OST count %d\n",
683                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
684                         lov_dump_lmm_v3(D_WARNING, lmm);
685                         return -EINVAL;
686                 }
687                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
688                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
689                         lov_dump_lmm_v3(D_WARNING, lmm);
690                         return -EINVAL;
691                 }
692         }
693
694         return 0;
695 }
696
697 const struct lsm_operations lsm_v3_ops = {
698         .lsm_free            = lsm_free_plain,
699         .lsm_destroy         = lsm_destroy_plain,
700         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
701         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
702         .lsm_revalidate         = lsm_revalidate_plain,
703         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
704         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
705         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
706         .lsm_lmm_verify         = lsm_lmm_verify_v3,
707         .lsm_unpackmd           = lsm_unpackmd_v3,
708 };
709