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->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 LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE)));
255 obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
257 lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
258 obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
260 buf = (char *)&osi->osi_obdo;
261 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
263 rc = out_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size,
264 (const char **)&buf);
269 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
270 const struct lu_attr *attr, struct thandle *th,
271 struct lustre_capa *capa)
273 CDEBUG(D_INFO, "attr set object "DFID"\n",
274 PFID(&dt->do_lu.lo_header->loh_fid));
279 static void osp_md_object_read_lock(const struct lu_env *env,
280 struct dt_object *dt, unsigned role)
282 struct osp_object *obj = dt2osp_obj(dt);
284 LASSERT(obj->opo_owner != env);
285 down_read_nested(&obj->opo_sem, role);
287 LASSERT(obj->opo_owner == NULL);
290 static void osp_md_object_write_lock(const struct lu_env *env,
291 struct dt_object *dt, unsigned role)
293 struct osp_object *obj = dt2osp_obj(dt);
295 down_write_nested(&obj->opo_sem, role);
297 LASSERT(obj->opo_owner == NULL);
298 obj->opo_owner = env;
301 static void osp_md_object_read_unlock(const struct lu_env *env,
302 struct dt_object *dt)
304 struct osp_object *obj = dt2osp_obj(dt);
306 up_read(&obj->opo_sem);
309 static void osp_md_object_write_unlock(const struct lu_env *env,
310 struct dt_object *dt)
312 struct osp_object *obj = dt2osp_obj(dt);
314 LASSERT(obj->opo_owner == env);
315 obj->opo_owner = NULL;
316 up_write(&obj->opo_sem);
319 static int osp_md_object_write_locked(const struct lu_env *env,
320 struct dt_object *dt)
322 struct osp_object *obj = dt2osp_obj(dt);
324 return obj->opo_owner == env;
327 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
328 struct dt_rec *rec, const struct dt_key *key,
329 struct lustre_capa *capa)
331 struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
332 struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
333 struct dt_device *dt_dev = &osp->opd_dt_dev;
334 struct update_request *update;
335 struct ptlrpc_request *req = NULL;
336 int size = strlen((char *)key) + 1;
338 struct update_reply *reply;
343 /* Because it needs send the update buffer right away,
344 * just create an update buffer, instead of attaching the
345 * update_remote list of the thandle.
347 update = out_create_update_req(dt_dev);
349 RETURN(PTR_ERR(update));
351 rc = out_insert_update(env, update, OBJ_INDEX_LOOKUP,
352 lu_object_fid(&dt->do_lu), 1, &size,
353 (const char **)&key);
355 CERROR("%s: Insert update error: rc = %d\n",
356 dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
360 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
364 reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
366 if (reply->ur_version != UPDATE_REPLY_V1) {
367 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
368 dt_dev->dd_lu_dev.ld_obd->obd_name,
369 reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
370 GOTO(out, rc = -EPROTO);
373 rc = update_get_reply_result(reply, NULL, 0);
375 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
376 dt_dev->dd_lu_dev.ld_obd->obd_name,
377 PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
381 rc = update_get_reply_buf(reply, lbuf, 0);
385 if (lbuf->lb_len != sizeof(*fid)) {
386 CERROR("%s: lookup "DFID" %s wrong size %d\n",
387 dt_dev->dd_lu_dev.ld_obd->obd_name,
388 PFID(lu_object_fid(&dt->do_lu)), (char *)key,
390 GOTO(out, rc = -EINVAL);
394 fid_le_to_cpu(fid, fid);
395 if (!fid_is_sane(fid)) {
396 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
397 dt_dev->dd_lu_dev.ld_obd->obd_name,
398 PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
399 GOTO(out, rc = -EINVAL);
402 memcpy(rec, fid, sizeof(*fid));
408 ptlrpc_req_finished(req);
410 out_destroy_update_req(update);
415 static int osp_md_declare_insert(const struct lu_env *env,
416 struct dt_object *dt,
417 const struct dt_rec *rec,
418 const struct dt_key *key,
421 struct update_request *update;
423 struct lu_fid *rec_fid = (struct lu_fid *)rec;
424 int size[2] = {strlen((char *)key) + 1,
426 const char *bufs[2] = {(char *)key, (char *)rec_fid};
429 update = out_find_create_update_loc(th, dt);
430 if (IS_ERR(update)) {
431 CERROR("%s: Get OSP update buf failed: rc = %d\n",
432 dt->do_lu.lo_dev->ld_obd->obd_name,
433 (int)PTR_ERR(update));
434 return PTR_ERR(update);
437 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
439 CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
440 dt->do_lu.lo_dev->ld_obd->obd_name,
441 PFID(fid), (char *)key, PFID(rec_fid));
443 fid_cpu_to_le(rec_fid, rec_fid);
445 rc = out_insert_update(env, update, OBJ_INDEX_INSERT, fid,
446 ARRAY_SIZE(size), size, bufs);
450 static int osp_md_index_insert(const struct lu_env *env,
451 struct dt_object *dt,
452 const struct dt_rec *rec,
453 const struct dt_key *key,
455 struct lustre_capa *capa,
461 static int osp_md_declare_delete(const struct lu_env *env,
462 struct dt_object *dt,
463 const struct dt_key *key,
466 struct update_request *update;
468 int size = strlen((char *)key) + 1;
471 update = out_find_create_update_loc(th, dt);
472 if (IS_ERR(update)) {
473 CERROR("%s: Get OSP update buf failed: rc = %d\n",
474 dt->do_lu.lo_dev->ld_obd->obd_name,
475 (int)PTR_ERR(update));
476 return PTR_ERR(update);
479 fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
481 rc = out_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
482 (const char **)&key);
487 static int osp_md_index_delete(const struct lu_env *env,
488 struct dt_object *dt,
489 const struct dt_key *key,
491 struct lustre_capa *capa)
493 CDEBUG(D_INFO, "index delete "DFID" %s\n",
494 PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
500 * Creates or initializes iterator context.
502 * Note: for OSP, these index iterate api is only used to check
503 * whether the directory is empty now (see mdd_dir_is_empty).
504 * Since dir_empty will be return by OBJ_ATTR_GET(see osp_attr_get/
505 * out_attr_get). So the implementation of these iterator is simplied
506 * to make mdd_dir_is_empty happy. The real iterator should be
507 * implemented, if we need it one day.
509 static struct dt_it *osp_it_init(const struct lu_env *env,
510 struct dt_object *dt,
512 struct lustre_capa *capa)
514 lu_object_get(&dt->do_lu);
515 return (struct dt_it *)dt;
518 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
520 struct dt_object *dt = (struct dt_object *)di;
521 lu_object_put(env, &dt->do_lu);
524 static int osp_it_get(const struct lu_env *env,
525 struct dt_it *di, const struct dt_key *key)
530 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
535 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
537 struct dt_object *dt = (struct dt_object *)di;
538 struct osp_object *o = dt2osp_obj(dt);
546 static struct dt_key *osp_it_key(const struct lu_env *env,
547 const struct dt_it *di)
553 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
559 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
560 struct dt_rec *lde, __u32 attr)
566 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
572 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
579 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
586 static const struct dt_index_operations osp_md_index_ops = {
587 .dio_lookup = osp_md_index_lookup,
588 .dio_declare_insert = osp_md_declare_insert,
589 .dio_insert = osp_md_index_insert,
590 .dio_declare_delete = osp_md_declare_delete,
591 .dio_delete = osp_md_index_delete,
599 .key_size = osp_it_key_size,
601 .store = osp_it_store,
603 .key_rec = osp_it_key_rec,
607 static int osp_md_index_try(const struct lu_env *env,
608 struct dt_object *dt,
609 const struct dt_index_features *feat)
611 dt->do_index_ops = &osp_md_index_ops;
615 static int osp_md_object_lock(const struct lu_env *env,
616 struct dt_object *dt,
617 struct lustre_handle *lh,
618 struct ldlm_enqueue_info *einfo,
621 struct osp_thread_info *info = osp_env_info(env);
622 struct ldlm_res_id *res_id = &info->osi_resid;
623 struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
624 struct osp_device *osp = dt2osp_dev(dt_dev);
625 struct ptlrpc_request *req = NULL;
630 fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
632 mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
633 LDLM_FL_BLOCK_GRANTED, res_id,
635 (ldlm_policy_data_t *)policy,
636 einfo->ei_mode, lh, 0);
640 req = ldlm_enqueue_pack(osp->opd_exp, 0);
642 RETURN(PTR_ERR(req));
644 rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
645 (const ldlm_policy_data_t *)policy,
646 &flags, NULL, 0, LVB_T_NONE, lh, 0);
648 ptlrpc_req_finished(req);
650 return rc == ELDLM_OK ? 0 : -EIO;
653 struct dt_object_operations osp_md_obj_ops = {
654 .do_read_lock = osp_md_object_read_lock,
655 .do_write_lock = osp_md_object_write_lock,
656 .do_read_unlock = osp_md_object_read_unlock,
657 .do_write_unlock = osp_md_object_write_unlock,
658 .do_write_locked = osp_md_object_write_locked,
659 .do_declare_create = osp_md_declare_object_create,
660 .do_create = osp_md_object_create,
661 .do_declare_ref_add = osp_md_declare_ref_add,
662 .do_ref_add = osp_md_object_ref_add,
663 .do_declare_ref_del = osp_md_declare_object_ref_del,
664 .do_ref_del = osp_md_object_ref_del,
665 .do_declare_destroy = osp_declare_object_destroy,
666 .do_destroy = osp_object_destroy,
667 .do_ah_init = osp_md_ah_init,
668 .do_attr_get = osp_attr_get,
669 .do_declare_attr_set = osp_md_declare_attr_set,
670 .do_attr_set = osp_md_attr_set,
671 .do_xattr_get = osp_xattr_get,
672 .do_declare_xattr_set = osp_declare_xattr_set,
673 .do_xattr_set = osp_xattr_set,
674 .do_index_try = osp_md_index_try,
675 .do_object_lock = osp_md_object_lock,