X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosp%2Fosp_md_object.c;h=a66dd6586896621bb5c29486fd06f04378b256e7;hb=9ff2d957982160103b5d885c9a532ad45bdf8d4d;hp=fd262ac969a6510c7ba797a0c4c68f8eefc26ab2;hpb=a52aa3b4d9dd5efb87bc983c13b9f58d61ce8a8e;p=fs%2Flustre-release.git diff --git a/lustre/osp/osp_md_object.c b/lustre/osp/osp_md_object.c index fd262ac..a66dd65 100644 --- a/lustre/osp/osp_md_object.c +++ b/lustre/osp/osp_md_object.c @@ -25,7 +25,28 @@ /* * lustre/osp/osp_md_object.c * - * Lustre MDT Proxy Device + * OST/MDT proxy device (OSP) Metadata methods + * + * This file implements methods for remote MD object, which include + * dt_object_operations, dt_index_operations and dt_body_operations. + * + * If there are multiple MDTs in one filesystem, one operation might + * include modifications in several MDTs. In such cases, clients + * send the RPC to the master MDT, then the operation is decomposed into + * object updates which will be dispatched to OSD or OSP. The local updates + * go to local OSD and the remote updates go to OSP. In OSP, these remote + * object updates will be packed into an update RPC, sent to the remote MDT + * and handled by Object Update Target (OUT). + * + * In DNE phase I, because of missing complete recovery solution, updates + * will be executed in order and synchronously. + * 1. The transaction is created. + * 2. In transaction declare, it collects and packs remote + * updates (in osp_md_declare_xxx()). + * 3. In transaction start, it sends these remote updates + * to remote MDTs, which will execute these updates synchronously. + * 4. In transaction execute phase, the local updates will be executed + * synchronously. * * Author: Di Wang */ @@ -38,22 +59,35 @@ static const char dot[] = "."; static const char dotdot[] = ".."; -static int osp_md_declare_object_create(const struct lu_env *env, - struct dt_object *dt, - struct lu_attr *attr, - struct dt_allocation_hint *hint, - struct dt_object_format *dof, - struct thandle *th) +/** + * Implementation of dt_object_operations::do_declare_create + * + * Insert object create update into the RPC, which will be sent during + * transaction start. Note: if the object has already been created, + * we must add object destroy updates ahead of create updates, so it will + * destroy then recreate the object. + * + * \param[in] env execution environment + * \param[in] dt remote object to be created + * \param[in] attr attribute of the created object + * \param[in] hint creation hint + * \param[in] dof creation format information + * \param[in] th the transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ +int osp_md_declare_object_create(const struct lu_env *env, + struct dt_object *dt, + struct lu_attr *attr, + struct dt_allocation_hint *hint, + struct dt_object_format *dof, + struct thandle *th) { - struct osp_thread_info *osi = osp_env_info(env); - struct update_request *update; - struct lu_fid *fid1; - int sizes[2] = {sizeof(struct obdo), 0}; - char *bufs[2] = {NULL, NULL}; - int buf_count; - int rc; + struct dt_update_request *update; + int rc; - update = out_find_create_update_loc(th, dt); + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -61,26 +95,6 @@ static int osp_md_declare_object_create(const struct lu_env *env, return PTR_ERR(update); } - osi->osi_obdo.o_valid = 0; - LASSERT(S_ISDIR(attr->la_mode)); - obdo_from_la(&osi->osi_obdo, attr, attr->la_valid); - lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo); - obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo); - - bufs[0] = (char *)&osi->osi_obdo; - buf_count = 1; - fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu); - if (hint->dah_parent) { - struct lu_fid *fid2; - struct lu_fid *tmp_fid = &osi->osi_fid; - - fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu); - fid_cpu_to_le(tmp_fid, fid2); - sizes[1] = sizeof(*tmp_fid); - bufs[1] = (char *)tmp_fid; - buf_count++; - } - if (lu_object_exists(&dt->do_lu)) { /* If the object already exists, we needs to destroy * this orphan object first. @@ -96,23 +110,27 @@ static int osp_md_declare_object_create(const struct lu_env *env, * but find the object already exists */ CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n", - dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1)); + dt->do_lu.lo_dev->ld_obd->obd_name, + PFID(lu_object_fid(&dt->do_lu))); - rc = out_insert_update(env, update, OBJ_REF_DEL, fid1, 0, - NULL, NULL); + rc = out_ref_del_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), + update->dur_batchid); if (rc != 0) GOTO(out, rc); if (S_ISDIR(lu_object_attr(&dt->do_lu))) { /* decrease for ".." */ - rc = out_insert_update(env, update, OBJ_REF_DEL, fid1, - 0, NULL, NULL); + rc = out_ref_del_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), + update->dur_batchid); if (rc != 0) GOTO(out, rc); } - rc = out_insert_update(env, update, OBJ_DESTROY, fid1, 0, NULL, - NULL); + rc = out_object_destroy_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), + update->dur_batchid); if (rc != 0) GOTO(out, rc); @@ -122,8 +140,11 @@ static int osp_md_declare_object_create(const struct lu_env *env, update_inc_batchid(update); } - rc = out_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes, - (const char **)bufs); + rc = out_create_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), attr, hint, dof, + update->dur_batchid); + if (rc != 0) + GOTO(out, rc); out: if (rc) CERROR("%s: Insert update error: rc = %d\n", @@ -132,14 +153,26 @@ out: return rc; } -static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt, - struct lu_attr *attr, - struct dt_allocation_hint *hint, - struct dt_object_format *dof, - struct thandle *th) +/** + * Implementation of dt_object_operations::do_create + * + * It sets necessary flags for created object. In DNE phase I, + * remote updates are actually executed during transaction start, + * i.e. the object has already been created when calling this method. + * + * \param[in] env execution environment + * \param[in] dt object to be created + * \param[in] attr attribute of the created object + * \param[in] hint creation hint + * \param[in] dof creation format information + * \param[in] th the transaction handle + * + * \retval only return 0 for now + */ +int osp_md_object_create(const struct lu_env *env, struct dt_object *dt, + struct lu_attr *attr, struct dt_allocation_hint *hint, + struct dt_object_format *dof, struct thandle *th) { - struct osp_object *obj = dt2osp_obj(dt); - CDEBUG(D_INFO, "create object "DFID"\n", PFID(&dt->do_lu.lo_header->loh_fid)); @@ -147,20 +180,33 @@ static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt, * if creation reaches here, it means the object has been created * successfully */ dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT); - obj->opo_empty = 1; + dt2osp_obj(dt)->opo_non_exist = 0; return 0; } +/** + * Implementation of dt_object_operations::do_declare_ref_del + * + * Declare decreasing the reference count of the remote object, i.e. insert + * decreasing object reference count update into the RPC, which will be sent + * during transaction start. + * + * \param[in] env execution environment + * \param[in] dt object to decrease the reference count. + * \param[in] th the transaction handle of refcount decrease. + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ static int osp_md_declare_object_ref_del(const struct lu_env *env, struct dt_object *dt, struct thandle *th) { - struct update_request *update; - struct lu_fid *fid; - int rc; + struct dt_update_request *update; + int rc; - update = out_find_create_update_loc(th, dt); + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -168,13 +214,25 @@ static int osp_md_declare_object_ref_del(const struct lu_env *env, return PTR_ERR(update); } - fid = (struct lu_fid *)lu_object_fid(&dt->do_lu); - - rc = out_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL); - + rc = out_ref_del_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), + update->dur_batchid); return rc; } +/** + * Implementation of dt_object_operations::do_ref_del + * + * Do nothing in this method for now. In DNE phase I, remote updates are + * actually executed during transaction start, i.e. the object reference + * count has already been decreased when calling this method. + * + * \param[in] env execution environment + * \param[in] dt object to decrease the reference count + * \param[in] th the transaction handle + * + * \retval only return 0 for now + */ static int osp_md_object_ref_del(const struct lu_env *env, struct dt_object *dt, struct thandle *th) @@ -185,14 +243,26 @@ static int osp_md_object_ref_del(const struct lu_env *env, return 0; } +/** + * Implementation of dt_object_operations::do_declare_ref_del + * + * Declare increasing the reference count of the remote object, + * i.e. insert increasing object reference count update into RPC. + * + * \param[in] env execution environment + * \param[in] dt object on which to increase the reference count. + * \param[in] th the transaction handle. + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ static int osp_md_declare_ref_add(const struct lu_env *env, struct dt_object *dt, struct thandle *th) { - struct update_request *update; - struct lu_fid *fid; - int rc; + struct dt_update_request *update; + int rc; - update = out_find_create_update_loc(th, dt); + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -200,15 +270,27 @@ static int osp_md_declare_ref_add(const struct lu_env *env, return PTR_ERR(update); } - fid = (struct lu_fid *)lu_object_fid(&dt->do_lu); - - rc = out_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL); + rc = out_ref_add_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), + update->dur_batchid); return rc; } -static int osp_md_object_ref_add(const struct lu_env *env, - struct dt_object *dt, +/** + * Implementation of dt_object_operations::do_ref_add + * + * Do nothing in this method for now. In DNE phase I, remote updates are + * actually executed during transaction start, i.e. the object reference + * count has already been increased when calling this method. + * + * \param[in] env execution environment + * \param[in] dt object on which to increase the reference count + * \param[in] th the transaction handle + * + * \retval only return 0 for now + */ +static int osp_md_object_ref_add(const struct lu_env *env, struct dt_object *dt, struct thandle *th) { CDEBUG(D_INFO, "ref add object "DFID"\n", @@ -217,6 +299,20 @@ static int osp_md_object_ref_add(const struct lu_env *env, return 0; } +/** + * Implementation of dt_object_operations::do_ah_init + * + * Initialize the allocation hint for object creation, which is usually called + * before the creation, and these hints (parent and child mode) will be sent to + * the remote Object Update Target (OUT) and used in the object create process, + * same as OSD object creation. + * + * \param[in] env execution environment + * \param[in] ah the hint to be initialized + * \param[in] parent the parent of the object + * \param[in] child the object to be created + * \param[in] child_mode the mode of the created object + */ static void osp_md_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah, struct dt_object *parent, @@ -225,24 +321,28 @@ static void osp_md_ah_init(const struct lu_env *env, { LASSERT(ah); - memset(ah, 0, sizeof(*ah)); ah->dah_parent = parent; ah->dah_mode = child_mode; } -static int osp_md_declare_attr_set(const struct lu_env *env, - struct dt_object *dt, - const struct lu_attr *attr, - struct thandle *th) +/** + * Add attr_set sub-request into the OUT RPC. + * + * \param[in] env execution environment + * \param[in] dt object on which to set attributes + * \param[in] attr attributes to be set + * \param[in] th the transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ +int __osp_md_attr_set(const struct lu_env *env, struct dt_object *dt, + const struct lu_attr *attr, struct thandle *th) { - struct osp_thread_info *osi = osp_env_info(env); - struct update_request *update; - struct lu_fid *fid; - int size = sizeof(struct obdo); - char *buf; - int rc; - - update = out_find_create_update_loc(th, dt); + struct dt_update_request *update; + int rc; + + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -250,32 +350,90 @@ static int osp_md_declare_attr_set(const struct lu_env *env, return PTR_ERR(update); } - osi->osi_obdo.o_valid = 0; - LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE))); - obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr, - attr->la_valid); - lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo); - obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo); + rc = out_attr_set_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), attr, + update->dur_batchid); + + return rc; +} + +/** + * Implementation of dt_object_operations::do_declare_attr_get + * + * Declare setting attributes to the specified remote object. + * + * If the transaction is a remote transaction, then add the modification + * sub-request into the OUT RPC here, and such OUT RPC will be triggered + * when transaction start. + * + * \param[in] env execution environment + * \param[in] dt object on which to set attributes + * \param[in] attr attributes to be set + * \param[in] th the transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ +int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt, + const struct lu_attr *attr, struct thandle *th) +{ + int rc = 0; - buf = (char *)&osi->osi_obdo; - fid = (struct lu_fid *)lu_object_fid(&dt->do_lu); + CDEBUG(D_INFO, "declare attr set object "DFID"\n", + PFID(&dt->do_lu.lo_header->loh_fid)); - rc = out_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size, - (const char **)&buf); + if (!is_only_remote_trans(th)) + rc = __osp_md_attr_set(env, dt, attr, th); return rc; } -static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt, - const struct lu_attr *attr, struct thandle *th, - struct lustre_capa *capa) +/** + * Implementation of dt_object_operations::do_attr_set + * + * Set attributes to the specified remote object. + * + * If the transaction is a remote transaction, then related modification + * sub-request has been added in the declare phase and related OUT RPC + * has been triggered at transaction start. Otherwise, the modification + * sub-request will be added here, and related OUT RPC will be triggered + * when transaction stop. + * + * \param[in] env execution environment + * \param[in] dt object to set attributes + * \param[in] attr attributes to be set + * \param[in] th the transaction handle + * \param[in] capa capability of setting attributes (not yet implemented). + * + * \retval only return 0 for now + */ +int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt, + const struct lu_attr *attr, struct thandle *th, + struct lustre_capa *capa) { + int rc = 0; + CDEBUG(D_INFO, "attr set object "DFID"\n", PFID(&dt->do_lu.lo_header->loh_fid)); - RETURN(0); + if (is_only_remote_trans(th)) + rc = __osp_md_attr_set(env, dt, attr, th); + + RETURN(rc); } +/** + * Implementation of dt_object_operations::do_read_lock + * + * osp_md_object_{read,write}_lock() will only lock the remote object in the + * local cache, which uses the semaphore (opo_sem) inside the osp_object to + * lock the object. Note: it will not lock the object in the whole cluster, + * which relies on the LDLM lock. + * + * \param[in] env execution environment + * \param[in] dt object to be locked + * \param[in] role lock role from MDD layer, see mdd_object_role(). + */ static void osp_md_object_read_lock(const struct lu_env *env, struct dt_object *dt, unsigned role) { @@ -287,6 +445,15 @@ static void osp_md_object_read_lock(const struct lu_env *env, LASSERT(obj->opo_owner == NULL); } +/** + * Implementation of dt_object_operations::do_write_lock + * + * Lock the remote object in write mode. + * + * \param[in] env execution environment + * \param[in] dt object to be locked + * \param[in] role lock role from MDD layer, see mdd_object_role(). + */ static void osp_md_object_write_lock(const struct lu_env *env, struct dt_object *dt, unsigned role) { @@ -298,6 +465,14 @@ static void osp_md_object_write_lock(const struct lu_env *env, obj->opo_owner = env; } +/** + * Implementation of dt_object_operations::do_read_unlock + * + * Unlock the read lock of remote object. + * + * \param[in] env execution environment + * \param[in] dt object to be unlocked + */ static void osp_md_object_read_unlock(const struct lu_env *env, struct dt_object *dt) { @@ -306,6 +481,14 @@ static void osp_md_object_read_unlock(const struct lu_env *env, up_read(&obj->opo_sem); } +/** + * Implementation of dt_object_operations::do_write_unlock + * + * Unlock the write lock of remote object. + * + * \param[in] env execution environment + * \param[in] dt object to be unlocked + */ static void osp_md_object_write_unlock(const struct lu_env *env, struct dt_object *dt) { @@ -316,6 +499,14 @@ static void osp_md_object_write_unlock(const struct lu_env *env, up_write(&obj->opo_sem); } +/** + * Implementation of dt_object_operations::do_write_locked + * + * Test if the object is locked in write mode. + * + * \param[in] env execution environment + * \param[in] dt object to be tested + */ static int osp_md_object_write_locked(const struct lu_env *env, struct dt_object *dt) { @@ -324,6 +515,21 @@ static int osp_md_object_write_locked(const struct lu_env *env, return obj->opo_owner == env; } +/** + * Implementation of dt_index_operations::dio_lookup + * + * Look up record by key under a remote index object. It packs lookup update + * into RPC, sends to the remote OUT and waits for the lookup result. + * + * \param[in] env execution environment + * \param[in] dt index object to lookup + * \param[out] rec record in which to return lookup result + * \param[in] key key of index which will be looked up + * \param[in] capa capability of lookup (not yet implemented) + * + * \retval 1 if the lookup succeeds. + * \retval negative errno if the lookup fails. + */ static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt, struct dt_rec *rec, const struct dt_key *key, struct lustre_capa *capa) @@ -331,27 +537,24 @@ static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt, struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2; struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev); struct dt_device *dt_dev = &osp->opd_dt_dev; - struct update_request *update; - struct ptlrpc_request *req = NULL; - int size = strlen((char *)key) + 1; - int rc; - struct update_reply *reply; - struct lu_fid *fid; - + struct dt_update_request *update; + struct object_update_reply *reply; + struct ptlrpc_request *req = NULL; + struct lu_fid *fid; + int rc; ENTRY; /* Because it needs send the update buffer right away, * just create an update buffer, instead of attaching the * update_remote list of the thandle. */ - update = out_create_update_req(dt_dev); + update = dt_update_request_create(dt_dev); if (IS_ERR(update)) RETURN(PTR_ERR(update)); - rc = out_insert_update(env, update, OBJ_INDEX_LOOKUP, - lu_object_fid(&dt->do_lu), 1, &size, - (const char **)&key); - if (rc) { + rc = out_index_lookup_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), rec, key); + if (rc != 0) { CERROR("%s: Insert update error: rc = %d\n", dt_dev->dd_lu_dev.ld_obd->obd_name, rc); GOTO(out, rc); @@ -361,24 +564,17 @@ static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt, if (rc < 0) GOTO(out, rc); - reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY, - UPDATE_BUFFER_SIZE); - if (reply->ur_version != UPDATE_REPLY_V1) { + reply = req_capsule_server_sized_get(&req->rq_pill, + &RMF_OUT_UPDATE_REPLY, + OUT_UPDATE_REPLY_SIZE); + if (reply->ourp_magic != UPDATE_REPLY_MAGIC) { CERROR("%s: Wrong version %x expected %x: rc = %d\n", dt_dev->dd_lu_dev.ld_obd->obd_name, - reply->ur_version, UPDATE_REPLY_V1, -EPROTO); + reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO); GOTO(out, rc = -EPROTO); } - rc = update_get_reply_result(reply, NULL, 0); - if (rc < 0) { - CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n", - dt_dev->dd_lu_dev.ld_obd->obd_name, - PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc); - GOTO(out, rc); - } - - rc = update_get_reply_buf(reply, lbuf, 0); + rc = object_update_result_data_get(reply, lbuf, 0); if (rc < 0) GOTO(out, rc); @@ -391,7 +587,8 @@ static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt, } fid = lbuf->lb_buf; - fid_le_to_cpu(fid, fid); + if (ptlrpc_rep_need_swab(req)) + lustre_swab_lu_fid(fid); if (!fid_is_sane(fid)) { CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n", dt_dev->dd_lu_dev.ld_obd->obd_name, @@ -407,26 +604,36 @@ out: if (req != NULL) ptlrpc_req_finished(req); - out_destroy_update_req(update); + dt_update_request_destroy(update); return rc; } +/** + * Implementation of dt_index_operations::dio_declare_insert + * + * Declare the index insert of the remote object, i.e. pack index insert update + * into the RPC, which will be sent during transaction start. + * + * \param[in] env execution environment + * \param[in] dt object for which to insert index + * \param[in] rec record of the index which will be inserted + * \param[in] key key of the index which will be inserted + * \param[in] th the transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ static int osp_md_declare_insert(const struct lu_env *env, struct dt_object *dt, const struct dt_rec *rec, const struct dt_key *key, struct thandle *th) { - struct update_request *update; - struct lu_fid *fid; - struct lu_fid *rec_fid = (struct lu_fid *)rec; - int size[2] = {strlen((char *)key) + 1, - sizeof(*rec_fid)}; - const char *bufs[2] = {(char *)key, (char *)rec_fid}; - int rc; + struct dt_update_request *update; + int rc; - update = out_find_create_update_loc(th, dt); + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -434,19 +641,29 @@ static int osp_md_declare_insert(const struct lu_env *env, return PTR_ERR(update); } - fid = (struct lu_fid *)lu_object_fid(&dt->do_lu); - - CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n", - dt->do_lu.lo_dev->ld_obd->obd_name, - PFID(fid), (char *)key, PFID(rec_fid)); - - fid_cpu_to_le(rec_fid, rec_fid); - - rc = out_insert_update(env, update, OBJ_INDEX_INSERT, fid, - ARRAY_SIZE(size), size, bufs); + rc = out_index_insert_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), rec, key, + update->dur_batchid); return rc; } +/** + * Implementation of dt_index_operations::dio_insert + * + * Do nothing in this method for now. In DNE phase I, remote updates + * are actually executed during transaction start, i.e. the index has + * already been inserted when calling this method. + * + * \param[in] env execution environment + * \param[in] dt object for which to insert index + * \param[in] rec record of the index to be inserted + * \param[in] key key of the index to be inserted + * \param[in] th the transaction handle + * \param[in] capa capability of insert (not yet implemented) + * \param[in] ignore_quota quota enforcement for insert + * + * \retval only return 0 for now + */ static int osp_md_index_insert(const struct lu_env *env, struct dt_object *dt, const struct dt_rec *rec, @@ -458,17 +675,29 @@ static int osp_md_index_insert(const struct lu_env *env, return 0; } +/** + * Implementation of dt_index_operations::dio_declare_delete + * + * Declare the index delete of the remote object, i.e. insert index delete + * update into the RPC, which will be sent during transaction start. + * + * \param[in] env execution environment + * \param[in] dt object for which to delete index + * \param[in] key key of the index + * \param[in] th the transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ static int osp_md_declare_delete(const struct lu_env *env, struct dt_object *dt, const struct dt_key *key, struct thandle *th) { - struct update_request *update; - struct lu_fid *fid; - int size = strlen((char *)key) + 1; - int rc; + struct dt_update_request *update; + int rc; - update = out_find_create_update_loc(th, dt); + update = dt_update_request_find_or_create(th, dt); if (IS_ERR(update)) { CERROR("%s: Get OSP update buf failed: rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name, @@ -476,14 +705,27 @@ static int osp_md_declare_delete(const struct lu_env *env, return PTR_ERR(update); } - fid = (struct lu_fid *)lu_object_fid(&dt->do_lu); - - rc = out_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size, - (const char **)&key); - + rc = out_index_delete_pack(env, &update->dur_buf, + lu_object_fid(&dt->do_lu), key, + update->dur_batchid); return rc; } +/** + * Implementation of dt_index_operations::dio_delete + * + * Do nothing in this method for now. Because in DNE phase I, remote updates + * are actually executed during transaction start, i.e. the index has already + * been deleted when calling this method. + * + * \param[in] env execution environment + * \param[in] dt object for which to delete index + * \param[in] key key of the index which will be deleted + * \param[in] th the transaction handle + * \param[in] capa capability of delete (not yet implemented) + * + * \retval only return 0 for now + */ static int osp_md_index_delete(const struct lu_env *env, struct dt_object *dt, const struct dt_key *key, @@ -497,93 +739,157 @@ static int osp_md_index_delete(const struct lu_env *env, } /** - * Creates or initializes iterator context. + * Implementation of dt_index_operations::dio_it.next + * + * Advance the pointer of the iterator to the next entry. It shares a similar + * internal implementation with osp_orphan_it_next(), which is being used for + * remote orphan index object. This method will be used for remote directory. * - * Note: for OSP, these index iterate api is only used to check - * whether the directory is empty now (see mdd_dir_is_empty). - * Since dir_empty will be return by OBJ_ATTR_GET(see osp_attr_get/ - * out_attr_get). So the implementation of these iterator is simplied - * to make mdd_dir_is_empty happy. The real iterator should be - * implemented, if we need it one day. + * \param[in] env execution environment + * \param[in] di iterator of this iteration + * + * \retval 0 if the pointer is advanced successfuly. + * \retval 1 if it reaches to the end of the index object. + * \retval negative errno if the pointer cannot be advanced. */ -static struct dt_it *osp_it_init(const struct lu_env *env, - struct dt_object *dt, - __u32 attr, - struct lustre_capa *capa) -{ - lu_object_get(&dt->do_lu); - return (struct dt_it *)dt; -} - -static void osp_it_fini(const struct lu_env *env, struct dt_it *di) -{ - struct dt_object *dt = (struct dt_object *)di; - lu_object_put(env, &dt->do_lu); -} - -static int osp_it_get(const struct lu_env *env, - struct dt_it *di, const struct dt_key *key) +int osp_md_index_it_next(const struct lu_env *env, struct dt_it *di) { - return 1; -} - -static void osp_it_put(const struct lu_env *env, struct dt_it *di) -{ - return; -} + struct osp_it *it = (struct osp_it *)di; + struct lu_idxpage *idxpage; + struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent; + int rc; + ENTRY; -static int osp_it_next(const struct lu_env *env, struct dt_it *di) -{ - struct dt_object *dt = (struct dt_object *)di; - struct osp_object *o = dt2osp_obj(dt); +again: + idxpage = it->ooi_cur_idxpage; + if (idxpage != NULL) { + if (idxpage->lip_nr == 0) + RETURN(1); + + it->ooi_pos_ent++; + if (ent == NULL) { + it->ooi_ent = + (struct lu_dirent *)idxpage->lip_entries; + RETURN(0); + } else if (le16_to_cpu(ent->lde_reclen) != 0 && + it->ooi_pos_ent < idxpage->lip_nr) { + ent = (struct lu_dirent *)(((char *)ent) + + le16_to_cpu(ent->lde_reclen)); + it->ooi_ent = ent; + RETURN(0); + } else { + it->ooi_ent = NULL; + } + } - if (o->opo_empty) - return 1; + rc = osp_it_next_page(env, di); + if (rc == 0) + goto again; - return 0; + RETURN(rc); } +/** + * Implementation of dt_index_operations::dio_it.key + * + * Get the key at current iterator poisiton. These iteration methods + * (dio_it) will only be used for iterating the remote directory, so + * the key is the name of the directory entry. + * + * \param[in] env execution environment + * \param[in] di iterator of this iteration + * + * \retval name of the current entry + */ static struct dt_key *osp_it_key(const struct lu_env *env, const struct dt_it *di) { - LBUG(); - return NULL; + struct osp_it *it = (struct osp_it *)di; + struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent; + + return (struct dt_key *)ent->lde_name; } +/** + * Implementation of dt_index_operations::dio_it.key_size + * + * Get the key size at current iterator poisiton. These iteration methods + * (dio_it) will only be used for iterating the remote directory, so the key + * size is the name size of the directory entry. + * + * \param[in] env execution environment + * \param[in] di iterator of this iteration + * + * \retval name size of the current entry + */ + static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di) { - LBUG(); - return 0; -} + struct osp_it *it = (struct osp_it *)di; + struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent; -static int osp_it_rec(const struct lu_env *env, const struct dt_it *di, - struct dt_rec *lde, __u32 attr) -{ - LBUG(); - return 0; + return (int)le16_to_cpu(ent->lde_namelen); } -static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di) +/** + * Implementation of dt_index_operations::dio_it.rec + * + * Get the record at current iterator position. These iteration methods + * (dio_it) will only be used for iterating the remote directory, so it + * uses lu_dirent_calc_size() to calculate the record size. + * + * \param[in] env execution environment + * \param[in] di iterator of this iteration + * \param[out] rec the record to be returned + * \param[in] attr attributes of the index object, so it knows + * how to pack the entry. + * + * \retval only return 0 for now + */ +static int osp_md_index_it_rec(const struct lu_env *env, const struct dt_it *di, + struct dt_rec *rec, __u32 attr) { - LBUG(); + struct osp_it *it = (struct osp_it *)di; + struct lu_dirent *ent = (struct lu_dirent *)it->ooi_ent; + int reclen; + + reclen = lu_dirent_calc_size(le16_to_cpu(ent->lde_namelen), attr); + memcpy(rec, ent, reclen); return 0; } +/** + * Implementation of dt_index_operations::dio_it.load + * + * Locate the iteration cursor to the specified position (cookie). + * + * \param[in] env pointer to the thread context + * \param[in] di pointer to the iteration structure + * \param[in] hash the specified position + * + * \retval positive number for locating to the exactly position + * or the next + * \retval 0 for arriving at the end of the iteration + * \retval negative error number on failure + */ static int osp_it_load(const struct lu_env *env, const struct dt_it *di, __u64 hash) { - LBUG(); - return 0; -} + struct osp_it *it = (struct osp_it *)di; + int rc; -static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di, - void *key_rec) -{ - LBUG(); - return 0; + it->ooi_next = hash; + rc = osp_md_index_it_next(env, (struct dt_it *)di); + if (rc == 1) + return 0; + + if (rc == 0) + return 1; + + return rc; } -static const struct dt_index_operations osp_md_index_ops = { +const struct dt_index_operations osp_md_index_ops = { .dio_lookup = osp_md_index_lookup, .dio_declare_insert = osp_md_declare_insert, .dio_insert = osp_md_index_insert, @@ -594,16 +900,30 @@ static const struct dt_index_operations osp_md_index_ops = { .fini = osp_it_fini, .get = osp_it_get, .put = osp_it_put, - .next = osp_it_next, + .next = osp_md_index_it_next, .key = osp_it_key, .key_size = osp_it_key_size, - .rec = osp_it_rec, + .rec = osp_md_index_it_rec, .store = osp_it_store, .load = osp_it_load, .key_rec = osp_it_key_rec, } }; +/** + * Implementation of dt_object_operations::do_index_try + * + * Try to initialize the index API pointer for the given object. This + * is the entry point of the index API, i.e. we must call this method + * to initialize the index object before calling other index methods. + * + * \param[in] env execution environment + * \param[in] dt index object to be initialized + * \param[in] feat the index feature of the object + * + * \retval 0 if the initialization succeeds. + * \retval negative errno if the initialization fails. + */ static int osp_md_index_try(const struct lu_env *env, struct dt_object *dt, const struct dt_index_features *feat) @@ -612,27 +932,41 @@ static int osp_md_index_try(const struct lu_env *env, return 0; } +/** + * Implementation of dt_object_operations::do_object_lock + * + * Enqueue a lock (by ldlm_cli_enqueue()) of remote object on the remote MDT, + * which will lock the object in the global namespace. + * + * \param[in] env execution environment + * \param[in] dt object to be locked + * \param[out] lh lock handle + * \param[in] einfo enqueue information + * \param[in] policy lock policy + * + * \retval ELDLM_OK if locking the object succeeds. + * \retval negative errno if locking fails. + */ static int osp_md_object_lock(const struct lu_env *env, struct dt_object *dt, struct lustre_handle *lh, struct ldlm_enqueue_info *einfo, - void *policy) + ldlm_policy_data_t *policy) { - struct osp_thread_info *info = osp_env_info(env); - struct ldlm_res_id *res_id = &info->osi_resid; - struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev); - struct osp_device *osp = dt2osp_dev(dt_dev); - struct ptlrpc_request *req = NULL; - int rc = 0; - __u64 flags = 0; - ldlm_mode_t mode; + struct ldlm_res_id *res_id; + struct dt_device *dt_dev = lu2dt_dev(dt->do_lu.lo_dev); + struct osp_device *osp = dt2osp_dev(dt_dev); + struct ptlrpc_request *req; + int rc = 0; + __u64 flags = 0; + ldlm_mode_t mode; - fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id); + res_id = einfo->ei_res_id; + LASSERT(res_id != NULL); mode = ldlm_lock_match(osp->opd_obd->obd_namespace, LDLM_FL_BLOCK_GRANTED, res_id, - einfo->ei_type, - (ldlm_policy_data_t *)policy, + einfo->ei_type, policy, einfo->ei_mode, lh, 0); if (mode > 0) return ELDLM_OK; @@ -650,6 +984,31 @@ static int osp_md_object_lock(const struct lu_env *env, return rc == ELDLM_OK ? 0 : -EIO; } +/** + * Implementation of dt_object_operations::do_object_unlock + * + * Cancel a lock of a remote object. + * + * \param[in] env execution environment + * \param[in] dt object to be unlocked + * \param[in] einfo lock enqueue information + * \param[in] policy lock policy + * + * \retval Only return 0 for now. + */ +static int osp_md_object_unlock(const struct lu_env *env, + struct dt_object *dt, + struct ldlm_enqueue_info *einfo, + ldlm_policy_data_t *policy) +{ + struct lustre_handle *lockh = einfo->ei_cbdata; + + /* unlock finally */ + ldlm_lock_decref(lockh, einfo->ei_mode); + + return 0; +} + struct dt_object_operations osp_md_obj_ops = { .do_read_lock = osp_md_object_read_lock, .do_write_lock = osp_md_object_write_lock, @@ -671,6 +1030,79 @@ struct dt_object_operations osp_md_obj_ops = { .do_xattr_get = osp_xattr_get, .do_declare_xattr_set = osp_declare_xattr_set, .do_xattr_set = osp_xattr_set, + .do_declare_xattr_del = osp_declare_xattr_del, + .do_xattr_del = osp_xattr_del, .do_index_try = osp_md_index_try, .do_object_lock = osp_md_object_lock, + .do_object_unlock = osp_md_object_unlock, +}; + +/** + * Implementation of dt_body_operations::dbo_declare_write + * + * Declare an object write. In DNE phase I, it will pack the write + * object update into the RPC. + * + * \param[in] env execution environment + * \param[in] dt object to be written + * \param[in] buf buffer to write which includes an embedded size field + * \param[in] pos offet in the object to start writing at + * \param[in] th transaction handle + * + * \retval 0 if the insertion succeeds. + * \retval negative errno if the insertion fails. + */ +static ssize_t osp_md_declare_write(const struct lu_env *env, + struct dt_object *dt, + const struct lu_buf *buf, + loff_t pos, struct thandle *th) +{ + struct dt_update_request *update; + ssize_t rc; + + update = dt_update_request_find_or_create(th, dt); + if (IS_ERR(update)) { + CERROR("%s: Get OSP update buf failed: rc = %d\n", + dt->do_lu.lo_dev->ld_obd->obd_name, + (int)PTR_ERR(update)); + return PTR_ERR(update); + } + + rc = out_write_pack(env, &update->dur_buf, lu_object_fid(&dt->do_lu), + buf, pos, update->dur_batchid); + + return rc; + +} + +/** + * Implementation of dt_body_operations::dbo_write + * + * Return the buffer size. In DNE phase I, remote updates + * are actually executed during transaction start, the buffer has + * already been written when this method is being called. + * + * \param[in] env execution environment + * \param[in] dt object to be written + * \param[in] buf buffer to write which includes an embedded size field + * \param[in] pos offet in the object to start writing at + * \param[in] th transaction handle + * \param[in] capa capability of the write (not yet implemented) + * \param[in] ignore_quota quota enforcement for this write + * + * \retval the buffer size in bytes. + */ +static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt, + const struct lu_buf *buf, loff_t *pos, + struct thandle *handle, + struct lustre_capa *capa, int ignore_quota) +{ + *pos += buf->lb_len; + return buf->lb_len; +} + +/* These body operation will be used to write symlinks during migration etc */ +struct dt_body_operations osp_md_body_ops = { + .dbo_declare_write = osp_md_declare_write, + .dbo_write = osp_md_write, };