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>
33 #define DEBUG_SUBSYSTEM S_MDS
35 #include <lustre_log.h>
36 #include <lustre_update.h>
37 #include "osp_internal.h"
38 static const char dot[] = ".";
39 static const char dotdot[] = "..";
41 static int osp_prep_update_req(const struct lu_env *env,
42 struct osp_device *osp,
43 struct update_buf *ubuf, int ubuf_len,
44 struct ptlrpc_request **reqp)
46 struct obd_import *imp;
47 struct ptlrpc_request *req;
48 struct update_buf *tmp;
52 imp = osp->opd_obd->u.cli.cl_import;
55 req = ptlrpc_request_alloc(imp, &RQF_UPDATE_OBJ);
59 req_capsule_set_size(&req->rq_pill, &RMF_UPDATE, RCL_CLIENT,
62 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, UPDATE_OBJ);
64 ptlrpc_req_finished(req);
68 req_capsule_set_size(&req->rq_pill, &RMF_UPDATE_REPLY, RCL_SERVER,
71 tmp = req_capsule_client_get(&req->rq_pill, &RMF_UPDATE);
72 memcpy(tmp, ubuf, ubuf_len);
74 ptlrpc_request_set_replen(req);
81 static int osp_remote_sync(const struct lu_env *env, struct dt_device *dt,
82 struct update_request *update,
83 struct ptlrpc_request **reqp)
85 struct osp_device *osp = dt2osp_dev(dt);
86 struct ptlrpc_request *req = NULL;
90 rc = osp_prep_update_req(env, osp, update->ur_buf, UPDATE_BUFFER_SIZE,
95 /* Note: some dt index api might return non-zero result here, like
96 * osd_index_ea_lookup, so we should only check rc < 0 here */
97 rc = ptlrpc_queue_wait(req);
99 ptlrpc_req_finished(req);
111 ptlrpc_req_finished(req);
117 * Create a new update request for the device.
119 static struct update_request
120 *osp_create_update_req(struct dt_device *dt)
122 struct update_request *update;
124 OBD_ALLOC_PTR(update);
126 return ERR_PTR(-ENOMEM);
128 OBD_ALLOC_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
129 if (update->ur_buf == NULL) {
130 OBD_FREE_PTR(update);
131 return ERR_PTR(-ENOMEM);
134 CFS_INIT_LIST_HEAD(&update->ur_list);
136 update->ur_batchid = 0;
137 update->ur_buf->ub_magic = UPDATE_BUFFER_MAGIC;
138 update->ur_buf->ub_count = 0;
143 static void osp_destroy_update_req(struct update_request *update)
148 cfs_list_del(&update->ur_list);
149 if (update->ur_buf != NULL)
150 OBD_FREE_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
152 OBD_FREE_PTR(update);
156 int osp_trans_stop(const struct lu_env *env, struct thandle *th)
160 rc = th->th_current_request->ur_rc;
161 osp_destroy_update_req(th->th_current_request);
162 th->th_current_request = NULL;
168 * In DNE phase I, all remote updates will be packed into RPC (the format
169 * description is in lustre_idl.h) during declare phase, all of updates
170 * are attached to the transaction, one entry per OSP. Then in trans start,
171 * LOD will walk through these entries and send these UPDATEs to the remote
172 * MDT to be executed synchronously.
174 int osp_trans_start(const struct lu_env *env, struct dt_device *dt,
177 struct update_request *update;
180 /* In phase I, if the transaction includes remote updates, the local
181 * update should be synchronized, so it will set th_sync = 1 */
182 update = th->th_current_request;
183 LASSERT(update != NULL && update->ur_dt == dt);
184 if (update->ur_buf->ub_count > 0) {
185 rc = osp_remote_sync(env, dt, update, NULL);
193 * Insert the update into the th_bufs for the device.
195 static int osp_insert_update(const struct lu_env *env,
196 struct update_request *update, int op,
197 struct lu_fid *fid, int count,
198 int *lens, char **bufs)
200 struct update_buf *ubuf = update->ur_buf;
201 struct update *obj_update;
208 obj_update = (struct update *)((char *)ubuf +
209 cfs_size_round(update_buf_size(ubuf)));
211 /* Check update size to make sure it can fit into the buffer */
212 update_length = cfs_size_round(offsetof(struct update,
214 for (i = 0; i < count; i++)
215 update_length += cfs_size_round(lens[i]);
217 if (cfs_size_round(update_buf_size(ubuf)) + update_length >
218 UPDATE_BUFFER_SIZE || ubuf->ub_count >= UPDATE_MAX_OPS) {
219 CERROR("%s: insert up %p, idx %d cnt %d len %lu: rc = %d\n",
220 update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf,
221 update_length, ubuf->ub_count, update_buf_size(ubuf),
226 if (count > UPDATE_BUF_COUNT) {
227 CERROR("%s: Insert too much params %d "DFID" op %d: rc = %d\n",
228 update->ur_dt->dd_lu_dev.ld_obd->obd_name, count,
229 PFID(fid), op, -E2BIG);
233 /* fill the update into the update buffer */
234 fid_cpu_to_le(&obj_update->u_fid, fid);
235 obj_update->u_type = cpu_to_le32(op);
236 obj_update->u_batchid = update->ur_batchid;
237 for (i = 0; i < count; i++)
238 obj_update->u_lens[i] = cpu_to_le32(lens[i]);
240 ptr = (char *)obj_update +
241 cfs_size_round(offsetof(struct update, u_bufs[0]));
242 for (i = 0; i < count; i++)
243 LOGL(bufs[i], lens[i], ptr);
247 CDEBUG(D_INFO, "%s: %p "DFID" idx %d: op %d params %d:%lu\n",
248 update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf, PFID(fid),
249 ubuf->ub_count, op, count, update_buf_size(ubuf));
254 static struct update_request
255 *osp_find_update(struct thandle *th, struct dt_device *dt_dev)
257 struct update_request *update;
259 /* Because transaction api does not proivde the interface
260 * to transfer the update from LOD to OSP, we need walk
261 * remote update list to find the update, this probably
262 * should move to LOD layer, when update can be part of
263 * the trancation api parameter. XXX */
264 cfs_list_for_each_entry(update, &th->th_remote_update_list, ur_list) {
265 if (update->ur_dt == dt_dev)
271 static inline void osp_md_add_update_batchid(struct update_request *update)
273 update->ur_batchid++;
277 * Find one loc in th_dev/dev_obj_update for the update,
278 * Because only one thread can access this thandle, no need
281 static struct update_request
282 *osp_find_create_update_loc(struct thandle *th, struct dt_object *dt)
284 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
285 struct update_request *update;
288 update = osp_find_update(th, dt_dev);
292 update = osp_create_update_req(dt_dev);
296 cfs_list_add_tail(&update->ur_list, &th->th_remote_update_list);
301 static int osp_get_attr_from_req(const struct lu_env *env,
302 struct ptlrpc_request *req,
303 struct lu_attr *attr, int index)
305 struct update_reply *reply;
306 struct obdo *lobdo = &osp_env_info(env)->osi_obdo;
310 LASSERT(attr != NULL);
312 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
314 if (reply->ur_version != UPDATE_REPLY_V1)
317 size = update_get_reply_buf(reply, (void **)&wobdo, index);
318 if (size != sizeof(struct obdo))
321 obdo_le_to_cpu(wobdo, wobdo);
322 lustre_get_wire_obdo(NULL, lobdo, wobdo);
323 la_from_obdo(attr, lobdo, lobdo->o_valid);
328 static int osp_md_declare_object_create(const struct lu_env *env,
329 struct dt_object *dt,
330 struct lu_attr *attr,
331 struct dt_allocation_hint *hint,
332 struct dt_object_format *dof,
335 struct osp_thread_info *osi = osp_env_info(env);
336 struct update_request *update;
338 int sizes[2] = {sizeof(struct obdo), 0};
339 char *bufs[2] = {NULL, NULL};
344 update = osp_find_create_update_loc(th, dt);
345 if (IS_ERR(update)) {
346 CERROR("%s: Get OSP update buf failed: rc = %d\n",
347 dt->do_lu.lo_dev->ld_obd->obd_name,
348 (int)PTR_ERR(update));
349 return PTR_ERR(update);
352 osi->osi_obdo.o_valid = 0;
353 LASSERT(S_ISDIR(attr->la_mode));
354 obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
355 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
356 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
358 bufs[0] = (char *)&osi->osi_obdo;
360 fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
361 if (hint->dah_parent) {
363 struct lu_fid *tmp_fid = &osi->osi_fid;
365 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
366 fid_cpu_to_le(tmp_fid, fid2);
367 sizes[1] = sizeof(*tmp_fid);
368 bufs[1] = (char *)tmp_fid;
372 if (lu_object_exists(&dt->do_lu)) {
373 /* If the object already exists, we needs to destroy
374 * this orphan object first.
376 * The scenario might happen in this case
378 * 1. client send remote create to MDT0.
379 * 2. MDT0 send create update to MDT1.
380 * 3. MDT1 finished create synchronously.
381 * 4. MDT0 failed and reboot.
382 * 5. client resend remote create to MDT0.
383 * 6. MDT0 tries to resend create update to MDT1,
384 * but find the object already exists
386 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
387 dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
389 rc = osp_insert_update(env, update, OBJ_REF_DEL, fid1, 0,
394 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
395 /* decrease for ".." */
396 rc = osp_insert_update(env, update, OBJ_REF_DEL, fid1,
402 rc = osp_insert_update(env, update, OBJ_DESTROY, fid1, 0, NULL,
407 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
408 /* Increase batchid to add this orphan object deletion
409 * to separate transaction */
410 osp_md_add_update_batchid(update);
413 rc = osp_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes,
417 CERROR("%s: Insert update error: rc = %d\n",
418 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
423 static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
424 struct lu_attr *attr,
425 struct dt_allocation_hint *hint,
426 struct dt_object_format *dof,
429 struct osp_object *obj = dt2osp_obj(dt);
431 CDEBUG(D_INFO, "create object "DFID"\n",
432 PFID(&dt->do_lu.lo_header->loh_fid));
434 /* Because the create update RPC will be sent during declare phase,
435 * if creation reaches here, it means the object has been created
437 dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
443 static int osp_md_declare_object_ref_del(const struct lu_env *env,
444 struct dt_object *dt,
447 struct update_request *update;
451 update = osp_find_create_update_loc(th, dt);
452 if (IS_ERR(update)) {
453 CERROR("%s: Get OSP update buf failed: rc = %d\n",
454 dt->do_lu.lo_dev->ld_obd->obd_name,
455 (int)PTR_ERR(update));
456 return PTR_ERR(update);
459 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
461 rc = osp_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL);
466 static int osp_md_object_ref_del(const struct lu_env *env,
467 struct dt_object *dt,
470 CDEBUG(D_INFO, "ref del object "DFID"\n",
471 PFID(&dt->do_lu.lo_header->loh_fid));
476 static int osp_md_declare_ref_add(const struct lu_env *env,
477 struct dt_object *dt, struct thandle *th)
479 struct update_request *update;
483 update = osp_find_create_update_loc(th, dt);
484 if (IS_ERR(update)) {
485 CERROR("%s: Get OSP update buf failed: rc = %d\n",
486 dt->do_lu.lo_dev->ld_obd->obd_name,
487 (int)PTR_ERR(update));
488 return PTR_ERR(update);
491 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
493 rc = osp_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL);
498 static int osp_md_object_ref_add(const struct lu_env *env,
499 struct dt_object *dt,
502 CDEBUG(D_INFO, "ref add object "DFID"\n",
503 PFID(&dt->do_lu.lo_header->loh_fid));
508 static void osp_md_ah_init(const struct lu_env *env,
509 struct dt_allocation_hint *ah,
510 struct dt_object *parent,
511 struct dt_object *child,
516 memset(ah, 0, sizeof(*ah));
517 ah->dah_parent = parent;
518 ah->dah_mode = child_mode;
521 static int osp_md_declare_attr_set(const struct lu_env *env,
522 struct dt_object *dt,
523 const struct lu_attr *attr,
526 struct osp_thread_info *osi = osp_env_info(env);
527 struct update_request *update;
529 int size = sizeof(struct obdo);
533 update = osp_find_create_update_loc(th, dt);
534 if (IS_ERR(update)) {
535 CERROR("%s: Get OSP update buf failed: %d\n",
536 dt->do_lu.lo_dev->ld_obd->obd_name,
537 (int)PTR_ERR(update));
538 return PTR_ERR(update);
541 osi->osi_obdo.o_valid = 0;
542 LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE)));
543 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
545 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
546 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
548 buf = (char *)&osi->osi_obdo;
549 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
551 rc = osp_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size, &buf);
556 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
557 const struct lu_attr *attr, struct thandle *th,
558 struct lustre_capa *capa)
560 CDEBUG(D_INFO, "attr set object "DFID"\n",
561 PFID(&dt->do_lu.lo_header->loh_fid));
566 static int osp_md_declare_xattr_set(const struct lu_env *env,
567 struct dt_object *dt,
568 const struct lu_buf *buf,
569 const char *name, int flag,
572 struct update_request *update;
574 int sizes[3] = {strlen(name), buf->lb_len,
576 char *bufs[3] = {(char *)name, (char *)buf->lb_buf };
579 LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
580 update = osp_find_create_update_loc(th, dt);
581 if (IS_ERR(update)) {
582 CERROR("%s: Get OSP update buf failed: rc = %d\n",
583 dt->do_lu.lo_dev->ld_obd->obd_name,
584 (int)PTR_ERR(update));
585 return PTR_ERR(update);
588 flag = cpu_to_le32(flag);
589 bufs[2] = (char *)&flag;
591 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
592 rc = osp_insert_update(env, update, OBJ_XATTR_SET, fid,
593 ARRAY_SIZE(sizes), sizes, bufs);
598 static int osp_md_xattr_set(const struct lu_env *env, struct dt_object *dt,
599 const struct lu_buf *buf, const char *name, int fl,
600 struct thandle *th, struct lustre_capa *capa)
602 CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
603 PFID(&dt->do_lu.lo_header->loh_fid));
608 static int osp_md_xattr_get(const struct lu_env *env, struct dt_object *dt,
609 struct lu_buf *buf, const char *name,
610 struct lustre_capa *capa)
612 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
613 struct update_request *update = NULL;
614 struct ptlrpc_request *req = NULL;
618 struct update_reply *reply;
622 /* Because it needs send the update buffer right away,
623 * just create an update buffer, instead of attaching the
624 * update_remote list of the thandle.
626 update = osp_create_update_req(dt_dev);
628 RETURN(PTR_ERR(update));
630 LASSERT(name != NULL);
631 buf_len = strlen(name);
632 rc = osp_insert_update(env, update, OBJ_XATTR_GET,
633 (struct lu_fid *)lu_object_fid(&dt->do_lu),
634 1, &buf_len, (char **)&name);
636 CERROR("%s: Insert update error: rc = %d\n",
637 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
640 dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
642 rc = osp_remote_sync(env, dt_dev, update, &req);
646 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
648 if (reply->ur_version != UPDATE_REPLY_V1) {
649 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
650 dt_dev->dd_lu_dev.ld_obd->obd_name,
651 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
652 GOTO(out, rc = -EPROTO);
655 size = update_get_reply_buf(reply, &ea_buf, 0);
657 GOTO(out, rc = size);
659 LASSERT(size > 0 && size < PAGE_CACHE_SIZE);
660 LASSERT(ea_buf != NULL);
663 if (buf->lb_buf != NULL)
664 memcpy(buf->lb_buf, ea_buf, size);
667 ptlrpc_req_finished(req);
670 osp_destroy_update_req(update);
675 static void osp_md_object_read_lock(const struct lu_env *env,
676 struct dt_object *dt, unsigned role)
678 struct osp_object *obj = dt2osp_obj(dt);
680 LASSERT(obj->opo_owner != env);
681 down_read_nested(&obj->opo_sem, role);
683 LASSERT(obj->opo_owner == NULL);
686 static void osp_md_object_write_lock(const struct lu_env *env,
687 struct dt_object *dt, unsigned role)
689 struct osp_object *obj = dt2osp_obj(dt);
691 down_write_nested(&obj->opo_sem, role);
693 LASSERT(obj->opo_owner == NULL);
694 obj->opo_owner = env;
697 static void osp_md_object_read_unlock(const struct lu_env *env,
698 struct dt_object *dt)
700 struct osp_object *obj = dt2osp_obj(dt);
702 up_read(&obj->opo_sem);
705 static void osp_md_object_write_unlock(const struct lu_env *env,
706 struct dt_object *dt)
708 struct osp_object *obj = dt2osp_obj(dt);
710 LASSERT(obj->opo_owner == env);
711 obj->opo_owner = NULL;
712 up_write(&obj->opo_sem);
715 static int osp_md_object_write_locked(const struct lu_env *env,
716 struct dt_object *dt)
718 struct osp_object *obj = dt2osp_obj(dt);
720 return obj->opo_owner == env;
723 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
724 struct dt_rec *rec, const struct dt_key *key,
725 struct lustre_capa *capa)
727 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
728 struct update_request *update;
729 struct ptlrpc_request *req = NULL;
730 int size = strlen((char *)key) + 1;
731 char *name = (char *)key;
733 struct update_reply *reply;
738 /* Because it needs send the update buffer right away,
739 * just create an update buffer, instead of attaching the
740 * update_remote list of the thandle.
742 update = osp_create_update_req(dt_dev);
744 RETURN(PTR_ERR(update));
746 rc = osp_insert_update(env, update, OBJ_INDEX_LOOKUP,
747 (struct lu_fid *)lu_object_fid(&dt->do_lu),
748 1, &size, (char **)&name);
750 CERROR("%s: Insert update error: rc = %d\n",
751 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
755 rc = osp_remote_sync(env, dt_dev, update, &req);
757 CERROR("%s: lookup "DFID" %s failed: rc = %d\n",
758 dt_dev->dd_lu_dev.ld_obd->obd_name,
759 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
763 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
765 if (reply->ur_version != UPDATE_REPLY_V1) {
766 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
767 dt_dev->dd_lu_dev.ld_obd->obd_name,
768 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
769 GOTO(out, rc = -EPROTO);
772 rc = update_get_reply_result(reply, NULL, 0);
774 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
775 dt_dev->dd_lu_dev.ld_obd->obd_name,
776 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
780 size = update_get_reply_buf(reply, (void **)&fid, 0);
782 GOTO(out, rc = size);
784 if (size != sizeof(struct lu_fid)) {
785 CERROR("%s: lookup "DFID" %s wrong size %d: rc = %d\n",
786 dt_dev->dd_lu_dev.ld_obd->obd_name,
787 PFID(lu_object_fid(&dt->do_lu)), (char *)key, size, rc);
788 GOTO(out, rc = -EINVAL);
791 fid_le_to_cpu(fid, fid);
792 if (!fid_is_sane(fid)) {
793 CERROR("%s: lookup "DFID" %s invalid fid "DFID": rc = %d\n",
794 dt_dev->dd_lu_dev.ld_obd->obd_name,
795 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid),
797 GOTO(out, rc = -EINVAL);
799 memcpy(rec, fid, sizeof(*fid));
802 ptlrpc_req_finished(req);
805 osp_destroy_update_req(update);
810 static int osp_md_declare_insert(const struct lu_env *env,
811 struct dt_object *dt,
812 const struct dt_rec *rec,
813 const struct dt_key *key,
816 struct update_request *update;
818 struct lu_fid *rec_fid = (struct lu_fid *)rec;
819 int size[2] = {strlen((char *)key) + 1,
821 char *bufs[2] = {(char *)key, (char *)rec_fid};
824 update = osp_find_create_update_loc(th, dt);
825 if (IS_ERR(update)) {
826 CERROR("%s: Get OSP update buf failed: rc = %d\n",
827 dt->do_lu.lo_dev->ld_obd->obd_name,
828 (int)PTR_ERR(update));
829 return PTR_ERR(update);
832 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
834 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
835 dt->do_lu.lo_dev->ld_obd->obd_name,
836 PFID(fid), (char *)key, PFID(rec_fid));
838 fid_cpu_to_le(rec_fid, rec_fid);
840 rc = osp_insert_update(env, update, OBJ_INDEX_INSERT, fid,
841 ARRAY_SIZE(size), size, bufs);
845 static int osp_md_index_insert(const struct lu_env *env,
846 struct dt_object *dt,
847 const struct dt_rec *rec,
848 const struct dt_key *key,
850 struct lustre_capa *capa,
856 static int osp_md_declare_delete(const struct lu_env *env,
857 struct dt_object *dt,
858 const struct dt_key *key,
861 struct update_request *update;
863 int size = strlen((char *)key) + 1;
864 char *buf = (char *)key;
867 update = osp_find_create_update_loc(th, dt);
868 if (IS_ERR(update)) {
869 CERROR("%s: Get OSP update buf failed: rc = %d\n",
870 dt->do_lu.lo_dev->ld_obd->obd_name,
871 (int)PTR_ERR(update));
872 return PTR_ERR(update);
875 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
877 rc = osp_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
883 static int osp_md_index_delete(const struct lu_env *env,
884 struct dt_object *dt,
885 const struct dt_key *key,
887 struct lustre_capa *capa)
889 CDEBUG(D_INFO, "index delete "DFID" %s\n",
890 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
896 * Creates or initializes iterator context.
898 * Note: for OSP, these index iterate api is only used to check
899 * whether the directory is empty now (see mdd_dir_is_empty).
900 * Since dir_empty will be return by OBJ_ATTR_GET(see osp_md_attr_get/
901 * out_attr_get). So the implementation of these iterator is simplied
902 * to make mdd_dir_is_empty happy. The real iterator should be
903 * implemented, if we need it one day.
905 static struct dt_it *osp_it_init(const struct lu_env *env,
906 struct dt_object *dt,
908 struct lustre_capa *capa)
910 lu_object_get(&dt->do_lu);
911 return (struct dt_it *)dt;
914 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
916 struct dt_object *dt = (struct dt_object *)di;
917 lu_object_put(env, &dt->do_lu);
920 static int osp_it_get(const struct lu_env *env,
921 struct dt_it *di, const struct dt_key *key)
926 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
931 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
933 struct dt_object *dt = (struct dt_object *)di;
934 struct osp_object *o = dt2osp_obj(dt);
942 static struct dt_key *osp_it_key(const struct lu_env *env,
943 const struct dt_it *di)
949 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
955 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
956 struct dt_rec *lde, __u32 attr)
962 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
968 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
975 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
982 static const struct dt_index_operations osp_md_index_ops = {
983 .dio_lookup = osp_md_index_lookup,
984 .dio_declare_insert = osp_md_declare_insert,
985 .dio_insert = osp_md_index_insert,
986 .dio_declare_delete = osp_md_declare_delete,
987 .dio_delete = osp_md_index_delete,
995 .key_size = osp_it_key_size,
997 .store = osp_it_store,
999 .key_rec = osp_it_key_rec,
1003 static int osp_md_index_try(const struct lu_env *env,
1004 struct dt_object *dt,
1005 const struct dt_index_features *feat)
1007 dt->do_index_ops = &osp_md_index_ops;
1011 static int osp_md_attr_get(const struct lu_env *env,
1012 struct dt_object *dt, struct lu_attr *attr,
1013 struct lustre_capa *capa)
1015 struct osp_object *obj = dt2osp_obj(dt);
1016 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1017 struct update_request *update = NULL;
1018 struct ptlrpc_request *req = NULL;
1022 /* Because it needs send the update buffer right away,
1023 * just create an update buffer, instead of attaching the
1024 * update_remote list of the thandle.
1026 update = osp_create_update_req(dt_dev);
1028 RETURN(PTR_ERR(update));
1030 rc = osp_insert_update(env, update, OBJ_ATTR_GET,
1031 (struct lu_fid *)lu_object_fid(&dt->do_lu),
1034 CERROR("%s: Insert update error: rc = %d\n",
1035 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
1038 dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1040 rc = osp_remote_sync(env, dt_dev, update, &req);
1044 rc = osp_get_attr_from_req(env, req, attr, 0);
1048 if (attr->la_flags == 1)
1054 ptlrpc_req_finished(req);
1057 osp_destroy_update_req(update);
1062 static int osp_md_declare_object_destroy(const struct lu_env *env,
1063 struct dt_object *dt,
1066 struct osp_object *o = dt2osp_obj(dt);
1071 * track objects to be destroyed via llog
1073 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1078 static int osp_md_object_destroy(const struct lu_env *env,
1079 struct dt_object *dt, struct thandle *th)
1081 struct osp_object *o = dt2osp_obj(dt);
1086 * once transaction is committed put proper command on
1087 * the queue going to our OST
1089 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1091 /* not needed in cache any more */
1092 set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1097 static int osp_md_object_lock(const struct lu_env *env,
1098 struct dt_object *dt,
1099 struct lustre_handle *lh,
1100 struct ldlm_enqueue_info *einfo,
1103 struct osp_thread_info *info = osp_env_info(env);
1104 struct ldlm_res_id *res_id = &info->osi_resid;
1105 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1106 struct osp_device *osp = dt2osp_dev(dt_dev);
1107 struct ptlrpc_request *req = NULL;
1112 fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
1114 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
1115 LDLM_FL_BLOCK_GRANTED, res_id,
1117 (ldlm_policy_data_t *)policy,
1118 einfo->ei_mode, lh, 0);
1122 req = ldlm_enqueue_pack(osp->opd_exp, 0);
1124 RETURN(PTR_ERR(req));
1126 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
1127 (const ldlm_policy_data_t *)policy,
1128 &flags, NULL, 0, LVB_T_NONE, lh, 0);
1130 ptlrpc_req_finished(req);
1132 return rc == ELDLM_OK ? 0 : -EIO;
1135 struct dt_object_operations osp_md_obj_ops = {
1136 .do_read_lock = osp_md_object_read_lock,
1137 .do_write_lock = osp_md_object_write_lock,
1138 .do_read_unlock = osp_md_object_read_unlock,
1139 .do_write_unlock = osp_md_object_write_unlock,
1140 .do_write_locked = osp_md_object_write_locked,
1141 .do_declare_create = osp_md_declare_object_create,
1142 .do_create = osp_md_object_create,
1143 .do_declare_ref_add = osp_md_declare_ref_add,
1144 .do_ref_add = osp_md_object_ref_add,
1145 .do_declare_ref_del = osp_md_declare_object_ref_del,
1146 .do_ref_del = osp_md_object_ref_del,
1147 .do_declare_destroy = osp_md_declare_object_destroy,
1148 .do_destroy = osp_md_object_destroy,
1149 .do_ah_init = osp_md_ah_init,
1150 .do_attr_get = osp_md_attr_get,
1151 .do_declare_attr_set = osp_md_declare_attr_set,
1152 .do_attr_set = osp_md_attr_set,
1153 .do_declare_xattr_set = osp_md_declare_xattr_set,
1154 .do_xattr_set = osp_md_xattr_set,
1155 .do_xattr_get = osp_md_xattr_get,
1156 .do_index_try = osp_md_index_try,
1157 .do_object_lock = osp_md_object_lock,