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 update_reply *reply,
211 struct lu_attr *attr,
212 struct osp_object *obj, int index)
214 struct osp_thread_info *osi = osp_env_info(env);
215 struct lu_buf *rbuf = &osi->osi_lb2;
216 struct obdo *lobdo = &osi->osi_obdo;
220 rc = update_get_reply_buf(reply, rbuf, index);
224 wobdo = rbuf->lb_buf;
225 if (rbuf->lb_len != sizeof(*wobdo))
228 obdo_le_to_cpu(wobdo, wobdo);
229 lustre_get_wire_obdo(NULL, lobdo, wobdo);
230 spin_lock(&obj->opo_lock);
231 if (obj->opo_ooa != NULL) {
232 la_from_obdo(&obj->opo_ooa->ooa_attr, lobdo, lobdo->o_valid);
234 *attr = obj->opo_ooa->ooa_attr;
236 LASSERT(attr != NULL);
238 la_from_obdo(attr, lobdo, lobdo->o_valid);
240 spin_unlock(&obj->opo_lock);
245 static int osp_attr_get_interpterer(const struct lu_env *env,
246 struct update_reply *reply,
247 struct osp_object *obj,
248 void *data, int index, int rc)
250 struct lu_attr *attr = data;
252 LASSERT(obj->opo_ooa != NULL);
255 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
256 obj->opo_non_exist = 0;
258 return osp_get_attr_from_reply(env, reply, NULL, obj, index);
261 osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
262 obj->opo_non_exist = 1;
265 spin_lock(&obj->opo_lock);
267 spin_unlock(&obj->opo_lock);
273 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
274 struct lustre_capa *capa)
276 struct osp_object *obj = dt2osp_obj(dt);
277 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
278 struct update_request *update;
281 if (obj->opo_ooa == NULL) {
282 rc = osp_oac_init(obj);
287 mutex_lock(&osp->opd_async_requests_mutex);
288 update = osp_find_or_create_async_update_request(osp);
290 rc = PTR_ERR(update);
292 rc = osp_insert_async_update(env, update, OBJ_ATTR_GET, obj, 0,
294 &obj->opo_ooa->ooa_attr,
295 osp_attr_get_interpterer);
296 mutex_unlock(&osp->opd_async_requests_mutex);
301 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
302 struct lu_attr *attr, struct lustre_capa *capa)
304 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
305 struct osp_object *obj = dt2osp_obj(dt);
306 struct dt_device *dev = &osp->opd_dt_dev;
307 struct update_request *update;
308 struct update_reply *reply;
309 struct ptlrpc_request *req = NULL;
313 if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
316 if (obj->opo_ooa != NULL) {
317 spin_lock(&obj->opo_lock);
318 if (obj->opo_ooa->ooa_attr.la_valid != 0) {
319 *attr = obj->opo_ooa->ooa_attr;
320 spin_unlock(&obj->opo_lock);
324 spin_unlock(&obj->opo_lock);
327 update = out_create_update_req(dev);
329 RETURN(PTR_ERR(update));
331 rc = out_insert_update(env, update, OBJ_ATTR_GET,
332 lu_object_fid(&dt->do_lu), 0, NULL, NULL);
334 CERROR("%s: Insert update error "DFID": rc = %d\n",
335 dev->dd_lu_dev.ld_obd->obd_name,
336 PFID(lu_object_fid(&dt->do_lu)), rc);
341 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
344 osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
345 obj->opo_non_exist = 1;
347 CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
348 dev->dd_lu_dev.ld_obd->obd_name,
349 PFID(lu_object_fid(&dt->do_lu)), rc);
355 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
356 obj->opo_non_exist = 0;
357 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
359 if (reply == NULL || reply->ur_version != UPDATE_REPLY_V1)
360 GOTO(out, rc = -EPROTO);
362 rc = osp_get_attr_from_reply(env, reply, attr, obj, 0);
366 if (!is_ost_obj(&dt->do_lu)) {
367 if (attr->la_flags == 1)
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)))
435 * track all UID/GID changes via llog
437 rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
438 if (rc != 0 || o->opo_ooa == NULL)
441 la = &o->opo_ooa->ooa_attr;
442 spin_lock(&o->opo_lock);
443 if (attr->la_valid & LA_UID) {
444 la->la_uid = attr->la_uid;
445 la->la_valid |= LA_UID;
448 if (attr->la_valid & LA_GID) {
449 la->la_gid = attr->la_gid;
450 la->la_valid |= LA_GID;
452 spin_unlock(&o->opo_lock);
457 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
458 const struct lu_attr *attr, struct thandle *th,
459 struct lustre_capa *capa)
461 struct osp_object *o = dt2osp_obj(dt);
466 /* we're interested in uid/gid changes only */
467 if (!(attr->la_valid & (LA_UID | LA_GID)))
470 /* new object, the very first ->attr_set()
471 * initializing attributes needs no logging
472 * all subsequent one are subject to the
473 * logging and synchronization with OST */
480 * once transaction is committed put proper command on
481 * the queue going to our OST
483 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
485 /* XXX: send new uid/gid to OST ASAP? */
490 static int osp_xattr_get_interpterer(const struct lu_env *env,
491 struct update_reply *reply,
492 struct osp_object *obj,
493 void *data, int index, int rc)
495 struct osp_object_attr *ooa = obj->opo_ooa;
496 struct osp_xattr_entry *oxe = data;
497 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
499 LASSERT(ooa != NULL);
502 int len = sizeof(*oxe) + oxe->oxe_namelen + 1;
504 rc = update_get_reply_buf(reply, rbuf, index);
505 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
506 spin_lock(&obj->opo_lock);
508 spin_unlock(&obj->opo_lock);
509 osp_oac_xattr_put(oxe);
511 return rc < 0 ? rc : -ERANGE;
514 spin_lock(&obj->opo_lock);
515 oxe->oxe_vallen = rbuf->lb_len;
516 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
519 spin_unlock(&obj->opo_lock);
520 } else if (rc == -ENOENT || rc == -ENODATA) {
521 spin_lock(&obj->opo_lock);
524 spin_unlock(&obj->opo_lock);
526 spin_lock(&obj->opo_lock);
528 spin_unlock(&obj->opo_lock);
531 osp_oac_xattr_put(oxe);
536 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
537 struct lu_buf *buf, const char *name,
538 struct lustre_capa *capa)
540 struct osp_object *obj = dt2osp_obj(dt);
541 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
542 struct update_request *update;
543 struct osp_xattr_entry *oxe;
544 int namelen = strlen(name);
547 LASSERT(buf != NULL);
548 LASSERT(name != NULL);
550 /* If only for xattr size, return directly. */
551 if (unlikely(buf->lb_len == 0))
554 if (obj->opo_ooa == NULL) {
555 rc = osp_oac_init(obj);
560 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
564 mutex_lock(&osp->opd_async_requests_mutex);
565 update = osp_find_or_create_async_update_request(osp);
566 if (IS_ERR(update)) {
567 rc = PTR_ERR(update);
568 mutex_unlock(&osp->opd_async_requests_mutex);
569 osp_oac_xattr_put(oxe);
571 rc = osp_insert_async_update(env, update, OBJ_XATTR_GET, obj,
572 1, &namelen, &name, oxe,
573 osp_xattr_get_interpterer);
575 mutex_unlock(&osp->opd_async_requests_mutex);
576 osp_oac_xattr_put(oxe);
578 /* XXX: Currently, we trigger the batched async OUT
579 * RPC via dt_declare_xattr_get(). It is not
580 * perfect solution, but works well now.
582 * We will improve it in the future. */
583 update = osp->opd_async_requests;
584 if (update != NULL && update->ur_buf != NULL &&
585 update->ur_buf->ub_count > 0) {
586 osp->opd_async_requests = NULL;
587 mutex_unlock(&osp->opd_async_requests_mutex);
588 rc = osp_unplug_async_update(env, osp, update);
590 mutex_unlock(&osp->opd_async_requests_mutex);
598 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
599 struct lu_buf *buf, const char *name,
600 struct lustre_capa *capa)
602 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
603 struct osp_object *obj = dt2osp_obj(dt);
604 struct dt_device *dev = &osp->opd_dt_dev;
605 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
606 struct update_request *update = NULL;
607 struct ptlrpc_request *req = NULL;
608 struct update_reply *reply;
609 struct osp_xattr_entry *oxe = NULL;
610 const char *dname = dt->do_lu.lo_dev->ld_obd->obd_name;
615 LASSERT(buf != NULL);
616 LASSERT(name != NULL);
618 if (unlikely(obj->opo_non_exist))
621 oxe = osp_oac_xattr_find(obj, name);
623 spin_lock(&obj->opo_lock);
624 if (oxe->oxe_ready) {
626 GOTO(unlock, rc = -ENODATA);
628 if (buf->lb_buf == NULL)
629 GOTO(unlock, rc = oxe->oxe_vallen);
631 if (buf->lb_len < oxe->oxe_vallen)
632 GOTO(unlock, rc = -ERANGE);
634 memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
636 GOTO(unlock, rc = oxe->oxe_vallen);
639 spin_unlock(&obj->opo_lock);
640 osp_oac_xattr_put(oxe);
644 spin_unlock(&obj->opo_lock);
647 update = out_create_update_req(dev);
649 GOTO(out, rc = PTR_ERR(update));
651 namelen = strlen(name);
652 rc = out_insert_update(env, update, OBJ_XATTR_GET,
653 lu_object_fid(&dt->do_lu), 1, &namelen, &name);
655 CERROR("%s: Insert update error "DFID": rc = %d\n",
656 dname, PFID(lu_object_fid(&dt->do_lu)), rc);
661 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
663 if (obj->opo_ooa == NULL)
667 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
670 CWARN("%s: Fail to add xattr (%s) to cache for "
671 DFID" (1): rc = %d\n", dname, name,
672 PFID(lu_object_fid(&dt->do_lu)), rc);
677 spin_lock(&obj->opo_lock);
678 if (rc == -ENOENT || rc == -ENODATA) {
684 spin_unlock(&obj->opo_lock);
689 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
691 if (reply->ur_version != UPDATE_REPLY_V1) {
692 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
693 dname, reply->ur_version, UPDATE_REPLY_V1,
694 PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
696 GOTO(out, rc = -EPROTO);
699 rc = update_get_reply_buf(reply, rbuf, 0);
703 LASSERT(rbuf->lb_len > 0 && rbuf->lb_len < PAGE_CACHE_SIZE);
705 if (buf->lb_buf == NULL)
706 GOTO(out, rc = rbuf->lb_len);
708 if (unlikely(buf->lb_len < rbuf->lb_len))
709 GOTO(out, rc = -ERANGE);
711 memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
713 if (obj->opo_ooa == NULL)
717 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
719 CWARN("%s: Fail to add xattr (%s) to "
720 "cache for "DFID" (2): rc = %d\n",
721 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
727 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
728 struct osp_xattr_entry *old = oxe;
729 struct osp_xattr_entry *tmp;
731 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
732 osp_oac_xattr_put(oxe);
735 CWARN("%s: Fail to update xattr (%s) to "
736 "cache for "DFID": rc = %d\n",
737 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
738 spin_lock(&obj->opo_lock);
740 spin_unlock(&obj->opo_lock);
745 /* Drop the ref for entry on list. */
746 osp_oac_xattr_put(old);
749 spin_lock(&obj->opo_lock);
750 oxe->oxe_vallen = rbuf->lb_len;
751 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
754 spin_unlock(&obj->opo_lock);
760 ptlrpc_req_finished(req);
762 if (update != NULL && !IS_ERR(update))
763 out_destroy_update_req(update);
766 osp_oac_xattr_put(oxe);
771 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
772 const struct lu_buf *buf, const char *name,
773 int flag, struct thandle *th)
775 struct osp_object *o = dt2osp_obj(dt);
776 struct update_request *update;
778 struct osp_xattr_entry *oxe;
779 int sizes[3] = {strlen(name), buf->lb_len,
781 char *bufs[3] = {(char *)name, (char *)buf->lb_buf };
784 LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
786 update = out_find_create_update_loc(th, dt);
787 if (IS_ERR(update)) {
788 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
789 dt->do_lu.lo_dev->ld_obd->obd_name,
790 PFID(lu_object_fid(&dt->do_lu)),
791 (int)PTR_ERR(update));
793 return PTR_ERR(update);
796 flag = cpu_to_le32(flag);
797 bufs[2] = (char *)&flag;
799 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
800 rc = out_insert_update(env, update, OBJ_XATTR_SET, fid,
801 ARRAY_SIZE(sizes), sizes, (const char **)bufs);
802 if (rc != 0 || o->opo_ooa == NULL)
805 oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
807 CWARN("%s: Fail to add xattr (%s) to cache for "DFID
808 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
809 name, PFID(lu_object_fid(&dt->do_lu)), rc);
814 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
815 struct osp_xattr_entry *old = oxe;
816 struct osp_xattr_entry *tmp;
818 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
819 osp_oac_xattr_put(oxe);
822 CWARN("%s: Fail to update xattr (%s) to cache for "DFID
823 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
824 name, PFID(lu_object_fid(&dt->do_lu)), rc);
825 spin_lock(&o->opo_lock);
827 spin_unlock(&o->opo_lock);
832 /* Drop the ref for entry on list. */
833 osp_oac_xattr_put(old);
836 spin_lock(&o->opo_lock);
837 oxe->oxe_vallen = buf->lb_len;
838 memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
841 spin_unlock(&o->opo_lock);
842 osp_oac_xattr_put(oxe);
847 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
848 const struct lu_buf *buf, const char *name, int fl,
849 struct thandle *th, struct lustre_capa *capa)
851 CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
852 PFID(&dt->do_lu.lo_header->loh_fid));
857 static int osp_declare_object_create(const struct lu_env *env,
858 struct dt_object *dt,
859 struct lu_attr *attr,
860 struct dt_allocation_hint *hint,
861 struct dt_object_format *dof,
864 struct osp_thread_info *osi = osp_env_info(env);
865 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
866 struct osp_object *o = dt2osp_obj(dt);
867 const struct lu_fid *fid;
872 /* should happen to non-0 OSP only so that at least one object
873 * has been already declared in the scenario and LOD should
875 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
878 LASSERT(d->opd_last_used_oid_file);
879 fid = lu_object_fid(&dt->do_lu);
882 * There can be gaps in precreated ids and record to unlink llog
883 * XXX: we do not handle gaps yet, implemented before solution
884 * was found to be racy, so we disabled that. there is no
885 * point in making useless but expensive llog declaration.
887 /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
889 if (unlikely(!fid_is_zero(fid))) {
890 /* replay case: caller knows fid */
891 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
892 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
893 sizeof(osi->osi_id), osi->osi_off,
899 * in declaration we need to reserve object so that we don't block
900 * awaiting precreation RPC to complete
902 rc = osp_precreate_reserve(env, d);
904 * we also need to declare update to local "last used id" file for
905 * recovery if object isn't used for a reason, we need to release
906 * reservation, this can be made in osd_object_release()
909 /* mark id is reserved: in create we don't want to talk
911 LASSERT(o->opo_reserved == 0);
914 /* common for all OSPs file hystorically */
915 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
916 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
917 sizeof(osi->osi_id), osi->osi_off,
920 /* not needed in the cache anymore */
921 set_bit(LU_OBJECT_HEARD_BANSHEE,
922 &dt->do_lu.lo_header->loh_flags);
927 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
928 struct lu_attr *attr,
929 struct dt_allocation_hint *hint,
930 struct dt_object_format *dof, struct thandle *th)
932 struct osp_thread_info *osi = osp_env_info(env);
933 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
934 struct osp_object *o = dt2osp_obj(dt);
936 struct lu_fid *fid = &osi->osi_fid;
939 o->opo_non_exist = 0;
940 if (o->opo_reserved) {
941 /* regular case, fid is assigned holding trunsaction open */
942 osp_object_assign_fid(env, d, o);
945 memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
947 LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
950 if (!o->opo_reserved) {
951 /* special case, id was assigned outside of transaction
952 * see comments in osp_declare_attr_set */
953 LASSERT(d->opd_pre != NULL);
954 spin_lock(&d->opd_pre_lock);
955 osp_update_last_fid(d, fid);
956 spin_unlock(&d->opd_pre_lock);
959 CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
961 /* If the precreate ends, it means it will be ready to rollover to
962 * the new sequence soon, all the creation should be synchronized,
963 * otherwise during replay, the replay fid will be inconsistent with
964 * last_used/create fid */
965 if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
969 * it's OK if the import is inactive by this moment - id was created
970 * by OST earlier, we just need to maintain it consistently on the disk
971 * once import is reconnected, OSP will claim this and other objects
972 * used and OST either keep them, if they exist or recreate
975 /* we might have lost precreated objects */
976 if (unlikely(d->opd_gap_count) > 0) {
977 LASSERT(d->opd_pre != NULL);
978 spin_lock(&d->opd_pre_lock);
979 if (d->opd_gap_count > 0) {
980 int count = d->opd_gap_count;
982 ostid_set_id(&osi->osi_oi,
983 fid_oid(&d->opd_gap_start_fid));
984 d->opd_gap_count = 0;
985 spin_unlock(&d->opd_pre_lock);
987 CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
988 PFID(&d->opd_gap_start_fid), count);
989 /* real gap handling is disabled intil ORI-692 will be
990 * fixed, now we only report gaps */
992 spin_unlock(&d->opd_pre_lock);
996 /* new object, the very first ->attr_set()
997 * initializing attributes needs no logging */
1000 /* Only need update last_used oid file, seq file will only be update
1001 * during seq rollover */
1002 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1003 &d->opd_last_used_fid.f_oid, d->opd_index);
1005 rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1008 CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1009 d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1014 int osp_declare_object_destroy(const struct lu_env *env,
1015 struct dt_object *dt, struct thandle *th)
1017 struct osp_object *o = dt2osp_obj(dt);
1023 * track objects to be destroyed via llog
1025 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1030 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1033 struct osp_object *o = dt2osp_obj(dt);
1038 o->opo_non_exist = 1;
1040 * once transaction is committed put proper command on
1041 * the queue going to our OST
1043 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1045 /* not needed in cache any more */
1046 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1051 struct dt_object_operations osp_obj_ops = {
1052 .do_declare_attr_get = osp_declare_attr_get,
1053 .do_attr_get = osp_attr_get,
1054 .do_declare_attr_set = osp_declare_attr_set,
1055 .do_attr_set = osp_attr_set,
1056 .do_declare_xattr_get = osp_declare_xattr_get,
1057 .do_xattr_get = osp_xattr_get,
1058 .do_declare_xattr_set = osp_declare_xattr_set,
1059 .do_xattr_set = osp_xattr_set,
1060 .do_declare_create = osp_declare_object_create,
1061 .do_create = osp_object_create,
1062 .do_declare_destroy = osp_declare_object_destroy,
1063 .do_destroy = osp_object_destroy,
1066 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1067 const struct lu_object_conf *conf)
1069 struct osp_object *po = lu2osp_obj(o);
1073 spin_lock_init(&po->opo_lock);
1074 o->lo_header->loh_attr |= LOHA_REMOTE;
1076 if (is_ost_obj(o)) {
1077 po->opo_obj.do_ops = &osp_obj_ops;
1079 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1081 po->opo_obj.do_ops = &osp_md_obj_ops;
1082 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1085 o->lo_header->loh_attr |=
1086 LOHA_EXISTS | (la->la_mode & S_IFMT);
1087 if (rc == -ENOENT) {
1088 po->opo_non_exist = 1;
1095 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1097 struct osp_object *obj = lu2osp_obj(o);
1098 struct lu_object_header *h = o->lo_header;
1100 dt_object_fini(&obj->opo_obj);
1101 lu_object_header_fini(h);
1102 if (obj->opo_ooa != NULL) {
1103 struct osp_xattr_entry *oxe;
1104 struct osp_xattr_entry *tmp;
1107 list_for_each_entry_safe(oxe, tmp,
1108 &obj->opo_ooa->ooa_xattr_list,
1110 list_del(&oxe->oxe_list);
1111 count = atomic_read(&oxe->oxe_ref);
1112 LASSERTF(count == 1,
1113 "Still has %d users on the xattr entry %.*s\n",
1114 count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1116 OBD_FREE(oxe, oxe->oxe_buflen);
1118 OBD_FREE_PTR(obj->opo_ooa);
1120 OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1123 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1125 struct osp_object *po = lu2osp_obj(o);
1126 struct osp_device *d = lu2osp_dev(o->lo_dev);
1131 * release reservation if object was declared but not created
1132 * this may require lu_object_put() in LOD
1134 if (unlikely(po->opo_reserved)) {
1135 LASSERT(d->opd_pre != NULL);
1136 LASSERT(d->opd_pre_reserved > 0);
1137 spin_lock(&d->opd_pre_lock);
1138 d->opd_pre_reserved--;
1139 spin_unlock(&d->opd_pre_lock);
1141 /* not needed in cache any more */
1142 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1146 /* XXX: Currently, NOT cache OST-object on MDT because:
1147 * 1. it is not often accessed on MDT.
1148 * 2. avoid up layer (such as LFSCK) to load too many
1149 * once-used OST-objects. */
1150 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1155 static int osp_object_print(const struct lu_env *env, void *cookie,
1156 lu_printer_t p, const struct lu_object *l)
1158 const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1160 return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1163 static int osp_object_invariant(const struct lu_object *o)
1168 struct lu_object_operations osp_lu_obj_ops = {
1169 .loo_object_init = osp_object_init,
1170 .loo_object_free = osp_object_free,
1171 .loo_object_release = osp_object_release,
1172 .loo_object_print = osp_object_print,
1173 .loo_object_invariant = osp_object_invariant