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);
68 bufs[0] = (char *)&osi->osi_obdo;
70 fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
71 if (hint != NULL && hint->dah_parent) {
74 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
75 sizes[1] = sizeof(*fid2);
76 bufs[1] = (char *)fid2;
80 if (lu_object_exists(&dt->do_lu)) {
81 /* If the object already exists, we needs to destroy
82 * this orphan object first.
84 * The scenario might happen in this case
86 * 1. client send remote create to MDT0.
87 * 2. MDT0 send create update to MDT1.
88 * 3. MDT1 finished create synchronously.
89 * 4. MDT0 failed and reboot.
90 * 5. client resend remote create to MDT0.
91 * 6. MDT0 tries to resend create update to MDT1,
92 * but find the object already exists
94 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
95 dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
97 rc = out_insert_update(env, update, OUT_REF_DEL, fid1, 0,
102 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
103 /* decrease for ".." */
104 rc = out_insert_update(env, update, OUT_REF_DEL, fid1,
110 rc = out_insert_update(env, update, OUT_DESTROY, fid1, 0, NULL,
115 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
116 /* Increase batchid to add this orphan object deletion
117 * to separate transaction */
118 update_inc_batchid(update);
121 rc = out_insert_update(env, update, OUT_CREATE, fid1, buf_count, sizes,
122 (const char **)bufs);
125 CERROR("%s: Insert update error: rc = %d\n",
126 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
131 int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
132 struct lu_attr *attr, struct dt_allocation_hint *hint,
133 struct dt_object_format *dof, struct thandle *th)
135 CDEBUG(D_INFO, "create object "DFID"\n",
136 PFID(&dt->do_lu.lo_header->loh_fid));
138 /* Because the create update RPC will be sent during declare phase,
139 * if creation reaches here, it means the object has been created
141 dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
146 static int osp_md_declare_object_ref_del(const struct lu_env *env,
147 struct dt_object *dt,
150 struct dt_update_request *update;
154 update = out_find_create_update_loc(th, dt);
155 if (IS_ERR(update)) {
156 CERROR("%s: Get OSP update buf failed: rc = %d\n",
157 dt->do_lu.lo_dev->ld_obd->obd_name,
158 (int)PTR_ERR(update));
159 return PTR_ERR(update);
162 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
164 rc = out_insert_update(env, update, OUT_REF_DEL, fid, 0, NULL, NULL);
169 static int osp_md_object_ref_del(const struct lu_env *env,
170 struct dt_object *dt,
173 CDEBUG(D_INFO, "ref del object "DFID"\n",
174 PFID(&dt->do_lu.lo_header->loh_fid));
179 static int osp_md_declare_ref_add(const struct lu_env *env,
180 struct dt_object *dt, struct thandle *th)
182 struct dt_update_request *update;
186 update = out_find_create_update_loc(th, dt);
187 if (IS_ERR(update)) {
188 CERROR("%s: Get OSP update buf failed: rc = %d\n",
189 dt->do_lu.lo_dev->ld_obd->obd_name,
190 (int)PTR_ERR(update));
191 return PTR_ERR(update);
194 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
196 rc = out_insert_update(env, update, OUT_REF_ADD, fid, 0, NULL, NULL);
201 static int osp_md_object_ref_add(const struct lu_env *env,
202 struct dt_object *dt,
205 CDEBUG(D_INFO, "ref add object "DFID"\n",
206 PFID(&dt->do_lu.lo_header->loh_fid));
211 static void osp_md_ah_init(const struct lu_env *env,
212 struct dt_allocation_hint *ah,
213 struct dt_object *parent,
214 struct dt_object *child,
219 ah->dah_parent = parent;
220 ah->dah_mode = child_mode;
223 int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
224 const struct lu_attr *attr, struct thandle *th)
226 struct osp_thread_info *osi = osp_env_info(env);
227 struct dt_update_request *update;
229 int size = sizeof(struct obdo);
233 update = out_find_create_update_loc(th, dt);
234 if (IS_ERR(update)) {
235 CERROR("%s: Get OSP update buf failed: %d\n",
236 dt->do_lu.lo_dev->ld_obd->obd_name,
237 (int)PTR_ERR(update));
238 return PTR_ERR(update);
241 osi->osi_obdo.o_valid = 0;
242 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
244 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
246 buf = (char *)&osi->osi_obdo;
247 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
249 rc = out_insert_update(env, update, OUT_ATTR_SET, fid, 1, &size,
250 (const char **)&buf);
255 int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
256 const struct lu_attr *attr, struct thandle *th,
257 struct lustre_capa *capa)
259 CDEBUG(D_INFO, "attr set object "DFID"\n",
260 PFID(&dt->do_lu.lo_header->loh_fid));
265 static void osp_md_object_read_lock(const struct lu_env *env,
266 struct dt_object *dt, unsigned role)
268 struct osp_object *obj = dt2osp_obj(dt);
270 LASSERT(obj->opo_owner != env);
271 down_read_nested(&obj->opo_sem, role);
273 LASSERT(obj->opo_owner == NULL);
276 static void osp_md_object_write_lock(const struct lu_env *env,
277 struct dt_object *dt, unsigned role)
279 struct osp_object *obj = dt2osp_obj(dt);
281 down_write_nested(&obj->opo_sem, role);
283 LASSERT(obj->opo_owner == NULL);
284 obj->opo_owner = env;
287 static void osp_md_object_read_unlock(const struct lu_env *env,
288 struct dt_object *dt)
290 struct osp_object *obj = dt2osp_obj(dt);
292 up_read(&obj->opo_sem);
295 static void osp_md_object_write_unlock(const struct lu_env *env,
296 struct dt_object *dt)
298 struct osp_object *obj = dt2osp_obj(dt);
300 LASSERT(obj->opo_owner == env);
301 obj->opo_owner = NULL;
302 up_write(&obj->opo_sem);
305 static int osp_md_object_write_locked(const struct lu_env *env,
306 struct dt_object *dt)
308 struct osp_object *obj = dt2osp_obj(dt);
310 return obj->opo_owner == env;
313 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
314 struct dt_rec *rec, const struct dt_key *key,
315 struct lustre_capa *capa)
317 struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
318 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
319 struct dt_device *dt_dev = &osp->opd_dt_dev;
320 struct dt_update_request *update;
321 struct object_update_reply *reply;
322 struct ptlrpc_request *req = NULL;
323 int size = strlen((char *)key) + 1;
328 /* Because it needs send the update buffer right away,
329 * just create an update buffer, instead of attaching the
330 * update_remote list of the thandle.
332 update = out_create_update_req(dt_dev);
334 RETURN(PTR_ERR(update));
336 rc = out_insert_update(env, update, OUT_INDEX_LOOKUP,
337 lu_object_fid(&dt->do_lu),
338 1, &size, (const char **)&key);
340 CERROR("%s: Insert update error: rc = %d\n",
341 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
345 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
349 reply = req_capsule_server_sized_get(&req->rq_pill,
350 &RMF_OUT_UPDATE_REPLY,
351 OUT_UPDATE_REPLY_SIZE);
352 if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
353 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
354 dt_dev->dd_lu_dev.ld_obd->obd_name,
355 reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO);
356 GOTO(out, rc = -EPROTO);
359 rc = object_update_result_data_get(reply, lbuf, 0);
361 GOTO(out, rc = size);
363 if (lbuf->lb_len != sizeof(*fid)) {
364 CERROR("%s: lookup "DFID" %s wrong size %d\n",
365 dt_dev->dd_lu_dev.ld_obd->obd_name,
366 PFID(lu_object_fid(&dt->do_lu)), (char *)key,
368 GOTO(out, rc = -EINVAL);
372 if (ptlrpc_rep_need_swab(req))
373 lustre_swab_lu_fid(fid);
374 if (!fid_is_sane(fid)) {
375 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
376 dt_dev->dd_lu_dev.ld_obd->obd_name,
377 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
378 GOTO(out, rc = -EINVAL);
381 memcpy(rec, fid, sizeof(*fid));
387 ptlrpc_req_finished(req);
389 out_destroy_update_req(update);
394 static int osp_md_declare_insert(const struct lu_env *env,
395 struct dt_object *dt,
396 const struct dt_rec *rec,
397 const struct dt_key *key,
400 struct dt_update_request *update;
402 struct lu_fid *rec_fid = (struct lu_fid *)rec;
403 int size[2] = {strlen((char *)key) + 1,
405 const char *bufs[2] = {(char *)key, (char *)rec_fid};
408 update = out_find_create_update_loc(th, dt);
409 if (IS_ERR(update)) {
410 CERROR("%s: Get OSP update buf failed: rc = %d\n",
411 dt->do_lu.lo_dev->ld_obd->obd_name,
412 (int)PTR_ERR(update));
413 return PTR_ERR(update);
416 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
418 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
419 dt->do_lu.lo_dev->ld_obd->obd_name,
420 PFID(fid), (char *)key, PFID(rec_fid));
422 fid_cpu_to_le(rec_fid, rec_fid);
424 rc = out_insert_update(env, update, OUT_INDEX_INSERT, fid,
425 ARRAY_SIZE(size), size, bufs);
429 static int osp_md_index_insert(const struct lu_env *env,
430 struct dt_object *dt,
431 const struct dt_rec *rec,
432 const struct dt_key *key,
434 struct lustre_capa *capa,
440 static int osp_md_declare_delete(const struct lu_env *env,
441 struct dt_object *dt,
442 const struct dt_key *key,
445 struct dt_update_request *update;
447 int size = strlen((char *)key) + 1;
450 update = out_find_create_update_loc(th, dt);
451 if (IS_ERR(update)) {
452 CERROR("%s: Get OSP update buf failed: rc = %d\n",
453 dt->do_lu.lo_dev->ld_obd->obd_name,
454 (int)PTR_ERR(update));
455 return PTR_ERR(update);
458 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
460 rc = out_insert_update(env, update, OUT_INDEX_DELETE, fid, 1, &size,
461 (const char **)&key);
466 static int osp_md_index_delete(const struct lu_env *env,
467 struct dt_object *dt,
468 const struct dt_key *key,
470 struct lustre_capa *capa)
472 CDEBUG(D_INFO, "index delete "DFID" %s\n",
473 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
478 int osp_md_index_it_next(const struct lu_env *env, struct dt_it *di)
480 struct osp_it *it = (struct osp_it *)di;
481 struct lu_idxpage *idxpage;
482 struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent;
487 idxpage = it->ooi_cur_idxpage;
488 if (idxpage != NULL) {
489 if (idxpage->lip_nr == 0)
495 (struct lu_dirent *)idxpage->lip_entries;
497 } else if (le16_to_cpu(ent->lde_reclen) != 0 &&
498 it->ooi_pos_ent < idxpage->lip_nr) {
499 ent = (struct lu_dirent *)(((char *)ent) +
500 le16_to_cpu(ent->lde_reclen));
508 rc = osp_it_next_page(env, di);
515 static struct dt_key *osp_it_key(const struct lu_env *env,
516 const struct dt_it *di)
522 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
528 static int osp_md_index_it_rec(const struct lu_env *env, const struct dt_it *di,
529 struct dt_rec *rec, __u32 attr)
531 struct osp_it *it = (struct osp_it *)di;
532 struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent;
535 reclen = lu_dirent_calc_size(le16_to_cpu(ent->lde_namelen), attr);
536 memcpy(rec, ent, reclen);
540 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
547 const struct dt_index_operations osp_md_index_ops = {
548 .dio_lookup = osp_md_index_lookup,
549 .dio_declare_insert = osp_md_declare_insert,
550 .dio_insert = osp_md_index_insert,
551 .dio_declare_delete = osp_md_declare_delete,
552 .dio_delete = osp_md_index_delete,
558 .next = osp_md_index_it_next,
560 .key_size = osp_it_key_size,
561 .rec = osp_md_index_it_rec,
562 .store = osp_it_store,
564 .key_rec = osp_it_key_rec,
568 static int osp_md_index_try(const struct lu_env *env,
569 struct dt_object *dt,
570 const struct dt_index_features *feat)
572 dt->do_index_ops = &osp_md_index_ops;
576 static int osp_md_object_lock(const struct lu_env *env,
577 struct dt_object *dt,
578 struct lustre_handle *lh,
579 struct ldlm_enqueue_info *einfo,
580 ldlm_policy_data_t *policy)
582 struct ldlm_res_id *res_id;
583 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
584 struct osp_device *osp = dt2osp_dev(dt_dev);
585 struct ptlrpc_request *req;
590 res_id = einfo->ei_res_id;
591 LASSERT(res_id != NULL);
593 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
594 LDLM_FL_BLOCK_GRANTED, res_id,
595 einfo->ei_type, policy,
596 einfo->ei_mode, lh, 0);
600 req = ldlm_enqueue_pack(osp->opd_exp, 0);
602 RETURN(PTR_ERR(req));
604 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
605 (const ldlm_policy_data_t *)policy,
606 &flags, NULL, 0, LVB_T_NONE, lh, 0);
608 ptlrpc_req_finished(req);
610 return rc == ELDLM_OK ? 0 : -EIO;
613 static int osp_md_object_unlock(const struct lu_env *env,
614 struct dt_object *dt,
615 struct ldlm_enqueue_info *einfo,
616 ldlm_policy_data_t *policy)
618 struct lustre_handle *lockh = einfo->ei_cbdata;
621 ldlm_lock_decref(lockh, einfo->ei_mode);
626 struct dt_object_operations osp_md_obj_ops = {
627 .do_read_lock = osp_md_object_read_lock,
628 .do_write_lock = osp_md_object_write_lock,
629 .do_read_unlock = osp_md_object_read_unlock,
630 .do_write_unlock = osp_md_object_write_unlock,
631 .do_write_locked = osp_md_object_write_locked,
632 .do_declare_create = osp_md_declare_object_create,
633 .do_create = osp_md_object_create,
634 .do_declare_ref_add = osp_md_declare_ref_add,
635 .do_ref_add = osp_md_object_ref_add,
636 .do_declare_ref_del = osp_md_declare_object_ref_del,
637 .do_ref_del = osp_md_object_ref_del,
638 .do_declare_destroy = osp_declare_object_destroy,
639 .do_destroy = osp_object_destroy,
640 .do_ah_init = osp_md_ah_init,
641 .do_attr_get = osp_attr_get,
642 .do_declare_attr_set = osp_md_declare_attr_set,
643 .do_attr_set = osp_md_attr_set,
644 .do_xattr_get = osp_xattr_get,
645 .do_declare_xattr_set = osp_declare_xattr_set,
646 .do_xattr_set = osp_xattr_set,
647 .do_declare_xattr_del = osp_declare_xattr_del,
648 .do_xattr_del = osp_xattr_del,
649 .do_index_try = osp_md_index_try,
650 .do_object_lock = osp_md_object_lock,
651 .do_object_unlock = osp_md_object_unlock,
654 static ssize_t osp_md_declare_write(const struct lu_env *env,
655 struct dt_object *dt,
656 const struct lu_buf *buf,
657 loff_t pos, struct thandle *th)
659 struct dt_update_request *update;
661 int sizes[2] = {buf->lb_len, sizeof(pos)};
662 const char *bufs[2] = {(char *)buf->lb_buf,
666 update = out_find_create_update_loc(th, dt);
667 if (IS_ERR(update)) {
668 CERROR("%s: Get OSP update buf failed: rc = %d\n",
669 dt->do_lu.lo_dev->ld_obd->obd_name,
670 (int)PTR_ERR(update));
671 return PTR_ERR(update);
674 pos = cpu_to_le64(pos);
675 bufs[1] = (char *)&pos;
676 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
677 rc = out_insert_update(env, update, OUT_WRITE, fid,
678 ARRAY_SIZE(sizes), sizes, bufs);
684 static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
685 const struct lu_buf *buf, loff_t *pos,
686 struct thandle *handle,
687 struct lustre_capa *capa, int ignore_quota)
692 /* These body operation will be used to write symlinks during migration etc */
693 struct dt_body_operations osp_md_body_ops = {
694 .dbo_declare_write = osp_md_declare_write,
695 .dbo_write = osp_md_write,