4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License version 2 only,
8 * as published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License version 2 for more details. A copy is
14 * included in the COPYING file that accompanied this code.
16 * You should have received a copy of the GNU General Public License
17 * along with this program; if not, write to the Free Software
18 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23 * Copyright 2009 Sun Microsystems, Inc. All rights reserved
24 * Use is subject to license terms.
26 * Copyright (c) 2011, 2012, Intel, Inc.
29 * lustre/lod/lod_object.c
31 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
35 # define EXPORT_SYMTAB
37 #define DEBUG_SUBSYSTEM S_MDS
40 #include <obd_class.h>
41 #include <lustre_ver.h>
42 #include <obd_support.h>
43 #include <lprocfs_status.h>
45 #include <lustre_fid.h>
46 #include <lustre_param.h>
47 #include <lustre_fid.h>
50 #include "lod_internal.h"
52 extern cfs_mem_cache_t *lod_object_kmem;
53 static const struct dt_body_operations lod_body_lnk_ops;
55 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
56 struct dt_rec *rec, const struct dt_key *key,
57 struct lustre_capa *capa)
59 struct dt_object *next = dt_object_child(dt);
60 return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
63 static int lod_declare_index_insert(const struct lu_env *env,
65 const struct dt_rec *rec,
66 const struct dt_key *key,
67 struct thandle *handle)
69 return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
72 static int lod_index_insert(const struct lu_env *env,
74 const struct dt_rec *rec,
75 const struct dt_key *key,
77 struct lustre_capa *capa,
80 return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
83 static int lod_declare_index_delete(const struct lu_env *env,
85 const struct dt_key *key,
88 return dt_declare_delete(env, dt_object_child(dt), key, th);
91 static int lod_index_delete(const struct lu_env *env,
93 const struct dt_key *key,
95 struct lustre_capa *capa)
97 return dt_delete(env, dt_object_child(dt), key, th, capa);
100 static struct dt_it *lod_it_init(const struct lu_env *env,
101 struct dt_object *dt, __u32 attr,
102 struct lustre_capa *capa)
104 struct dt_object *next = dt_object_child(dt);
106 return next->do_index_ops->dio_it.init(env, next, attr, capa);
109 static struct dt_index_operations lod_index_ops = {
110 .dio_lookup = lod_index_lookup,
111 .dio_declare_insert = lod_declare_index_insert,
112 .dio_insert = lod_index_insert,
113 .dio_declare_delete = lod_declare_index_delete,
114 .dio_delete = lod_index_delete,
120 static void lod_object_read_lock(const struct lu_env *env,
121 struct dt_object *dt, unsigned role)
123 dt_read_lock(env, dt_object_child(dt), role);
126 static void lod_object_write_lock(const struct lu_env *env,
127 struct dt_object *dt, unsigned role)
129 dt_write_lock(env, dt_object_child(dt), role);
132 static void lod_object_read_unlock(const struct lu_env *env,
133 struct dt_object *dt)
135 dt_read_unlock(env, dt_object_child(dt));
138 static void lod_object_write_unlock(const struct lu_env *env,
139 struct dt_object *dt)
141 dt_write_unlock(env, dt_object_child(dt));
144 static int lod_object_write_locked(const struct lu_env *env,
145 struct dt_object *dt)
147 return dt_write_locked(env, dt_object_child(dt));
150 static int lod_attr_get(const struct lu_env *env,
151 struct dt_object *dt,
152 struct lu_attr *attr,
153 struct lustre_capa *capa)
155 return dt_attr_get(env, dt_object_child(dt), attr, capa);
158 static int lod_declare_attr_set(const struct lu_env *env,
159 struct dt_object *dt,
160 const struct lu_attr *attr,
161 struct thandle *handle)
163 struct dt_object *next = dt_object_child(dt);
168 * declare setattr on the local object
170 rc = dt_declare_attr_set(env, next, attr, handle);
175 static int lod_attr_set(const struct lu_env *env,
176 struct dt_object *dt,
177 const struct lu_attr *attr,
178 struct thandle *handle,
179 struct lustre_capa *capa)
181 struct dt_object *next = dt_object_child(dt);
186 * apply changes to the local object
188 rc = dt_attr_set(env, next, attr, handle, capa);
195 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
196 struct lu_buf *buf, const char *name,
197 struct lustre_capa *capa)
199 return dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
203 * LOV xattr is a storage for striping, and LOD owns this xattr.
204 * but LOD allows others to control striping to some extent
206 * - to set new defined striping
207 * - to set new semi-defined striping
208 * - number of stripes is defined
209 * - number of stripes + osts are defined
212 static int lod_declare_xattr_set(const struct lu_env *env,
213 struct dt_object *dt,
214 const struct lu_buf *buf,
215 const char *name, int fl,
218 struct dt_object *next = dt_object_child(dt);
222 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
227 static int lod_xattr_set(const struct lu_env *env,
228 struct dt_object *dt, const struct lu_buf *buf,
229 const char *name, int fl, struct thandle *th,
230 struct lustre_capa *capa)
232 struct dt_object *next = dt_object_child(dt);
237 * behave transparantly for all other EAs
239 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
244 static int lod_declare_xattr_del(const struct lu_env *env,
245 struct dt_object *dt, const char *name,
248 return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
251 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
252 const char *name, struct thandle *th,
253 struct lustre_capa *capa)
255 return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
258 static int lod_xattr_list(const struct lu_env *env,
259 struct dt_object *dt, struct lu_buf *buf,
260 struct lustre_capa *capa)
262 return dt_xattr_list(env, dt_object_child(dt), buf, capa);
265 int lod_object_set_pool(struct lod_object *o, char *pool)
270 len = strlen(o->ldo_pool);
271 OBD_FREE(o->ldo_pool, len + 1);
276 OBD_ALLOC(o->ldo_pool, len + 1);
277 if (o->ldo_pool == NULL)
279 strcpy(o->ldo_pool, pool);
285 * used to transfer default striping data to the object being created
287 static void lod_ah_init(const struct lu_env *env,
288 struct dt_allocation_hint *ah,
289 struct dt_object *parent,
290 struct dt_object *child,
291 cfs_umode_t child_mode)
293 struct dt_object *nextc;
294 struct dt_object *nextp = NULL;
295 struct lod_object *lc;
301 nextp = dt_object_child(parent);
303 nextc = dt_object_child(child);
304 lc = lod_dt_obj(child);
306 LASSERT(lc->ldo_stripenr == 0);
307 LASSERT(lc->ldo_stripe == NULL);
310 * local object may want some hints
311 * in case of late striping creation, ->ah_init()
312 * can be called with local object existing
314 if (!dt_object_exists(nextc))
315 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
320 static int lod_declare_object_create(const struct lu_env *env,
321 struct dt_object *dt,
322 struct lu_attr *attr,
323 struct dt_allocation_hint *hint,
324 struct dt_object_format *dof,
327 struct dt_object *next = dt_object_child(dt);
334 LASSERT(!dt_object_exists(next));
337 * first of all, we declare creation of local object
339 rc = dt_declare_create(env, next, attr, hint, dof, th);
343 if (dof->dof_type == DFT_SYM)
344 dt->do_body_ops = &lod_body_lnk_ops;
350 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
351 struct lu_attr *attr,
352 struct dt_allocation_hint *hint,
353 struct dt_object_format *dof, struct thandle *th)
355 struct dt_object *next = dt_object_child(dt);
359 /* create local object */
360 rc = dt_create(env, next, attr, hint, dof, th);
365 static int lod_declare_object_destroy(const struct lu_env *env,
366 struct dt_object *dt,
369 struct dt_object *next = dt_object_child(dt);
374 * we declare destroy for the local object
376 rc = dt_declare_destroy(env, next, th);
383 static int lod_object_destroy(const struct lu_env *env,
384 struct dt_object *dt, struct thandle *th)
386 struct dt_object *next = dt_object_child(dt);
390 /* destroy local object */
391 rc = dt_destroy(env, next, th);
398 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
399 const struct dt_index_features *feat)
401 struct dt_object *next = dt_object_child(dt);
405 LASSERT(next->do_ops);
406 LASSERT(next->do_ops->do_index_try);
408 rc = next->do_ops->do_index_try(env, next, feat);
409 if (next->do_index_ops && dt->do_index_ops == NULL) {
410 dt->do_index_ops = &lod_index_ops;
411 /* XXX: iterators don't accept device, so bypass LOD */
412 /* will be fixed with DNE */
413 if (lod_index_ops.dio_it.fini == NULL) {
414 lod_index_ops.dio_it = next->do_index_ops->dio_it;
415 lod_index_ops.dio_it.init = lod_it_init;
422 static int lod_declare_ref_add(const struct lu_env *env,
423 struct dt_object *dt, struct thandle *th)
425 return dt_declare_ref_add(env, dt_object_child(dt), th);
428 static int lod_ref_add(const struct lu_env *env,
429 struct dt_object *dt, struct thandle *th)
431 return dt_ref_add(env, dt_object_child(dt), th);
434 static int lod_declare_ref_del(const struct lu_env *env,
435 struct dt_object *dt, struct thandle *th)
437 return dt_declare_ref_del(env, dt_object_child(dt), th);
440 static int lod_ref_del(const struct lu_env *env,
441 struct dt_object *dt, struct thandle *th)
443 return dt_ref_del(env, dt_object_child(dt), th);
446 static struct obd_capa *lod_capa_get(const struct lu_env *env,
447 struct dt_object *dt,
448 struct lustre_capa *old, __u64 opc)
450 return dt_capa_get(env, dt_object_child(dt), old, opc);
453 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
455 return dt_object_sync(env, dt_object_child(dt));
458 struct dt_object_operations lod_obj_ops = {
459 .do_read_lock = lod_object_read_lock,
460 .do_write_lock = lod_object_write_lock,
461 .do_read_unlock = lod_object_read_unlock,
462 .do_write_unlock = lod_object_write_unlock,
463 .do_write_locked = lod_object_write_locked,
464 .do_attr_get = lod_attr_get,
465 .do_declare_attr_set = lod_declare_attr_set,
466 .do_attr_set = lod_attr_set,
467 .do_xattr_get = lod_xattr_get,
468 .do_declare_xattr_set = lod_declare_xattr_set,
469 .do_xattr_set = lod_xattr_set,
470 .do_declare_xattr_del = lod_declare_xattr_del,
471 .do_xattr_del = lod_xattr_del,
472 .do_xattr_list = lod_xattr_list,
473 .do_ah_init = lod_ah_init,
474 .do_declare_create = lod_declare_object_create,
475 .do_create = lod_object_create,
476 .do_declare_destroy = lod_declare_object_destroy,
477 .do_destroy = lod_object_destroy,
478 .do_index_try = lod_index_try,
479 .do_declare_ref_add = lod_declare_ref_add,
480 .do_ref_add = lod_ref_add,
481 .do_declare_ref_del = lod_declare_ref_del,
482 .do_ref_del = lod_ref_del,
483 .do_capa_get = lod_capa_get,
484 .do_object_sync = lod_object_sync,
487 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
488 struct lu_buf *buf, loff_t *pos,
489 struct lustre_capa *capa)
491 struct dt_object *next = dt_object_child(dt);
492 return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
495 static ssize_t lod_declare_write(const struct lu_env *env,
496 struct dt_object *dt,
497 const loff_t size, loff_t pos,
500 return dt_declare_record_write(env, dt_object_child(dt),
504 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
505 const struct lu_buf *buf, loff_t *pos,
506 struct thandle *th, struct lustre_capa *capa, int iq)
508 struct dt_object *next = dt_object_child(dt);
510 return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
513 static const struct dt_body_operations lod_body_lnk_ops = {
514 .dbo_read = lod_read,
515 .dbo_declare_write = lod_declare_write,
516 .dbo_write = lod_write
519 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
520 const struct lu_object_conf *conf)
522 struct lod_device *d = lu2lod_dev(o->lo_dev);
523 struct lu_object *below;
524 struct lu_device *under;
528 * create local object
530 under = &d->lod_child->dd_lu_dev;
531 below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
535 lu_object_add(o, below);
540 void lod_object_free_striping(const struct lu_env *env, struct lod_object *o)
545 * ->start is called once all slices are initialized, including header's
546 * cache for mode (object type). using the type we can initialize ops
548 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
550 if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
551 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
555 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
557 struct lod_object *mo = lu2lod_obj(o);
560 * release all underlying object pinned
563 lod_object_free_striping(env, mo);
565 lod_object_set_pool(mo, NULL);
568 OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
571 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
573 /* XXX: shouldn't we release everything here in case if object
574 * creation failed before? */
577 static int lod_object_print(const struct lu_env *env, void *cookie,
578 lu_printer_t p, const struct lu_object *l)
580 struct lod_object *o = lu2lod_obj((struct lu_object *) l);
582 return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
585 struct lu_object_operations lod_lu_obj_ops = {
586 .loo_object_init = lod_object_init,
587 .loo_object_start = lod_object_start,
588 .loo_object_free = lod_object_free,
589 .loo_object_release = lod_object_release,
590 .loo_object_print = lod_object_print,