lustre/target/out_lib.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   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.
   9  *
  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).
  15  *
  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
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2014, Intel Corporation.
  24  */
  25 /*
  26  * lustre/target/out_lib.c
  27  *
  28  * Author: Di Wang <di.wang@intel.com>
  29  * Author: Fan, Yong <fan.yong@intel.com>
  30  */
  31
  32 #define DEBUG_SUBSYSTEM S_CLASS
  33
  34 #include <lu_target.h>
  35 #include <lustre_update.h>
  36 #include <obd.h>
  37 #include <obd_class.h>
  38
  39 #define OUT_UPDATE_BUFFER_SIZE_ADD      4096
  40 #define OUT_UPDATE_BUFFER_SIZE_MAX      (256 * 4096)  /* 1MB update size now */
  41 static inline struct object_update_request *
  42 object_update_request_alloc(size_t size)
  43 {
  44         struct object_update_request *ourq;
  45
  46         OBD_ALLOC_LARGE(ourq, size);
  47         if (ourq == NULL)
  48                 RETURN(ERR_PTR(-ENOMEM));
  49
  50         ourq->ourq_magic = UPDATE_REQUEST_MAGIC;
  51         ourq->ourq_count = 0;
  52
  53         RETURN(ourq);
  54 }
  55
  56 static inline void
  57 object_update_request_free(struct object_update_request *ourq,
  58                            size_t ourq_size)
  59 {
  60         if (ourq != NULL)
  61                 OBD_FREE_LARGE(ourq, ourq_size);
  62 }
  63
  64 /**
  65  * Allocate and initialize dt_update_request
  66  *
  67  * dt_update_request is being used to track updates being executed on
  68  * this dt_device(OSD or OSP). The update buffer will be 4k initially,
  69  * and increased if needed.
  70  *
  71  * \param [in] dt       dt device
  72  *
  73  * \retval              dt_update_request being allocated if succeed
  74  * \retval              ERR_PTR(errno) if failed
  75  */
  76 struct dt_update_request *dt_update_request_create(struct dt_device *dt)
  77 {
  78         struct dt_update_request *dt_update;
  79         struct object_update_request *ourq;
  80
  81         OBD_ALLOC_PTR(dt_update);
  82         if (!dt_update)
  83                 return ERR_PTR(-ENOMEM);
  84
  85         ourq = object_update_request_alloc(OUT_UPDATE_INIT_BUFFER_SIZE);
  86         if (IS_ERR(ourq)) {
  87                 OBD_FREE_PTR(dt_update);
  88                 return ERR_CAST(ourq);
  89         }
  90
  91         dt_update->dur_buf.ub_req = ourq;
  92         dt_update->dur_buf.ub_req_size = OUT_UPDATE_INIT_BUFFER_SIZE;
  93
  94         dt_update->dur_dt = dt;
  95         dt_update->dur_batchid = 0;
  96         INIT_LIST_HEAD(&dt_update->dur_cb_items);
  97
  98         return dt_update;
  99 }
 100 EXPORT_SYMBOL(dt_update_request_create);
 101
 102 /**
 103  * Destroy dt_update_request
 104  *
 105  * \param [in] dt_update        dt_update_request being destroyed
 106  */
 107 void dt_update_request_destroy(struct dt_update_request *dt_update)
 108 {
 109         if (dt_update == NULL)
 110                 return;
 111
 112         object_update_request_free(dt_update->dur_buf.ub_req,
 113                                    dt_update->dur_buf.ub_req_size);
 114         OBD_FREE_PTR(dt_update);
 115
 116         return;
 117 }
 118 EXPORT_SYMBOL(dt_update_request_destroy);
 119
 120 /**
 121  * resize update buffer
 122  *
 123  * Extend the update buffer by new_size.
 124  *
 125  * \param[in] ubuf      update buffer to be extended
 126  * \param[in] new_size  new size of the update buffer
 127  *
 128  * \retval              0 if extending succeeds.
 129  * \retval              negative errno if extending fails.
 130  */
 131 static int update_buffer_resize(struct update_buffer *ubuf, size_t new_size)
 132 {
 133         struct object_update_request *ureq;
 134
 135         if (new_size > ubuf->ub_req_size)
 136                 return 0;
 137
 138         OBD_ALLOC_LARGE(ureq, new_size);
 139         if (ureq == NULL)
 140                 return -ENOMEM;
 141
 142         memcpy(ureq, ubuf->ub_req, ubuf->ub_req_size);
 143
 144         OBD_FREE_LARGE(ubuf->ub_req, ubuf->ub_req_size);
 145
 146         ubuf->ub_req = ureq;
 147         ubuf->ub_req_size = new_size;
 148
 149         return 0;
 150 }
 151
 152 /**
 153  * Pack the header of object_update_request
 154  *
 155  * Packs updates into the update_buffer header, which will either be sent to
 156  * the remote MDT or stored in the local update log. The maximum update buffer
 157  * size is 1MB for now.
 158  *
 159  * \param[in] env       execution environment
 160  * \param[in] ubuf      update bufer which it will pack the update in
 161  * \param[in] op        update operation
 162  * \param[in] fid       object FID for this update
 163  * \param[in] param_count       parameters count for this update
 164  * \param[in] lens      each parameters length of this update
 165  * \param[in] batchid   batchid(transaction no) of this update
 166  *
 167  * \retval              0 pack update succeed.
 168  * \retval              negative errno pack update failed.
 169  **/
 170 static struct object_update *
 171 out_update_header_pack(const struct lu_env *env, struct update_buffer *ubuf,
 172                        enum update_type op, const struct lu_fid *fid,
 173                        int params_count, __u16 *param_sizes, __u64 batchid)
 174 {
 175         struct object_update_request    *ureq = ubuf->ub_req;
 176         size_t                          ureq_size = ubuf->ub_req_size;
 177         struct object_update            *obj_update;
 178         struct object_update_param      *param;
 179         size_t                          update_size;
 180         int                             rc = 0;
 181         unsigned int                    i;
 182         ENTRY;
 183
 184         /* Check update size to make sure it can fit into the buffer */
 185         ureq_size = object_update_request_size(ureq);
 186         update_size = offsetof(struct object_update, ou_params[0]);
 187         for (i = 0; i < params_count; i++)
 188                 update_size += cfs_size_round(param_sizes[i] + sizeof(*param));
 189
 190         if (unlikely(cfs_size_round(ureq_size + update_size) >
 191                      ubuf->ub_req_size)) {
 192                 size_t new_size = ubuf->ub_req_size;
 193
 194                 /* enlarge object update request size */
 195                 while (new_size <
 196                        cfs_size_round(ureq_size + update_size))
 197                         new_size += OUT_UPDATE_BUFFER_SIZE_ADD;
 198                 if (new_size >= OUT_UPDATE_BUFFER_SIZE_MAX)
 199                         RETURN(ERR_PTR(-E2BIG));
 200
 201                 rc = update_buffer_resize(ubuf, new_size);
 202                 if (rc < 0)
 203                         RETURN(ERR_PTR(rc));
 204
 205                 ureq = ubuf->ub_req;
 206         }
 207
 208         /* fill the update into the update buffer */
 209         obj_update = (struct object_update *)((char *)ureq + ureq_size);
 210         obj_update->ou_fid = *fid;
 211         obj_update->ou_type = op;
 212         obj_update->ou_params_count = (__u16)params_count;
 213         obj_update->ou_batchid = batchid;
 214         param = &obj_update->ou_params[0];
 215         for (i = 0; i < params_count; i++) {
 216                 param->oup_len = param_sizes[i];
 217                 param = (struct object_update_param *)((char *)param +
 218                          object_update_param_size(param));
 219         }
 220         ureq->ourq_count++;
 221
 222         CDEBUG(D_INFO, "%p "DFID" idx %u: op %d params %d:%d\n",
 223                ureq, PFID(fid), ureq->ourq_count, op, params_count,
 224                (int)update_size);
 225
 226         RETURN(obj_update);
 227 }
 228
 229 /**
 230  * Packs one update into the update_buffer.
 231  *
 232  * \param[in] env       execution environment
 233  * \param[in] ubuf      bufer where update will be packed
 234  * \param[in] op        update operation (enum update_type)
 235  * \param[in] fid       object FID for this update
 236  * \param[in] param_count       number of parameters for this update
 237  * \param[in] param_sizes       array of parameters length of this update
 238  * \param[in] param_bufs        parameter buffers
 239  * \param[in] batchid   transaction no of this update, plus mdt_index, which
 240  *                      will be globally unique
 241  *
 242  * \retval              = 0 if updates packing succeeds
 243  * \retval              negative errno if updates packing fails
 244  **/
 245 int out_update_pack(const struct lu_env *env, struct update_buffer *ubuf,
 246                     enum update_type op, const struct lu_fid *fid,
 247                     int params_count, __u16 *param_sizes,
 248                     const void **param_bufs, __u64 batchid)
 249 {
 250         struct object_update            *update;
 251         struct object_update_param      *param;
 252         unsigned int                    i;
 253         ENTRY;
 254
 255         update = out_update_header_pack(env, ubuf, op, fid, params_count,
 256                                         param_sizes, batchid);
 257         if (IS_ERR(update))
 258                 RETURN(PTR_ERR(update));
 259
 260         param = &update->ou_params[0];
 261         for (i = 0; i < params_count; i++) {
 262                 memcpy(&param->oup_buf[0], param_bufs[i], param_sizes[i]);
 263                 param = (struct object_update_param *)((char *)param +
 264                          object_update_param_size(param));
 265         }
 266
 267         RETURN(0);
 268 }
 269 EXPORT_SYMBOL(out_update_pack);
 270
 271 /**
 272  * Pack various updates into the update_buffer.
 273  *
 274  * The following functions pack different updates into the update_buffer
 275  * So parameters of these API is basically same as its correspondent OSD/OSP
 276  * API, for detail description of these parameters see osd_handler.c or
 277  * osp_md_object.c.
 278  *
 279  * \param[in] env       execution environment
 280  * \param[in] ubuf      update buffer
 281  * \param[in] fid       fid of this object for the update
 282  * \param[in] batchid   batch id of this update
 283  *
 284  * \retval              0 if insertion succeeds.
 285  * \retval              negative errno if insertion fails.
 286  */
 287 int out_create_pack(const struct lu_env *env, struct update_buffer *ubuf,
 288                     const struct lu_fid *fid, struct lu_attr *attr,
 289                     struct dt_allocation_hint *hint,
 290                     struct dt_object_format *dof, __u64 batchid)
 291 {
 292         struct obdo             *obdo;
 293         __u16                   sizes[2] = {sizeof(*obdo), 0};
 294         int                     buf_count = 1;
 295         const struct lu_fid     *fid1 = NULL;
 296         struct object_update    *update;
 297         ENTRY;
 298
 299         if (hint != NULL && hint->dah_parent) {
 300                 fid1 = lu_object_fid(&hint->dah_parent->do_lu);
 301                 sizes[1] = sizeof(*fid1);
 302                 buf_count++;
 303         }
 304
 305         update = out_update_header_pack(env, ubuf, OUT_CREATE, fid,
 306                                         buf_count, sizes, batchid);
 307         if (IS_ERR(update))
 308                 RETURN(PTR_ERR(update));
 309
 310         obdo = object_update_param_get(update, 0, NULL);
 311         obdo->o_valid = 0;
 312         obdo_from_la(obdo, attr, attr->la_valid);
 313         lustre_set_wire_obdo(NULL, obdo, obdo);
 314         if (fid1 != NULL) {
 315                 struct lu_fid *fid;
 316                 fid = object_update_param_get(update, 1, NULL);
 317                 fid_cpu_to_le(fid, fid1);
 318         }
 319
 320         RETURN(0);
 321 }
 322 EXPORT_SYMBOL(out_create_pack);
 323
 324 int out_ref_del_pack(const struct lu_env *env, struct update_buffer *ubuf,
 325                      const struct lu_fid *fid, __u64 batchid)
 326 {
 327         return out_update_pack(env, ubuf, OUT_REF_DEL, fid, 0, NULL, NULL,
 328                                batchid);
 329 }
 330 EXPORT_SYMBOL(out_ref_del_pack);
 331
 332 int out_ref_add_pack(const struct lu_env *env, struct update_buffer *ubuf,
 333                      const struct lu_fid *fid, __u64 batchid)
 334 {
 335         return out_update_pack(env, ubuf, OUT_REF_ADD, fid, 0, NULL, NULL,
 336                                batchid);
 337 }
 338 EXPORT_SYMBOL(out_ref_add_pack);
 339
 340 int out_attr_set_pack(const struct lu_env *env, struct update_buffer *ubuf,
 341                       const struct lu_fid *fid, const struct lu_attr *attr,
 342                       __u64 batchid)
 343 {
 344         struct object_update    *update;
 345         struct obdo             *obdo;
 346         __u16                   size = sizeof(*obdo);
 347         ENTRY;
 348
 349         update = out_update_header_pack(env, ubuf, OUT_ATTR_SET, fid, 1,
 350                                         &size, batchid);
 351         if (IS_ERR(update))
 352                 RETURN(PTR_ERR(update));
 353
 354         obdo = object_update_param_get(update, 0, NULL);
 355         obdo->o_valid = 0;
 356         obdo_from_la(obdo, attr, attr->la_valid);
 357         lustre_set_wire_obdo(NULL, obdo, obdo);
 358
 359         RETURN(0);
 360 }
 361 EXPORT_SYMBOL(out_attr_set_pack);
 362
 363 int out_xattr_set_pack(const struct lu_env *env, struct update_buffer *ubuf,
 364                        const struct lu_fid *fid, const struct lu_buf *buf,
 365                        const char *name, int flag, __u64 batchid)
 366 {
 367         __u16   sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 368         const void *bufs[3] = {(char *)name, (char *)buf->lb_buf,
 369                                (char *)&flag};
 370
 371         return out_update_pack(env, ubuf, OUT_XATTR_SET, fid,
 372                                ARRAY_SIZE(sizes), sizes, bufs, batchid);
 373 }
 374 EXPORT_SYMBOL(out_xattr_set_pack);
 375
 376 int out_xattr_del_pack(const struct lu_env *env, struct update_buffer *ubuf,
 377                        const struct lu_fid *fid, const char *name,
 378                        __u64 batchid)
 379 {
 380         __u16   size = strlen(name) + 1;
 381
 382         return out_update_pack(env, ubuf, OUT_XATTR_DEL, fid, 1, &size,
 383                                (const void **)&name, batchid);
 384 }
 385 EXPORT_SYMBOL(out_xattr_del_pack);
 386
 387
 388 int out_index_insert_pack(const struct lu_env *env, struct update_buffer *ubuf,
 389                           const struct lu_fid *fid, const struct dt_rec *rec,
 390                           const struct dt_key *key, __u64 batchid)
 391 {
 392         struct dt_insert_rec       *rec1 = (struct dt_insert_rec *)rec;
 393         struct lu_fid              rec_fid;
 394         __u32                       type = cpu_to_le32(rec1->rec_type);
 395         __u16                       sizes[3] = { strlen((char *)key) + 1,
 396                                                 sizeof(rec_fid),
 397                                                 sizeof(type) };
 398         const void                 *bufs[3] = { (char *)key,
 399                                                 (char *)&rec_fid,
 400                                                 (char *)&type };
 401
 402         fid_cpu_to_le(&rec_fid, rec1->rec_fid);
 403
 404         return out_update_pack(env, ubuf, OUT_INDEX_INSERT, fid,
 405                                ARRAY_SIZE(sizes), sizes, bufs, batchid);
 406 }
 407 EXPORT_SYMBOL(out_index_insert_pack);
 408
 409 int out_index_delete_pack(const struct lu_env *env, struct update_buffer *ubuf,
 410                           const struct lu_fid *fid, const struct dt_key *key,
 411                           __u64 batchid)
 412 {
 413         __u16   size = strlen((char *)key) + 1;
 414         const void *buf = key;
 415
 416         return out_update_pack(env, ubuf, OUT_INDEX_DELETE, fid, 1, &size,
 417                                &buf, batchid);
 418 }
 419 EXPORT_SYMBOL(out_index_delete_pack);
 420
 421 int out_object_destroy_pack(const struct lu_env *env,
 422                             struct update_buffer *ubuf,
 423                             const struct lu_fid *fid, __u64 batchid)
 424 {
 425         return out_update_pack(env, ubuf, OUT_DESTROY, fid, 0, NULL, NULL,
 426                                batchid);
 427 }
 428 EXPORT_SYMBOL(out_object_destroy_pack);
 429
 430 int out_write_pack(const struct lu_env *env, struct update_buffer *ubuf,
 431                    const struct lu_fid *fid, const struct lu_buf *buf,
 432                    loff_t pos, __u64 batchid)
 433 {
 434         __u16           sizes[2] = {buf->lb_len, sizeof(pos)};
 435         const void      *bufs[2] = {(char *)buf->lb_buf, (char *)&pos};
 436         int             rc;
 437
 438         pos = cpu_to_le64(pos);
 439
 440         rc = out_update_pack(env, ubuf, OUT_WRITE, fid, ARRAY_SIZE(sizes),
 441                              sizes, bufs, batchid);
 442         return rc;
 443 }
 444 EXPORT_SYMBOL(out_write_pack);
 445
 446 /**
 447  * Pack various readonly updates into the update_buffer.
 448  *
 449  * The following update funcs are only used by read-only ops, lookup,
 450  * getattr etc, so it does not need transaction here. Currently they
 451  * are only used by OSP.
 452  *
 453  * \param[in] env       execution environment
 454  * \param[in] fid       fid of this object for the update
 455  * \param[in] ubuf      update buffer
 456  *
 457  * \retval              0 if packing succeeds.
 458  * \retval              negative errno if packing fails.
 459  */
 460 int out_index_lookup_pack(const struct lu_env *env, struct update_buffer *ubuf,
 461                           const struct lu_fid *fid, struct dt_rec *rec,
 462                           const struct dt_key *key)
 463 {
 464         const void      *name = key;
 465         __u16           size = strlen((char *)name) + 1;
 466
 467         return out_update_pack(env, ubuf, OUT_INDEX_LOOKUP, fid, 1, &size,
 468                                &name, 0);
 469 }
 470 EXPORT_SYMBOL(out_index_lookup_pack);
 471
 472 int out_attr_get_pack(const struct lu_env *env, struct update_buffer *ubuf,
 473                       const struct lu_fid *fid)
 474 {
 475         return out_update_pack(env, ubuf, OUT_ATTR_GET, fid, 0, NULL, NULL, 0);
 476 }
 477 EXPORT_SYMBOL(out_attr_get_pack);
 478
 479 int out_xattr_get_pack(const struct lu_env *env, struct update_buffer *ubuf,
 480                        const struct lu_fid *fid, const char *name)
 481 {
 482         __u16 size;
 483
 484         LASSERT(name != NULL);
 485         size = strlen(name) + 1;
 486         return out_update_pack(env, ubuf, OUT_XATTR_GET, fid, 1, &size,
 487                                (const void **)&name, 0);
 488 }
 489 EXPORT_SYMBOL(out_xattr_get_pack);