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 static 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 LASSERT(S_ISDIR(attr->la_mode));
66 obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
67 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
68 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
70 bufs[0] = (char *)&osi->osi_obdo;
72 fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
73 if (hint != NULL && hint->dah_parent) {
75 struct lu_fid *tmp_fid = &osi->osi_fid;
77 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
78 fid_cpu_to_le(tmp_fid, fid2);
79 sizes[1] = sizeof(*tmp_fid);
80 bufs[1] = (char *)tmp_fid;
84 if (lu_object_exists(&dt->do_lu)) {
85 /* If the object already exists, we needs to destroy
86 * this orphan object first.
88 * The scenario might happen in this case
90 * 1. client send remote create to MDT0.
91 * 2. MDT0 send create update to MDT1.
92 * 3. MDT1 finished create synchronously.
93 * 4. MDT0 failed and reboot.
94 * 5. client resend remote create to MDT0.
95 * 6. MDT0 tries to resend create update to MDT1,
96 * but find the object already exists
98 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
99 dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
101 rc = out_insert_update(env, update, OBJ_REF_DEL, fid1, 0,
106 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
107 /* decrease for ".." */
108 rc = out_insert_update(env, update, OBJ_REF_DEL, fid1,
114 rc = out_insert_update(env, update, OBJ_DESTROY, fid1, 0, NULL,
119 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
120 /* Increase batchid to add this orphan object deletion
121 * to separate transaction */
122 update_inc_batchid(update);
125 rc = out_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes,
126 (const char **)bufs);
129 CERROR("%s: Insert update error: rc = %d\n",
130 dt->do_lu.lo_dev->ld_obd->obd_name, rc);
135 static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
136 struct lu_attr *attr,
137 struct dt_allocation_hint *hint,
138 struct dt_object_format *dof,
141 struct osp_object *obj = dt2osp_obj(dt);
143 CDEBUG(D_INFO, "create object "DFID"\n",
144 PFID(&dt->do_lu.lo_header->loh_fid));
146 /* Because the create update RPC will be sent during declare phase,
147 * if creation reaches here, it means the object has been created
149 dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
155 static int osp_md_declare_object_ref_del(const struct lu_env *env,
156 struct dt_object *dt,
159 struct update_request *update;
163 update = out_find_create_update_loc(th, dt);
164 if (IS_ERR(update)) {
165 CERROR("%s: Get OSP update buf failed: rc = %d\n",
166 dt->do_lu.lo_dev->ld_obd->obd_name,
167 (int)PTR_ERR(update));
168 return PTR_ERR(update);
171 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
173 rc = out_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL);
178 static int osp_md_object_ref_del(const struct lu_env *env,
179 struct dt_object *dt,
182 CDEBUG(D_INFO, "ref del object "DFID"\n",
183 PFID(&dt->do_lu.lo_header->loh_fid));
188 static int osp_md_declare_ref_add(const struct lu_env *env,
189 struct dt_object *dt, struct thandle *th)
191 struct update_request *update;
195 update = out_find_create_update_loc(th, dt);
196 if (IS_ERR(update)) {
197 CERROR("%s: Get OSP update buf failed: rc = %d\n",
198 dt->do_lu.lo_dev->ld_obd->obd_name,
199 (int)PTR_ERR(update));
200 return PTR_ERR(update);
203 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
205 rc = out_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL);
210 static int osp_md_object_ref_add(const struct lu_env *env,
211 struct dt_object *dt,
214 CDEBUG(D_INFO, "ref add object "DFID"\n",
215 PFID(&dt->do_lu.lo_header->loh_fid));
220 static void osp_md_ah_init(const struct lu_env *env,
221 struct dt_allocation_hint *ah,
222 struct dt_object *parent,
223 struct dt_object *child,
228 memset(ah, 0, sizeof(*ah));
229 ah->dah_parent = parent;
230 ah->dah_mode = child_mode;
233 static int osp_md_declare_attr_set(const struct lu_env *env,
234 struct dt_object *dt,
235 const struct lu_attr *attr,
238 struct osp_thread_info *osi = osp_env_info(env);
239 struct update_request *update;
241 int size = sizeof(struct obdo);
245 update = out_find_create_update_loc(th, dt);
246 if (IS_ERR(update)) {
247 CERROR("%s: Get OSP update buf failed: %d\n",
248 dt->do_lu.lo_dev->ld_obd->obd_name,
249 (int)PTR_ERR(update));
250 return PTR_ERR(update);
253 osi->osi_obdo.o_valid = 0;
254 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
256 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
257 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
259 buf = (char *)&osi->osi_obdo;
260 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
262 rc = out_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size,
263 (const char **)&buf);
268 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
269 const struct lu_attr *attr, struct thandle *th,
270 struct lustre_capa *capa)
272 CDEBUG(D_INFO, "attr set object "DFID"\n",
273 PFID(&dt->do_lu.lo_header->loh_fid));
278 static void osp_md_object_read_lock(const struct lu_env *env,
279 struct dt_object *dt, unsigned role)
281 struct osp_object *obj = dt2osp_obj(dt);
283 LASSERT(obj->opo_owner != env);
284 down_read_nested(&obj->opo_sem, role);
286 LASSERT(obj->opo_owner == NULL);
289 static void osp_md_object_write_lock(const struct lu_env *env,
290 struct dt_object *dt, unsigned role)
292 struct osp_object *obj = dt2osp_obj(dt);
294 down_write_nested(&obj->opo_sem, role);
296 LASSERT(obj->opo_owner == NULL);
297 obj->opo_owner = env;
300 static void osp_md_object_read_unlock(const struct lu_env *env,
301 struct dt_object *dt)
303 struct osp_object *obj = dt2osp_obj(dt);
305 up_read(&obj->opo_sem);
308 static void osp_md_object_write_unlock(const struct lu_env *env,
309 struct dt_object *dt)
311 struct osp_object *obj = dt2osp_obj(dt);
313 LASSERT(obj->opo_owner == env);
314 obj->opo_owner = NULL;
315 up_write(&obj->opo_sem);
318 static int osp_md_object_write_locked(const struct lu_env *env,
319 struct dt_object *dt)
321 struct osp_object *obj = dt2osp_obj(dt);
323 return obj->opo_owner == env;
326 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
327 struct dt_rec *rec, const struct dt_key *key,
328 struct lustre_capa *capa)
330 struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
331 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
332 struct dt_device *dt_dev = &osp->opd_dt_dev;
333 struct update_request *update;
334 struct ptlrpc_request *req = NULL;
335 int size = strlen((char *)key) + 1;
337 struct update_reply *reply;
342 /* Because it needs send the update buffer right away,
343 * just create an update buffer, instead of attaching the
344 * update_remote list of the thandle.
346 update = out_create_update_req(dt_dev);
348 RETURN(PTR_ERR(update));
350 rc = out_insert_update(env, update, OBJ_INDEX_LOOKUP,
351 lu_object_fid(&dt->do_lu), 1, &size,
352 (const char **)&key);
354 CERROR("%s: Insert update error: rc = %d\n",
355 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
359 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
363 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
365 if (reply->ur_version != UPDATE_REPLY_V1) {
366 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
367 dt_dev->dd_lu_dev.ld_obd->obd_name,
368 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
369 GOTO(out, rc = -EPROTO);
372 rc = update_get_reply_result(reply, NULL, 0);
374 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
375 dt_dev->dd_lu_dev.ld_obd->obd_name,
376 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
380 rc = update_get_reply_buf(reply, lbuf, 0);
384 if (lbuf->lb_len != sizeof(*fid)) {
385 CERROR("%s: lookup "DFID" %s wrong size %d\n",
386 dt_dev->dd_lu_dev.ld_obd->obd_name,
387 PFID(lu_object_fid(&dt->do_lu)), (char *)key,
389 GOTO(out, rc = -EINVAL);
393 fid_le_to_cpu(fid, fid);
394 if (!fid_is_sane(fid)) {
395 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
396 dt_dev->dd_lu_dev.ld_obd->obd_name,
397 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
398 GOTO(out, rc = -EINVAL);
401 memcpy(rec, fid, sizeof(*fid));
407 ptlrpc_req_finished(req);
409 out_destroy_update_req(update);
414 static int osp_md_declare_insert(const struct lu_env *env,
415 struct dt_object *dt,
416 const struct dt_rec *rec,
417 const struct dt_key *key,
420 struct update_request *update;
422 struct lu_fid *rec_fid = (struct lu_fid *)rec;
423 int size[2] = {strlen((char *)key) + 1,
425 const char *bufs[2] = {(char *)key, (char *)rec_fid};
428 update = out_find_create_update_loc(th, dt);
429 if (IS_ERR(update)) {
430 CERROR("%s: Get OSP update buf failed: rc = %d\n",
431 dt->do_lu.lo_dev->ld_obd->obd_name,
432 (int)PTR_ERR(update));
433 return PTR_ERR(update);
436 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
438 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
439 dt->do_lu.lo_dev->ld_obd->obd_name,
440 PFID(fid), (char *)key, PFID(rec_fid));
442 fid_cpu_to_le(rec_fid, rec_fid);
444 rc = out_insert_update(env, update, OBJ_INDEX_INSERT, fid,
445 ARRAY_SIZE(size), size, bufs);
449 static int osp_md_index_insert(const struct lu_env *env,
450 struct dt_object *dt,
451 const struct dt_rec *rec,
452 const struct dt_key *key,
454 struct lustre_capa *capa,
460 static int osp_md_declare_delete(const struct lu_env *env,
461 struct dt_object *dt,
462 const struct dt_key *key,
465 struct update_request *update;
467 int size = strlen((char *)key) + 1;
470 update = out_find_create_update_loc(th, dt);
471 if (IS_ERR(update)) {
472 CERROR("%s: Get OSP update buf failed: rc = %d\n",
473 dt->do_lu.lo_dev->ld_obd->obd_name,
474 (int)PTR_ERR(update));
475 return PTR_ERR(update);
478 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
480 rc = out_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
481 (const char **)&key);
486 static int osp_md_index_delete(const struct lu_env *env,
487 struct dt_object *dt,
488 const struct dt_key *key,
490 struct lustre_capa *capa)
492 CDEBUG(D_INFO, "index delete "DFID" %s\n",
493 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
499 * Creates or initializes iterator context.
501 * Note: for OSP, these index iterate api is only used to check
502 * whether the directory is empty now (see mdd_dir_is_empty).
503 * Since dir_empty will be return by OBJ_ATTR_GET(see osp_attr_get/
504 * out_attr_get). So the implementation of these iterator is simplied
505 * to make mdd_dir_is_empty happy. The real iterator should be
506 * implemented, if we need it one day.
508 static struct dt_it *osp_it_init(const struct lu_env *env,
509 struct dt_object *dt,
511 struct lustre_capa *capa)
513 lu_object_get(&dt->do_lu);
514 return (struct dt_it *)dt;
517 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
519 struct dt_object *dt = (struct dt_object *)di;
520 lu_object_put(env, &dt->do_lu);
523 static int osp_it_get(const struct lu_env *env,
524 struct dt_it *di, const struct dt_key *key)
529 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
534 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
536 struct dt_object *dt = (struct dt_object *)di;
537 struct osp_object *o = dt2osp_obj(dt);
545 static struct dt_key *osp_it_key(const struct lu_env *env,
546 const struct dt_it *di)
552 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
558 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
559 struct dt_rec *lde, __u32 attr)
565 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
571 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
578 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
585 static const struct dt_index_operations osp_md_index_ops = {
586 .dio_lookup = osp_md_index_lookup,
587 .dio_declare_insert = osp_md_declare_insert,
588 .dio_insert = osp_md_index_insert,
589 .dio_declare_delete = osp_md_declare_delete,
590 .dio_delete = osp_md_index_delete,
598 .key_size = osp_it_key_size,
600 .store = osp_it_store,
602 .key_rec = osp_it_key_rec,
606 static int osp_md_index_try(const struct lu_env *env,
607 struct dt_object *dt,
608 const struct dt_index_features *feat)
610 dt->do_index_ops = &osp_md_index_ops;
614 static int osp_md_object_lock(const struct lu_env *env,
615 struct dt_object *dt,
616 struct lustre_handle *lh,
617 struct ldlm_enqueue_info *einfo,
620 struct osp_thread_info *info = osp_env_info(env);
621 struct ldlm_res_id *res_id = &info->osi_resid;
622 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
623 struct osp_device *osp = dt2osp_dev(dt_dev);
624 struct ptlrpc_request *req = NULL;
629 fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
631 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
632 LDLM_FL_BLOCK_GRANTED, res_id,
634 (ldlm_policy_data_t *)policy,
635 einfo->ei_mode, lh, 0);
639 req = ldlm_enqueue_pack(osp->opd_exp, 0);
641 RETURN(PTR_ERR(req));
643 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
644 (const ldlm_policy_data_t *)policy,
645 &flags, NULL, 0, LVB_T_NONE, lh, 0);
647 ptlrpc_req_finished(req);
649 return rc == ELDLM_OK ? 0 : -EIO;
652 struct dt_object_operations osp_md_obj_ops = {
653 .do_read_lock = osp_md_object_read_lock,
654 .do_write_lock = osp_md_object_write_lock,
655 .do_read_unlock = osp_md_object_read_unlock,
656 .do_write_unlock = osp_md_object_write_unlock,
657 .do_write_locked = osp_md_object_write_locked,
658 .do_declare_create = osp_md_declare_object_create,
659 .do_create = osp_md_object_create,
660 .do_declare_ref_add = osp_md_declare_ref_add,
661 .do_ref_add = osp_md_object_ref_add,
662 .do_declare_ref_del = osp_md_declare_object_ref_del,
663 .do_ref_del = osp_md_object_ref_del,
664 .do_declare_destroy = osp_declare_object_destroy,
665 .do_destroy = osp_object_destroy,
666 .do_ah_init = osp_md_ah_init,
667 .do_attr_get = osp_attr_get,
668 .do_declare_attr_set = osp_md_declare_attr_set,
669 .do_attr_set = osp_md_attr_set,
670 .do_xattr_get = osp_xattr_get,
671 .do_declare_xattr_set = osp_declare_xattr_set,
672 .do_xattr_set = osp_xattr_set,
673 .do_index_try = osp_md_index_try,
674 .do_object_lock = osp_md_object_lock,