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 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, OBJ_REF_DEL, fid1, 0,
105 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
106 /* decrease for ".." */
107 rc = out_insert_update(env, update, OBJ_REF_DEL, fid1,
113 rc = out_insert_update(env, update, OBJ_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, OBJ_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 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, OBJ_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 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, OBJ_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 memset(ah, 0, sizeof(*ah));
226 ah->dah_parent = parent;
227 ah->dah_mode = child_mode;
230 int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
231 const struct lu_attr *attr, struct thandle *th)
233 struct osp_thread_info *osi = osp_env_info(env);
234 struct update_request *update;
236 int size = sizeof(struct obdo);
240 update = out_find_create_update_loc(th, dt);
241 if (IS_ERR(update)) {
242 CERROR("%s: Get OSP update buf failed: %d\n",
243 dt->do_lu.lo_dev->ld_obd->obd_name,
244 (int)PTR_ERR(update));
245 return PTR_ERR(update);
248 osi->osi_obdo.o_valid = 0;
249 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
251 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
252 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
254 buf = (char *)&osi->osi_obdo;
255 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
257 rc = out_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size,
258 (const char **)&buf);
263 int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
264 const struct lu_attr *attr, struct thandle *th,
265 struct lustre_capa *capa)
267 CDEBUG(D_INFO, "attr set object "DFID"\n",
268 PFID(&dt->do_lu.lo_header->loh_fid));
273 static void osp_md_object_read_lock(const struct lu_env *env,
274 struct dt_object *dt, unsigned role)
276 struct osp_object *obj = dt2osp_obj(dt);
278 LASSERT(obj->opo_owner != env);
279 down_read_nested(&obj->opo_sem, role);
281 LASSERT(obj->opo_owner == NULL);
284 static void osp_md_object_write_lock(const struct lu_env *env,
285 struct dt_object *dt, unsigned role)
287 struct osp_object *obj = dt2osp_obj(dt);
289 down_write_nested(&obj->opo_sem, role);
291 LASSERT(obj->opo_owner == NULL);
292 obj->opo_owner = env;
295 static void osp_md_object_read_unlock(const struct lu_env *env,
296 struct dt_object *dt)
298 struct osp_object *obj = dt2osp_obj(dt);
300 up_read(&obj->opo_sem);
303 static void osp_md_object_write_unlock(const struct lu_env *env,
304 struct dt_object *dt)
306 struct osp_object *obj = dt2osp_obj(dt);
308 LASSERT(obj->opo_owner == env);
309 obj->opo_owner = NULL;
310 up_write(&obj->opo_sem);
313 static int osp_md_object_write_locked(const struct lu_env *env,
314 struct dt_object *dt)
316 struct osp_object *obj = dt2osp_obj(dt);
318 return obj->opo_owner == env;
321 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
322 struct dt_rec *rec, const struct dt_key *key,
323 struct lustre_capa *capa)
325 struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
326 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
327 struct dt_device *dt_dev = &osp->opd_dt_dev;
328 struct update_request *update;
329 struct ptlrpc_request *req = NULL;
330 int size = strlen((char *)key) + 1;
332 struct update_reply *reply;
337 /* Because it needs send the update buffer right away,
338 * just create an update buffer, instead of attaching the
339 * update_remote list of the thandle.
341 update = out_create_update_req(dt_dev);
343 RETURN(PTR_ERR(update));
345 rc = out_insert_update(env, update, OBJ_INDEX_LOOKUP,
346 lu_object_fid(&dt->do_lu), 1, &size,
347 (const char **)&key);
349 CERROR("%s: Insert update error: rc = %d\n",
350 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
354 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
358 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
360 if (reply->ur_version != UPDATE_REPLY_V1) {
361 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
362 dt_dev->dd_lu_dev.ld_obd->obd_name,
363 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
364 GOTO(out, rc = -EPROTO);
367 rc = update_get_reply_result(reply, NULL, 0);
369 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
370 dt_dev->dd_lu_dev.ld_obd->obd_name,
371 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
375 rc = update_get_reply_buf(reply, lbuf, 0);
379 if (lbuf->lb_len != sizeof(*fid)) {
380 CERROR("%s: lookup "DFID" %s wrong size %d\n",
381 dt_dev->dd_lu_dev.ld_obd->obd_name,
382 PFID(lu_object_fid(&dt->do_lu)), (char *)key,
384 GOTO(out, rc = -EINVAL);
388 fid_le_to_cpu(fid, fid);
389 if (!fid_is_sane(fid)) {
390 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
391 dt_dev->dd_lu_dev.ld_obd->obd_name,
392 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
393 GOTO(out, rc = -EINVAL);
396 memcpy(rec, fid, sizeof(*fid));
402 ptlrpc_req_finished(req);
404 out_destroy_update_req(update);
409 static int osp_md_declare_insert(const struct lu_env *env,
410 struct dt_object *dt,
411 const struct dt_rec *rec,
412 const struct dt_key *key,
415 struct update_request *update;
417 struct lu_fid *rec_fid = (struct lu_fid *)rec;
418 int size[2] = {strlen((char *)key) + 1,
420 const char *bufs[2] = {(char *)key, (char *)rec_fid};
423 update = out_find_create_update_loc(th, dt);
424 if (IS_ERR(update)) {
425 CERROR("%s: Get OSP update buf failed: rc = %d\n",
426 dt->do_lu.lo_dev->ld_obd->obd_name,
427 (int)PTR_ERR(update));
428 return PTR_ERR(update);
431 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
433 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
434 dt->do_lu.lo_dev->ld_obd->obd_name,
435 PFID(fid), (char *)key, PFID(rec_fid));
437 fid_cpu_to_le(rec_fid, rec_fid);
439 rc = out_insert_update(env, update, OBJ_INDEX_INSERT, fid,
440 ARRAY_SIZE(size), size, bufs);
444 static int osp_md_index_insert(const struct lu_env *env,
445 struct dt_object *dt,
446 const struct dt_rec *rec,
447 const struct dt_key *key,
449 struct lustre_capa *capa,
455 static int osp_md_declare_delete(const struct lu_env *env,
456 struct dt_object *dt,
457 const struct dt_key *key,
460 struct update_request *update;
462 int size = strlen((char *)key) + 1;
465 update = out_find_create_update_loc(th, dt);
466 if (IS_ERR(update)) {
467 CERROR("%s: Get OSP update buf failed: rc = %d\n",
468 dt->do_lu.lo_dev->ld_obd->obd_name,
469 (int)PTR_ERR(update));
470 return PTR_ERR(update);
473 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
475 rc = out_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
476 (const char **)&key);
481 static int osp_md_index_delete(const struct lu_env *env,
482 struct dt_object *dt,
483 const struct dt_key *key,
485 struct lustre_capa *capa)
487 CDEBUG(D_INFO, "index delete "DFID" %s\n",
488 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
494 * Creates or initializes iterator context.
496 * Note: for OSP, these index iterate api is only used to check
497 * whether the directory is empty now (see mdd_dir_is_empty).
498 * Since dir_empty will be return by OBJ_ATTR_GET(see osp_attr_get/
499 * out_attr_get). So the implementation of these iterator is simplied
500 * to make mdd_dir_is_empty happy. The real iterator should be
501 * implemented, if we need it one day.
503 static struct dt_it *osp_it_init(const struct lu_env *env,
504 struct dt_object *dt,
506 struct lustre_capa *capa)
508 lu_object_get(&dt->do_lu);
509 return (struct dt_it *)dt;
512 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
514 struct dt_object *dt = (struct dt_object *)di;
515 lu_object_put(env, &dt->do_lu);
518 static int osp_it_get(const struct lu_env *env,
519 struct dt_it *di, const struct dt_key *key)
524 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
529 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
531 struct dt_object *dt = (struct dt_object *)di;
532 struct osp_object *o = dt2osp_obj(dt);
540 static struct dt_key *osp_it_key(const struct lu_env *env,
541 const struct dt_it *di)
547 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
553 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
554 struct dt_rec *lde, __u32 attr)
560 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
566 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
573 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
580 static const struct dt_index_operations osp_md_index_ops = {
581 .dio_lookup = osp_md_index_lookup,
582 .dio_declare_insert = osp_md_declare_insert,
583 .dio_insert = osp_md_index_insert,
584 .dio_declare_delete = osp_md_declare_delete,
585 .dio_delete = osp_md_index_delete,
593 .key_size = osp_it_key_size,
595 .store = osp_it_store,
597 .key_rec = osp_it_key_rec,
601 static int osp_md_index_try(const struct lu_env *env,
602 struct dt_object *dt,
603 const struct dt_index_features *feat)
605 dt->do_index_ops = &osp_md_index_ops;
609 static int osp_md_object_lock(const struct lu_env *env,
610 struct dt_object *dt,
611 struct lustre_handle *lh,
612 struct ldlm_enqueue_info *einfo,
615 struct osp_thread_info *info = osp_env_info(env);
616 struct ldlm_res_id *res_id = &info->osi_resid;
617 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
618 struct osp_device *osp = dt2osp_dev(dt_dev);
619 struct ptlrpc_request *req = NULL;
624 fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
626 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
627 LDLM_FL_BLOCK_GRANTED, res_id,
629 (ldlm_policy_data_t *)policy,
630 einfo->ei_mode, lh, 0);
634 req = ldlm_enqueue_pack(osp->opd_exp, 0);
636 RETURN(PTR_ERR(req));
638 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
639 (const ldlm_policy_data_t *)policy,
640 &flags, NULL, 0, LVB_T_NONE, lh, 0);
642 ptlrpc_req_finished(req);
644 return rc == ELDLM_OK ? 0 : -EIO;
647 struct dt_object_operations osp_md_obj_ops = {
648 .do_read_lock = osp_md_object_read_lock,
649 .do_write_lock = osp_md_object_write_lock,
650 .do_read_unlock = osp_md_object_read_unlock,
651 .do_write_unlock = osp_md_object_write_unlock,
652 .do_write_locked = osp_md_object_write_locked,
653 .do_declare_create = osp_md_declare_object_create,
654 .do_create = osp_md_object_create,
655 .do_declare_ref_add = osp_md_declare_ref_add,
656 .do_ref_add = osp_md_object_ref_add,
657 .do_declare_ref_del = osp_md_declare_object_ref_del,
658 .do_ref_del = osp_md_object_ref_del,
659 .do_declare_destroy = osp_declare_object_destroy,
660 .do_destroy = osp_object_destroy,
661 .do_ah_init = osp_md_ah_init,
662 .do_attr_get = osp_attr_get,
663 .do_declare_attr_set = osp_md_declare_attr_set,
664 .do_attr_set = osp_md_attr_set,
665 .do_xattr_get = osp_xattr_get,
666 .do_declare_xattr_set = osp_declare_xattr_set,
667 .do_xattr_set = osp_xattr_set,
668 .do_index_try = osp_md_index_try,
669 .do_object_lock = osp_md_object_lock,