Whamcloud - gitweb
LU-5451 lod: improve weird FID handling
[fs/lustre-release.git] / lustre / lod / lod_dev.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  2009 Sun Microsystems, Inc. All rights reserved
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2013, Intel Corporation.
27  *
28  */
29 /*
30  * This file is part of Lustre, http://www.lustre.org/
31  * Lustre is a trademark of Sun Microsystems, Inc.
32  *
33  * lustre/lod/lod_dev.c
34  *
35  * Lustre Logical Object Device
36  *
37  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
38  * Author: Mikhail Pershin <mike.pershin@intel.com>
39  */
40 /**
41  * The Logical Object Device (LOD) layer manages access to striped
42  * objects (both regular files and directories). It implements the DT
43  * device and object APIs and is responsible for creating, storing,
44  * and loading striping information as an extended attribute of the
45  * underlying OSD object. LOD is the server side analog of the LOV and
46  * LMV layers on the client side.
47  *
48  * Metadata LU object stack (layers of the same compound LU object,
49  * all have the same FID):
50  *
51  *        MDT
52  *         |      MD API
53  *        MDD
54  *         |      DT API
55  *        LOD
56  *       /   \    DT API
57  *     OSD   OSP
58  *
59  * During LOD object initialization the localness or remoteness of the
60  * object FID dictates the choice between OSD and OSP.
61  *
62  * An LOD object (file or directory) with N stripes (each has a
63  * different FID):
64  *
65  *          LOD
66  *           |
67  *   +---+---+---+...+
68  *   |   |   |   |   |
69  *   S0  S1  S2  S3  S(N-1)  OS[DP] objects, seen as DT objects by LOD
70  *
71  * When upper layers must access an object's stripes (which are
72  * themselves OST or MDT LU objects) LOD finds these objects by their
73  * FIDs and stores them as an array of DT object pointers on the
74  * object. Declarations and operations on LOD objects are received by
75  * LOD (as DT object operations) and performed on the underlying
76  * OS[DP] object and (as needed) on the stripes. From the perspective
77  * of LOD, a stripe-less file (created by mknod() or open with
78  * O_LOV_DELAY_CREATE) is an object which does not yet have stripes,
79  * while a non-striped directory (created by mkdir()) is an object
80  * which will never have stripes.
81  *
82  * The LOD layer also implements a small subset of the OBD device API
83  * to support MDT stack initialization and finalization (an MDD device
84  * connects and disconnects itself to and from the underlying LOD
85  * device), and pool management. In turn LOD uses the OBD device API
86  * to connect it self to the underlying OSD, and to connect itself to
87  * OSP devices representing the MDTs and OSTs that bear the stripes of
88  * its objects.
89  */
90
91 #define DEBUG_SUBSYSTEM S_MDS
92
93 #include <obd_class.h>
94 #include <md_object.h>
95 #include <lustre_fid.h>
96 #include <lustre_param.h>
97 #include <lustre_update.h>
98
99 #include "lod_internal.h"
100
101 /*
102  * Lookup target by FID.
103  *
104  * Lookup MDT/OST target index by FID. Type of the target can be
105  * specific or any.
106  *
107  * \param[in] env               LU environment provided by the caller
108  * \param[in] lod               lod device
109  * \param[in] fid               FID
110  * \param[out] tgt              result target index
111  * \param[in] type              expected type of the target:
112  *                              LU_SEQ_RANGE_{MDT,OST,ANY}
113  *
114  * \retval 0                    on success
115  * \retval negative             negated errno on error
116  **/
117 int lod_fld_lookup(const struct lu_env *env, struct lod_device *lod,
118                    const struct lu_fid *fid, __u32 *tgt, int *type)
119 {
120         struct lu_seq_range     range = { 0 };
121         struct lu_server_fld    *server_fld;
122         int rc = 0;
123         ENTRY;
124
125         if (!fid_is_sane(fid)) {
126                 CERROR("%s: invalid FID "DFID"\n", lod2obd(lod)->obd_name,
127                        PFID(fid));
128                 RETURN(-EIO);
129         }
130
131         if (fid_is_idif(fid)) {
132                 *tgt = fid_idif_ost_idx(fid);
133                 *type = LU_SEQ_RANGE_OST;
134                 RETURN(rc);
135         }
136
137         if (!lod->lod_initialized || (!fid_seq_in_fldb(fid_seq(fid)))) {
138                 LASSERT(lu_site2seq(lod2lu_dev(lod)->ld_site) != NULL);
139
140                 *tgt = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
141                 *type = LU_SEQ_RANGE_MDT;
142                 RETURN(rc);
143         }
144
145         server_fld = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_server_fld;
146         fld_range_set_type(&range, *type);
147         rc = fld_server_lookup(env, server_fld, fid_seq(fid), &range);
148         if (rc)
149                 RETURN(rc);
150
151         *tgt = range.lsr_index;
152         *type = range.lsr_flags;
153
154         CDEBUG(D_INFO, "LOD: got tgt %x for sequence: "
155                LPX64"\n", *tgt, fid_seq(fid));
156
157         RETURN(rc);
158 }
159
160 extern struct lu_object_operations lod_lu_obj_ops;
161 extern struct dt_object_operations lod_obj_ops;
162
163 /* Slab for OSD object allocation */
164 struct kmem_cache *lod_object_kmem;
165
166 static struct lu_kmem_descr lod_caches[] = {
167         {
168                 .ckd_cache = &lod_object_kmem,
169                 .ckd_name  = "lod_obj",
170                 .ckd_size  = sizeof(struct lod_object)
171         },
172         {
173                 .ckd_cache = NULL
174         }
175 };
176
177 static struct lu_device *lod_device_fini(const struct lu_env *env,
178                                          struct lu_device *d);
179
180 /**
181  * Implementation of lu_device_operations::ldo_object_alloc() for LOD
182  *
183  * Allocates and initializes LOD's slice in the given object.
184  *
185  * see include/lu_object.h for the details.
186  */
187 struct lu_object *lod_object_alloc(const struct lu_env *env,
188                                    const struct lu_object_header *hdr,
189                                    struct lu_device *dev)
190 {
191         struct lod_object       *lod_obj;
192         struct lu_object        *lu_obj;
193         ENTRY;
194
195         OBD_SLAB_ALLOC_PTR_GFP(lod_obj, lod_object_kmem, GFP_NOFS);
196         if (lod_obj == NULL)
197                 RETURN(ERR_PTR(-ENOMEM));
198
199         lu_obj = lod2lu_obj(lod_obj);
200         dt_object_init(&lod_obj->ldo_obj, NULL, dev);
201         lod_obj->ldo_obj.do_ops = &lod_obj_ops;
202         lu_obj->lo_ops = &lod_lu_obj_ops;
203
204         RETURN(lu_obj);
205 }
206
207 /**
208  * Cleanup table of target's descriptors.
209  *
210  * The function goes through all the targets in the given table
211  * and apply given configuration command on to the targets.
212  * Used to cleanup the targets at unmount.
213  *
214  * \param[in] env               LU environment provided by the caller
215  * \param[in] lod               lod device
216  * \param[in] ltd               target's table to go through
217  * \param[in] lcfg              configuration command to apply
218  *
219  * \retval 0                    on success
220  * \retval negative             negated errno on error
221  **/
222 static int lod_cleanup_desc_tgts(const struct lu_env *env,
223                                  struct lod_device *lod,
224                                  struct lod_tgt_descs *ltd,
225                                  struct lustre_cfg *lcfg)
226 {
227         struct lu_device  *next;
228         int rc = 0;
229         unsigned int i;
230
231         lod_getref(ltd);
232         if (ltd->ltd_tgts_size <= 0) {
233                 lod_putref(lod, ltd);
234                 return 0;
235         }
236         cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) {
237                 struct lod_tgt_desc *tgt;
238                 int rc1;
239
240                 tgt = LTD_TGT(ltd, i);
241                 LASSERT(tgt && tgt->ltd_tgt);
242                 next = &tgt->ltd_tgt->dd_lu_dev;
243                 rc1 = next->ld_ops->ldo_process_config(env, next, lcfg);
244                 if (rc1) {
245                         CERROR("%s: error cleaning up LOD index %u: cmd %#x"
246                                ": rc = %d\n", lod2obd(lod)->obd_name, i,
247                                lcfg->lcfg_command, rc1);
248                         rc = rc1;
249                 }
250         }
251         lod_putref(lod, ltd);
252         return rc;
253 }
254
255 /**
256  * Extract MDT target index from a device name.
257  *
258  * a helper function to extract index from the given device name
259  * like "fsname-MDTxxxx-mdtlov"
260  *
261  * \param[in] lodname   device name
262  * \param[out] index    extracted index
263  *
264  * \retval 0            on success
265  * \retval -EINVAL      if the name is invalid
266  */
267 static int lodname2mdt_index(char *lodname, long *index)
268 {
269         char *ptr, *tmp;
270
271         ptr = strrchr(lodname, '-');
272         if (ptr == NULL) {
273                 CERROR("invalid MDT index in '%s'\n", lodname);
274                 return -EINVAL;
275         }
276
277         if (strncmp(ptr, "-mdtlov", 7) != 0) {
278                 CERROR("invalid MDT index in '%s'\n", lodname);
279                 return -EINVAL;
280         }
281
282         if ((unsigned long)ptr - (unsigned long)lodname <= 8) {
283                 CERROR("invalid MDT index in '%s'\n", lodname);
284                 return -EINVAL;
285         }
286
287         if (strncmp(ptr - 8, "-MDT", 4) != 0) {
288                 CERROR("invalid MDT index in '%s'\n", lodname);
289                 return -EINVAL;
290         }
291
292         *index = simple_strtol(ptr - 4, &tmp, 16);
293         if (*tmp != '-' || *index > INT_MAX || *index < 0) {
294                 CERROR("invalid MDT index in '%s'\n", lodname);
295                 return -EINVAL;
296         }
297         return 0;
298 }
299
300 /**
301  * Implementation of lu_device_operations::ldo_process_config() for LOD
302  *
303  * The method is called by the configuration subsystem during setup,
304  * cleanup and when the configuration changes. The method processes
305  * few specific commands like adding/removing the targets, changing
306  * the runtime parameters.
307
308  * \param[in] env               LU environment provided by the caller
309  * \param[in] dev               lod device
310  * \param[in] lcfg              configuration command to apply
311  *
312  * \retval 0                    on success
313  * \retval negative             negated errno on error
314  *
315  * The examples are below.
316  *
317  * Add osc config log:
318  * marker  20 (flags=0x01, v2.2.49.56) lustre-OST0001  'add osc'
319  * add_uuid  nid=192.168.122.162@tcp(0x20000c0a87aa2)  0:  1:nidxxx
320  * attach    0:lustre-OST0001-osc-MDT0001  1:osc  2:lustre-MDT0001-mdtlov_UUID
321  * setup     0:lustre-OST0001-osc-MDT0001  1:lustre-OST0001_UUID  2:nid
322  * lov_modify_tgts add 0:lustre-MDT0001-mdtlov  1:lustre-OST0001_UUID  2:1  3:1
323  * marker  20 (flags=0x02, v2.2.49.56) lustre-OST0001  'add osc'
324  *
325  * Add mdc config log:
326  * marker  10 (flags=0x01, v2.2.49.56) lustre-MDT0000  'add osp'
327  * add_uuid  nid=192.168.122.162@tcp(0x20000c0a87aa2)  0:  1:nid
328  * attach 0:lustre-MDT0000-osp-MDT0001  1:osp  2:lustre-MDT0001-mdtlov_UUID
329  * setup     0:lustre-MDT0000-osp-MDT0001  1:lustre-MDT0000_UUID  2:nid
330  * modify_mdc_tgts add 0:lustre-MDT0001  1:lustre-MDT0000_UUID  2:0  3:1
331  * marker  10 (flags=0x02, v2.2.49.56) lustre-MDT0000_UUID  'add osp'
332  */
333 static int lod_process_config(const struct lu_env *env,
334                               struct lu_device *dev,
335                               struct lustre_cfg *lcfg)
336 {
337         struct lod_device *lod = lu2lod_dev(dev);
338         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
339         char              *arg1;
340         int                rc = 0;
341         ENTRY;
342
343         switch(lcfg->lcfg_command) {
344         case LCFG_LOV_DEL_OBD:
345         case LCFG_LOV_ADD_INA:
346         case LCFG_LOV_ADD_OBD:
347         case LCFG_ADD_MDC: {
348                 __u32 index;
349                 __u32 mdt_index;
350                 int gen;
351                 /* lov_modify_tgts add  0:lov_mdsA  1:osp  2:0  3:1
352                  * modify_mdc_tgts add  0:lustre-MDT0001
353                  *                    1:lustre-MDT0001-mdc0002
354                  *                    2:2  3:1*/
355                 arg1 = lustre_cfg_string(lcfg, 1);
356
357                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
358                         GOTO(out, rc = -EINVAL);
359                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
360                         GOTO(out, rc = -EINVAL);
361
362                 if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD) {
363                         char *mdt;
364                         mdt = strstr(lustre_cfg_string(lcfg, 0), "-MDT");
365                         /* 1.8 configs don't have "-MDT0000" at the end */
366                         if (mdt == NULL) {
367                                 mdt_index = 0;
368                         } else {
369                                 long long_index;
370                                 rc = lodname2mdt_index(
371                                         lustre_cfg_string(lcfg, 0),
372                                         &long_index);
373                                 if (rc != 0)
374                                         GOTO(out, rc);
375                                 mdt_index = long_index;
376                         }
377                         rc = lod_add_device(env, lod, arg1, index, gen,
378                                             mdt_index, LUSTRE_OSC_NAME, 1);
379                 } else if (lcfg->lcfg_command == LCFG_ADD_MDC) {
380                         mdt_index = index;
381                         rc = lod_add_device(env, lod, arg1, index, gen,
382                                             mdt_index, LUSTRE_MDC_NAME, 1);
383                 } else if (lcfg->lcfg_command == LCFG_LOV_ADD_INA) {
384                         /*FIXME: Add mdt_index for LCFG_LOV_ADD_INA*/
385                         mdt_index = 0;
386                         rc = lod_add_device(env, lod, arg1, index, gen,
387                                             mdt_index, LUSTRE_OSC_NAME, 0);
388                 } else {
389                         rc = lod_del_device(env, lod,
390                                             &lod->lod_ost_descs,
391                                             arg1, index, gen, true);
392                 }
393
394                 break;
395         }
396
397         case LCFG_PARAM: {
398                 struct obd_device *obd = lod2obd(lod);
399
400                 rc = class_process_proc_param(PARAM_LOV, obd->obd_vars,
401                                               lcfg, obd);
402                 if (rc > 0)
403                         rc = 0;
404                 GOTO(out, rc);
405         }
406         case LCFG_CLEANUP:
407         case LCFG_PRE_CLEANUP: {
408                 lu_dev_del_linkage(dev->ld_site, dev);
409                 lod_cleanup_desc_tgts(env, lod, &lod->lod_mdt_descs, lcfg);
410                 lod_cleanup_desc_tgts(env, lod, &lod->lod_ost_descs, lcfg);
411                 if (lcfg->lcfg_command == LCFG_PRE_CLEANUP)
412                         break;
413                 /*
414                  * do cleanup on underlying storage only when
415                  * all OSPs are cleaned up, as they use that OSD as well
416                  */
417                 next = &lod->lod_child->dd_lu_dev;
418                 rc = next->ld_ops->ldo_process_config(env, next, lcfg);
419                 if (rc)
420                         CERROR("%s: can't process %u: %d\n",
421                                lod2obd(lod)->obd_name, lcfg->lcfg_command, rc);
422
423                 rc = obd_disconnect(lod->lod_child_exp);
424                 if (rc)
425                         CERROR("error in disconnect from storage: %d\n", rc);
426                 break;
427         }
428         default:
429                CERROR("%s: unknown command %u\n", lod2obd(lod)->obd_name,
430                       lcfg->lcfg_command);
431                rc = -EINVAL;
432                break;
433         }
434
435 out:
436         RETURN(rc);
437 }
438
439 /**
440  * Implementation of lu_device_operations::ldo_recovery_complete() for LOD
441  *
442  * The method is called once the recovery is complete. This implementation
443  * distributes the notification to all the known targets.
444  *
445  * see include/lu_object.h for the details
446  */
447 static int lod_recovery_complete(const struct lu_env *env,
448                                  struct lu_device *dev)
449 {
450         struct lod_device   *lod = lu2lod_dev(dev);
451         struct lu_device    *next = &lod->lod_child->dd_lu_dev;
452         unsigned int         i;
453         int                  rc;
454         ENTRY;
455
456         LASSERT(lod->lod_recovery_completed == 0);
457         lod->lod_recovery_completed = 1;
458
459         rc = next->ld_ops->ldo_recovery_complete(env, next);
460
461         lod_getref(&lod->lod_ost_descs);
462         if (lod->lod_osts_size > 0) {
463                 cfs_foreach_bit(lod->lod_ost_bitmap, i) {
464                         struct lod_tgt_desc *tgt;
465                         tgt = OST_TGT(lod, i);
466                         LASSERT(tgt && tgt->ltd_tgt);
467                         next = &tgt->ltd_ost->dd_lu_dev;
468                         rc = next->ld_ops->ldo_recovery_complete(env, next);
469                         if (rc)
470                                 CERROR("%s: can't complete recovery on #%d:"
471                                         "%d\n", lod2obd(lod)->obd_name, i, rc);
472                 }
473         }
474         lod_putref(lod, &lod->lod_ost_descs);
475         RETURN(rc);
476 }
477
478 /**
479  * Implementation of lu_device_operations::ldo_prepare() for LOD
480  *
481  * see include/lu_object.h for the details.
482  */
483 static int lod_prepare(const struct lu_env *env, struct lu_device *pdev,
484                        struct lu_device *cdev)
485 {
486         struct lod_device   *lod = lu2lod_dev(cdev);
487         struct lu_device    *next = &lod->lod_child->dd_lu_dev;
488         int                  rc;
489         ENTRY;
490
491         rc = next->ld_ops->ldo_prepare(env, pdev, next);
492         if (rc != 0) {
493                 CERROR("%s: prepare bottom error: rc = %d\n",
494                        lod2obd(lod)->obd_name, rc);
495                 RETURN(rc);
496         }
497
498         lod->lod_initialized = 1;
499
500         RETURN(rc);
501 }
502
503 const struct lu_device_operations lod_lu_ops = {
504         .ldo_object_alloc       = lod_object_alloc,
505         .ldo_process_config     = lod_process_config,
506         .ldo_recovery_complete  = lod_recovery_complete,
507         .ldo_prepare            = lod_prepare,
508 };
509
510 /**
511  * Implementation of dt_device_operations::dt_root_get() for LOD
512  *
513  * see include/dt_object.h for the details.
514  */
515 static int lod_root_get(const struct lu_env *env,
516                         struct dt_device *dev, struct lu_fid *f)
517 {
518         return dt_root_get(env, dt2lod_dev(dev)->lod_child, f);
519 }
520
521 /**
522  * Implementation of dt_device_operations::dt_statfs() for LOD
523  *
524  * see include/dt_object.h for the details.
525  */
526 static int lod_statfs(const struct lu_env *env,
527                       struct dt_device *dev, struct obd_statfs *sfs)
528 {
529         return dt_statfs(env, dt2lod_dev(dev)->lod_child, sfs);
530 }
531
532 /**
533  * Implementation of dt_device_operations::dt_trans_create() for LOD
534  *
535  * Creates a transaction using local (to this node) OSD.
536  *
537  * see include/dt_object.h for the details.
538  */
539 static struct thandle *lod_trans_create(const struct lu_env *env,
540                                         struct dt_device *dev)
541 {
542         struct thandle *th;
543
544         th = dt_trans_create(env, dt2lod_dev(dev)->lod_child);
545         if (IS_ERR(th))
546                 return th;
547
548         return th;
549 }
550
551 /**
552  * Implementation of dt_device_operations::dt_trans_start() for LOD
553  *
554  * Starts the set of local transactions using the targets involved
555  * in declare phase. Initial support for the distributed transactions.
556  *
557  * see include/dt_object.h for the details.
558  */
559 static int lod_trans_start(const struct lu_env *env, struct dt_device *dev,
560                            struct thandle *th)
561 {
562         struct lod_device *lod = dt2lod_dev((struct dt_device *) dev);
563         int rc = 0;
564
565         if (unlikely(th->th_update != NULL)) {
566                 struct thandle_update *tu = th->th_update;
567                 struct dt_update_request *update;
568
569                 list_for_each_entry(update, &tu->tu_remote_update_list,
570                                     dur_list) {
571                         LASSERT(update->dur_dt != NULL);
572                         rc = dt_trans_start(env, update->dur_dt, th);
573                         if (rc != 0)
574                                 return rc;
575                 }
576         }
577         return dt_trans_start(env, lod->lod_child, th);
578 }
579
580 /**
581  * Implementation of dt_device_operations::dt_trans_stop() for LOD
582  *
583  * Stops the set of local transactions using the targets involved
584  * in declare phase. Initial support for the distributed transactions.
585  *
586  * see include/dt_object.h for the details.
587  */
588 static int lod_trans_stop(const struct lu_env *env, struct dt_device *dt,
589                           struct thandle *th)
590 {
591         struct thandle_update           *tu = th->th_update;
592         struct dt_update_request        *update;
593         struct dt_update_request        *tmp;
594         int                             rc2 = 0;
595         int                             rc;
596         ENTRY;
597
598         rc = dt_trans_stop(env, th->th_dev, th);
599         if (likely(tu == NULL))
600                 RETURN(rc);
601
602         list_for_each_entry_safe(update, tmp,
603                                  &tu->tu_remote_update_list,
604                                  dur_list) {
605                 /* update will be freed inside dt_trans_stop */
606                 rc2 = dt_trans_stop(env, update->dur_dt, th);
607                 if (unlikely(rc2 != 0 && rc == 0))
608                         rc = rc2;
609         }
610
611         RETURN(rc);
612 }
613
614 /**
615  * Implementation of dt_device_operations::dt_conf_get() for LOD
616  *
617  * Currently returns the configuration provided by the local OSD.
618  *
619  * see include/dt_object.h for the details.
620  */
621 static void lod_conf_get(const struct lu_env *env,
622                          const struct dt_device *dev,
623                          struct dt_device_param *param)
624 {
625         dt_conf_get(env, dt2lod_dev((struct dt_device *)dev)->lod_child, param);
626 }
627
628 /**
629  * Implementation of dt_device_operations::dt_sync() for LOD
630  *
631  * Syncs all known OST targets. Very very expensive and used
632  * rarely by LFSCK now. Should not be used in general.
633  *
634  * see include/dt_object.h for the details.
635  */
636 static int lod_sync(const struct lu_env *env, struct dt_device *dev)
637 {
638         struct lod_device   *lod = dt2lod_dev(dev);
639         struct lod_ost_desc *ost;
640         unsigned int         i;
641         int                  rc = 0;
642         ENTRY;
643
644         lod_getref(&lod->lod_ost_descs);
645         lod_foreach_ost(lod, i) {
646                 ost = OST_TGT(lod, i);
647                 LASSERT(ost && ost->ltd_ost);
648                 rc = dt_sync(env, ost->ltd_ost);
649                 if (rc) {
650                         CERROR("%s: can't sync %u: %d\n",
651                                lod2obd(lod)->obd_name, i, rc);
652                         break;
653                 }
654         }
655         lod_putref(lod, &lod->lod_ost_descs);
656         if (rc == 0)
657                 rc = dt_sync(env, lod->lod_child);
658
659         RETURN(rc);
660 }
661
662 /**
663  * Implementation of dt_device_operations::dt_ro() for LOD
664  *
665  * Turns local OSD read-only, used for the testing only.
666  *
667  * see include/dt_object.h for the details.
668  */
669 static int lod_ro(const struct lu_env *env, struct dt_device *dev)
670 {
671         return dt_ro(env, dt2lod_dev(dev)->lod_child);
672 }
673
674 /**
675  * Implementation of dt_device_operations::dt_commit_async() for LOD
676  *
677  * Asks local OSD to commit sooner.
678  *
679  * see include/dt_object.h for the details.
680  */
681 static int lod_commit_async(const struct lu_env *env, struct dt_device *dev)
682 {
683         return dt_commit_async(env, dt2lod_dev(dev)->lod_child);
684 }
685
686 /**
687  * Not used
688  */
689 static int lod_init_capa_ctxt(const struct lu_env *env, struct dt_device *dev,
690                               int mode, unsigned long timeout,
691                               __u32 alg, struct lustre_capa_key *keys)
692 {
693         struct dt_device *next = dt2lod_dev(dev)->lod_child;
694         return dt_init_capa_ctxt(env, next, mode, timeout, alg, keys);
695 }
696
697 static const struct dt_device_operations lod_dt_ops = {
698         .dt_root_get         = lod_root_get,
699         .dt_statfs           = lod_statfs,
700         .dt_trans_create     = lod_trans_create,
701         .dt_trans_start      = lod_trans_start,
702         .dt_trans_stop       = lod_trans_stop,
703         .dt_conf_get         = lod_conf_get,
704         .dt_sync             = lod_sync,
705         .dt_ro               = lod_ro,
706         .dt_commit_async     = lod_commit_async,
707         .dt_init_capa_ctxt   = lod_init_capa_ctxt,
708 };
709
710 /**
711  * Connect to a local OSD.
712  *
713  * Used to connect to the local OSD at mount. OSD name is taken from the
714  * configuration command passed. This connection is used to identify LU
715  * site and pin the OSD from early removal.
716  *
717  * \param[in] env               LU environment provided by the caller
718  * \param[in] lod               lod device
719  * \param[in] cfg               configuration command to apply
720  *
721  * \retval 0                    on success
722  * \retval negative             negated errno on error
723  **/
724 static int lod_connect_to_osd(const struct lu_env *env, struct lod_device *lod,
725                               struct lustre_cfg *cfg)
726 {
727         struct obd_connect_data *data = NULL;
728         struct obd_device       *obd;
729         char                    *nextdev = NULL, *p, *s;
730         int                      rc, len = 0;
731         ENTRY;
732
733         LASSERT(cfg);
734         LASSERT(lod->lod_child_exp == NULL);
735
736         /* compatibility hack: we still use old config logs
737          * which specify LOV, but we need to learn underlying
738          * OSD device, which is supposed to be:
739          *  <fsname>-MDTxxxx-osd
740          *
741          * 2.x MGS generates lines like the following:
742          *   #03 (176)lov_setup 0:lustre-MDT0000-mdtlov  1:(struct lov_desc)
743          * 1.8 MGS generates lines like the following:
744          *   #03 (168)lov_setup 0:lustre-mdtlov  1:(struct lov_desc)
745          *
746          * we use "-MDT" to differentiate 2.x from 1.8 */
747
748         if ((p = lustre_cfg_string(cfg, 0)) && strstr(p, "-mdtlov")) {
749                 len = strlen(p) + 1;
750                 OBD_ALLOC(nextdev, len);
751                 if (nextdev == NULL)
752                         GOTO(out, rc = -ENOMEM);
753
754                 strcpy(nextdev, p);
755                 s = strstr(nextdev, "-mdtlov");
756                 if (unlikely(s == NULL)) {
757                         CERROR("unable to parse device name %s\n",
758                                lustre_cfg_string(cfg, 0));
759                         GOTO(out, rc = -EINVAL);
760                 }
761
762                 if (strstr(nextdev, "-MDT")) {
763                         /* 2.x config */
764                         strcpy(s, "-osd");
765                 } else {
766                         /* 1.8 config */
767                         strcpy(s, "-MDT0000-osd");
768                 }
769         } else {
770                 CERROR("unable to parse device name %s\n",
771                        lustre_cfg_string(cfg, 0));
772                 GOTO(out, rc = -EINVAL);
773         }
774
775         OBD_ALLOC_PTR(data);
776         if (data == NULL)
777                 GOTO(out, rc = -ENOMEM);
778
779         obd = class_name2obd(nextdev);
780         if (obd == NULL) {
781                 CERROR("can not locate next device: %s\n", nextdev);
782                 GOTO(out, rc = -ENOTCONN);
783         }
784
785         data->ocd_connect_flags = OBD_CONNECT_VERSION;
786         data->ocd_version = LUSTRE_VERSION_CODE;
787
788         rc = obd_connect(env, &lod->lod_child_exp, obd, &obd->obd_uuid,
789                          data, NULL);
790         if (rc) {
791                 CERROR("cannot connect to next dev %s (%d)\n", nextdev, rc);
792                 GOTO(out, rc);
793         }
794
795         lod->lod_dt_dev.dd_lu_dev.ld_site =
796                 lod->lod_child_exp->exp_obd->obd_lu_dev->ld_site;
797         LASSERT(lod->lod_dt_dev.dd_lu_dev.ld_site);
798         lod->lod_child = lu2dt_dev(lod->lod_child_exp->exp_obd->obd_lu_dev);
799
800 out:
801         if (data)
802                 OBD_FREE_PTR(data);
803         if (nextdev)
804                 OBD_FREE(nextdev, len);
805         RETURN(rc);
806 }
807
808 /**
809  * Allocate and initialize target table.
810  *
811  * A helper function to initialize the target table and allocate
812  * a bitmap of the available targets.
813  *
814  * \param[in] ltd               target's table to initialize
815  *
816  * \retval 0                    on success
817  * \retval negative             negated errno on error
818  **/
819 static int lod_tgt_desc_init(struct lod_tgt_descs *ltd)
820 {
821         mutex_init(&ltd->ltd_mutex);
822         init_rwsem(&ltd->ltd_rw_sem);
823
824         /* the OST array and bitmap are allocated/grown dynamically as OSTs are
825          * added to the LOD, see lod_add_device() */
826         ltd->ltd_tgt_bitmap = CFS_ALLOCATE_BITMAP(32);
827         if (ltd->ltd_tgt_bitmap == NULL)
828                 RETURN(-ENOMEM);
829
830         ltd->ltd_tgts_size  = 32;
831         ltd->ltd_tgtnr      = 0;
832
833         ltd->ltd_death_row = 0;
834         ltd->ltd_refcount  = 0;
835         return 0;
836 }
837
838 /**
839  * Initialize LOD device at setup.
840  *
841  * Initializes the given LOD device using the original configuration command.
842  * The function initiates a connection to the local OSD and initializes few
843  * internal structures like pools, target tables, etc.
844  *
845  * \param[in] env               LU environment provided by the caller
846  * \param[in] lod               lod device
847  * \param[in] ldt               not used
848  * \param[in] cfg               configuration command
849  *
850  * \retval 0                    on success
851  * \retval negative             negated errno on error
852  **/
853 static int lod_init0(const struct lu_env *env, struct lod_device *lod,
854                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
855 {
856         struct dt_device_param ddp;
857         struct obd_device     *obd;
858         int                    rc;
859         ENTRY;
860
861         obd = class_name2obd(lustre_cfg_string(cfg, 0));
862         if (obd == NULL) {
863                 CERROR("Cannot find obd with name %s\n",
864                        lustre_cfg_string(cfg, 0));
865                 RETURN(-ENODEV);
866         }
867
868         obd->obd_lu_dev = &lod->lod_dt_dev.dd_lu_dev;
869         lod->lod_dt_dev.dd_lu_dev.ld_obd = obd;
870         lod->lod_dt_dev.dd_lu_dev.ld_ops = &lod_lu_ops;
871         lod->lod_dt_dev.dd_ops = &lod_dt_ops;
872
873         rc = lod_connect_to_osd(env, lod, cfg);
874         if (rc)
875                 RETURN(rc);
876
877         dt_conf_get(env, &lod->lod_dt_dev, &ddp);
878         lod->lod_osd_max_easize = ddp.ddp_max_ea_size;
879
880         /* setup obd to be used with old lov code */
881         rc = lod_pools_init(lod, cfg);
882         if (rc)
883                 GOTO(out_disconnect, rc);
884
885         rc = lod_procfs_init(lod);
886         if (rc)
887                 GOTO(out_pools, rc);
888
889         spin_lock_init(&lod->lod_desc_lock);
890         spin_lock_init(&lod->lod_connects_lock);
891         lod_tgt_desc_init(&lod->lod_mdt_descs);
892         lod_tgt_desc_init(&lod->lod_ost_descs);
893
894         RETURN(0);
895
896 out_pools:
897         lod_pools_fini(lod);
898 out_disconnect:
899         obd_disconnect(lod->lod_child_exp);
900         RETURN(rc);
901 }
902
903 /**
904  * Implementation of lu_device_type_operations::ldto_device_free() for LOD
905  *
906  * Releases the memory allocated for LOD device.
907  *
908  * see include/lu_object.h for the details.
909  */
910 static struct lu_device *lod_device_free(const struct lu_env *env,
911                                          struct lu_device *lu)
912 {
913         struct lod_device *lod = lu2lod_dev(lu);
914         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
915         ENTRY;
916
917         LASSERT(atomic_read(&lu->ld_ref) == 0);
918         dt_device_fini(&lod->lod_dt_dev);
919         OBD_FREE_PTR(lod);
920         RETURN(next);
921 }
922
923 /**
924  * Implementation of lu_device_type_operations::ldto_device_alloc() for LOD
925  *
926  * Allocates LOD device and calls the helpers to initialize it.
927  *
928  * see include/lu_object.h for the details.
929  */
930 static struct lu_device *lod_device_alloc(const struct lu_env *env,
931                                           struct lu_device_type *type,
932                                           struct lustre_cfg *lcfg)
933 {
934         struct lod_device *lod;
935         struct lu_device  *lu_dev;
936
937         OBD_ALLOC_PTR(lod);
938         if (lod == NULL) {
939                 lu_dev = ERR_PTR(-ENOMEM);
940         } else {
941                 int rc;
942
943                 lu_dev = lod2lu_dev(lod);
944                 dt_device_init(&lod->lod_dt_dev, type);
945                 rc = lod_init0(env, lod, type, lcfg);
946                 if (rc != 0) {
947                         lod_device_free(env, lu_dev);
948                         lu_dev = ERR_PTR(rc);
949                 }
950         }
951
952         return lu_dev;
953 }
954
955 /**
956  * Implementation of lu_device_type_operations::ldto_device_fini() for LOD
957  *
958  * Releases the internal resources used by LOD device.
959  *
960  * see include/lu_object.h for the details.
961  */
962 static struct lu_device *lod_device_fini(const struct lu_env *env,
963                                          struct lu_device *d)
964 {
965         struct lod_device *lod = lu2lod_dev(d);
966         int                rc;
967         ENTRY;
968
969         lod_pools_fini(lod);
970
971         lod_procfs_fini(lod);
972
973         rc = lod_fini_tgt(env, lod, &lod->lod_ost_descs, true);
974         if (rc)
975                 CERROR("%s:can not fini ost descs %d\n",
976                         lod2obd(lod)->obd_name, rc);
977
978         rc = lod_fini_tgt(env, lod, &lod->lod_mdt_descs, false);
979         if (rc)
980                 CERROR("%s:can not fini mdt descs %d\n",
981                         lod2obd(lod)->obd_name, rc);
982
983         RETURN(NULL);
984 }
985
986 /**
987  * Implementation of obd_ops::o_connect() for LOD
988  *
989  * Used to track all the users of this specific LOD device,
990  * so the device stays up until the last user disconnected.
991  *
992  * \param[in] env               LU environment provided by the caller
993  * \param[out] exp              export the caller will be using to access LOD
994  * \param[in] obd               OBD device representing LOD device
995  * \param[in] cluuid            unique identifier of the caller
996  * \param[in] data              not used
997  * \param[in] localdata         not used
998  *
999  * \retval 0                    on success
1000  * \retval negative             negated errno on error
1001  **/
1002 static int lod_obd_connect(const struct lu_env *env, struct obd_export **exp,
1003                            struct obd_device *obd, struct obd_uuid *cluuid,
1004                            struct obd_connect_data *data, void *localdata)
1005 {
1006         struct lod_device    *lod = lu2lod_dev(obd->obd_lu_dev);
1007         struct lustre_handle  conn;
1008         int                   rc;
1009         ENTRY;
1010
1011         CDEBUG(D_CONFIG, "connect #%d\n", lod->lod_connects);
1012
1013         rc = class_connect(&conn, obd, cluuid);
1014         if (rc)
1015                 RETURN(rc);
1016
1017         *exp = class_conn2export(&conn);
1018
1019         spin_lock(&lod->lod_connects_lock);
1020         lod->lod_connects++;
1021         /* at the moment we expect the only user */
1022         LASSERT(lod->lod_connects == 1);
1023         spin_unlock(&lod->lod_connects_lock);
1024
1025         RETURN(0);
1026 }
1027
1028 /**
1029  *
1030  * Implementation of obd_ops::o_disconnect() for LOD
1031  *
1032  * When the caller doesn't need to use this LOD instance, it calls
1033  * obd_disconnect() and LOD releases corresponding export/reference count.
1034  * Once all the users gone, LOD device is released.
1035  *
1036  * \param[in] exp               export provided to the caller in obd_connect()
1037  *
1038  * \retval 0                    on success
1039  * \retval negative             negated errno on error
1040  **/
1041 static int lod_obd_disconnect(struct obd_export *exp)
1042 {
1043         struct obd_device *obd = exp->exp_obd;
1044         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
1045         int                rc, release = 0;
1046         ENTRY;
1047
1048         /* Only disconnect the underlying layers on the final disconnect. */
1049         spin_lock(&lod->lod_connects_lock);
1050         lod->lod_connects--;
1051         if (lod->lod_connects != 0) {
1052                 /* why should there be more than 1 connect? */
1053                 spin_unlock(&lod->lod_connects_lock);
1054                 CERROR("%s: disconnect #%d\n", exp->exp_obd->obd_name,
1055                        lod->lod_connects);
1056                 goto out;
1057         }
1058         spin_unlock(&lod->lod_connects_lock);
1059
1060         /* the last user of lod has gone, let's release the device */
1061         release = 1;
1062
1063 out:
1064         rc = class_disconnect(exp); /* bz 9811 */
1065
1066         if (rc == 0 && release)
1067                 class_manual_cleanup(obd);
1068         RETURN(rc);
1069 }
1070
1071 LU_KEY_INIT(lod, struct lod_thread_info);
1072
1073 static void lod_key_fini(const struct lu_context *ctx,
1074                 struct lu_context_key *key, void *data)
1075 {
1076         struct lod_thread_info *info = data;
1077         /* allocated in lod_get_lov_ea
1078          * XXX: this is overload, a tread may have such store but used only
1079          * once. Probably better would be pool of such stores per LOD.
1080          */
1081         if (info->lti_ea_store) {
1082                 OBD_FREE_LARGE(info->lti_ea_store, info->lti_ea_store_size);
1083                 info->lti_ea_store = NULL;
1084                 info->lti_ea_store_size = 0;
1085         }
1086         lu_buf_free(&info->lti_linkea_buf);
1087         OBD_FREE_PTR(info);
1088 }
1089
1090 /* context key: lod_thread_key */
1091 LU_CONTEXT_KEY_DEFINE(lod, LCT_MD_THREAD);
1092
1093 LU_TYPE_INIT_FINI(lod, &lod_thread_key);
1094
1095 static struct lu_device_type_operations lod_device_type_ops = {
1096         .ldto_init           = lod_type_init,
1097         .ldto_fini           = lod_type_fini,
1098
1099         .ldto_start          = lod_type_start,
1100         .ldto_stop           = lod_type_stop,
1101
1102         .ldto_device_alloc   = lod_device_alloc,
1103         .ldto_device_free    = lod_device_free,
1104
1105         .ldto_device_fini    = lod_device_fini
1106 };
1107
1108 static struct lu_device_type lod_device_type = {
1109         .ldt_tags     = LU_DEVICE_DT,
1110         .ldt_name     = LUSTRE_LOD_NAME,
1111         .ldt_ops      = &lod_device_type_ops,
1112         .ldt_ctx_tags = LCT_MD_THREAD,
1113 };
1114
1115 /**
1116  * Implementation of obd_ops::o_get_info() for LOD
1117  *
1118  * Currently, there is only one supported key: KEY_OSP_CONNECTED , to provide
1119  * the caller binary status whether LOD has seen connection to any OST target.
1120  *
1121  * \param[in] env               LU environment provided by the caller
1122  * \param[in] exp               export of the caller
1123  * \param[in] keylen            len of the key
1124  * \param[in] key               the key
1125  * \param[in] vallen            not used
1126  * \param[in] val               not used
1127  * \param[in] lsm               not used
1128  *
1129  * \retval                      0 if a connection was seen
1130  * \retval                      -EAGAIN if LOD isn't running yet or no
1131  *                              connection has been seen yet
1132  * \retval                      -EINVAL if not supported key is requested
1133  **/
1134 static int lod_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1135                             __u32 keylen, void *key, __u32 *vallen, void *val,
1136                             struct lov_stripe_md *lsm)
1137 {
1138         int rc = -EINVAL;
1139
1140         if (KEY_IS(KEY_OSP_CONNECTED)) {
1141                 struct obd_device       *obd = exp->exp_obd;
1142                 struct lod_device       *d;
1143                 struct lod_ost_desc     *ost;
1144                 unsigned int            i;
1145                 int                     rc = 1;
1146
1147                 if (!obd->obd_set_up || obd->obd_stopping)
1148                         RETURN(-EAGAIN);
1149
1150                 d = lu2lod_dev(obd->obd_lu_dev);
1151                 lod_getref(&d->lod_ost_descs);
1152                 lod_foreach_ost(d, i) {
1153                         ost = OST_TGT(d, i);
1154                         LASSERT(ost && ost->ltd_ost);
1155
1156                         rc = obd_get_info(env, ost->ltd_exp, keylen, key,
1157                                           vallen, val, lsm);
1158                         /* one healthy device is enough */
1159                         if (rc == 0)
1160                                 break;
1161                 }
1162                 lod_putref(d, &d->lod_ost_descs);
1163                 RETURN(rc);
1164         }
1165
1166         RETURN(rc);
1167 }
1168
1169 static struct obd_ops lod_obd_device_ops = {
1170         .o_owner        = THIS_MODULE,
1171         .o_connect      = lod_obd_connect,
1172         .o_disconnect   = lod_obd_disconnect,
1173         .o_get_info     = lod_obd_get_info,
1174         .o_pool_new     = lod_pool_new,
1175         .o_pool_rem     = lod_pool_remove,
1176         .o_pool_add     = lod_pool_add,
1177         .o_pool_del     = lod_pool_del,
1178 };
1179
1180 static int __init lod_mod_init(void)
1181 {
1182         struct obd_type *type;
1183         int rc;
1184
1185         rc = lu_kmem_init(lod_caches);
1186         if (rc)
1187                 return rc;
1188
1189         rc = class_register_type(&lod_obd_device_ops, NULL, true, NULL,
1190                                  LUSTRE_LOD_NAME, &lod_device_type);
1191         if (rc) {
1192                 lu_kmem_fini(lod_caches);
1193                 return rc;
1194         }
1195
1196         /* create "lov" entry in procfs for compatibility purposes */
1197         type = class_search_type(LUSTRE_LOV_NAME);
1198         if (type != NULL && type->typ_procroot != NULL)
1199                 return rc;
1200
1201         type = class_search_type(LUSTRE_LOD_NAME);
1202         type->typ_procsym = lprocfs_seq_register("lov", proc_lustre_root,
1203                                                  NULL, NULL);
1204         if (IS_ERR(type->typ_procsym)) {
1205                 CERROR("lod: can't create compat entry \"lov\": %d\n",
1206                        (int)PTR_ERR(type->typ_procsym));
1207                 type->typ_procsym = NULL;
1208         }
1209         return rc;
1210 }
1211
1212 static void __exit lod_mod_exit(void)
1213 {
1214         class_unregister_type(LUSTRE_LOD_NAME);
1215         lu_kmem_fini(lod_caches);
1216 }
1217
1218 MODULE_AUTHOR("Whamcloud, Inc. <http://www.whamcloud.com/>");
1219 MODULE_DESCRIPTION("Lustre Logical Object Device ("LUSTRE_LOD_NAME")");
1220 MODULE_LICENSE("GPL");
1221
1222 module_init(lod_mod_init);
1223 module_exit(lod_mod_exit);
1224