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);
373 if (!is_ost_obj(&dt->do_lu)) {
374 if (attr->la_flags == 1)
384 ptlrpc_req_finished(req);
386 out_destroy_update_req(update);
391 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
392 const struct lu_attr *attr, struct thandle *th)
394 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
395 struct osp_object *o = dt2osp_obj(dt);
402 * Usually we don't allow server stack to manipulate size
403 * but there is a special case when striping is created
404 * late, after stripless file got truncated to non-zero.
406 * In this case we do the following:
408 * 1) grab id in declare - this can lead to leaked OST objects
409 * but we don't currently have proper mechanism and the only
410 * options we have are to do truncate RPC holding transaction
411 * open (very bad) or to grab id in declare at cost of leaked
412 * OST object in same very rare unfortunate case (just bad)
413 * notice 1.6-2.0 do assignment outside of running transaction
414 * all the time, meaning many more chances for leaked objects.
416 * 2) send synchronous truncate RPC with just assigned id
419 /* there are few places in MDD code still passing NULL
420 * XXX: to be fixed soon */
424 if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
425 fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
426 LASSERT(!dt_object_exists(dt));
427 osp_object_assign_fid(env, d, o);
428 rc = osp_object_truncate(env, dt, attr->la_size);
434 /* no need in logging for new objects being created */
438 if (!(attr->la_valid & (LA_UID | LA_GID)))
441 if (!is_only_remote_trans(th))
443 * track all UID/GID changes via llog
445 rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
447 /* It is for OST-object attr_set directly without updating
448 * local MDT-object attribute. It is usually used by LFSCK. */
449 rc = osp_md_declare_attr_set(env, dt, attr, th);
451 if (rc != 0 || o->opo_ooa == NULL)
454 la = &o->opo_ooa->ooa_attr;
455 spin_lock(&o->opo_lock);
456 if (attr->la_valid & LA_UID) {
457 la->la_uid = attr->la_uid;
458 la->la_valid |= LA_UID;
461 if (attr->la_valid & LA_GID) {
462 la->la_gid = attr->la_gid;
463 la->la_valid |= LA_GID;
465 spin_unlock(&o->opo_lock);
470 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
471 const struct lu_attr *attr, struct thandle *th,
472 struct lustre_capa *capa)
474 struct osp_object *o = dt2osp_obj(dt);
479 /* we're interested in uid/gid changes only */
480 if (!(attr->la_valid & (LA_UID | LA_GID)))
483 /* new object, the very first ->attr_set()
484 * initializing attributes needs no logging
485 * all subsequent one are subject to the
486 * logging and synchronization with OST */
492 if (!is_only_remote_trans(th))
494 * once transaction is committed put proper command on
495 * the queue going to our OST
497 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
498 /* XXX: send new uid/gid to OST ASAP? */
500 /* It is for OST-object attr_set directly without updating
501 * local MDT-object attribute. It is usually used by LFSCK. */
502 rc = osp_md_attr_set(env, dt, attr, th, capa);
507 static int osp_xattr_get_interpterer(const struct lu_env *env,
508 struct object_update_reply *reply,
509 struct ptlrpc_request *req,
510 struct osp_object *obj,
511 void *data, int index, int rc)
513 struct osp_object_attr *ooa = obj->opo_ooa;
514 struct osp_xattr_entry *oxe = data;
515 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
517 LASSERT(ooa != NULL);
520 int len = sizeof(*oxe) + oxe->oxe_namelen + 1;
522 rc = object_update_result_data_get(reply, rbuf, index);
523 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
524 spin_lock(&obj->opo_lock);
526 spin_unlock(&obj->opo_lock);
527 osp_oac_xattr_put(oxe);
529 return rc < 0 ? rc : -ERANGE;
532 spin_lock(&obj->opo_lock);
533 oxe->oxe_vallen = rbuf->lb_len;
534 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
537 spin_unlock(&obj->opo_lock);
538 } else if (rc == -ENOENT || rc == -ENODATA) {
539 spin_lock(&obj->opo_lock);
542 spin_unlock(&obj->opo_lock);
544 spin_lock(&obj->opo_lock);
546 spin_unlock(&obj->opo_lock);
549 osp_oac_xattr_put(oxe);
554 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
555 struct lu_buf *buf, const char *name,
556 struct lustre_capa *capa)
558 struct osp_object *obj = dt2osp_obj(dt);
559 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
560 struct dt_update_request *update;
561 struct osp_xattr_entry *oxe;
562 int namelen = strlen(name);
565 LASSERT(buf != NULL);
566 LASSERT(name != NULL);
568 /* If only for xattr size, return directly. */
569 if (unlikely(buf->lb_len == 0))
572 if (obj->opo_ooa == NULL) {
573 rc = osp_oac_init(obj);
578 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
582 mutex_lock(&osp->opd_async_requests_mutex);
583 update = osp_find_or_create_async_update_request(osp);
584 if (IS_ERR(update)) {
585 rc = PTR_ERR(update);
586 mutex_unlock(&osp->opd_async_requests_mutex);
587 osp_oac_xattr_put(oxe);
589 rc = osp_insert_async_update(env, update, OUT_XATTR_GET, obj,
590 1, &namelen, &name, oxe,
591 osp_xattr_get_interpterer);
593 mutex_unlock(&osp->opd_async_requests_mutex);
594 osp_oac_xattr_put(oxe);
596 /* XXX: Currently, we trigger the batched async OUT
597 * RPC via dt_declare_xattr_get(). It is not
598 * perfect solution, but works well now.
600 * We will improve it in the future. */
601 update = osp->opd_async_requests;
602 if (update != NULL && update->dur_req != NULL &&
603 update->dur_req->ourq_count > 0) {
604 osp->opd_async_requests = NULL;
605 mutex_unlock(&osp->opd_async_requests_mutex);
606 rc = osp_unplug_async_update(env, osp, update);
608 mutex_unlock(&osp->opd_async_requests_mutex);
616 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
617 struct lu_buf *buf, const char *name,
618 struct lustre_capa *capa)
620 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
621 struct osp_object *obj = dt2osp_obj(dt);
622 struct dt_device *dev = &osp->opd_dt_dev;
623 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
624 struct dt_update_request *update = NULL;
625 struct ptlrpc_request *req = NULL;
626 struct object_update_reply *reply;
627 struct osp_xattr_entry *oxe = NULL;
628 const char *dname = dt->do_lu.lo_dev->ld_obd->obd_name;
633 LASSERT(buf != NULL);
634 LASSERT(name != NULL);
636 if (unlikely(obj->opo_non_exist))
639 oxe = osp_oac_xattr_find(obj, name);
641 spin_lock(&obj->opo_lock);
642 if (oxe->oxe_ready) {
644 GOTO(unlock, rc = -ENODATA);
646 if (buf->lb_buf == NULL)
647 GOTO(unlock, rc = oxe->oxe_vallen);
649 if (buf->lb_len < oxe->oxe_vallen)
650 GOTO(unlock, rc = -ERANGE);
652 memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
654 GOTO(unlock, rc = oxe->oxe_vallen);
657 spin_unlock(&obj->opo_lock);
658 osp_oac_xattr_put(oxe);
662 spin_unlock(&obj->opo_lock);
665 update = out_create_update_req(dev);
667 GOTO(out, rc = PTR_ERR(update));
669 namelen = strlen(name) + 1;
670 rc = out_insert_update(env, update, OUT_XATTR_GET,
671 lu_object_fid(&dt->do_lu), 1, &namelen, &name);
673 CERROR("%s: Insert update error "DFID": rc = %d\n",
674 dname, PFID(lu_object_fid(&dt->do_lu)), rc);
679 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
681 if (obj->opo_ooa == NULL)
685 oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
688 CWARN("%s: Fail to add xattr (%s) to cache for "
689 DFID" (1): rc = %d\n", dname, name,
690 PFID(lu_object_fid(&dt->do_lu)), rc);
695 spin_lock(&obj->opo_lock);
696 if (rc == -ENOENT || rc == -ENODATA) {
702 spin_unlock(&obj->opo_lock);
707 reply = req_capsule_server_sized_get(&req->rq_pill,
708 &RMF_OUT_UPDATE_REPLY,
709 OUT_UPDATE_REPLY_SIZE);
710 if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
711 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
712 dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
713 PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
715 GOTO(out, rc = -EPROTO);
718 rc = object_update_result_data_get(reply, rbuf, 0);
722 if (buf->lb_buf == NULL)
723 GOTO(out, rc = rbuf->lb_len);
725 if (unlikely(buf->lb_len < rbuf->lb_len))
726 GOTO(out, rc = -ERANGE);
728 memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
730 if (obj->opo_ooa == NULL)
734 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
736 CWARN("%s: Fail to add xattr (%s) to "
737 "cache for "DFID" (2): rc = %d\n",
738 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
744 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
745 struct osp_xattr_entry *old = oxe;
746 struct osp_xattr_entry *tmp;
748 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
749 osp_oac_xattr_put(oxe);
752 CWARN("%s: Fail to update xattr (%s) to "
753 "cache for "DFID": rc = %d\n",
754 dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
755 spin_lock(&obj->opo_lock);
757 spin_unlock(&obj->opo_lock);
762 /* Drop the ref for entry on list. */
763 osp_oac_xattr_put(old);
766 spin_lock(&obj->opo_lock);
767 oxe->oxe_vallen = rbuf->lb_len;
768 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
771 spin_unlock(&obj->opo_lock);
777 ptlrpc_req_finished(req);
779 if (update != NULL && !IS_ERR(update))
780 out_destroy_update_req(update);
783 osp_oac_xattr_put(oxe);
788 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
789 const struct lu_buf *buf, const char *name,
790 int flag, struct thandle *th)
792 struct osp_object *o = dt2osp_obj(dt);
793 struct dt_update_request *update;
795 struct osp_xattr_entry *oxe;
796 int sizes[3] = {strlen(name), buf->lb_len,
798 char *bufs[3] = {(char *)name, (char *)buf->lb_buf };
801 LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
803 update = out_find_create_update_loc(th, dt);
804 if (IS_ERR(update)) {
805 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
806 dt->do_lu.lo_dev->ld_obd->obd_name,
807 PFID(lu_object_fid(&dt->do_lu)),
808 (int)PTR_ERR(update));
810 return PTR_ERR(update);
813 flag = cpu_to_le32(flag);
814 bufs[2] = (char *)&flag;
816 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
817 rc = out_insert_update(env, update, OUT_XATTR_SET, fid,
818 ARRAY_SIZE(sizes), sizes, (const char **)bufs);
819 if (rc != 0 || o->opo_ooa == NULL)
822 oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
824 CWARN("%s: Fail to add xattr (%s) to cache for "DFID
825 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
826 name, PFID(lu_object_fid(&dt->do_lu)), rc);
831 if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
832 struct osp_xattr_entry *old = oxe;
833 struct osp_xattr_entry *tmp;
835 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
836 osp_oac_xattr_put(oxe);
839 CWARN("%s: Fail to update xattr (%s) to cache for "DFID
840 ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
841 name, PFID(lu_object_fid(&dt->do_lu)), rc);
842 spin_lock(&o->opo_lock);
844 spin_unlock(&o->opo_lock);
849 /* Drop the ref for entry on list. */
850 osp_oac_xattr_put(old);
853 spin_lock(&o->opo_lock);
854 oxe->oxe_vallen = buf->lb_len;
855 memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
858 spin_unlock(&o->opo_lock);
859 osp_oac_xattr_put(oxe);
864 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
865 const struct lu_buf *buf, const char *name, int fl,
866 struct thandle *th, struct lustre_capa *capa)
868 CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
869 PFID(&dt->do_lu.lo_header->loh_fid));
874 static int osp_declare_object_create(const struct lu_env *env,
875 struct dt_object *dt,
876 struct lu_attr *attr,
877 struct dt_allocation_hint *hint,
878 struct dt_object_format *dof,
881 struct osp_thread_info *osi = osp_env_info(env);
882 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
883 struct osp_object *o = dt2osp_obj(dt);
884 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
889 if (is_only_remote_trans(th)) {
890 LASSERT(fid_is_sane(fid));
892 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
897 /* should happen to non-0 OSP only so that at least one object
898 * has been already declared in the scenario and LOD should
900 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
903 LASSERT(d->opd_last_used_oid_file);
906 * There can be gaps in precreated ids and record to unlink llog
907 * XXX: we do not handle gaps yet, implemented before solution
908 * was found to be racy, so we disabled that. there is no
909 * point in making useless but expensive llog declaration.
911 /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
913 if (unlikely(!fid_is_zero(fid))) {
914 /* replay case: caller knows fid */
915 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
916 osi->osi_lb.lb_len = sizeof(osi->osi_id);
917 osi->osi_lb.lb_buf = NULL;
918 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
919 &osi->osi_lb, osi->osi_off, th);
924 * in declaration we need to reserve object so that we don't block
925 * awaiting precreation RPC to complete
927 rc = osp_precreate_reserve(env, d);
929 * we also need to declare update to local "last used id" file for
930 * recovery if object isn't used for a reason, we need to release
931 * reservation, this can be made in osd_object_release()
934 /* mark id is reserved: in create we don't want to talk
936 LASSERT(o->opo_reserved == 0);
939 /* common for all OSPs file hystorically */
940 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
941 osi->osi_lb.lb_len = sizeof(osi->osi_id);
942 osi->osi_lb.lb_buf = NULL;
943 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
944 &osi->osi_lb, osi->osi_off, th);
946 /* not needed in the cache anymore */
947 set_bit(LU_OBJECT_HEARD_BANSHEE,
948 &dt->do_lu.lo_header->loh_flags);
953 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
954 struct lu_attr *attr,
955 struct dt_allocation_hint *hint,
956 struct dt_object_format *dof, struct thandle *th)
958 struct osp_thread_info *osi = osp_env_info(env);
959 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
960 struct osp_object *o = dt2osp_obj(dt);
962 struct lu_fid *fid = &osi->osi_fid;
965 if (is_only_remote_trans(th)) {
966 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
968 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
970 o->opo_non_exist = 0;
975 o->opo_non_exist = 0;
976 if (o->opo_reserved) {
977 /* regular case, fid is assigned holding trunsaction open */
978 osp_object_assign_fid(env, d, o);
981 memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
983 LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
986 if (!o->opo_reserved) {
987 /* special case, id was assigned outside of transaction
988 * see comments in osp_declare_attr_set */
989 LASSERT(d->opd_pre != NULL);
990 spin_lock(&d->opd_pre_lock);
991 osp_update_last_fid(d, fid);
992 spin_unlock(&d->opd_pre_lock);
995 CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
997 /* If the precreate ends, it means it will be ready to rollover to
998 * the new sequence soon, all the creation should be synchronized,
999 * otherwise during replay, the replay fid will be inconsistent with
1000 * last_used/create fid */
1001 if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1005 * it's OK if the import is inactive by this moment - id was created
1006 * by OST earlier, we just need to maintain it consistently on the disk
1007 * once import is reconnected, OSP will claim this and other objects
1008 * used and OST either keep them, if they exist or recreate
1011 /* we might have lost precreated objects */
1012 if (unlikely(d->opd_gap_count) > 0) {
1013 LASSERT(d->opd_pre != NULL);
1014 spin_lock(&d->opd_pre_lock);
1015 if (d->opd_gap_count > 0) {
1016 int count = d->opd_gap_count;
1018 ostid_set_id(&osi->osi_oi,
1019 fid_oid(&d->opd_gap_start_fid));
1020 d->opd_gap_count = 0;
1021 spin_unlock(&d->opd_pre_lock);
1023 CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1024 PFID(&d->opd_gap_start_fid), count);
1025 /* real gap handling is disabled intil ORI-692 will be
1026 * fixed, now we only report gaps */
1028 spin_unlock(&d->opd_pre_lock);
1032 /* new object, the very first ->attr_set()
1033 * initializing attributes needs no logging */
1036 /* Only need update last_used oid file, seq file will only be update
1037 * during seq rollover */
1038 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1039 &d->opd_last_used_fid.f_oid, d->opd_index);
1041 rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1044 CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1045 d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1050 int osp_declare_object_destroy(const struct lu_env *env,
1051 struct dt_object *dt, struct thandle *th)
1053 struct osp_object *o = dt2osp_obj(dt);
1059 * track objects to be destroyed via llog
1061 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1066 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1069 struct osp_object *o = dt2osp_obj(dt);
1074 o->opo_non_exist = 1;
1076 * once transaction is committed put proper command on
1077 * the queue going to our OST
1079 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1081 /* not needed in cache any more */
1082 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1087 struct osp_orphan_it {
1091 int ooi_total_npages;
1092 int ooi_valid_npages;
1093 unsigned int ooi_swab:1;
1095 struct dt_object *ooi_obj;
1096 struct lu_orphan_ent *ooi_ent;
1097 struct page *ooi_cur_page;
1098 struct lu_idxpage *ooi_cur_idxpage;
1099 struct page **ooi_pages;
1102 static int osp_orphan_index_lookup(const struct lu_env *env,
1103 struct dt_object *dt,
1105 const struct dt_key *key,
1106 struct lustre_capa *capa)
1111 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1112 struct dt_object *dt,
1113 const struct dt_rec *rec,
1114 const struct dt_key *key,
1115 struct thandle *handle)
1120 static int osp_orphan_index_insert(const struct lu_env *env,
1121 struct dt_object *dt,
1122 const struct dt_rec *rec,
1123 const struct dt_key *key,
1124 struct thandle *handle,
1125 struct lustre_capa *capa,
1131 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1132 struct dt_object *dt,
1133 const struct dt_key *key,
1134 struct thandle *handle)
1139 static int osp_orphan_index_delete(const struct lu_env *env,
1140 struct dt_object *dt,
1141 const struct dt_key *key,
1142 struct thandle *handle,
1143 struct lustre_capa *capa)
1148 static struct dt_it *osp_orphan_it_init(const struct lu_env *env,
1149 struct dt_object *dt,
1151 struct lustre_capa *capa)
1153 struct osp_orphan_it *it;
1157 return ERR_PTR(-ENOMEM);
1162 return (struct dt_it *)it;
1165 static void osp_orphan_it_fini(const struct lu_env *env,
1168 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1169 struct page **pages = it->ooi_pages;
1170 int npages = it->ooi_total_npages;
1173 if (pages != NULL) {
1174 for (i = 0; i < npages; i++) {
1175 if (pages[i] != NULL) {
1176 if (pages[i] == it->ooi_cur_page) {
1178 it->ooi_cur_page = NULL;
1180 __free_page(pages[i]);
1183 OBD_FREE(pages, npages * sizeof(*pages));
1188 static int osp_orphan_it_fetch(const struct lu_env *env,
1189 struct osp_orphan_it *it)
1191 struct lu_device *dev = it->ooi_obj->do_lu.lo_dev;
1192 struct osp_device *osp = lu2osp_dev(dev);
1193 struct page **pages;
1194 struct ptlrpc_request *req = NULL;
1195 struct ptlrpc_bulk_desc *desc;
1196 struct idx_info *ii;
1203 npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1204 npages /= PAGE_CACHE_SIZE;
1206 OBD_ALLOC(pages, npages * sizeof(*pages));
1210 it->ooi_pages = pages;
1211 it->ooi_total_npages = npages;
1212 for (i = 0; i < npages; i++) {
1213 pages[i] = alloc_page(GFP_IOFS);
1214 if (pages[i] == NULL)
1218 req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1223 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1225 ptlrpc_request_free(req);
1229 req->rq_request_portal = OUT_PORTAL;
1230 ptlrpc_at_set_req_timeout(req);
1232 desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
1235 ptlrpc_request_free(req);
1239 for (i = 0; i < npages; i++)
1240 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1242 ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1243 memset(ii, 0, sizeof(*ii));
1244 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1245 ii->ii_fid.f_oid = osp->opd_index;
1246 ii->ii_fid.f_ver = 0;
1247 ii->ii_magic = IDX_INFO_MAGIC;
1248 ii->ii_flags = II_FL_NOHASH;
1249 ii->ii_count = npages * LU_PAGE_COUNT;
1250 ii->ii_hash_start = it->ooi_next;
1252 osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
1254 ptlrpc_request_set_replen(req);
1255 rc = ptlrpc_queue_wait(req);
1259 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1260 req->rq_bulk->bd_nob_transferred);
1264 ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1265 if (ii->ii_magic != IDX_INFO_MAGIC)
1266 GOTO(out, rc = -EPROTO);
1268 npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1269 (PAGE_CACHE_SHIFT - LU_PAGE_SHIFT);
1270 if (npages > it->ooi_total_npages) {
1271 CERROR("%s: returned more pages than expected, %u > %u\n",
1272 osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1273 GOTO(out, rc = -EINVAL);
1276 it->ooi_valid_npages = npages;
1277 if (ptlrpc_rep_need_swab(req))
1280 it->ooi_next = ii->ii_hash_end;
1285 ptlrpc_req_finished(req);
1290 static int osp_orphan_it_next(const struct lu_env *env,
1293 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1294 struct lu_idxpage *idxpage;
1295 struct page **pages;
1301 idxpage = it->ooi_cur_idxpage;
1302 if (idxpage != NULL) {
1303 if (idxpage->lip_nr == 0)
1307 if (it->ooi_pos2 < idxpage->lip_nr) {
1309 (struct lu_orphan_ent *)idxpage->lip_entries +
1312 lustre_swab_orphan_ent(it->ooi_ent);
1316 it->ooi_cur_idxpage = NULL;
1320 if (it->ooi_pos1 < LU_PAGE_COUNT) {
1321 it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1322 LU_PAGE_SIZE * it->ooi_pos1;
1324 lustre_swab_lip_header(it->ooi_cur_idxpage);
1325 if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1326 struct osp_device *osp =
1327 lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1329 CERROR("%s: invalid magic (%x != %x) for page "
1330 "%d/%d while read layout orphan index\n",
1331 osp->opd_obd->obd_name,
1332 it->ooi_cur_idxpage->lip_magic,
1333 LIP_MAGIC, it->ooi_pos0, it->ooi_pos1);
1334 /* Skip this lu_page next time. */
1335 it->ooi_pos2 = idxpage->lip_nr - 1;
1342 kunmap(it->ooi_cur_page);
1343 it->ooi_cur_page = NULL;
1347 pages = it->ooi_pages;
1348 if (it->ooi_pos0 < it->ooi_valid_npages) {
1349 it->ooi_cur_page = kmap(pages[it->ooi_pos0]);
1354 for (i = 0; i < it->ooi_total_npages; i++) {
1355 if (pages[i] != NULL)
1356 __free_page(pages[i]);
1358 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1361 it->ooi_total_npages = 0;
1362 it->ooi_valid_npages = 0;
1365 it->ooi_cur_page = NULL;
1366 it->ooi_cur_idxpage = NULL;
1367 it->ooi_pages = NULL;
1370 if (it->ooi_next == II_END_OFF)
1373 rc = osp_orphan_it_fetch(env, it);
1380 static int osp_orphan_it_get(const struct lu_env *env,
1382 const struct dt_key *key)
1387 static void osp_orphan_it_put(const struct lu_env *env,
1392 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1393 const struct dt_it *di)
1395 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1396 struct lu_orphan_ent *ent = it->ooi_ent;
1398 if (likely(ent != NULL))
1399 return (struct dt_key *)(&ent->loe_key);
1404 static int osp_orphan_it_key_size(const struct lu_env *env,
1405 const struct dt_it *di)
1407 return sizeof(struct lu_fid);
1410 static int osp_orphan_it_rec(const struct lu_env *env,
1411 const struct dt_it *di,
1415 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1416 struct lu_orphan_ent *ent = it->ooi_ent;
1418 if (likely(ent != NULL)) {
1419 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1426 static __u64 osp_orphan_it_store(const struct lu_env *env,
1427 const struct dt_it *di)
1429 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1431 return it->ooi_next;
1435 * \retval +1: locate to the exactly position
1436 * \retval 0: cannot locate to the exactly position,
1437 * call next() to move to a valid position.
1438 * \retval -ve: on error
1440 static int osp_orphan_it_load(const struct lu_env *env,
1441 const struct dt_it *di,
1444 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1447 it->ooi_next = hash;
1448 rc = osp_orphan_it_next(env, (struct dt_it *)di);
1458 static int osp_orphan_it_key_rec(const struct lu_env *env,
1459 const struct dt_it *di,
1465 static const struct dt_index_operations osp_orphan_index_ops = {
1466 .dio_lookup = osp_orphan_index_lookup,
1467 .dio_declare_insert = osp_orphan_index_declare_insert,
1468 .dio_insert = osp_orphan_index_insert,
1469 .dio_declare_delete = osp_orphan_index_declare_delete,
1470 .dio_delete = osp_orphan_index_delete,
1472 .init = osp_orphan_it_init,
1473 .fini = osp_orphan_it_fini,
1474 .next = osp_orphan_it_next,
1475 .get = osp_orphan_it_get,
1476 .put = osp_orphan_it_put,
1477 .key = osp_orphan_it_key,
1478 .key_size = osp_orphan_it_key_size,
1479 .rec = osp_orphan_it_rec,
1480 .store = osp_orphan_it_store,
1481 .load = osp_orphan_it_load,
1482 .key_rec = osp_orphan_it_key_rec,
1486 static int osp_index_try(const struct lu_env *env,
1487 struct dt_object *dt,
1488 const struct dt_index_features *feat)
1490 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1492 if (fid_is_last_id(fid) && fid_is_idif(fid)) {
1493 dt->do_index_ops = &osp_orphan_index_ops;
1501 struct dt_object_operations osp_obj_ops = {
1502 .do_declare_attr_get = osp_declare_attr_get,
1503 .do_attr_get = osp_attr_get,
1504 .do_declare_attr_set = osp_declare_attr_set,
1505 .do_attr_set = osp_attr_set,
1506 .do_declare_xattr_get = osp_declare_xattr_get,
1507 .do_xattr_get = osp_xattr_get,
1508 .do_declare_xattr_set = osp_declare_xattr_set,
1509 .do_xattr_set = osp_xattr_set,
1510 .do_declare_create = osp_declare_object_create,
1511 .do_create = osp_object_create,
1512 .do_declare_destroy = osp_declare_object_destroy,
1513 .do_destroy = osp_object_destroy,
1514 .do_index_try = osp_index_try,
1517 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1518 const struct lu_object_conf *conf)
1520 struct osp_object *po = lu2osp_obj(o);
1524 spin_lock_init(&po->opo_lock);
1525 o->lo_header->loh_attr |= LOHA_REMOTE;
1527 if (is_ost_obj(o)) {
1528 po->opo_obj.do_ops = &osp_obj_ops;
1530 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1532 po->opo_obj.do_ops = &osp_md_obj_ops;
1533 po->opo_obj.do_body_ops = &osp_md_body_ops;
1534 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1537 o->lo_header->loh_attr |=
1538 LOHA_EXISTS | (la->la_mode & S_IFMT);
1539 if (rc == -ENOENT) {
1540 po->opo_non_exist = 1;
1543 init_rwsem(&po->opo_sem);
1548 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1550 struct osp_object *obj = lu2osp_obj(o);
1551 struct lu_object_header *h = o->lo_header;
1553 dt_object_fini(&obj->opo_obj);
1554 lu_object_header_fini(h);
1555 if (obj->opo_ooa != NULL) {
1556 struct osp_xattr_entry *oxe;
1557 struct osp_xattr_entry *tmp;
1560 list_for_each_entry_safe(oxe, tmp,
1561 &obj->opo_ooa->ooa_xattr_list,
1563 list_del(&oxe->oxe_list);
1564 count = atomic_read(&oxe->oxe_ref);
1565 LASSERTF(count == 1,
1566 "Still has %d users on the xattr entry %.*s\n",
1567 count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1569 OBD_FREE(oxe, oxe->oxe_buflen);
1571 OBD_FREE_PTR(obj->opo_ooa);
1573 OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1576 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1578 struct osp_object *po = lu2osp_obj(o);
1579 struct osp_device *d = lu2osp_dev(o->lo_dev);
1584 * release reservation if object was declared but not created
1585 * this may require lu_object_put() in LOD
1587 if (unlikely(po->opo_reserved)) {
1588 LASSERT(d->opd_pre != NULL);
1589 LASSERT(d->opd_pre_reserved > 0);
1590 spin_lock(&d->opd_pre_lock);
1591 d->opd_pre_reserved--;
1592 spin_unlock(&d->opd_pre_lock);
1594 /* not needed in cache any more */
1595 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1599 /* XXX: Currently, NOT cache OST-object on MDT because:
1600 * 1. it is not often accessed on MDT.
1601 * 2. avoid up layer (such as LFSCK) to load too many
1602 * once-used OST-objects. */
1603 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1608 static int osp_object_print(const struct lu_env *env, void *cookie,
1609 lu_printer_t p, const struct lu_object *l)
1611 const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1613 return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1616 static int osp_object_invariant(const struct lu_object *o)
1621 struct lu_object_operations osp_lu_obj_ops = {
1622 .loo_object_init = osp_object_init,
1623 .loo_object_free = osp_object_free,
1624 .loo_object_release = osp_object_release,
1625 .loo_object_print = osp_object_print,
1626 .loo_object_invariant = osp_object_invariant