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.gnu.org/licenses/gpl-2.0.html
23 * Copyright (c) 2013, Intel Corporation.
26 * lustre/osp/osp_md_object.c
28 * Lustre MDT Proxy Device
30 * Author: Di Wang <di.wang@intel.com>
34 # define EXPORT_SYMTAB
36 #define DEBUG_SUBSYSTEM S_MDS
38 #include <lustre_log.h>
39 #include <lustre_update.h>
40 #include "osp_internal.h"
41 static const char dot[] = ".";
42 static const char dotdot[] = "..";
44 static int osp_prep_update_req(const struct lu_env *env,
45 struct osp_device *osp,
46 struct update_buf *ubuf, int ubuf_len,
47 struct ptlrpc_request **reqp)
49 struct obd_import *imp;
50 struct ptlrpc_request *req;
51 struct update_buf *tmp;
55 imp = osp->opd_obd->u.cli.cl_import;
58 req = ptlrpc_request_alloc(imp, &RQF_UPDATE_OBJ);
62 req_capsule_set_size(&req->rq_pill, &RMF_UPDATE, RCL_CLIENT,
65 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, UPDATE_OBJ);
67 ptlrpc_req_finished(req);
71 req_capsule_set_size(&req->rq_pill, &RMF_UPDATE_REPLY, RCL_SERVER,
74 tmp = req_capsule_client_get(&req->rq_pill, &RMF_UPDATE);
75 memcpy(tmp, ubuf, ubuf_len);
77 ptlrpc_request_set_replen(req);
84 static int osp_remote_sync(const struct lu_env *env, struct dt_device *dt,
85 struct update_request *update,
86 struct ptlrpc_request **reqp)
88 struct osp_device *osp = dt2osp_dev(dt);
89 struct ptlrpc_request *req = NULL;
93 rc = osp_prep_update_req(env, osp, update->ur_buf,
94 UPDATE_BUFFER_SIZE, &req);
98 /* Note: some dt index api might return non-zero result here, like
99 * osd_index_ea_lookup, so we should only check rc < 0 here */
100 rc = ptlrpc_queue_wait(req);
102 ptlrpc_req_finished(req);
114 ptlrpc_req_finished(req);
120 * Create a new update request for the device.
122 static struct update_request
123 *osp_create_update_req(struct dt_device *dt)
125 struct update_request *update;
127 OBD_ALLOC_PTR(update);
129 return ERR_PTR(-ENOMEM);
131 OBD_ALLOC_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
132 if (update->ur_buf == NULL) {
133 OBD_FREE_PTR(update);
134 return ERR_PTR(-ENOMEM);
137 CFS_INIT_LIST_HEAD(&update->ur_list);
140 update->ur_buf->ub_magic = UPDATE_BUFFER_MAGIC;
141 update->ur_buf->ub_count = 0;
146 static void osp_destroy_update_req(struct update_request *update)
151 cfs_list_del(&update->ur_list);
152 if (update->ur_buf != NULL)
153 OBD_FREE_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
155 OBD_FREE_PTR(update);
159 int osp_trans_stop(const struct lu_env *env, struct thandle *th)
163 osp_destroy_update_req(th->th_current_request);
164 rc = th->th_current_request->ur_rc;
165 th->th_current_request = NULL;
171 * In DNE phase I, all remote updates will be packed into RPC (the format
172 * description is in lustre_idl.h) during declare phase, all of updates
173 * are attached to the transaction, one entry per OSP. Then in trans start,
174 * LOD will walk through these entries and send these UPDATEs to the remote
175 * MDT to be executed synchronously.
177 int osp_trans_start(const struct lu_env *env, struct dt_device *dt,
180 struct update_request *update;
183 /* In phase I, if the transaction includes remote updates, the local
184 * update should be synchronized, so it will set th_sync = 1 */
185 update = th->th_current_request;
186 LASSERT(update != NULL && update->ur_dt == dt);
187 if (update->ur_buf->ub_count > 0) {
188 rc = osp_remote_sync(env, dt, update, NULL);
196 * Insert the update into the th_bufs for the device.
198 static int osp_insert_update(const struct lu_env *env,
199 struct update_request *update, int op,
200 struct lu_fid *fid, int count, int *lens,
203 struct update_buf *ubuf = update->ur_buf;
204 struct update *obj_update;
211 obj_update = (struct update *)((char *)ubuf +
212 cfs_size_round(update_buf_size(ubuf)));
214 /* Check update size to make sure it can fit into the buffer */
215 update_length = cfs_size_round(offsetof(struct update,
217 for (i = 0; i < count; i++)
218 update_length += cfs_size_round(lens[i]);
220 if (cfs_size_round(update_buf_size(ubuf)) + update_length >
221 UPDATE_BUFFER_SIZE || ubuf->ub_count >= UPDATE_MAX_OPS) {
222 CERROR("%s: insert up %p, idx %d cnt %d len %lu: rc = %d\n",
223 update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf,
224 update_length, ubuf->ub_count, update_buf_size(ubuf),
229 if (count > UPDATE_BUF_COUNT) {
230 CERROR("%s: Insert too much params %d "DFID" op %d: rc = %d\n",
231 update->ur_dt->dd_lu_dev.ld_obd->obd_name, count,
232 PFID(fid), op, -E2BIG);
236 /* fill the update into the update buffer */
237 fid_cpu_to_le(&obj_update->u_fid, fid);
238 obj_update->u_type = cpu_to_le32(op);
239 for (i = 0; i < count; i++)
240 obj_update->u_lens[i] = cpu_to_le32(lens[i]);
242 ptr = (char *)obj_update +
243 cfs_size_round(offsetof(struct update, u_bufs[0]));
244 for (i = 0; i < count; i++)
245 LOGL(bufs[i], lens[i], ptr);
249 CDEBUG(D_INFO, "%s: %p "DFID" idx %d: op %d params %d:%lu\n",
250 update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf, PFID(fid),
251 ubuf->ub_count, op, count, update_buf_size(ubuf));
256 static struct update_request
257 *osp_find_update(struct thandle *th, struct dt_device *dt_dev)
259 struct update_request *update;
261 /* Because transaction api does not proivde the interface
262 * to transfer the update from LOD to OSP, we need walk
263 * remote update list to find the update, this probably
264 * should move to LOD layer, when update can be part of
265 * the trancation api parameter. XXX */
266 cfs_list_for_each_entry(update, &th->th_remote_update_list, ur_list) {
267 if (update->ur_dt == dt_dev)
274 * Find one loc in th_dev/dev_obj_update for the update,
275 * Because only one thread can access this thandle, no need
278 static struct update_request
279 *osp_find_create_update_loc(struct thandle *th, struct dt_object *dt)
281 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
282 struct update_request *update;
285 update = osp_find_update(th, dt_dev);
289 update = osp_create_update_req(dt_dev);
293 cfs_list_add_tail(&update->ur_list, &th->th_remote_update_list);
298 static int osp_get_attr_from_req(const struct lu_env *env,
299 struct ptlrpc_request *req,
300 struct lu_attr *attr, int index)
302 struct update_reply *reply;
303 struct obdo *lobdo = &osp_env_info(env)->osi_obdo;
307 LASSERT(attr != NULL);
309 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
311 if (reply->ur_version != UPDATE_REPLY_V1)
314 size = update_get_reply_buf(reply, (void **)&wobdo, index);
315 if (size != sizeof(struct obdo))
318 obdo_le_to_cpu(wobdo, wobdo);
319 lustre_get_wire_obdo(lobdo, wobdo);
320 la_from_obdo(attr, lobdo, lobdo->o_valid);
325 static int osp_md_declare_object_create(const struct lu_env *env,
326 struct dt_object *dt,
327 struct lu_attr *attr,
328 struct dt_allocation_hint *hint,
329 struct dt_object_format *dof,
332 struct osp_thread_info *osi = osp_env_info(env);
333 struct update_request *update;
335 int sizes[2] = {sizeof(struct obdo), 0};
336 char *bufs[2] = {NULL, NULL};
340 update = osp_find_create_update_loc(th, dt);
341 if (IS_ERR(update)) {
342 CERROR("%s: Get OSP update buf failed: rc = %d\n",
343 dt->do_lu.lo_dev->ld_obd->obd_name,
344 (int)PTR_ERR(update));
345 return PTR_ERR(update);
348 osi->osi_obdo.o_valid = 0;
349 LASSERT(S_ISDIR(attr->la_mode));
350 obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
351 lustre_set_wire_obdo(&osi->osi_obdo, &osi->osi_obdo);
352 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
354 bufs[0] = (char *)&osi->osi_obdo;
356 fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
357 if (hint->dah_parent) {
359 struct lu_fid *tmp_fid = &osi->osi_fid;
361 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
362 fid_cpu_to_le(tmp_fid, fid2);
363 sizes[1] = sizeof(*tmp_fid);
364 bufs[1] = (char *)tmp_fid;
368 rc = osp_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes,
371 CERROR("%s: Insert update error: rc = %d\n",
372 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
378 static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
379 struct lu_attr *attr,
380 struct dt_allocation_hint *hint,
381 struct dt_object_format *dof,
384 struct osp_object *obj = dt2osp_obj(dt);
386 CDEBUG(D_INFO, "create object "DFID"\n",
387 PFID(&dt->do_lu.lo_header->loh_fid));
389 /* Because the create update RPC will be sent during declare phase,
390 * if creation reaches here, it means the object has been created
392 dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
398 static int osp_md_declare_object_ref_del(const struct lu_env *env,
399 struct dt_object *dt,
402 struct update_request *update;
406 update = osp_find_create_update_loc(th, dt);
407 if (IS_ERR(update)) {
408 CERROR("%s: Get OSP update buf failed: rc = %d\n",
409 dt->do_lu.lo_dev->ld_obd->obd_name,
410 (int)PTR_ERR(update));
411 return PTR_ERR(update);
414 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
416 rc = osp_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL);
421 static int osp_md_object_ref_del(const struct lu_env *env,
422 struct dt_object *dt,
425 CDEBUG(D_INFO, "ref del object "DFID"\n",
426 PFID(&dt->do_lu.lo_header->loh_fid));
431 static int osp_md_declare_ref_add(const struct lu_env *env,
432 struct dt_object *dt, struct thandle *th)
434 struct update_request *update;
438 update = osp_find_create_update_loc(th, dt);
439 if (IS_ERR(update)) {
440 CERROR("%s: Get OSP update buf failed: rc = %d\n",
441 dt->do_lu.lo_dev->ld_obd->obd_name,
442 (int)PTR_ERR(update));
443 return PTR_ERR(update);
446 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
448 rc = osp_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL);
453 static int osp_md_object_ref_add(const struct lu_env *env,
454 struct dt_object *dt,
457 CDEBUG(D_INFO, "ref add object "DFID"\n",
458 PFID(&dt->do_lu.lo_header->loh_fid));
463 static void osp_md_ah_init(const struct lu_env *env,
464 struct dt_allocation_hint *ah,
465 struct dt_object *parent,
466 struct dt_object *child,
467 cfs_umode_t child_mode)
471 memset(ah, 0, sizeof(*ah));
472 ah->dah_parent = parent;
473 ah->dah_mode = child_mode;
476 static int osp_md_declare_attr_set(const struct lu_env *env,
477 struct dt_object *dt,
478 const struct lu_attr *attr,
481 struct osp_thread_info *osi = osp_env_info(env);
482 struct update_request *update;
484 int size = sizeof(struct obdo);
488 update = osp_find_create_update_loc(th, dt);
489 if (IS_ERR(update)) {
490 CERROR("%s: Get OSP update buf failed: %d\n",
491 dt->do_lu.lo_dev->ld_obd->obd_name,
492 (int)PTR_ERR(update));
493 return PTR_ERR(update);
496 osi->osi_obdo.o_valid = 0;
497 LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE)));
498 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
500 lustre_set_wire_obdo(&osi->osi_obdo, &osi->osi_obdo);
501 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
503 buf = (char *)&osi->osi_obdo;
504 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
506 rc = osp_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size, &buf);
511 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
512 const struct lu_attr *attr, struct thandle *th,
513 struct lustre_capa *capa)
515 CDEBUG(D_INFO, "attr set object "DFID"\n",
516 PFID(&dt->do_lu.lo_header->loh_fid));
521 static int osp_md_declare_xattr_set(const struct lu_env *env,
522 struct dt_object *dt,
523 const struct lu_buf *buf,
524 const char *name, int flag,
527 struct update_request *update;
529 int sizes[3] = {strlen(name), buf->lb_len,
531 char *bufs[3] = {(char *)name, (char *)buf->lb_buf };
534 update = osp_find_create_update_loc(th, dt);
535 if (IS_ERR(update)) {
536 CERROR("%s: Get OSP update buf failed: rc = %d\n",
537 dt->do_lu.lo_dev->ld_obd->obd_name,
538 (int)PTR_ERR(update));
539 return PTR_ERR(update);
542 flag = cpu_to_le32(flag);
543 bufs[2] = (char *)&flag;
545 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
546 rc = osp_insert_update(env, update, OBJ_XATTR_SET, fid,
547 ARRAY_SIZE(sizes), sizes, bufs);
552 static int osp_md_xattr_set(const struct lu_env *env, struct dt_object *dt,
553 const struct lu_buf *buf, const char *name, int fl,
554 struct thandle *th, struct lustre_capa *capa)
556 CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
557 PFID(&dt->do_lu.lo_header->loh_fid));
562 static int osp_md_xattr_get(const struct lu_env *env, struct dt_object *dt,
563 struct lu_buf *buf, const char *name,
564 struct lustre_capa *capa)
566 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
567 struct update_request *update = NULL;
568 struct ptlrpc_request *req = NULL;
572 struct update_reply *reply;
576 /* Because it needs send the update buffer right away,
577 * just create an update buffer, instead of attaching the
578 * update_remote list of the thandle.
580 update = osp_create_update_req(dt_dev);
582 RETURN(PTR_ERR(update));
584 LASSERT(name != NULL);
585 buf_len = strlen(name);
586 rc = osp_insert_update(env, update, OBJ_XATTR_GET,
587 (struct lu_fid *)lu_object_fid(&dt->do_lu),
588 1, &buf_len, (char **)&name);
590 CERROR("%s: Insert update error: rc = %d\n",
591 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
594 dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
596 rc = osp_remote_sync(env, dt_dev, update, &req);
600 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
602 if (reply->ur_version != UPDATE_REPLY_V1) {
603 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
604 dt_dev->dd_lu_dev.ld_obd->obd_name,
605 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
606 GOTO(out, rc = -EPROTO);
609 size = update_get_reply_buf(reply, &ea_buf, 0);
611 GOTO(out, rc = size);
613 LASSERT(size > 0 && size < CFS_PAGE_SIZE);
614 LASSERT(ea_buf != NULL);
617 memcpy(buf->lb_buf, ea_buf, size);
620 ptlrpc_req_finished(req);
623 osp_destroy_update_req(update);
628 static void osp_md_object_read_lock(const struct lu_env *env,
629 struct dt_object *dt, unsigned role)
631 struct osp_object *obj = dt2osp_obj(dt);
633 LASSERT(obj->opo_owner != env);
634 down_read_nested(&obj->opo_sem, role);
636 LASSERT(obj->opo_owner == NULL);
639 static void osp_md_object_write_lock(const struct lu_env *env,
640 struct dt_object *dt, unsigned role)
642 struct osp_object *obj = dt2osp_obj(dt);
644 down_write_nested(&obj->opo_sem, role);
646 LASSERT(obj->opo_owner == NULL);
647 obj->opo_owner = env;
650 static void osp_md_object_read_unlock(const struct lu_env *env,
651 struct dt_object *dt)
653 struct osp_object *obj = dt2osp_obj(dt);
655 up_read(&obj->opo_sem);
658 static void osp_md_object_write_unlock(const struct lu_env *env,
659 struct dt_object *dt)
661 struct osp_object *obj = dt2osp_obj(dt);
663 LASSERT(obj->opo_owner == env);
664 obj->opo_owner = NULL;
665 up_write(&obj->opo_sem);
668 static int osp_md_object_write_locked(const struct lu_env *env,
669 struct dt_object *dt)
671 struct osp_object *obj = dt2osp_obj(dt);
673 return obj->opo_owner == env;
676 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
677 struct dt_rec *rec, const struct dt_key *key,
678 struct lustre_capa *capa)
680 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
681 struct update_request *update;
682 struct ptlrpc_request *req = NULL;
683 int size = strlen((char *)key) + 1;
684 char *name = (char *)key;
686 struct update_reply *reply;
691 /* Because it needs send the update buffer right away,
692 * just create an update buffer, instead of attaching the
693 * update_remote list of the thandle.
695 update = osp_create_update_req(dt_dev);
697 RETURN(PTR_ERR(update));
699 rc = osp_insert_update(env, update, OBJ_INDEX_LOOKUP,
700 (struct lu_fid *)lu_object_fid(&dt->do_lu),
701 1, &size, (char **)&name);
703 CERROR("%s: Insert update error: rc = %d\n",
704 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
708 rc = osp_remote_sync(env, dt_dev, update, &req);
710 CERROR("%s: lookup "DFID" %s failed: rc = %d\n",
711 dt_dev->dd_lu_dev.ld_obd->obd_name,
712 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
716 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
718 if (reply->ur_version != UPDATE_REPLY_V1) {
719 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
720 dt_dev->dd_lu_dev.ld_obd->obd_name,
721 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
722 GOTO(out, rc = -EPROTO);
725 rc = update_get_reply_result(reply, NULL, 0);
727 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
728 dt_dev->dd_lu_dev.ld_obd->obd_name,
729 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
733 size = update_get_reply_buf(reply, (void **)&fid, 0);
735 GOTO(out, rc = size);
737 if (size != sizeof(struct lu_fid)) {
738 CERROR("%s: lookup "DFID" %s wrong size %d: rc = %d\n",
739 dt_dev->dd_lu_dev.ld_obd->obd_name,
740 PFID(lu_object_fid(&dt->do_lu)), (char *)key, size, rc);
741 GOTO(out, rc = -EINVAL);
744 fid_le_to_cpu(fid, fid);
745 if (!fid_is_sane(fid)) {
746 CERROR("%s: lookup "DFID" %s invalid fid "DFID": rc = %d\n",
747 dt_dev->dd_lu_dev.ld_obd->obd_name,
748 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid),
750 GOTO(out, rc = -EINVAL);
752 memcpy(rec, fid, sizeof(*fid));
755 ptlrpc_req_finished(req);
758 osp_destroy_update_req(update);
763 static int osp_md_declare_insert(const struct lu_env *env,
764 struct dt_object *dt,
765 const struct dt_rec *rec,
766 const struct dt_key *key,
769 struct update_request *update;
771 struct lu_fid *rec_fid = (struct lu_fid *)rec;
772 int size[2] = {strlen((char *)key) + 1,
774 char *bufs[2] = {(char *)key, (char *)rec_fid};
777 update = osp_find_create_update_loc(th, dt);
778 if (IS_ERR(update)) {
779 CERROR("%s: Get OSP update buf failed: rc = %d\n",
780 dt->do_lu.lo_dev->ld_obd->obd_name,
781 (int)PTR_ERR(update));
782 return PTR_ERR(update);
785 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
787 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
788 dt->do_lu.lo_dev->ld_obd->obd_name,
789 PFID(fid), (char *)key, PFID(rec_fid));
791 fid_cpu_to_le(rec_fid, rec_fid);
793 rc = osp_insert_update(env, update, OBJ_INDEX_INSERT, fid,
794 ARRAY_SIZE(size), size, bufs);
798 static int osp_md_index_insert(const struct lu_env *env,
799 struct dt_object *dt,
800 const struct dt_rec *rec,
801 const struct dt_key *key,
803 struct lustre_capa *capa,
809 static int osp_md_declare_delete(const struct lu_env *env,
810 struct dt_object *dt,
811 const struct dt_key *key,
814 struct update_request *update;
816 int size = strlen((char *)key) + 1;
817 char *buf = (char *)key;
820 update = osp_find_create_update_loc(th, dt);
821 if (IS_ERR(update)) {
822 CERROR("%s: Get OSP update buf failed: rc = %d\n",
823 dt->do_lu.lo_dev->ld_obd->obd_name,
824 (int)PTR_ERR(update));
825 return PTR_ERR(update);
828 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
830 rc = osp_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
836 static int osp_md_index_delete(const struct lu_env *env,
837 struct dt_object *dt,
838 const struct dt_key *key,
840 struct lustre_capa *capa)
842 CDEBUG(D_INFO, "index delete "DFID" %s\n",
843 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
849 * Creates or initializes iterator context.
851 * Note: for OSP, these index iterate api is only used to check
852 * whether the directory is empty now (see mdd_dir_is_empty).
853 * Since dir_empty will be return by OBJ_ATTR_GET(see osp_md_attr_get/
854 * out_attr_get). So the implementation of these iterator is simplied
855 * to make mdd_dir_is_empty happy. The real iterator should be
856 * implemented, if we need it one day.
858 static struct dt_it *osp_it_init(const struct lu_env *env,
859 struct dt_object *dt,
861 struct lustre_capa *capa)
863 lu_object_get(&dt->do_lu);
864 return (struct dt_it *)dt;
867 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
869 struct dt_object *dt = (struct dt_object *)di;
870 lu_object_put(env, &dt->do_lu);
873 static int osp_it_get(const struct lu_env *env,
874 struct dt_it *di, const struct dt_key *key)
879 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
884 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
886 struct dt_object *dt = (struct dt_object *)di;
887 struct osp_object *osp_obj = dt2osp_obj(dt);
888 if (osp_obj->opo_empty)
893 static struct dt_key *osp_it_key(const struct lu_env *env,
894 const struct dt_it *di)
900 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
906 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
907 struct dt_rec *lde, __u32 attr)
913 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
919 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
926 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
933 static const struct dt_index_operations osp_md_index_ops = {
934 .dio_lookup = osp_md_index_lookup,
935 .dio_declare_insert = osp_md_declare_insert,
936 .dio_insert = osp_md_index_insert,
937 .dio_declare_delete = osp_md_declare_delete,
938 .dio_delete = osp_md_index_delete,
946 .key_size = osp_it_key_size,
948 .store = osp_it_store,
950 .key_rec = osp_it_key_rec,
954 static int osp_md_index_try(const struct lu_env *env,
955 struct dt_object *dt,
956 const struct dt_index_features *feat)
958 dt->do_index_ops = &osp_md_index_ops;
962 static int osp_md_attr_get(const struct lu_env *env,
963 struct dt_object *dt, struct lu_attr *attr,
964 struct lustre_capa *capa)
966 struct osp_object *obj = dt2osp_obj(dt);
967 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
968 struct update_request *update = NULL;
969 struct ptlrpc_request *req = NULL;
973 /* Because it needs send the update buffer right away,
974 * just create an update buffer, instead of attaching the
975 * update_remote list of the thandle.
977 update = osp_create_update_req(dt_dev);
979 RETURN(PTR_ERR(update));
981 rc = osp_insert_update(env, update, OBJ_ATTR_GET,
982 (struct lu_fid *)lu_object_fid(&dt->do_lu),
985 CERROR("%s: Insert update error: rc = %d\n",
986 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
989 dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
991 rc = osp_remote_sync(env, dt_dev, update, &req);
995 rc = osp_get_attr_from_req(env, req, attr, 0);
999 if (attr->la_flags == 1)
1005 ptlrpc_req_finished(req);
1008 osp_destroy_update_req(update);
1013 static int osp_md_declare_object_destroy(const struct lu_env *env,
1014 struct dt_object *dt,
1017 struct osp_object *o = dt2osp_obj(dt);
1022 * track objects to be destroyed via llog
1024 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1029 static int osp_md_object_destroy(const struct lu_env *env,
1030 struct dt_object *dt, struct thandle *th)
1032 struct osp_object *o = dt2osp_obj(dt);
1037 * once transaction is committed put proper command on
1038 * the queue going to our OST
1040 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1042 /* not needed in cache any more */
1043 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1048 struct dt_object_operations osp_md_obj_ops = {
1049 .do_read_lock = osp_md_object_read_lock,
1050 .do_write_lock = osp_md_object_write_lock,
1051 .do_read_unlock = osp_md_object_read_unlock,
1052 .do_write_unlock = osp_md_object_write_unlock,
1053 .do_write_locked = osp_md_object_write_locked,
1054 .do_declare_create = osp_md_declare_object_create,
1055 .do_create = osp_md_object_create,
1056 .do_declare_ref_add = osp_md_declare_ref_add,
1057 .do_ref_add = osp_md_object_ref_add,
1058 .do_declare_ref_del = osp_md_declare_object_ref_del,
1059 .do_ref_del = osp_md_object_ref_del,
1060 .do_declare_destroy = osp_md_declare_object_destroy,
1061 .do_destroy = osp_md_object_destroy,
1062 .do_ah_init = osp_md_ah_init,
1063 .do_attr_get = osp_md_attr_get,
1064 .do_declare_attr_set = osp_md_declare_attr_set,
1065 .do_attr_set = osp_md_attr_set,
1066 .do_declare_xattr_set = osp_md_declare_xattr_set,
1067 .do_xattr_set = osp_md_xattr_set,
1068 .do_xattr_get = osp_md_xattr_get,
1069 .do_index_try = osp_md_index_try,
1072 static int osp_md_object_lock(const struct lu_env *env,
1073 struct dt_object *dt,
1074 struct lustre_handle *lh,
1075 struct ldlm_enqueue_info *einfo,
1078 struct osp_thread_info *info = osp_env_info(env);
1079 struct ldlm_res_id *res_id = &info->osi_resid;
1080 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1081 struct osp_device *osp = dt2osp_dev(dt_dev);
1082 struct ptlrpc_request *req = NULL;
1087 fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
1089 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
1090 LDLM_FL_BLOCK_GRANTED, res_id,
1092 (ldlm_policy_data_t *)policy,
1093 einfo->ei_mode, lh, 0);
1097 req = ldlm_enqueue_pack(osp->opd_exp, 0);
1099 RETURN(PTR_ERR(req));
1101 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
1102 (const ldlm_policy_data_t *)policy,
1103 &flags, NULL, 0, LVB_T_NONE, lh, 0);
1105 ptlrpc_req_finished(req);
1107 return rc == ELDLM_OK ? 0 : -EIO;
1110 struct dt_lock_operations osp_md_lock_ops = {
1111 .do_object_lock = osp_md_object_lock,