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 "osp_internal.h"
38 static const char dot[] = ".";
39 static const char dotdot[] = "..";
41 int osp_md_declare_object_create(const struct lu_env *env,
44 struct dt_allocation_hint *hint,
45 struct dt_object_format *dof,
48 struct osp_thread_info *osi = osp_env_info(env);
49 struct dt_update_request *update;
51 int sizes[2] = {sizeof(struct obdo), 0};
52 char *bufs[2] = {NULL, NULL};
56 update = out_find_create_update_loc(th, dt);
58 CERROR("%s: Get OSP update buf failed: rc = %d\n",
59 dt->do_lu.lo_dev->ld_obd->obd_name,
60 (int)PTR_ERR(update));
61 return PTR_ERR(update);
64 osi->osi_obdo.o_valid = 0;
65 obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
66 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
67 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
69 bufs[0] = (char *)&osi->osi_obdo;
71 fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
72 if (hint != NULL && hint->dah_parent) {
74 struct lu_fid *tmp_fid = &osi->osi_fid;
76 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
77 fid_cpu_to_le(tmp_fid, fid2);
78 sizes[1] = sizeof(*tmp_fid);
79 bufs[1] = (char *)tmp_fid;
83 if (lu_object_exists(&dt->do_lu)) {
84 /* If the object already exists, we needs to destroy
85 * this orphan object first.
87 * The scenario might happen in this case
89 * 1. client send remote create to MDT0.
90 * 2. MDT0 send create update to MDT1.
91 * 3. MDT1 finished create synchronously.
92 * 4. MDT0 failed and reboot.
93 * 5. client resend remote create to MDT0.
94 * 6. MDT0 tries to resend create update to MDT1,
95 * but find the object already exists
97 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
98 dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
100 rc = out_insert_update(env, update, OUT_REF_DEL, fid1, 0,
105 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
106 /* decrease for ".." */
107 rc = out_insert_update(env, update, OUT_REF_DEL, fid1,
113 rc = out_insert_update(env, update, OUT_DESTROY, fid1, 0, NULL,
118 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
119 /* Increase batchid to add this orphan object deletion
120 * to separate transaction */
121 update_inc_batchid(update);
124 rc = out_insert_update(env, update, OUT_CREATE, fid1, buf_count, sizes,
125 (const char **)bufs);
128 CERROR("%s: Insert update error: rc = %d\n",
129 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
134 int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
135 struct lu_attr *attr, struct dt_allocation_hint *hint,
136 struct dt_object_format *dof, struct thandle *th)
138 struct osp_object *obj = dt2osp_obj(dt);
140 CDEBUG(D_INFO, "create object "DFID"\n",
141 PFID(&dt->do_lu.lo_header->loh_fid));
143 /* Because the create update RPC will be sent during declare phase,
144 * if creation reaches here, it means the object has been created
146 dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
152 static int osp_md_declare_object_ref_del(const struct lu_env *env,
153 struct dt_object *dt,
156 struct dt_update_request *update;
160 update = out_find_create_update_loc(th, dt);
161 if (IS_ERR(update)) {
162 CERROR("%s: Get OSP update buf failed: rc = %d\n",
163 dt->do_lu.lo_dev->ld_obd->obd_name,
164 (int)PTR_ERR(update));
165 return PTR_ERR(update);
168 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
170 rc = out_insert_update(env, update, OUT_REF_DEL, fid, 0, NULL, NULL);
175 static int osp_md_object_ref_del(const struct lu_env *env,
176 struct dt_object *dt,
179 CDEBUG(D_INFO, "ref del object "DFID"\n",
180 PFID(&dt->do_lu.lo_header->loh_fid));
185 static int osp_md_declare_ref_add(const struct lu_env *env,
186 struct dt_object *dt, struct thandle *th)
188 struct dt_update_request *update;
192 update = out_find_create_update_loc(th, dt);
193 if (IS_ERR(update)) {
194 CERROR("%s: Get OSP update buf failed: rc = %d\n",
195 dt->do_lu.lo_dev->ld_obd->obd_name,
196 (int)PTR_ERR(update));
197 return PTR_ERR(update);
200 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
202 rc = out_insert_update(env, update, OUT_REF_ADD, fid, 0, NULL, NULL);
207 static int osp_md_object_ref_add(const struct lu_env *env,
208 struct dt_object *dt,
211 CDEBUG(D_INFO, "ref add object "DFID"\n",
212 PFID(&dt->do_lu.lo_header->loh_fid));
217 static void osp_md_ah_init(const struct lu_env *env,
218 struct dt_allocation_hint *ah,
219 struct dt_object *parent,
220 struct dt_object *child,
225 ah->dah_parent = parent;
226 ah->dah_mode = child_mode;
229 int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
230 const struct lu_attr *attr, struct thandle *th)
232 struct osp_thread_info *osi = osp_env_info(env);
233 struct dt_update_request *update;
235 int size = sizeof(struct obdo);
239 update = out_find_create_update_loc(th, dt);
240 if (IS_ERR(update)) {
241 CERROR("%s: Get OSP update buf failed: %d\n",
242 dt->do_lu.lo_dev->ld_obd->obd_name,
243 (int)PTR_ERR(update));
244 return PTR_ERR(update);
247 osi->osi_obdo.o_valid = 0;
248 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
250 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
251 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
253 buf = (char *)&osi->osi_obdo;
254 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
256 rc = out_insert_update(env, update, OUT_ATTR_SET, fid, 1, &size,
257 (const char **)&buf);
262 int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
263 const struct lu_attr *attr, struct thandle *th,
264 struct lustre_capa *capa)
266 CDEBUG(D_INFO, "attr set object "DFID"\n",
267 PFID(&dt->do_lu.lo_header->loh_fid));
272 static void osp_md_object_read_lock(const struct lu_env *env,
273 struct dt_object *dt, unsigned role)
275 struct osp_object *obj = dt2osp_obj(dt);
277 LASSERT(obj->opo_owner != env);
278 down_read_nested(&obj->opo_sem, role);
280 LASSERT(obj->opo_owner == NULL);
283 static void osp_md_object_write_lock(const struct lu_env *env,
284 struct dt_object *dt, unsigned role)
286 struct osp_object *obj = dt2osp_obj(dt);
288 down_write_nested(&obj->opo_sem, role);
290 LASSERT(obj->opo_owner == NULL);
291 obj->opo_owner = env;
294 static void osp_md_object_read_unlock(const struct lu_env *env,
295 struct dt_object *dt)
297 struct osp_object *obj = dt2osp_obj(dt);
299 up_read(&obj->opo_sem);
302 static void osp_md_object_write_unlock(const struct lu_env *env,
303 struct dt_object *dt)
305 struct osp_object *obj = dt2osp_obj(dt);
307 LASSERT(obj->opo_owner == env);
308 obj->opo_owner = NULL;
309 up_write(&obj->opo_sem);
312 static int osp_md_object_write_locked(const struct lu_env *env,
313 struct dt_object *dt)
315 struct osp_object *obj = dt2osp_obj(dt);
317 return obj->opo_owner == env;
320 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
321 struct dt_rec *rec, const struct dt_key *key,
322 struct lustre_capa *capa)
324 struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
325 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
326 struct dt_device *dt_dev = &osp->opd_dt_dev;
327 struct dt_update_request *update;
328 struct object_update_reply *reply;
329 struct ptlrpc_request *req = NULL;
330 int size = strlen((char *)key) + 1;
335 /* Because it needs send the update buffer right away,
336 * just create an update buffer, instead of attaching the
337 * update_remote list of the thandle.
339 update = out_create_update_req(dt_dev);
341 RETURN(PTR_ERR(update));
343 rc = out_insert_update(env, update, OUT_INDEX_LOOKUP,
344 lu_object_fid(&dt->do_lu),
345 1, &size, (const char **)&key);
347 CERROR("%s: Insert update error: rc = %d\n",
348 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
352 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
356 reply = req_capsule_server_sized_get(&req->rq_pill,
357 &RMF_OUT_UPDATE_REPLY,
358 OUT_UPDATE_REPLY_SIZE);
359 if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
360 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
361 dt_dev->dd_lu_dev.ld_obd->obd_name,
362 reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO);
363 GOTO(out, rc = -EPROTO);
366 rc = object_update_result_data_get(reply, lbuf, 0);
368 GOTO(out, rc = size);
370 if (lbuf->lb_len != sizeof(*fid)) {
371 CERROR("%s: lookup "DFID" %s wrong size %d\n",
372 dt_dev->dd_lu_dev.ld_obd->obd_name,
373 PFID(lu_object_fid(&dt->do_lu)), (char *)key,
375 GOTO(out, rc = -EINVAL);
379 fid_le_to_cpu(fid, fid);
380 if (!fid_is_sane(fid)) {
381 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
382 dt_dev->dd_lu_dev.ld_obd->obd_name,
383 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
384 GOTO(out, rc = -EINVAL);
387 memcpy(rec, fid, sizeof(*fid));
393 ptlrpc_req_finished(req);
395 out_destroy_update_req(update);
400 static int osp_md_declare_insert(const struct lu_env *env,
401 struct dt_object *dt,
402 const struct dt_rec *rec,
403 const struct dt_key *key,
406 struct dt_update_request *update;
408 struct lu_fid *rec_fid = (struct lu_fid *)rec;
409 int size[2] = {strlen((char *)key) + 1,
411 const char *bufs[2] = {(char *)key, (char *)rec_fid};
414 update = out_find_create_update_loc(th, dt);
415 if (IS_ERR(update)) {
416 CERROR("%s: Get OSP update buf failed: rc = %d\n",
417 dt->do_lu.lo_dev->ld_obd->obd_name,
418 (int)PTR_ERR(update));
419 return PTR_ERR(update);
422 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
424 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
425 dt->do_lu.lo_dev->ld_obd->obd_name,
426 PFID(fid), (char *)key, PFID(rec_fid));
428 fid_cpu_to_le(rec_fid, rec_fid);
430 rc = out_insert_update(env, update, OUT_INDEX_INSERT, fid,
431 ARRAY_SIZE(size), size, bufs);
435 static int osp_md_index_insert(const struct lu_env *env,
436 struct dt_object *dt,
437 const struct dt_rec *rec,
438 const struct dt_key *key,
440 struct lustre_capa *capa,
446 static int osp_md_declare_delete(const struct lu_env *env,
447 struct dt_object *dt,
448 const struct dt_key *key,
451 struct dt_update_request *update;
453 int size = strlen((char *)key) + 1;
456 update = out_find_create_update_loc(th, dt);
457 if (IS_ERR(update)) {
458 CERROR("%s: Get OSP update buf failed: rc = %d\n",
459 dt->do_lu.lo_dev->ld_obd->obd_name,
460 (int)PTR_ERR(update));
461 return PTR_ERR(update);
464 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
466 rc = out_insert_update(env, update, OUT_INDEX_DELETE, fid, 1, &size,
467 (const char **)&key);
472 static int osp_md_index_delete(const struct lu_env *env,
473 struct dt_object *dt,
474 const struct dt_key *key,
476 struct lustre_capa *capa)
478 CDEBUG(D_INFO, "index delete "DFID" %s\n",
479 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
485 * Creates or initializes iterator context.
487 * Note: for OSP, these index iterate api is only used to check
488 * whether the directory is empty now (see mdd_dir_is_empty).
489 * Since dir_empty will be return by OUT_ATTR_GET(see osp_attr_get/
490 * out_attr_get). So the implementation of these iterator is simplied
491 * to make mdd_dir_is_empty happy. The real iterator should be
492 * implemented, if we need it one day.
494 static struct dt_it *osp_it_init(const struct lu_env *env,
495 struct dt_object *dt,
497 struct lustre_capa *capa)
499 lu_object_get(&dt->do_lu);
500 return (struct dt_it *)dt;
503 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
505 struct dt_object *dt = (struct dt_object *)di;
506 lu_object_put(env, &dt->do_lu);
509 static int osp_it_get(const struct lu_env *env,
510 struct dt_it *di, const struct dt_key *key)
515 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
520 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
522 struct dt_object *dt = (struct dt_object *)di;
523 struct osp_object *o = dt2osp_obj(dt);
531 static struct dt_key *osp_it_key(const struct lu_env *env,
532 const struct dt_it *di)
538 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
544 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
545 struct dt_rec *lde, __u32 attr)
551 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
557 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
564 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
571 static const struct dt_index_operations osp_md_index_ops = {
572 .dio_lookup = osp_md_index_lookup,
573 .dio_declare_insert = osp_md_declare_insert,
574 .dio_insert = osp_md_index_insert,
575 .dio_declare_delete = osp_md_declare_delete,
576 .dio_delete = osp_md_index_delete,
584 .key_size = osp_it_key_size,
586 .store = osp_it_store,
588 .key_rec = osp_it_key_rec,
592 static int osp_md_index_try(const struct lu_env *env,
593 struct dt_object *dt,
594 const struct dt_index_features *feat)
596 dt->do_index_ops = &osp_md_index_ops;
600 static int osp_md_object_lock(const struct lu_env *env,
601 struct dt_object *dt,
602 struct lustre_handle *lh,
603 struct ldlm_enqueue_info *einfo,
604 ldlm_policy_data_t *policy)
606 struct ldlm_res_id *res_id;
607 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
608 struct osp_device *osp = dt2osp_dev(dt_dev);
609 struct ptlrpc_request *req;
614 res_id = einfo->ei_res_id;
615 LASSERT(res_id != NULL);
617 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
618 LDLM_FL_BLOCK_GRANTED, res_id,
619 einfo->ei_type, policy,
620 einfo->ei_mode, lh, 0);
624 req = ldlm_enqueue_pack(osp->opd_exp, 0);
626 RETURN(PTR_ERR(req));
628 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
629 (const ldlm_policy_data_t *)policy,
630 &flags, NULL, 0, LVB_T_NONE, lh, 0);
632 ptlrpc_req_finished(req);
634 return rc == ELDLM_OK ? 0 : -EIO;
637 static int osp_md_object_unlock(const struct lu_env *env,
638 struct dt_object *dt,
639 struct ldlm_enqueue_info *einfo,
640 ldlm_policy_data_t *policy)
642 struct lustre_handle *lockh = einfo->ei_cbdata;
645 ldlm_lock_decref(lockh, einfo->ei_mode);
650 struct dt_object_operations osp_md_obj_ops = {
651 .do_read_lock = osp_md_object_read_lock,
652 .do_write_lock = osp_md_object_write_lock,
653 .do_read_unlock = osp_md_object_read_unlock,
654 .do_write_unlock = osp_md_object_write_unlock,
655 .do_write_locked = osp_md_object_write_locked,
656 .do_declare_create = osp_md_declare_object_create,
657 .do_create = osp_md_object_create,
658 .do_declare_ref_add = osp_md_declare_ref_add,
659 .do_ref_add = osp_md_object_ref_add,
660 .do_declare_ref_del = osp_md_declare_object_ref_del,
661 .do_ref_del = osp_md_object_ref_del,
662 .do_declare_destroy = osp_declare_object_destroy,
663 .do_destroy = osp_object_destroy,
664 .do_ah_init = osp_md_ah_init,
665 .do_attr_get = osp_attr_get,
666 .do_declare_attr_set = osp_md_declare_attr_set,
667 .do_attr_set = osp_md_attr_set,
668 .do_xattr_get = osp_xattr_get,
669 .do_declare_xattr_set = osp_declare_xattr_set,
670 .do_xattr_set = osp_xattr_set,
671 .do_index_try = osp_md_index_try,
672 .do_object_lock = osp_md_object_lock,
673 .do_object_unlock = osp_md_object_unlock,
676 static ssize_t osp_md_declare_write(const struct lu_env *env,
677 struct dt_object *dt,
678 const struct lu_buf *buf,
679 loff_t pos, struct thandle *th)
681 struct dt_update_request *update;
683 int sizes[2] = {buf->lb_len, sizeof(pos)};
684 const char *bufs[2] = {(char *)buf->lb_buf,
688 update = out_find_create_update_loc(th, dt);
689 if (IS_ERR(update)) {
690 CERROR("%s: Get OSP update buf failed: rc = %d\n",
691 dt->do_lu.lo_dev->ld_obd->obd_name,
692 (int)PTR_ERR(update));
693 return PTR_ERR(update);
696 pos = cpu_to_le64(pos);
697 bufs[1] = (char *)&pos;
698 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
699 rc = out_insert_update(env, update, OUT_WRITE, fid,
700 ARRAY_SIZE(sizes), sizes, bufs);
706 static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
707 const struct lu_buf *buf, loff_t *pos,
708 struct thandle *handle,
709 struct lustre_capa *capa, int ignore_quota)
714 /* These body operation will be used to write symlinks during migration etc */
715 struct dt_body_operations osp_md_body_ops = {
716 .dbo_declare_write = osp_md_declare_write,
717 .dbo_write = osp_md_write,