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, but
11 * WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * General Public License version 2 for more details (a copy is included
14 * in the LICENSE file that accompanied this code).
16 * You should have received a copy of the GNU General Public License
17 * version 2 along with this program; If not, see
18 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
20 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21 * CA 95054 USA or visit www.sun.com if you need additional information or
27 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28 * Use is subject to license terms.
30 * Copyright (c) 2012, 2013, Intel Corporation.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lustre/osp/osp_object.c
38 * Lustre OST Proxy Device
40 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
41 * Author: Mikhail Pershin <mike.tappro@intel.com>
44 #define DEBUG_SUBSYSTEM S_MDS
46 #include "osp_internal.h"
48 static inline bool is_ost_obj(struct lu_object *lo)
50 return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
53 static void osp_object_assign_fid(const struct lu_env *env,
54 struct osp_device *d, struct osp_object *o)
56 struct osp_thread_info *osi = osp_env_info(env);
58 LASSERT(fid_is_zero(lu_object_fid(&o->opo_obj.do_lu)));
59 LASSERT(o->opo_reserved);
62 osp_precreate_get_fid(env, d, &osi->osi_fid);
64 lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
67 static int osp_oac_init(struct osp_object *obj)
69 struct osp_object_attr *ooa;
75 INIT_LIST_HEAD(&ooa->ooa_xattr_list);
76 spin_lock(&obj->opo_lock);
77 if (likely(obj->opo_ooa == NULL)) {
79 spin_unlock(&obj->opo_lock);
81 spin_unlock(&obj->opo_lock);
88 static struct osp_xattr_entry *
89 osp_oac_xattr_find_locked(struct osp_object_attr *ooa,
90 const char *name, int namelen, bool unlink)
92 struct osp_xattr_entry *oxe;
94 list_for_each_entry(oxe, &ooa->ooa_xattr_list, oxe_list) {
95 if (namelen == oxe->oxe_namelen &&
96 strncmp(name, oxe->oxe_buf, namelen) == 0) {
98 list_del_init(&oxe->oxe_list);
100 atomic_inc(&oxe->oxe_ref);
109 static struct osp_xattr_entry *osp_oac_xattr_find(struct osp_object *obj,
112 struct osp_xattr_entry *oxe = NULL;
114 spin_lock(&obj->opo_lock);
115 if (obj->opo_ooa != NULL)
116 oxe = osp_oac_xattr_find_locked(obj->opo_ooa, name,
117 strlen(name), false);
118 spin_unlock(&obj->opo_lock);
123 static struct osp_xattr_entry *
124 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, int len)
126 struct osp_object_attr *ooa = obj->opo_ooa;
127 struct osp_xattr_entry *oxe;
128 struct osp_xattr_entry *tmp = NULL;
129 int namelen = strlen(name);
130 int size = sizeof(*oxe) + namelen + 1 + len;
132 LASSERT(ooa != NULL);
134 oxe = osp_oac_xattr_find(obj, name);
138 OBD_ALLOC(oxe, size);
139 if (unlikely(oxe == NULL))
142 INIT_LIST_HEAD(&oxe->oxe_list);
143 oxe->oxe_buflen = size;
144 oxe->oxe_namelen = namelen;
145 memcpy(oxe->oxe_buf, name, namelen);
146 oxe->oxe_value = oxe->oxe_buf + namelen + 1;
147 /* One ref is for the caller, the other is for the entry on the list. */
148 atomic_set(&oxe->oxe_ref, 2);
150 spin_lock(&obj->opo_lock);
151 tmp = osp_oac_xattr_find_locked(ooa, name, namelen, false);
153 list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
154 spin_unlock(&obj->opo_lock);
164 static struct osp_xattr_entry *
165 osp_oac_xattr_replace(struct osp_object *obj,
166 struct osp_xattr_entry **poxe, int len)
168 struct osp_object_attr *ooa = obj->opo_ooa;
169 struct osp_xattr_entry *old = *poxe;
170 struct osp_xattr_entry *oxe;
171 struct osp_xattr_entry *tmp = NULL;
172 int namelen = old->oxe_namelen;
173 int size = sizeof(*oxe) + namelen + 1 + len;
175 LASSERT(ooa != NULL);
177 OBD_ALLOC(oxe, size);
178 if (unlikely(oxe == NULL))
181 INIT_LIST_HEAD(&oxe->oxe_list);
182 oxe->oxe_buflen = size;
183 oxe->oxe_namelen = namelen;
184 memcpy(oxe->oxe_buf, old->oxe_buf, namelen);
185 oxe->oxe_value = oxe->oxe_buf + namelen + 1;
186 /* One ref is for the caller, the other is for the entry on the list. */
187 atomic_set(&oxe->oxe_ref, 2);
189 spin_lock(&obj->opo_lock);
190 tmp = osp_oac_xattr_find_locked(ooa, oxe->oxe_buf, namelen, true);
191 list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
192 spin_unlock(&obj->opo_lock);
195 LASSERT(tmp != NULL);
200 static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
202 if (atomic_dec_and_test(&oxe->oxe_ref)) {
203 LASSERT(list_empty(&oxe->oxe_list));
205 OBD_FREE(oxe, oxe->oxe_buflen);
209 static int osp_get_attr_from_reply(const struct lu_env *env,
210 struct object_update_reply *reply,
211 struct ptlrpc_request *req,
212 struct lu_attr *attr,
213 struct osp_object *obj, int index)
215 struct osp_thread_info *osi = osp_env_info(env);
216 struct lu_buf *rbuf = &osi->osi_lb2;
217 struct obdo *lobdo = &osi->osi_obdo;
221 rc = object_update_result_data_get(reply, rbuf, index);
225 wobdo = rbuf->lb_buf;
226 if (rbuf->lb_len != sizeof(*wobdo))
229 LASSERT(req != NULL);
230 if (ptlrpc_req_need_swab(req))
231 lustre_swab_obdo(wobdo);
233 lustre_get_wire_obdo(NULL, lobdo, wobdo);
234 spin_lock(&obj->opo_lock);
235 if (obj->opo_ooa != NULL) {
236 la_from_obdo(&obj->opo_ooa->ooa_attr, lobdo, lobdo->o_valid);
238 *attr = obj->opo_ooa->ooa_attr;
240 LASSERT(attr != NULL);
242 la_from_obdo(attr, lobdo, lobdo->o_valid);
244 spin_unlock(&obj->opo_lock);
249 static int osp_attr_get_interpterer(const struct lu_env *env,
250 struct object_update_reply *reply,
251 struct ptlrpc_request *req,
252 struct osp_object *obj,
253 void *data, int index, int rc)
255 struct lu_attr *attr = data;
257 LASSERT(obj->opo_ooa != NULL);
260 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
261 obj->opo_non_exist = 0;
263 return osp_get_attr_from_reply(env, reply, req, NULL, obj,
267 osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
268 obj->opo_non_exist = 1;
271 spin_lock(&obj->opo_lock);
273 spin_unlock(&obj->opo_lock);
279 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
280 struct lustre_capa *capa)
282 struct osp_object *obj = dt2osp_obj(dt);
283 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
284 struct dt_update_request *update;
287 if (obj->opo_ooa == NULL) {
288 rc = osp_oac_init(obj);
293 mutex_lock(&osp->opd_async_requests_mutex);
294 update = osp_find_or_create_async_update_request(osp);
296 rc = PTR_ERR(update);
298 rc = osp_insert_async_update(env, update, OUT_ATTR_GET, obj, 0,
300 &obj->opo_ooa->ooa_attr,
301 osp_attr_get_interpterer);
302 mutex_unlock(&osp->opd_async_requests_mutex);
307 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
308 struct lu_attr *attr, struct lustre_capa *capa)
310 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
311 struct osp_object *obj = dt2osp_obj(dt);
312 struct dt_device *dev = &osp->opd_dt_dev;
313 struct dt_update_request *update;
314 struct object_update_reply *reply;
315 struct ptlrpc_request *req = NULL;
319 if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
322 if (obj->opo_ooa != NULL) {
323 spin_lock(&obj->opo_lock);
324 if (obj->opo_ooa->ooa_attr.la_valid != 0) {
325 *attr = obj->opo_ooa->ooa_attr;
326 spin_unlock(&obj->opo_lock);
330 spin_unlock(&obj->opo_lock);
333 update = out_create_update_req(dev);
335 RETURN(PTR_ERR(update));
337 rc = out_insert_update(env, update, OUT_ATTR_GET,
338 lu_object_fid(&dt->do_lu), 0, NULL, NULL);
340 CERROR("%s: Insert update error "DFID": rc = %d\n",
341 dev->dd_lu_dev.ld_obd->obd_name,
342 PFID(lu_object_fid(&dt->do_lu)), rc);
347 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
350 osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
351 obj->opo_non_exist = 1;
353 CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
354 dev->dd_lu_dev.ld_obd->obd_name,
355 PFID(lu_object_fid(&dt->do_lu)), rc);
361 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
362 obj->opo_non_exist = 0;
363 reply = req_capsule_server_sized_get(&req->rq_pill,
364 &RMF_OUT_UPDATE_REPLY,
365 OUT_UPDATE_REPLY_SIZE);
366 if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
367 GOTO(out, rc = -EPROTO);
369 rc = osp_get_attr_from_reply(env, reply, req, attr, obj, 0);
377 ptlrpc_req_finished(req);
379 out_destroy_update_req(update);
384 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
385 const struct lu_attr *attr, struct thandle *th)
387 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
388 struct osp_object *o = dt2osp_obj(dt);
395 * Usually we don't allow server stack to manipulate size
396 * but there is a special case when striping is created
397 * late, after stripless file got truncated to non-zero.
399 * In this case we do the following:
401 * 1) grab id in declare - this can lead to leaked OST objects
402 * but we don't currently have proper mechanism and the only
403 * options we have are to do truncate RPC holding transaction
404 * open (very bad) or to grab id in declare at cost of leaked
405 * OST object in same very rare unfortunate case (just bad)
406 * notice 1.6-2.0 do assignment outside of running transaction
407 * all the time, meaning many more chances for leaked objects.
409 * 2) send synchronous truncate RPC with just assigned id
412 /* there are few places in MDD code still passing NULL
413 * XXX: to be fixed soon */
417 if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
418 fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
419 LASSERT(!dt_object_exists(dt));
420 osp_object_assign_fid(env, d, o);
421 rc = osp_object_truncate(env, dt, attr->la_size);
427 /* no need in logging for new objects being created */
431 if (!(attr->la_valid & (LA_UID | LA_GID)))
434 if (!is_only_remote_trans(th))
436 * track all UID/GID changes via llog
438 rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
440 /* It is for OST-object attr_set directly without updating
441 * local MDT-object attribute. It is usually used by LFSCK. */
442 rc = osp_md_declare_attr_set(env, dt, attr, th);
444 if (rc != 0 || o->opo_ooa == NULL)
447 la = &o->opo_ooa->ooa_attr;
448 spin_lock(&o->opo_lock);
449 if (attr->la_valid & LA_UID) {
450 la->la_uid = attr->la_uid;
451 la->la_valid |= LA_UID;
454 if (attr->la_valid & LA_GID) {
455 la->la_gid = attr->la_gid;
456 la->la_valid |= LA_GID;
458 spin_unlock(&o->opo_lock);
463 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
464 const struct lu_attr *attr, struct thandle *th,
465 struct lustre_capa *capa)
467 struct osp_object *o = dt2osp_obj(dt);
472 /* we're interested in uid/gid changes only */
473 if (!(attr->la_valid & (LA_UID | LA_GID)))
476 /* new object, the very first ->attr_set()
477 * initializing attributes needs no logging
478 * all subsequent one are subject to the
479 * logging and synchronization with OST */
485 if (!is_only_remote_trans(th))
487 * once transaction is committed put proper command on
488 * the queue going to our OST
490 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
491 /* XXX: send new uid/gid to OST ASAP? */
493 /* It is for OST-object attr_set directly without updating
494 * local MDT-object attribute. It is usually used by LFSCK. */
495 rc = osp_md_attr_set(env, dt, attr, th, capa);
500 static int osp_xattr_get_interpterer(const struct lu_env *env,
501 struct object_update_reply *reply,
502 struct ptlrpc_request *req,
503 struct osp_object *obj,
504 void *data, int index, int rc)
506 struct osp_object_attr *ooa = obj->opo_ooa;
507 struct osp_xattr_entry *oxe = data;
508 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
510 LASSERT(ooa != NULL);
513 int len = sizeof(*oxe) + oxe->oxe_namelen + 1;
515 rc = object_update_result_data_get(reply, rbuf, index);
516 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
517 spin_lock(&obj->opo_lock);
519 spin_unlock(&obj->opo_lock);
520 osp_oac_xattr_put(oxe);
522 return rc < 0 ? rc : -ERANGE;
525 spin_lock(&obj->opo_lock);
526 oxe->oxe_vallen = rbuf->lb_len;
527 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
530 spin_unlock(&obj->opo_lock);
531 } else if (rc == -ENOENT || rc == -ENODATA) {
532 spin_lock(&obj->opo_lock);
535 spin_unlock(&obj->opo_lock);
537 spin_lock(&obj->opo_lock);
539 spin_unlock(&obj->opo_lock);
542 osp_oac_xattr_put(oxe);
547 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
548 struct lu_buf *buf, const char *name,
549 struct lustre_capa *capa)
551 struct osp_object *obj = dt2osp_obj(dt);
552 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
553 struct dt_update_request *update;
554 struct osp_xattr_entry *oxe;
555 int namelen = strlen(name);
558 LASSERT(buf != NULL);
559 LASSERT(name != NULL);
561 /* If only for xattr size, return directly. */
562 if (unlikely(buf->lb_len == 0))
565 if (obj->opo_ooa == NULL) {
566 rc = osp_oac_init(obj);
571 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
575 mutex_lock(&osp->opd_async_requests_mutex);
576 update = osp_find_or_create_async_update_request(osp);
577 if (IS_ERR(update)) {
578 rc = PTR_ERR(update);
579 mutex_unlock(&osp->opd_async_requests_mutex);
580 osp_oac_xattr_put(oxe);
582 rc = osp_insert_async_update(env, update, OUT_XATTR_GET, obj,
583 1, &namelen, &name, oxe,
584 osp_xattr_get_interpterer);
586 mutex_unlock(&osp->opd_async_requests_mutex);
587 osp_oac_xattr_put(oxe);
589 /* XXX: Currently, we trigger the batched async OUT
590 * RPC via dt_declare_xattr_get(). It is not
591 * perfect solution, but works well now.
593 * We will improve it in the future. */
594 update = osp->opd_async_requests;
595 if (update != NULL && update->dur_req != NULL &&
596 update->dur_req->ourq_count > 0) {
597 osp->opd_async_requests = NULL;
598 mutex_unlock(&osp->opd_async_requests_mutex);
599 rc = osp_unplug_async_update(env, osp, update);
601 mutex_unlock(&osp->opd_async_requests_mutex);
609 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
610 struct lu_buf *buf, const char *name,
611 struct lustre_capa *capa)
613 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
614 struct osp_object *obj = dt2osp_obj(dt);
615 struct dt_device *dev = &osp->opd_dt_dev;
616 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
617 struct dt_update_request *update = NULL;
618 struct ptlrpc_request *req = NULL;
619 struct object_update_reply *reply;
620 struct osp_xattr_entry *oxe = NULL;
621 const char *dname = dt->do_lu.lo_dev->ld_obd->obd_name;
626 LASSERT(buf != NULL);
627 LASSERT(name != NULL);
629 if (unlikely(obj->opo_non_exist))
632 oxe = osp_oac_xattr_find(obj, name);
634 spin_lock(&obj->opo_lock);
635 if (oxe->oxe_ready) {
637 GOTO(unlock, rc = -ENODATA);
639 if (buf->lb_buf == NULL)
640 GOTO(unlock, rc = oxe->oxe_vallen);
642 if (buf->lb_len < oxe->oxe_vallen)
643 GOTO(unlock, rc = -ERANGE);
645 memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
647 GOTO(unlock, rc = oxe->oxe_vallen);
650 spin_unlock(&obj->opo_lock);
651 osp_oac_xattr_put(oxe);
655 spin_unlock(&obj->opo_lock);
658 update = out_create_update_req(dev);
660 GOTO(out, rc = PTR_ERR(update));
662 namelen = strlen(name) + 1;
663 rc = out_insert_update(env, update, OUT_XATTR_GET,
664 lu_object_fid(&dt->do_lu), 1, &namelen, &name);
666 CERROR("%s: Insert update error "DFID": rc = %d\n",
667 dname, PFID(lu_object_fid(&dt->do_lu)), rc);
672 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
674 if (obj->opo_ooa == NULL)
678 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
681 CWARN("%s: Fail to add xattr (%s) to cache for "
682 DFID" (1): rc = %d\n", dname, name,
683 PFID(lu_object_fid(&dt->do_lu)), rc);
688 spin_lock(&obj->opo_lock);
689 if (rc == -ENOENT || rc == -ENODATA) {
695 spin_unlock(&obj->opo_lock);
700 reply = req_capsule_server_sized_get(&req->rq_pill,
701 &RMF_OUT_UPDATE_REPLY,
702 OUT_UPDATE_REPLY_SIZE);
703 if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
704 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
705 dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
706 PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
708 GOTO(out, rc = -EPROTO);
711 rc = object_update_result_data_get(reply, rbuf, 0);
715 if (buf->lb_buf == NULL)
716 GOTO(out, rc = rbuf->lb_len);
718 if (unlikely(buf->lb_len < rbuf->lb_len))
719 GOTO(out, rc = -ERANGE);
721 memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
723 if (obj->opo_ooa == NULL)
727 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
729 CWARN("%s: Fail to add xattr (%s) to "
730 "cache for "DFID" (2): rc = %d\n",
731 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
737 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
738 struct osp_xattr_entry *old = oxe;
739 struct osp_xattr_entry *tmp;
741 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
742 osp_oac_xattr_put(oxe);
745 CWARN("%s: Fail to update xattr (%s) to "
746 "cache for "DFID": rc = %d\n",
747 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
748 spin_lock(&obj->opo_lock);
750 spin_unlock(&obj->opo_lock);
755 /* Drop the ref for entry on list. */
756 osp_oac_xattr_put(old);
759 spin_lock(&obj->opo_lock);
760 oxe->oxe_vallen = rbuf->lb_len;
761 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
764 spin_unlock(&obj->opo_lock);
770 ptlrpc_req_finished(req);
772 if (update != NULL && !IS_ERR(update))
773 out_destroy_update_req(update);
776 osp_oac_xattr_put(oxe);
781 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
782 const struct lu_buf *buf, const char *name,
783 int flag, struct thandle *th)
785 struct osp_object *o = dt2osp_obj(dt);
786 struct dt_update_request *update;
788 struct osp_xattr_entry *oxe;
789 int sizes[3] = {strlen(name), buf->lb_len,
791 char *bufs[3] = {(char *)name, (char *)buf->lb_buf };
794 LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
796 update = out_find_create_update_loc(th, dt);
797 if (IS_ERR(update)) {
798 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
799 dt->do_lu.lo_dev->ld_obd->obd_name,
800 PFID(lu_object_fid(&dt->do_lu)),
801 (int)PTR_ERR(update));
803 return PTR_ERR(update);
806 flag = cpu_to_le32(flag);
807 bufs[2] = (char *)&flag;
809 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
810 rc = out_insert_update(env, update, OUT_XATTR_SET, fid,
811 ARRAY_SIZE(sizes), sizes, (const char **)bufs);
812 if (rc != 0 || o->opo_ooa == NULL)
815 oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
817 CWARN("%s: Fail to add xattr (%s) to cache for "DFID
818 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
819 name, PFID(lu_object_fid(&dt->do_lu)), rc);
824 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
825 struct osp_xattr_entry *old = oxe;
826 struct osp_xattr_entry *tmp;
828 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
829 osp_oac_xattr_put(oxe);
832 CWARN("%s: Fail to update xattr (%s) to cache for "DFID
833 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
834 name, PFID(lu_object_fid(&dt->do_lu)), rc);
835 spin_lock(&o->opo_lock);
837 spin_unlock(&o->opo_lock);
842 /* Drop the ref for entry on list. */
843 osp_oac_xattr_put(old);
846 spin_lock(&o->opo_lock);
847 oxe->oxe_vallen = buf->lb_len;
848 memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
851 spin_unlock(&o->opo_lock);
852 osp_oac_xattr_put(oxe);
857 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
858 const struct lu_buf *buf, const char *name, int fl,
859 struct thandle *th, struct lustre_capa *capa)
861 CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
862 PFID(&dt->do_lu.lo_header->loh_fid));
867 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
868 const char *name, struct thandle *th)
870 struct dt_update_request *update;
871 const struct lu_fid *fid;
872 int size = strlen(name);
875 update = out_find_create_update_loc(th, dt);
877 return PTR_ERR(update);
879 fid = lu_object_fid(&dt->do_lu);
881 rc = out_insert_update(env, update, OUT_XATTR_DEL, fid, 1, &size,
882 (const char **)&name);
887 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
888 const char *name, struct thandle *th,
889 struct lustre_capa *capa)
891 CDEBUG(D_INFO, "xattr %s del object "DFID"\n", name,
892 PFID(&dt->do_lu.lo_header->loh_fid));
897 static int osp_declare_object_create(const struct lu_env *env,
898 struct dt_object *dt,
899 struct lu_attr *attr,
900 struct dt_allocation_hint *hint,
901 struct dt_object_format *dof,
904 struct osp_thread_info *osi = osp_env_info(env);
905 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
906 struct osp_object *o = dt2osp_obj(dt);
907 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
912 if (is_only_remote_trans(th)) {
913 LASSERT(fid_is_sane(fid));
915 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
920 /* should happen to non-0 OSP only so that at least one object
921 * has been already declared in the scenario and LOD should
923 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
926 LASSERT(d->opd_last_used_oid_file);
929 * There can be gaps in precreated ids and record to unlink llog
930 * XXX: we do not handle gaps yet, implemented before solution
931 * was found to be racy, so we disabled that. there is no
932 * point in making useless but expensive llog declaration.
934 /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
936 if (unlikely(!fid_is_zero(fid))) {
937 /* replay case: caller knows fid */
938 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
939 osi->osi_lb.lb_len = sizeof(osi->osi_id);
940 osi->osi_lb.lb_buf = NULL;
941 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
942 &osi->osi_lb, osi->osi_off, th);
947 * in declaration we need to reserve object so that we don't block
948 * awaiting precreation RPC to complete
950 rc = osp_precreate_reserve(env, d);
952 * we also need to declare update to local "last used id" file for
953 * recovery if object isn't used for a reason, we need to release
954 * reservation, this can be made in osd_object_release()
957 /* mark id is reserved: in create we don't want to talk
959 LASSERT(o->opo_reserved == 0);
962 /* common for all OSPs file hystorically */
963 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
964 osi->osi_lb.lb_len = sizeof(osi->osi_id);
965 osi->osi_lb.lb_buf = NULL;
966 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
967 &osi->osi_lb, osi->osi_off, th);
969 /* not needed in the cache anymore */
970 set_bit(LU_OBJECT_HEARD_BANSHEE,
971 &dt->do_lu.lo_header->loh_flags);
976 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
977 struct lu_attr *attr,
978 struct dt_allocation_hint *hint,
979 struct dt_object_format *dof, struct thandle *th)
981 struct osp_thread_info *osi = osp_env_info(env);
982 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
983 struct osp_object *o = dt2osp_obj(dt);
985 struct lu_fid *fid = &osi->osi_fid;
988 if (is_only_remote_trans(th)) {
989 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
991 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
993 o->opo_non_exist = 0;
998 o->opo_non_exist = 0;
999 if (o->opo_reserved) {
1000 /* regular case, fid is assigned holding trunsaction open */
1001 osp_object_assign_fid(env, d, o);
1004 memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
1006 LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
1009 if (!o->opo_reserved) {
1010 /* special case, id was assigned outside of transaction
1011 * see comments in osp_declare_attr_set */
1012 LASSERT(d->opd_pre != NULL);
1013 spin_lock(&d->opd_pre_lock);
1014 osp_update_last_fid(d, fid);
1015 spin_unlock(&d->opd_pre_lock);
1018 CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
1020 /* If the precreate ends, it means it will be ready to rollover to
1021 * the new sequence soon, all the creation should be synchronized,
1022 * otherwise during replay, the replay fid will be inconsistent with
1023 * last_used/create fid */
1024 if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1028 * it's OK if the import is inactive by this moment - id was created
1029 * by OST earlier, we just need to maintain it consistently on the disk
1030 * once import is reconnected, OSP will claim this and other objects
1031 * used and OST either keep them, if they exist or recreate
1034 /* we might have lost precreated objects */
1035 if (unlikely(d->opd_gap_count) > 0) {
1036 LASSERT(d->opd_pre != NULL);
1037 spin_lock(&d->opd_pre_lock);
1038 if (d->opd_gap_count > 0) {
1039 int count = d->opd_gap_count;
1041 ostid_set_id(&osi->osi_oi,
1042 fid_oid(&d->opd_gap_start_fid));
1043 d->opd_gap_count = 0;
1044 spin_unlock(&d->opd_pre_lock);
1046 CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1047 PFID(&d->opd_gap_start_fid), count);
1048 /* real gap handling is disabled intil ORI-692 will be
1049 * fixed, now we only report gaps */
1051 spin_unlock(&d->opd_pre_lock);
1055 /* new object, the very first ->attr_set()
1056 * initializing attributes needs no logging */
1059 /* Only need update last_used oid file, seq file will only be update
1060 * during seq rollover */
1061 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1062 &d->opd_last_used_fid.f_oid, d->opd_index);
1064 rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1067 CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1068 d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1073 int osp_declare_object_destroy(const struct lu_env *env,
1074 struct dt_object *dt, struct thandle *th)
1076 struct osp_object *o = dt2osp_obj(dt);
1082 * track objects to be destroyed via llog
1084 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1089 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1092 struct osp_object *o = dt2osp_obj(dt);
1097 o->opo_non_exist = 1;
1099 * once transaction is committed put proper command on
1100 * the queue going to our OST
1102 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1104 /* not needed in cache any more */
1105 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1110 static int osp_orphan_index_lookup(const struct lu_env *env,
1111 struct dt_object *dt,
1113 const struct dt_key *key,
1114 struct lustre_capa *capa)
1119 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1120 struct dt_object *dt,
1121 const struct dt_rec *rec,
1122 const struct dt_key *key,
1123 struct thandle *handle)
1128 static int osp_orphan_index_insert(const struct lu_env *env,
1129 struct dt_object *dt,
1130 const struct dt_rec *rec,
1131 const struct dt_key *key,
1132 struct thandle *handle,
1133 struct lustre_capa *capa,
1139 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1140 struct dt_object *dt,
1141 const struct dt_key *key,
1142 struct thandle *handle)
1147 static int osp_orphan_index_delete(const struct lu_env *env,
1148 struct dt_object *dt,
1149 const struct dt_key *key,
1150 struct thandle *handle,
1151 struct lustre_capa *capa)
1156 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
1157 __u32 attr, struct lustre_capa *capa)
1163 return ERR_PTR(-ENOMEM);
1165 it->ooi_pos_ent = -1;
1168 return (struct dt_it *)it;
1171 void osp_it_fini(const struct lu_env *env, struct dt_it *di)
1173 struct osp_it *it = (struct osp_it *)di;
1174 struct page **pages = it->ooi_pages;
1175 int npages = it->ooi_total_npages;
1178 if (pages != NULL) {
1179 for (i = 0; i < npages; i++) {
1180 if (pages[i] != NULL) {
1181 if (pages[i] == it->ooi_cur_page) {
1183 it->ooi_cur_page = NULL;
1185 __free_page(pages[i]);
1188 OBD_FREE(pages, npages * sizeof(*pages));
1193 static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
1195 struct lu_device *dev = it->ooi_obj->do_lu.lo_dev;
1196 struct osp_device *osp = lu2osp_dev(dev);
1197 struct page **pages;
1198 struct ptlrpc_request *req = NULL;
1199 struct ptlrpc_bulk_desc *desc;
1200 struct idx_info *ii;
1207 npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1208 npages /= PAGE_CACHE_SIZE;
1210 OBD_ALLOC(pages, npages * sizeof(*pages));
1214 it->ooi_pages = pages;
1215 it->ooi_total_npages = npages;
1216 for (i = 0; i < npages; i++) {
1217 pages[i] = alloc_page(GFP_IOFS);
1218 if (pages[i] == NULL)
1222 req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1227 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1229 ptlrpc_request_free(req);
1233 req->rq_request_portal = OUT_PORTAL;
1234 ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1235 memset(ii, 0, sizeof(*ii));
1236 if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
1237 /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
1238 * ost_index, 0] with LAST_ID FID, so it needs to replace
1239 * the FID with orphan FID here */
1240 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1241 ii->ii_fid.f_oid = osp->opd_index;
1242 ii->ii_fid.f_ver = 0;
1243 ii->ii_flags = II_FL_NOHASH;
1245 ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
1246 ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
1249 ii->ii_magic = IDX_INFO_MAGIC;
1250 ii->ii_count = npages * LU_PAGE_COUNT;
1251 ii->ii_hash_start = it->ooi_next;
1253 osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
1255 ptlrpc_at_set_req_timeout(req);
1257 desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
1260 ptlrpc_request_free(req);
1264 for (i = 0; i < npages; i++)
1265 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1267 ptlrpc_request_set_replen(req);
1268 rc = ptlrpc_queue_wait(req);
1272 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1273 req->rq_bulk->bd_nob_transferred);
1278 ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1279 if (ii->ii_magic != IDX_INFO_MAGIC)
1280 GOTO(out, rc = -EPROTO);
1282 npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1283 (PAGE_CACHE_SHIFT - LU_PAGE_SHIFT);
1284 if (npages > it->ooi_total_npages) {
1285 CERROR("%s: returned more pages than expected, %u > %u\n",
1286 osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1287 GOTO(out, rc = -EINVAL);
1290 it->ooi_valid_npages = npages;
1291 if (ptlrpc_rep_need_swab(req))
1294 it->ooi_next = ii->ii_hash_end;
1297 ptlrpc_req_finished(req);
1302 int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
1304 struct osp_it *it = (struct osp_it *)di;
1305 struct lu_idxpage *idxpage;
1306 struct page **pages;
1312 idxpage = it->ooi_cur_idxpage;
1313 if (idxpage != NULL) {
1314 if (idxpage->lip_nr == 0)
1317 if (it->ooi_pos_ent < idxpage->lip_nr) {
1318 CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
1319 (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
1322 it->ooi_cur_idxpage = NULL;
1323 it->ooi_pos_lu_page++;
1325 if (it->ooi_pos_lu_page < LU_PAGE_COUNT) {
1326 it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1327 LU_PAGE_SIZE * it->ooi_pos_lu_page;
1329 lustre_swab_lip_header(it->ooi_cur_idxpage);
1330 if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1331 struct osp_device *osp =
1332 lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1334 CERROR("%s: invalid magic (%x != %x) for page "
1335 "%d/%d while read layout orphan index\n",
1336 osp->opd_obd->obd_name,
1337 it->ooi_cur_idxpage->lip_magic,
1338 LIP_MAGIC, it->ooi_pos_page,
1339 it->ooi_pos_lu_page);
1340 /* Skip this lu_page next time. */
1341 it->ooi_pos_ent = idxpage->lip_nr - 1;
1344 it->ooi_pos_ent = -1;
1348 kunmap(it->ooi_cur_page);
1349 it->ooi_cur_page = NULL;
1353 pages = it->ooi_pages;
1354 if (it->ooi_pos_page < it->ooi_valid_npages) {
1355 it->ooi_cur_page = kmap(pages[it->ooi_pos_page]);
1356 it->ooi_pos_lu_page = 0;
1360 for (i = 0; i < it->ooi_total_npages; i++) {
1361 if (pages[i] != NULL)
1362 __free_page(pages[i]);
1364 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1366 it->ooi_pos_page = 0;
1367 it->ooi_total_npages = 0;
1368 it->ooi_valid_npages = 0;
1371 it->ooi_cur_page = NULL;
1372 it->ooi_cur_idxpage = NULL;
1373 it->ooi_pages = NULL;
1376 if (it->ooi_next == II_END_OFF)
1379 rc = osp_it_fetch(env, it);
1386 int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
1388 struct osp_it *it = (struct osp_it *)di;
1389 struct lu_idxpage *idxpage;
1394 idxpage = it->ooi_cur_idxpage;
1395 if (idxpage != NULL) {
1396 if (idxpage->lip_nr == 0)
1400 if (it->ooi_pos_ent < idxpage->lip_nr) {
1402 (struct lu_orphan_ent *)idxpage->lip_entries +
1405 lustre_swab_orphan_ent(it->ooi_ent);
1410 rc = osp_it_next_page(env, di);
1417 int osp_it_get(const struct lu_env *env, struct dt_it *di,
1418 const struct dt_key *key)
1423 void osp_it_put(const struct lu_env *env, struct dt_it *di)
1427 struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1428 const struct dt_it *di)
1430 struct osp_it *it = (struct osp_it *)di;
1431 struct lu_orphan_ent *ent = (struct lu_orphan_ent *)it->ooi_ent;
1433 if (likely(ent != NULL))
1434 return (struct dt_key *)(&ent->loe_key);
1439 int osp_orphan_it_key_size(const struct lu_env *env, const struct dt_it *di)
1441 return sizeof(struct lu_fid);
1444 int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
1445 struct dt_rec *rec, __u32 attr)
1447 struct osp_it *it = (struct osp_it *)di;
1448 struct lu_orphan_ent *ent = (struct lu_orphan_ent *)it->ooi_ent;
1450 if (likely(ent != NULL)) {
1451 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1458 __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
1460 struct osp_it *it = (struct osp_it *)di;
1462 return it->ooi_next;
1466 * \retval +1: locate to the exactly position
1467 * \retval 0: cannot locate to the exactly position,
1468 * call next() to move to a valid position.
1469 * \retval -ve: on error
1471 int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
1474 struct osp_it *it = (struct osp_it *)di;
1477 it->ooi_next = hash;
1478 rc = osp_orphan_it_next(env, (struct dt_it *)di);
1488 int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
1494 static const struct dt_index_operations osp_orphan_index_ops = {
1495 .dio_lookup = osp_orphan_index_lookup,
1496 .dio_declare_insert = osp_orphan_index_declare_insert,
1497 .dio_insert = osp_orphan_index_insert,
1498 .dio_declare_delete = osp_orphan_index_declare_delete,
1499 .dio_delete = osp_orphan_index_delete,
1501 .init = osp_it_init,
1502 .fini = osp_it_fini,
1503 .next = osp_orphan_it_next,
1506 .key = osp_orphan_it_key,
1507 .key_size = osp_orphan_it_key_size,
1508 .rec = osp_orphan_it_rec,
1509 .store = osp_it_store,
1510 .load = osp_orphan_it_load,
1511 .key_rec = osp_it_key_rec,
1515 static int osp_index_try(const struct lu_env *env,
1516 struct dt_object *dt,
1517 const struct dt_index_features *feat)
1519 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1521 if (fid_is_last_id(fid) && fid_is_idif(fid))
1522 dt->do_index_ops = &osp_orphan_index_ops;
1524 dt->do_index_ops = &osp_md_index_ops;
1528 struct dt_object_operations osp_obj_ops = {
1529 .do_declare_attr_get = osp_declare_attr_get,
1530 .do_attr_get = osp_attr_get,
1531 .do_declare_attr_set = osp_declare_attr_set,
1532 .do_attr_set = osp_attr_set,
1533 .do_declare_xattr_get = osp_declare_xattr_get,
1534 .do_xattr_get = osp_xattr_get,
1535 .do_declare_xattr_set = osp_declare_xattr_set,
1536 .do_xattr_set = osp_xattr_set,
1537 .do_declare_create = osp_declare_object_create,
1538 .do_create = osp_object_create,
1539 .do_declare_destroy = osp_declare_object_destroy,
1540 .do_destroy = osp_object_destroy,
1541 .do_index_try = osp_index_try,
1544 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1545 const struct lu_object_conf *conf)
1547 struct osp_object *po = lu2osp_obj(o);
1551 spin_lock_init(&po->opo_lock);
1552 o->lo_header->loh_attr |= LOHA_REMOTE;
1554 if (is_ost_obj(o)) {
1555 po->opo_obj.do_ops = &osp_obj_ops;
1557 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1559 po->opo_obj.do_ops = &osp_md_obj_ops;
1560 po->opo_obj.do_body_ops = &osp_md_body_ops;
1561 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1564 o->lo_header->loh_attr |=
1565 LOHA_EXISTS | (la->la_mode & S_IFMT);
1566 if (rc == -ENOENT) {
1567 po->opo_non_exist = 1;
1570 init_rwsem(&po->opo_sem);
1575 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1577 struct osp_object *obj = lu2osp_obj(o);
1578 struct lu_object_header *h = o->lo_header;
1580 dt_object_fini(&obj->opo_obj);
1581 lu_object_header_fini(h);
1582 if (obj->opo_ooa != NULL) {
1583 struct osp_xattr_entry *oxe;
1584 struct osp_xattr_entry *tmp;
1587 list_for_each_entry_safe(oxe, tmp,
1588 &obj->opo_ooa->ooa_xattr_list,
1590 list_del(&oxe->oxe_list);
1591 count = atomic_read(&oxe->oxe_ref);
1592 LASSERTF(count == 1,
1593 "Still has %d users on the xattr entry %.*s\n",
1594 count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1596 OBD_FREE(oxe, oxe->oxe_buflen);
1598 OBD_FREE_PTR(obj->opo_ooa);
1600 OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1603 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1605 struct osp_object *po = lu2osp_obj(o);
1606 struct osp_device *d = lu2osp_dev(o->lo_dev);
1611 * release reservation if object was declared but not created
1612 * this may require lu_object_put() in LOD
1614 if (unlikely(po->opo_reserved)) {
1615 LASSERT(d->opd_pre != NULL);
1616 LASSERT(d->opd_pre_reserved > 0);
1617 spin_lock(&d->opd_pre_lock);
1618 d->opd_pre_reserved--;
1619 spin_unlock(&d->opd_pre_lock);
1621 /* not needed in cache any more */
1622 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1626 /* XXX: Currently, NOT cache OST-object on MDT because:
1627 * 1. it is not often accessed on MDT.
1628 * 2. avoid up layer (such as LFSCK) to load too many
1629 * once-used OST-objects. */
1630 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1635 static int osp_object_print(const struct lu_env *env, void *cookie,
1636 lu_printer_t p, const struct lu_object *l)
1638 const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1640 return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1643 static int osp_object_invariant(const struct lu_object *o)
1648 struct lu_object_operations osp_lu_obj_ops = {
1649 .loo_object_init = osp_object_init,
1650 .loo_object_free = osp_object_free,
1651 .loo_object_release = osp_object_release,
1652 .loo_object_print = osp_object_print,
1653 .loo_object_invariant = osp_object_invariant