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 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 = object_update_result_data_get(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 object_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 dt_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, OUT_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 dt_update_request *update;
308 struct object_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, OUT_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,
358 &RMF_OUT_UPDATE_REPLY,
359 OUT_UPDATE_REPLY_SIZE);
360 if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
361 GOTO(out, rc = -EPROTO);
363 rc = osp_get_attr_from_reply(env, reply, attr, obj, 0);
367 if (!is_ost_obj(&dt->do_lu)) {
368 if (attr->la_flags == 1)
378 ptlrpc_req_finished(req);
380 out_destroy_update_req(update);
385 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
386 const struct lu_attr *attr, struct thandle *th)
388 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
389 struct osp_object *o = dt2osp_obj(dt);
396 * Usually we don't allow server stack to manipulate size
397 * but there is a special case when striping is created
398 * late, after stripless file got truncated to non-zero.
400 * In this case we do the following:
402 * 1) grab id in declare - this can lead to leaked OST objects
403 * but we don't currently have proper mechanism and the only
404 * options we have are to do truncate RPC holding transaction
405 * open (very bad) or to grab id in declare at cost of leaked
406 * OST object in same very rare unfortunate case (just bad)
407 * notice 1.6-2.0 do assignment outside of running transaction
408 * all the time, meaning many more chances for leaked objects.
410 * 2) send synchronous truncate RPC with just assigned id
413 /* there are few places in MDD code still passing NULL
414 * XXX: to be fixed soon */
418 if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
419 fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
420 LASSERT(!dt_object_exists(dt));
421 osp_object_assign_fid(env, d, o);
422 rc = osp_object_truncate(env, dt, attr->la_size);
428 /* no need in logging for new objects being created */
432 if (!(attr->la_valid & (LA_UID | LA_GID)))
435 if (!is_only_remote_trans(th))
437 * track all UID/GID changes via llog
439 rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
441 /* It is for OST-object attr_set directly without updating
442 * local MDT-object attribute. It is usually used by LFSCK. */
443 rc = osp_md_declare_attr_set(env, dt, attr, th);
445 if (rc != 0 || o->opo_ooa == NULL)
448 la = &o->opo_ooa->ooa_attr;
449 spin_lock(&o->opo_lock);
450 if (attr->la_valid & LA_UID) {
451 la->la_uid = attr->la_uid;
452 la->la_valid |= LA_UID;
455 if (attr->la_valid & LA_GID) {
456 la->la_gid = attr->la_gid;
457 la->la_valid |= LA_GID;
459 spin_unlock(&o->opo_lock);
464 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
465 const struct lu_attr *attr, struct thandle *th,
466 struct lustre_capa *capa)
468 struct osp_object *o = dt2osp_obj(dt);
473 /* we're interested in uid/gid changes only */
474 if (!(attr->la_valid & (LA_UID | LA_GID)))
477 /* new object, the very first ->attr_set()
478 * initializing attributes needs no logging
479 * all subsequent one are subject to the
480 * logging and synchronization with OST */
486 if (!is_only_remote_trans(th))
488 * once transaction is committed put proper command on
489 * the queue going to our OST
491 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
492 /* XXX: send new uid/gid to OST ASAP? */
494 /* It is for OST-object attr_set directly without updating
495 * local MDT-object attribute. It is usually used by LFSCK. */
496 rc = osp_md_attr_set(env, dt, attr, th, capa);
501 static int osp_xattr_get_interpterer(const struct lu_env *env,
502 struct object_update_reply *reply,
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 static int osp_declare_object_create(const struct lu_env *env,
868 struct dt_object *dt,
869 struct lu_attr *attr,
870 struct dt_allocation_hint *hint,
871 struct dt_object_format *dof,
874 struct osp_thread_info *osi = osp_env_info(env);
875 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
876 struct osp_object *o = dt2osp_obj(dt);
877 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
882 if (is_only_remote_trans(th)) {
883 LASSERT(fid_is_sane(fid));
885 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
890 /* should happen to non-0 OSP only so that at least one object
891 * has been already declared in the scenario and LOD should
893 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
896 LASSERT(d->opd_last_used_oid_file);
899 * There can be gaps in precreated ids and record to unlink llog
900 * XXX: we do not handle gaps yet, implemented before solution
901 * was found to be racy, so we disabled that. there is no
902 * point in making useless but expensive llog declaration.
904 /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
906 if (unlikely(!fid_is_zero(fid))) {
907 /* replay case: caller knows fid */
908 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
909 osi->osi_lb.lb_len = sizeof(osi->osi_id);
910 osi->osi_lb.lb_buf = NULL;
911 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
912 &osi->osi_lb, osi->osi_off, th);
917 * in declaration we need to reserve object so that we don't block
918 * awaiting precreation RPC to complete
920 rc = osp_precreate_reserve(env, d);
922 * we also need to declare update to local "last used id" file for
923 * recovery if object isn't used for a reason, we need to release
924 * reservation, this can be made in osd_object_release()
927 /* mark id is reserved: in create we don't want to talk
929 LASSERT(o->opo_reserved == 0);
932 /* common for all OSPs file hystorically */
933 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
934 osi->osi_lb.lb_len = sizeof(osi->osi_id);
935 osi->osi_lb.lb_buf = NULL;
936 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
937 &osi->osi_lb, osi->osi_off, th);
939 /* not needed in the cache anymore */
940 set_bit(LU_OBJECT_HEARD_BANSHEE,
941 &dt->do_lu.lo_header->loh_flags);
946 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
947 struct lu_attr *attr,
948 struct dt_allocation_hint *hint,
949 struct dt_object_format *dof, struct thandle *th)
951 struct osp_thread_info *osi = osp_env_info(env);
952 struct osp_device *d = lu2osp_dev(dt->do_lu.lo_dev);
953 struct osp_object *o = dt2osp_obj(dt);
955 struct lu_fid *fid = &osi->osi_fid;
958 if (is_only_remote_trans(th)) {
959 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
961 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
963 o->opo_non_exist = 0;
968 o->opo_non_exist = 0;
969 if (o->opo_reserved) {
970 /* regular case, fid is assigned holding trunsaction open */
971 osp_object_assign_fid(env, d, o);
974 memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
976 LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
979 if (!o->opo_reserved) {
980 /* special case, id was assigned outside of transaction
981 * see comments in osp_declare_attr_set */
982 LASSERT(d->opd_pre != NULL);
983 spin_lock(&d->opd_pre_lock);
984 osp_update_last_fid(d, fid);
985 spin_unlock(&d->opd_pre_lock);
988 CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
990 /* If the precreate ends, it means it will be ready to rollover to
991 * the new sequence soon, all the creation should be synchronized,
992 * otherwise during replay, the replay fid will be inconsistent with
993 * last_used/create fid */
994 if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
998 * it's OK if the import is inactive by this moment - id was created
999 * by OST earlier, we just need to maintain it consistently on the disk
1000 * once import is reconnected, OSP will claim this and other objects
1001 * used and OST either keep them, if they exist or recreate
1004 /* we might have lost precreated objects */
1005 if (unlikely(d->opd_gap_count) > 0) {
1006 LASSERT(d->opd_pre != NULL);
1007 spin_lock(&d->opd_pre_lock);
1008 if (d->opd_gap_count > 0) {
1009 int count = d->opd_gap_count;
1011 ostid_set_id(&osi->osi_oi,
1012 fid_oid(&d->opd_gap_start_fid));
1013 d->opd_gap_count = 0;
1014 spin_unlock(&d->opd_pre_lock);
1016 CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1017 PFID(&d->opd_gap_start_fid), count);
1018 /* real gap handling is disabled intil ORI-692 will be
1019 * fixed, now we only report gaps */
1021 spin_unlock(&d->opd_pre_lock);
1025 /* new object, the very first ->attr_set()
1026 * initializing attributes needs no logging */
1029 /* Only need update last_used oid file, seq file will only be update
1030 * during seq rollover */
1031 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1032 &d->opd_last_used_fid.f_oid, d->opd_index);
1034 rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1037 CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1038 d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1043 int osp_declare_object_destroy(const struct lu_env *env,
1044 struct dt_object *dt, struct thandle *th)
1046 struct osp_object *o = dt2osp_obj(dt);
1052 * track objects to be destroyed via llog
1054 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1059 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1062 struct osp_object *o = dt2osp_obj(dt);
1067 o->opo_non_exist = 1;
1069 * once transaction is committed put proper command on
1070 * the queue going to our OST
1072 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1074 /* not needed in cache any more */
1075 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1080 struct osp_orphan_it {
1084 int ooi_total_npages;
1085 int ooi_valid_npages;
1086 unsigned int ooi_swab:1;
1088 struct dt_object *ooi_obj;
1089 struct lu_orphan_ent *ooi_ent;
1090 struct page *ooi_cur_page;
1091 struct lu_idxpage *ooi_cur_idxpage;
1092 struct page **ooi_pages;
1095 static int osp_orphan_index_lookup(const struct lu_env *env,
1096 struct dt_object *dt,
1098 const struct dt_key *key,
1099 struct lustre_capa *capa)
1104 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1105 struct dt_object *dt,
1106 const struct dt_rec *rec,
1107 const struct dt_key *key,
1108 struct thandle *handle)
1113 static int osp_orphan_index_insert(const struct lu_env *env,
1114 struct dt_object *dt,
1115 const struct dt_rec *rec,
1116 const struct dt_key *key,
1117 struct thandle *handle,
1118 struct lustre_capa *capa,
1124 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1125 struct dt_object *dt,
1126 const struct dt_key *key,
1127 struct thandle *handle)
1132 static int osp_orphan_index_delete(const struct lu_env *env,
1133 struct dt_object *dt,
1134 const struct dt_key *key,
1135 struct thandle *handle,
1136 struct lustre_capa *capa)
1141 static struct dt_it *osp_orphan_it_init(const struct lu_env *env,
1142 struct dt_object *dt,
1144 struct lustre_capa *capa)
1146 struct osp_orphan_it *it;
1150 return ERR_PTR(-ENOMEM);
1155 return (struct dt_it *)it;
1158 static void osp_orphan_it_fini(const struct lu_env *env,
1161 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1162 struct page **pages = it->ooi_pages;
1163 int npages = it->ooi_total_npages;
1166 if (pages != NULL) {
1167 for (i = 0; i < npages; i++) {
1168 if (pages[i] != NULL) {
1169 if (pages[i] == it->ooi_cur_page) {
1171 it->ooi_cur_page = NULL;
1173 __free_page(pages[i]);
1176 OBD_FREE(pages, npages * sizeof(*pages));
1181 static int osp_orphan_it_fetch(const struct lu_env *env,
1182 struct osp_orphan_it *it)
1184 struct lu_device *dev = it->ooi_obj->do_lu.lo_dev;
1185 struct osp_device *osp = lu2osp_dev(dev);
1186 struct page **pages;
1187 struct ptlrpc_request *req = NULL;
1188 struct ptlrpc_bulk_desc *desc;
1189 struct idx_info *ii;
1196 npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1197 npages /= PAGE_CACHE_SIZE;
1199 OBD_ALLOC(pages, npages * sizeof(*pages));
1203 it->ooi_pages = pages;
1204 it->ooi_total_npages = npages;
1205 for (i = 0; i < npages; i++) {
1206 pages[i] = alloc_page(GFP_IOFS);
1207 if (pages[i] == NULL)
1211 req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1216 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1218 ptlrpc_request_free(req);
1222 req->rq_request_portal = OST_IDX_PORTAL;
1223 ptlrpc_at_set_req_timeout(req);
1225 desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
1228 ptlrpc_request_free(req);
1232 for (i = 0; i < npages; i++)
1233 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1235 ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1236 memset(ii, 0, sizeof(*ii));
1237 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1238 ii->ii_fid.f_oid = osp->opd_index;
1239 ii->ii_fid.f_ver = 0;
1240 ii->ii_magic = IDX_INFO_MAGIC;
1241 ii->ii_flags = II_FL_NOHASH;
1242 ii->ii_count = npages * LU_PAGE_COUNT;
1243 ii->ii_hash_start = it->ooi_next;
1245 osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
1247 ptlrpc_request_set_replen(req);
1248 rc = ptlrpc_queue_wait(req);
1252 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1253 req->rq_bulk->bd_nob_transferred);
1257 ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1258 if (ii->ii_magic != IDX_INFO_MAGIC)
1259 GOTO(out, rc = -EPROTO);
1261 npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1262 (PAGE_CACHE_SHIFT - LU_PAGE_SHIFT);
1263 if (npages > it->ooi_total_npages) {
1264 CERROR("%s: returned more pages than expected, %u > %u\n",
1265 osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1266 GOTO(out, rc = -EINVAL);
1269 it->ooi_valid_npages = npages;
1270 if (ptlrpc_rep_need_swab(req))
1273 it->ooi_next = ii->ii_hash_end;
1278 ptlrpc_req_finished(req);
1283 static int osp_orphan_it_next(const struct lu_env *env,
1286 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1287 struct lu_idxpage *idxpage;
1288 struct page **pages;
1294 idxpage = it->ooi_cur_idxpage;
1295 if (idxpage != NULL) {
1296 if (idxpage->lip_nr == 0)
1300 if (it->ooi_pos2 < idxpage->lip_nr) {
1302 (struct lu_orphan_ent *)idxpage->lip_entries +
1305 lustre_swab_orphan_ent(it->ooi_ent);
1309 it->ooi_cur_idxpage = NULL;
1313 if (it->ooi_pos1 < LU_PAGE_COUNT) {
1314 it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1315 LU_PAGE_SIZE * it->ooi_pos1;
1317 lustre_swab_lip_header(it->ooi_cur_idxpage);
1318 if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1319 struct osp_device *osp =
1320 lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1322 CERROR("%s: invalid magic (%x != %x) for page "
1323 "%d/%d while read layout orphan index\n",
1324 osp->opd_obd->obd_name,
1325 it->ooi_cur_idxpage->lip_magic,
1326 LIP_MAGIC, it->ooi_pos0, it->ooi_pos1);
1327 /* Skip this lu_page next time. */
1328 it->ooi_pos2 = idxpage->lip_nr - 1;
1335 kunmap(it->ooi_cur_page);
1336 it->ooi_cur_page = NULL;
1340 pages = it->ooi_pages;
1341 if (it->ooi_pos0 < it->ooi_valid_npages) {
1342 it->ooi_cur_page = kmap(pages[it->ooi_pos0]);
1347 for (i = 0; i < it->ooi_total_npages; i++) {
1348 if (pages[i] != NULL)
1349 __free_page(pages[i]);
1351 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1354 it->ooi_total_npages = 0;
1355 it->ooi_valid_npages = 0;
1358 it->ooi_cur_page = NULL;
1359 it->ooi_cur_idxpage = NULL;
1360 it->ooi_pages = NULL;
1363 if (it->ooi_next == II_END_OFF)
1366 rc = osp_orphan_it_fetch(env, it);
1373 static int osp_orphan_it_get(const struct lu_env *env,
1375 const struct dt_key *key)
1380 static void osp_orphan_it_put(const struct lu_env *env,
1385 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1386 const struct dt_it *di)
1388 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1389 struct lu_orphan_ent *ent = it->ooi_ent;
1391 if (likely(ent != NULL))
1392 return (struct dt_key *)(&ent->loe_key);
1397 static int osp_orphan_it_key_size(const struct lu_env *env,
1398 const struct dt_it *di)
1400 return sizeof(struct lu_fid);
1403 static int osp_orphan_it_rec(const struct lu_env *env,
1404 const struct dt_it *di,
1408 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1409 struct lu_orphan_ent *ent = it->ooi_ent;
1411 if (likely(ent != NULL)) {
1412 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1419 static __u64 osp_orphan_it_store(const struct lu_env *env,
1420 const struct dt_it *di)
1422 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1424 return it->ooi_next;
1428 * \retval +1: locate to the exactly position
1429 * \retval 0: cannot locate to the exactly position,
1430 * call next() to move to a valid position.
1431 * \retval -ve: on error
1433 static int osp_orphan_it_load(const struct lu_env *env,
1434 const struct dt_it *di,
1437 struct osp_orphan_it *it = (struct osp_orphan_it *)di;
1440 it->ooi_next = hash;
1441 rc = osp_orphan_it_next(env, (struct dt_it *)di);
1451 static int osp_orphan_it_key_rec(const struct lu_env *env,
1452 const struct dt_it *di,
1458 static const struct dt_index_operations osp_orphan_index_ops = {
1459 .dio_lookup = osp_orphan_index_lookup,
1460 .dio_declare_insert = osp_orphan_index_declare_insert,
1461 .dio_insert = osp_orphan_index_insert,
1462 .dio_declare_delete = osp_orphan_index_declare_delete,
1463 .dio_delete = osp_orphan_index_delete,
1465 .init = osp_orphan_it_init,
1466 .fini = osp_orphan_it_fini,
1467 .next = osp_orphan_it_next,
1468 .get = osp_orphan_it_get,
1469 .put = osp_orphan_it_put,
1470 .key = osp_orphan_it_key,
1471 .key_size = osp_orphan_it_key_size,
1472 .rec = osp_orphan_it_rec,
1473 .store = osp_orphan_it_store,
1474 .load = osp_orphan_it_load,
1475 .key_rec = osp_orphan_it_key_rec,
1479 static int osp_index_try(const struct lu_env *env,
1480 struct dt_object *dt,
1481 const struct dt_index_features *feat)
1483 const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1485 if (fid_is_last_id(fid) && fid_is_idif(fid)) {
1486 dt->do_index_ops = &osp_orphan_index_ops;
1494 struct dt_object_operations osp_obj_ops = {
1495 .do_declare_attr_get = osp_declare_attr_get,
1496 .do_attr_get = osp_attr_get,
1497 .do_declare_attr_set = osp_declare_attr_set,
1498 .do_attr_set = osp_attr_set,
1499 .do_declare_xattr_get = osp_declare_xattr_get,
1500 .do_xattr_get = osp_xattr_get,
1501 .do_declare_xattr_set = osp_declare_xattr_set,
1502 .do_xattr_set = osp_xattr_set,
1503 .do_declare_create = osp_declare_object_create,
1504 .do_create = osp_object_create,
1505 .do_declare_destroy = osp_declare_object_destroy,
1506 .do_destroy = osp_object_destroy,
1507 .do_index_try = osp_index_try,
1510 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1511 const struct lu_object_conf *conf)
1513 struct osp_object *po = lu2osp_obj(o);
1517 spin_lock_init(&po->opo_lock);
1518 o->lo_header->loh_attr |= LOHA_REMOTE;
1520 if (is_ost_obj(o)) {
1521 po->opo_obj.do_ops = &osp_obj_ops;
1523 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1525 po->opo_obj.do_ops = &osp_md_obj_ops;
1526 po->opo_obj.do_body_ops = &osp_md_body_ops;
1527 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1530 o->lo_header->loh_attr |=
1531 LOHA_EXISTS | (la->la_mode & S_IFMT);
1532 if (rc == -ENOENT) {
1533 po->opo_non_exist = 1;
1536 init_rwsem(&po->opo_sem);
1541 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1543 struct osp_object *obj = lu2osp_obj(o);
1544 struct lu_object_header *h = o->lo_header;
1546 dt_object_fini(&obj->opo_obj);
1547 lu_object_header_fini(h);
1548 if (obj->opo_ooa != NULL) {
1549 struct osp_xattr_entry *oxe;
1550 struct osp_xattr_entry *tmp;
1553 list_for_each_entry_safe(oxe, tmp,
1554 &obj->opo_ooa->ooa_xattr_list,
1556 list_del(&oxe->oxe_list);
1557 count = atomic_read(&oxe->oxe_ref);
1558 LASSERTF(count == 1,
1559 "Still has %d users on the xattr entry %.*s\n",
1560 count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1562 OBD_FREE(oxe, oxe->oxe_buflen);
1564 OBD_FREE_PTR(obj->opo_ooa);
1566 OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1569 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1571 struct osp_object *po = lu2osp_obj(o);
1572 struct osp_device *d = lu2osp_dev(o->lo_dev);
1577 * release reservation if object was declared but not created
1578 * this may require lu_object_put() in LOD
1580 if (unlikely(po->opo_reserved)) {
1581 LASSERT(d->opd_pre != NULL);
1582 LASSERT(d->opd_pre_reserved > 0);
1583 spin_lock(&d->opd_pre_lock);
1584 d->opd_pre_reserved--;
1585 spin_unlock(&d->opd_pre_lock);
1587 /* not needed in cache any more */
1588 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1592 /* XXX: Currently, NOT cache OST-object on MDT because:
1593 * 1. it is not often accessed on MDT.
1594 * 2. avoid up layer (such as LFSCK) to load too many
1595 * once-used OST-objects. */
1596 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1601 static int osp_object_print(const struct lu_env *env, void *cookie,
1602 lu_printer_t p, const struct lu_object *l)
1604 const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1606 return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1609 static int osp_object_invariant(const struct lu_object *o)
1614 struct lu_object_operations osp_lu_obj_ops = {
1615 .loo_object_init = osp_object_init,
1616 .loo_object_free = osp_object_free,
1617 .loo_object_release = osp_object_release,
1618 .loo_object_print = osp_object_print,
1619 .loo_object_invariant = osp_object_invariant