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) 2014, 2015, Intel Corporation.
27 * lustre/target/update_records.c
29 * This file implement the methods to pack updates as update records, which
30 * will be written to the disk as llog record, and might be used during
33 * For cross-MDT operation, all of updates of the operation needs to be
34 * recorded in the disk, then during recovery phase, the recovery thread
35 * will retrieve and redo these updates if it needed.
37 * See comments above struct update_records for the format of update_records.
39 * Author: Di Wang <di.wang@intel.com>
41 #define DEBUG_SUBSYSTEM S_CLASS
43 #include <lu_target.h>
44 #include <lustre_obdo.h>
45 #include <lustre_update.h>
47 #include <obd_class.h>
49 #include "tgt_internal.h"
51 #define UPDATE_RECORDS_BUFFER_SIZE 8192
52 #define UPDATE_PARAMS_BUFFER_SIZE 8192
56 * Dump all of updates in the update_records, mostly for debugging purpose.
58 * \param[in] records update records to be dumpped
59 * \param[in] mask debug level mask
60 * \param[in] dump_params if dump all of updates the updates.
63 void update_records_dump(const struct update_records *records,
64 unsigned int mask, bool dump_updates)
66 const struct update_ops *ops;
67 const struct update_op *op = NULL;
68 struct update_params *params = NULL;
71 CDEBUG(mask, "master transno = %llu batchid = %llu flags = %x"
72 " ops = %d params = %d\n", records->ur_master_transno,
73 records->ur_batchid, records->ur_flags, records->ur_update_count,
74 records->ur_param_count);
76 if (records->ur_update_count == 0)
82 ops = &records->ur_ops;
83 if (records->ur_param_count > 0)
84 params = update_records_get_params(records);
86 op = &ops->uops_op[0];
87 for (i = 0; i < records->ur_update_count; i++,
88 op = update_op_next_op(op)) {
91 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
92 PFID(&op->uop_fid), update_op_str(op->uop_type),
98 for (j = 0; j < op->uop_param_count; j++) {
99 struct object_update_param *param;
101 param = update_params_get_param(params,
102 (unsigned int)op->uop_params_off[j],
103 records->ur_param_count);
107 CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
108 param, j, op->uop_params_off[j], param->oup_len);
114 * Pack parameters to update records
116 * Find and insert parameter to update records, if the parameter
117 * already exists in \a params, then just return the offset of this
118 * parameter, otherwise insert the parameter and return its offset
120 * \param[in] params update params in which to insert parameter
121 * \param[in] new_param parameters to be inserted.
122 * \param[in] new_param_size the size of \a new_param
124 * \retval index inside \a params if parameter insertion
126 * \retval negative errno if it fails.
128 static unsigned int update_records_param_pack(struct update_params *params,
129 const void *new_param,
130 size_t new_param_size,
131 unsigned int *param_count)
133 struct object_update_param *param;
136 for (i = 0; i < *param_count; i++) {
137 struct object_update_param *param;
139 param = update_params_get_param(params, i, *param_count);
140 if ((new_param == NULL && param->oup_len == new_param_size) ||
141 (param->oup_len == new_param_size &&
142 memcmp(param->oup_buf, new_param, new_param_size) == 0))
143 /* Found the parameter and return its index */
147 param = (struct object_update_param *)((char *)params +
148 update_params_size(params, *param_count));
150 param->oup_len = new_param_size;
151 if (new_param != NULL)
152 memcpy(param->oup_buf, new_param, new_param_size);
154 *param_count = *param_count + 1;
156 return *param_count - 1;
160 * Pack update to update records
162 * Pack the update and its parameters to the update records. First it will
163 * insert parameters, get the offset of these parameter, then fill the
164 * update with these offset. If insertion exceed the maximum size of
165 * current update records, it will return -E2BIG here, and the caller might
166 * extend the update_record size \see lod_updates_pack.
168 * \param[in] env execution environment
169 * \param[in] fid FID of the update.
170 * \param[in] op_type operation type of the update
171 * \param[in] ops ur_ops in update records
172 * \param[in|out] op_count pointer to the count of ops
173 * \param[in|out] max_op_size maximum size of the update
174 * \param[in] params ur_params in update records
175 * \param[in|out] param_count pointer to the count of params
176 * \param[in|out] max_param_size maximum size of the parameter
177 * \param[in] param_bufs buffers of parameters
178 * \param[in] params_buf_count the count of the parameter buffers
179 * \param[in] param_size sizes of parameters
181 * \retval 0 if packing succeeds
182 * \retval negative errno if packing fails
184 static int update_records_update_pack(const struct lu_env *env,
185 const struct lu_fid *fid,
186 enum update_type op_type,
187 struct update_ops *ops,
188 unsigned int *op_count,
190 struct update_params *params,
191 unsigned int *param_count,
192 size_t *max_param_size,
193 unsigned int param_bufs_count,
194 const void **param_bufs,
197 struct update_op *op;
198 size_t total_param_sizes = 0;
202 /* Check whether the packing exceeding the maximum update size */
203 if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
204 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
205 *max_op_size, update_op_size(param_bufs_count));
206 *max_op_size = update_op_size(param_bufs_count);
210 for (i = 0; i < param_bufs_count; i++)
212 cfs_size_round(sizeof(struct object_update_param) +
215 /* Check whether the packing exceeding the maximum parameter size */
216 if (unlikely(*max_param_size < total_param_sizes)) {
217 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
218 *max_param_size, total_param_sizes);
220 *max_param_size = total_param_sizes;
224 op = update_ops_get_op(ops, *op_count, *op_count);
226 op->uop_type = op_type;
227 op->uop_param_count = param_bufs_count;
228 for (i = 0; i < param_bufs_count; i++) {
229 index = update_records_param_pack(params, param_bufs[i],
230 param_sizes[i], param_count);
234 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
235 update_op_str(op_type), i, index, param_sizes[i]);
237 op->uop_params_off[i] = index;
239 CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
240 *op_count, PFID(fid), update_op_str(op_type), *param_count);
242 *op_count = *op_count + 1;
248 * Calculate update_records size
250 * Calculate update_records size by param_count and param_sizes array.
252 * \param[in] param_count the count of parameters
253 * \param[in] sizes the size array of these parameters
255 * \retval the size of this update
257 static size_t update_records_update_size(__u32 param_count, size_t *sizes)
262 /* Check whether the packing exceeding the maximum update size */
263 size = update_op_size(param_count);
265 for (i = 0; i < param_count; i++)
266 size += cfs_size_round(sizeof(struct object_update_param) +
273 * Calculate create update size
275 * \param[in] env execution environment
276 * \param[in] ops ur_ops in update records
277 * \param[in] fid FID of the object to be created
278 * \param[in] attr attribute of the object to be created
279 * \param[in] hint creation hint
280 * \param[in] dof creation format information
282 * \retval size of create update.
284 size_t update_records_create_size(const struct lu_env *env,
285 const struct lu_fid *fid,
286 const struct lu_attr *attr,
287 const struct dt_allocation_hint *hint,
288 struct dt_object_format *dof)
294 sizes[param_count] = sizeof(struct obdo);
298 if (hint != NULL && hint->dah_parent != NULL) {
299 sizes[param_count] = sizeof(*fid);
303 return update_records_update_size(param_count, sizes);
305 EXPORT_SYMBOL(update_records_create_size);
310 * Pack create update into update records.
312 * \param[in] env execution environment
313 * \param[in] ops ur_ops in update records
314 * \param[in|out] op_count pointer to the count of ops
315 * \param[in|out] max_op_size maximum size of the update
316 * \param[in] params ur_params in update records
317 * \param[in|out] param_count pointer to the count of params
318 * \param[in|out] max_param_size maximum size of the parameter
319 * \param[in] fid FID of the object to be created
320 * \param[in] attr attribute of the object to be created
321 * \param[in] hint creation hint
322 * \param[in] dof creation format information
324 * \retval 0 if packing succeeds.
325 * \retval negative errno if packing fails.
327 int update_records_create_pack(const struct lu_env *env,
328 struct update_ops *ops,
329 unsigned int *op_count,
330 size_t *max_ops_size,
331 struct update_params *params,
332 unsigned int *param_count,
333 size_t *max_param_size,
334 const struct lu_fid *fid,
335 const struct lu_attr *attr,
336 const struct dt_allocation_hint *hint,
337 struct dt_object_format *dof)
342 const struct lu_fid *parent_fid = NULL;
343 struct lu_fid tmp_fid;
348 obdo = &update_env_info(env)->uti_obdo;
350 obdo_from_la(obdo, attr, attr->la_valid);
351 lustre_set_wire_obdo(NULL, obdo, obdo);
352 bufs[buf_count] = obdo;
353 sizes[buf_count] = sizeof(*obdo);
357 if (hint != NULL && hint->dah_parent != NULL) {
358 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
359 fid_cpu_to_le(&tmp_fid, parent_fid);
360 bufs[buf_count] = &tmp_fid;
361 sizes[buf_count] = sizeof(tmp_fid);
365 rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
366 max_ops_size, params, param_count,
367 max_param_size, buf_count, bufs, sizes);
370 EXPORT_SYMBOL(update_records_create_pack);
373 * Calculate attr set update size
375 * \param[in] env execution environment
376 * \param[in] ops ur_ops in update records
377 * \param[in] fid FID of the object to set attr
378 * \param[in] attr attribute of attr set
380 * \retval size of attr set update.
382 size_t update_records_attr_set_size(const struct lu_env *env,
383 const struct lu_fid *fid,
384 const struct lu_attr *attr)
386 size_t size = sizeof(struct obdo);
388 return update_records_update_size(1, &size);
390 EXPORT_SYMBOL(update_records_attr_set_size);
393 * Pack attr set update
395 * Pack attr_set update into update records.
397 * \param[in] env execution environment
398 * \param[in] ops ur_ops in update records
399 * \param[in|out] op_count pointer to the count of ops
400 * \param[in|out] max_op_size maximum size of the update
401 * \param[in] params ur_params in update records
402 * \param[in|out] param_count pointer to the count of params
403 * \param[in|out] max_param_size maximum size of the parameter
404 * \param[in] fid FID of the object to set attr
405 * \param[in] attr attribute of attr set
407 * \retval 0 if packing succeeds.
408 * \retval negative errno if packing fails.
410 int update_records_attr_set_pack(const struct lu_env *env,
411 struct update_ops *ops,
412 unsigned int *op_count,
413 size_t *max_ops_size,
414 struct update_params *params,
415 unsigned int *param_count,
416 size_t *max_param_size,
417 const struct lu_fid *fid,
418 const struct lu_attr *attr)
420 struct obdo *obdo = &update_env_info(env)->uti_obdo;
421 size_t size = sizeof(*obdo);
424 obdo_from_la(obdo, attr, attr->la_valid);
425 lustre_set_wire_obdo(NULL, obdo, obdo);
426 return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
427 max_ops_size, params, param_count,
429 (const void **)&obdo, &size);
431 EXPORT_SYMBOL(update_records_attr_set_pack);
434 * Calculate ref add update size
436 * \param[in] env execution environment
437 * \param[in] fid FID of the object to add reference
439 * \retval size of ref_add udpate.
441 size_t update_records_ref_add_size(const struct lu_env *env,
442 const struct lu_fid *fid)
444 return update_records_update_size(0, NULL);
446 EXPORT_SYMBOL(update_records_ref_add_size);
449 * Pack ref add update
451 * Pack ref add update into update records.
453 * \param[in] env execution environment
454 * \param[in] ops ur_ops in update records
455 * \param[in|out] op_count pointer to the count of ops
456 * \param[in|out] max_op_size maximum size of the update
457 * \param[in] params ur_params in update records
458 * \param[in|out] param_count pointer to the count of params
459 * \param[in|out] max_param_size maximum size of the parameter
460 * \param[in] fid FID of the object to add reference
462 * \retval 0 if packing succeeds.
463 * \retval negative errno if packing fails.
465 int update_records_ref_add_pack(const struct lu_env *env,
466 struct update_ops *ops,
467 unsigned int *op_count,
468 size_t *max_ops_size,
469 struct update_params *params,
470 unsigned int *param_count,
471 size_t *max_param_size,
472 const struct lu_fid *fid)
474 return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
475 max_ops_size, params, param_count,
476 max_param_size, 0, NULL, NULL);
478 EXPORT_SYMBOL(update_records_ref_add_pack);
483 * Pack no op update into update records. Note: no op means
484 * the update does not need do anything, which is only used
485 * in test case to verify large size record.
487 * \param[in] env execution environment
488 * \param[in] ops ur_ops in update records
489 * \param[in|out] op_count pointer to the count of ops
490 * \param[in|out] max_op_size maximum size of the update
491 * \param[in] params ur_params in update records
492 * \param[in|out] param_count pointer to the count of params
493 * \param[in|out] max_param_size maximum size of the parameter
494 * \param[in] fid FID of the object to add reference
496 * \retval 0 if packing succeeds.
497 * \retval negative errno if packing fails.
499 int update_records_noop_pack(const struct lu_env *env,
500 struct update_ops *ops,
501 unsigned int *op_count,
502 size_t *max_ops_size,
503 struct update_params *params,
504 unsigned int *param_count,
505 size_t *max_param_size,
506 const struct lu_fid *fid)
508 return update_records_update_pack(env, fid, OUT_NOOP, ops, op_count,
509 max_ops_size, params, param_count,
510 max_param_size, 0, NULL, NULL);
512 EXPORT_SYMBOL(update_records_noop_pack);
515 * Calculate ref del update size
517 * \param[in] env execution environment
518 * \param[in] fid FID of the object to delete reference
520 * \retval size of ref_del update.
522 size_t update_records_ref_del_size(const struct lu_env *env,
523 const struct lu_fid *fid)
525 return update_records_update_size(0, NULL);
527 EXPORT_SYMBOL(update_records_ref_del_size);
530 * Pack ref del update
532 * Pack ref del update into update records.
534 * \param[in] env execution environment
535 * \param[in] ops ur_ops in update records
536 * \param[in|out] op_count pointer to the count of ops
537 * \param[in|out] max_op_size maximum size of the update
538 * \param[in] params ur_params in update records
539 * \param[in] param_count pointer to the count of params
540 * \param[in|out] max_param_size maximum size of the parameter
541 * \param[in] fid FID of the object to delete reference
543 * \retval 0 if packing succeeds.
544 * \retval negative errno if packing fails.
546 int update_records_ref_del_pack(const struct lu_env *env,
547 struct update_ops *ops,
548 unsigned int *op_count,
549 size_t *max_ops_size,
550 struct update_params *params,
551 unsigned int *param_count,
552 size_t *max_param_size,
553 const struct lu_fid *fid)
555 return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
556 max_ops_size, params, param_count,
557 max_param_size, 0, NULL, NULL);
559 EXPORT_SYMBOL(update_records_ref_del_pack);
562 * Calculate object destroy update size
564 * \param[in] env execution environment
565 * \param[in] fid FID of the object to delete reference
567 * \retval size of object destroy update.
569 size_t update_records_object_destroy_size(const struct lu_env *env,
570 const struct lu_fid *fid)
572 return update_records_update_size(0, NULL);
574 EXPORT_SYMBOL(update_records_object_destroy_size);
577 * Pack object destroy update
579 * Pack object destroy update into update records.
581 * \param[in] env execution environment
582 * \param[in] ops ur_ops in update records
583 * \param[in|out] op_count pointer to the count of ops
584 * \param[in|out] max_op_size maximum size of the update
585 * \param[in] params ur_params in update records
586 * \param[in|out] param_count pointer to the count of params
587 * \param[in|out] max_param_size maximum size of the parameter
588 * \param[in] fid FID of the object to delete reference
590 * \retval 0 if packing succeeds.
591 * \retval negative errno if packing fails.
593 int update_records_object_destroy_pack(const struct lu_env *env,
594 struct update_ops *ops,
595 unsigned int *op_count,
596 size_t *max_ops_size,
597 struct update_params *params,
598 unsigned int *param_count,
599 size_t *max_param_size,
600 const struct lu_fid *fid)
602 return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
603 max_ops_size, params, param_count,
604 max_param_size, 0, NULL, NULL);
606 EXPORT_SYMBOL(update_records_object_destroy_pack);
609 * Calculate index insert update size
611 * \param[in] env execution environment
612 * \param[in] fid FID of the object to insert index
613 * \param[in] rec record of insertion
614 * \param[in] key key of insertion
616 * \retval the size of index insert update.
618 size_t update_records_index_insert_size(const struct lu_env *env,
619 const struct lu_fid *fid,
620 const struct dt_rec *rec,
621 const struct dt_key *key)
623 size_t sizes[3] = { strlen((const char *)key) + 1,
624 sizeof(struct lu_fid),
626 return update_records_update_size(3, sizes);
628 EXPORT_SYMBOL(update_records_index_insert_size);
631 * Pack index insert update
633 * Pack index insert update into update records.
635 * \param[in] env execution environment
636 * \param[in] ops ur_ops in update records
637 * \param[in] op_count pointer to the count of ops
638 * \param[in|out] max_op_size maximum size of the update
639 * \param[in] params ur_params in update records
640 * \param[in] param_count pointer to the count of params
641 * \param[in|out] max_param_size maximum size of the parameter
642 * \param[in] fid FID of the object to insert index
643 * \param[in] rec record of insertion
644 * \param[in] key key of insertion
646 * \retval 0 if packing succeeds.
647 * \retval negative errno if packing fails.
649 int update_records_index_insert_pack(const struct lu_env *env,
650 struct update_ops *ops,
651 unsigned int *op_count,
652 size_t *max_ops_size,
653 struct update_params *params,
654 unsigned int *param_count,
655 size_t *max_param_size,
656 const struct lu_fid *fid,
657 const struct dt_rec *rec,
658 const struct dt_key *key)
660 struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
661 struct lu_fid rec_fid;
662 __u32 type = cpu_to_le32(rec1->rec_type);
663 size_t sizes[3] = { strlen((const char *)key) + 1,
666 const void *bufs[3] = { key,
670 fid_cpu_to_le(&rec_fid, rec1->rec_fid);
672 return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
673 op_count, max_ops_size, params,
674 param_count, max_param_size,
677 EXPORT_SYMBOL(update_records_index_insert_pack);
680 * Calculate index delete update size
682 * \param[in] env execution environment
683 * \param[in] fid FID of the object to delete index
684 * \param[in] key key of deletion
686 * \retval the size of index delete update
688 size_t update_records_index_delete_size(const struct lu_env *env,
689 const struct lu_fid *fid,
690 const struct dt_key *key)
692 size_t size = strlen((const char *)key) + 1;
694 return update_records_update_size(1, &size);
696 EXPORT_SYMBOL(update_records_index_delete_size);
699 * Pack index delete update
701 * Pack index delete update into update records.
703 * \param[in] env execution environment
704 * \param[in] ops ur_ops in update records
705 * \param[in|out] op_count pointer to the count of ops
706 * \param[in|out] max_op_size maximum size of the update
707 * \param[in] params ur_params in update records
708 * \param[in|ount] param_count pointer to the count of params
709 * \param[in|out] max_param_size maximum size of the parameter
710 * \param[in] fid FID of the object to delete index
711 * \param[in] key key of deletion
713 * \retval 0 if packing succeeds.
714 * \retval negative errno if packing fails.
716 int update_records_index_delete_pack(const struct lu_env *env,
717 struct update_ops *ops,
718 unsigned int *op_count,
719 size_t *max_ops_size,
720 struct update_params *params,
721 unsigned int *param_count,
722 size_t *max_param_size,
723 const struct lu_fid *fid,
724 const struct dt_key *key)
726 size_t size = strlen((const char *)key) + 1;
728 return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
729 op_count, max_ops_size, params,
730 param_count, max_param_size,
731 1, (const void **)&key, &size);
733 EXPORT_SYMBOL(update_records_index_delete_pack);
736 * Calculate xattr set size
738 * \param[in] env execution environment
739 * \param[in] fid FID of the object to set xattr
740 * \param[in] buf xattr to be set
741 * \param[in] name name of the xattr
742 * \param[in] flag flag for setting xattr
744 * \retval size of xattr set update.
746 size_t update_records_xattr_set_size(const struct lu_env *env,
747 const struct lu_fid *fid,
748 const struct lu_buf *buf,
749 const char *name, __u32 flag)
751 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
753 return update_records_update_size(3, sizes);
755 EXPORT_SYMBOL(update_records_xattr_set_size);
758 * Pack xattr set update
760 * Pack xattr set update into update records.
762 * \param[in] env execution environment
763 * \param[in] ops ur_ops in update records
764 * \param[in|out] op_count pointer to the count of ops
765 * \param[in|out] max_op_size maximum size of the update
766 * \param[in] params ur_params in update records
767 * \param[in|out] param_count pointer to the count of params
768 * \param[in|out] max_param_size maximum size of the parameter
769 * \param[in] fid FID of the object to set xattr
770 * \param[in] buf xattr to be set
771 * \param[in] name name of the xattr
772 * \param[in] flag flag for setting xattr
774 * \retval 0 if packing succeeds.
775 * \retval negative errno if packing fails.
777 int update_records_xattr_set_pack(const struct lu_env *env,
778 struct update_ops *ops,
779 unsigned int *op_count,
780 size_t *max_ops_size,
781 struct update_params *params,
782 unsigned int *param_count,
783 size_t *max_param_size,
784 const struct lu_fid *fid,
785 const struct lu_buf *buf, const char *name,
788 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
789 const void *bufs[3] = {name, buf->lb_buf, &flag};
791 flag = cpu_to_le32(flag);
793 return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
794 op_count, max_ops_size, params,
795 param_count, max_param_size,
798 EXPORT_SYMBOL(update_records_xattr_set_pack);
801 * Calculate xattr delete update size.
803 * \param[in] env execution environment
804 * \param[in] fid FID of the object to delete xattr
805 * \param[in] name name of the xattr
807 * \retval size of xattr delet updatee.
809 size_t update_records_xattr_del_size(const struct lu_env *env,
810 const struct lu_fid *fid,
813 size_t size = strlen(name) + 1;
815 return update_records_update_size(1, &size);
817 EXPORT_SYMBOL(update_records_xattr_del_size);
820 * Pack xattr delete update
822 * Pack xattr delete update into update records.
824 * \param[in] env execution environment
825 * \param[in] ops ur_ops in update records
826 * \param[in|out] op_count pointer to the count of ops
827 * \param[in|out] max_op_size maximum size of the update
828 * \param[in] params ur_params in update records
829 * \param[in|out] param_count pointer to the count of params
830 * \param[in|out] max_param_size maximum size of the parameter
831 * \param[in] fid FID of the object to delete xattr
832 * \param[in] name name of the xattr
834 * \retval 0 if packing succeeds.
835 * \retval negative errno if packing fails.
837 int update_records_xattr_del_pack(const struct lu_env *env,
838 struct update_ops *ops,
839 unsigned int *op_count,
840 size_t *max_ops_size,
841 struct update_params *params,
842 unsigned int *param_count,
843 size_t *max_param_size,
844 const struct lu_fid *fid,
847 size_t size = strlen(name) + 1;
849 return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
850 op_count, max_ops_size, params,
851 param_count, max_param_size,
852 1, (const void **)&name, &size);
854 EXPORT_SYMBOL(update_records_xattr_del_pack);
857 * Calculate write update size
859 * \param[in] env execution environment
860 * \param[in] fid FID of the object to write into
861 * \param[in] buf buffer to write which includes an embedded size field
862 * \param[in] pos offet in the object to start writing at
864 * \retval size of write udpate.
866 size_t update_records_write_size(const struct lu_env *env,
867 const struct lu_fid *fid,
868 const struct lu_buf *buf,
871 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
873 return update_records_update_size(2, sizes);
875 EXPORT_SYMBOL(update_records_write_size);
880 * Pack write update into update records.
882 * \param[in] env execution environment
883 * \param[in] ops ur_ops in update records
884 * \param[in|out] op_count pointer to the count of ops
885 * \param[in|out] max_op_size maximum size of the update
886 * \param[in] params ur_params in update records
887 * \param[in|out] param_count pointer to the count of params
888 * \param[in|out] max_param_size maximum size of the parameter
889 * \param[in] fid FID of the object to write into
890 * \param[in] buf buffer to write which includes an embedded size field
891 * \param[in] pos offet in the object to start writing at
893 * \retval 0 if packing succeeds.
894 * \retval negative errno if packing fails.
896 int update_records_write_pack(const struct lu_env *env,
897 struct update_ops *ops,
898 unsigned int *op_count,
899 size_t *max_ops_size,
900 struct update_params *params,
901 unsigned int *param_count,
902 size_t *max_param_size,
903 const struct lu_fid *fid,
904 const struct lu_buf *buf,
907 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
908 const void *bufs[2] = {buf->lb_buf, &pos};
910 pos = cpu_to_le64(pos);
912 return update_records_update_pack(env, fid, OUT_WRITE, ops,
913 op_count, max_ops_size, params,
914 param_count, max_param_size,
917 EXPORT_SYMBOL(update_records_write_pack);
920 * Calculate size of punch update.
922 * \param[in] env execution environment
923 * \param[in] fid FID of the object to write into
924 * \param[in] start start offset of punch
925 * \param[in] end end offet of punch
927 * \retval size of update punch.
929 size_t update_records_punch_size(const struct lu_env *env,
930 const struct lu_fid *fid,
931 __u64 start, __u64 end)
933 size_t sizes[2] = {sizeof(start), sizeof(end)};
935 return update_records_update_size(2, sizes);
937 EXPORT_SYMBOL(update_records_punch_size);
942 * Pack punch update into update records.
944 * \param[in] env execution environment
945 * \param[in] ops ur_ops in update records
946 * \param[in|out] op_count pointer to the count of ops
947 * \param[in|out] max_op_size maximum size of the update
948 * \param[in] params ur_params in update records
949 * \param[in|out] param_count pointer to the count of params
950 * \param[in|out] max_param_size maximum size of the parameter
951 * \param[in] fid FID of the object to write into
952 * \param[in] start start offset of punch
953 * \param[in] end end offet of punch
955 * \retval 0 if packing succeeds.
956 * \retval negative errno if packing fails.
958 int update_records_punch_pack(const struct lu_env *env,
959 struct update_ops *ops,
960 unsigned int *op_count,
961 size_t *max_ops_size,
962 struct update_params *params,
963 unsigned int *param_count,
964 size_t *max_param_size,
965 const struct lu_fid *fid,
966 __u64 start, __u64 end)
968 size_t sizes[2] = {sizeof(start), sizeof(end)};
969 const void *bufs[2] = {&start, &end};
971 start = cpu_to_le64(start);
972 end = cpu_to_le64(end);
974 return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
975 max_ops_size, params, param_count,
976 max_param_size, 2, bufs, sizes);
978 EXPORT_SYMBOL(update_records_punch_pack);
981 * Create update records in thandle_update_records
983 * Allocate update_records for thandle_update_records, the initial size
986 * \param[in] tur thandle_update_records where update_records will be
988 * \retval 0 if allocation succeeds.
989 * \retval negative errno if allocation fails.
991 static int tur_update_records_create(struct thandle_update_records *tur)
993 if (tur->tur_update_records != NULL)
996 OBD_ALLOC_LARGE(tur->tur_update_records,
997 UPDATE_RECORDS_BUFFER_SIZE);
999 if (tur->tur_update_records == NULL)
1002 tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
1008 * Extend update records
1010 * Extend update_records to the new size in thandle_update_records.
1012 * \param[in] tur thandle_update_records where update_records will be
1014 * \retval 0 if extension succeeds.
1015 * \retval negative errno if extension fails.
1017 int tur_update_records_extend(struct thandle_update_records *tur,
1020 struct llog_update_record *record;
1022 OBD_ALLOC_LARGE(record, new_size);
1026 if (tur->tur_update_records != NULL) {
1027 memcpy(record, tur->tur_update_records,
1028 tur->tur_update_records_buf_size);
1029 OBD_FREE_LARGE(tur->tur_update_records,
1030 tur->tur_update_records_buf_size);
1033 tur->tur_update_records = record;
1034 tur->tur_update_records_buf_size = new_size;
1038 EXPORT_SYMBOL(tur_update_records_extend);
1041 * Extend update records
1043 * Extend update records in thandle to make sure it is able to hold
1044 * the update with certain update_op and params size.
1046 * \param [in] tur thandle_update_records to be extend
1047 * \param [in] new_op_size update_op size of the update record
1048 * \param [in] new_param_size params size of the update record
1050 * \retval 0 if the update_records is being extended.
1051 * \retval negative errno if the update_records is not being
1054 int tur_update_extend(struct thandle_update_records *tur,
1055 size_t new_op_size, size_t new_param_size)
1063 record_size = llog_update_record_size(tur->tur_update_records);
1064 /* extend update records buffer */
1065 if (new_op_size >= (tur->tur_update_records_buf_size - record_size)) {
1066 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
1067 rc = tur_update_records_extend(tur,
1068 tur->tur_update_records_buf_size +
1074 /* extend parameters buffer */
1075 params_size = update_params_size(tur->tur_update_params,
1076 tur->tur_update_param_count);
1077 if (new_param_size >= (tur->tur_update_params_buf_size -
1079 extend_size = round_up(new_param_size,
1080 UPDATE_PARAMS_BUFFER_SIZE);
1081 rc = tur_update_params_extend(tur,
1082 tur->tur_update_params_buf_size +
1090 EXPORT_SYMBOL(tur_update_extend);
1093 * Create update params in thandle_update_records
1095 * Allocate update_params for thandle_update_records, the initial size
1098 * \param[in] tur thandle_update_records where update_params will be
1100 * \retval 0 if allocation succeeds.
1101 * \retval negative errno if allocation fails.
1103 static int tur_update_params_create(struct thandle_update_records *tur)
1105 if (tur->tur_update_params != NULL)
1108 OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
1109 if (tur->tur_update_params == NULL)
1112 tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
1117 * Extend update params
1119 * Extend update_params to the new size in thandle_update_records.
1121 * \param[in] tur thandle_update_records where update_params will be
1123 * \retval 0 if extension succeeds.
1124 * \retval negative errno if extension fails.
1126 int tur_update_params_extend(struct thandle_update_records *tur,
1129 struct update_params *params;
1131 OBD_ALLOC_LARGE(params, new_size);
1135 if (tur->tur_update_params != NULL) {
1136 memcpy(params, tur->tur_update_params,
1137 tur->tur_update_params_buf_size);
1138 OBD_FREE_LARGE(tur->tur_update_params,
1139 tur->tur_update_params_buf_size);
1142 tur->tur_update_params = params;
1143 tur->tur_update_params_buf_size = new_size;
1147 EXPORT_SYMBOL(tur_update_params_extend);
1150 * Check and prepare whether it needs to record update.
1152 * Checks if the transaction needs to record updates, and if it
1153 * does, then initialize the update record buffer in the transaction.
1155 * \param[in] env execution environment
1156 * \param[in] th transaction handle
1158 * \retval 0 if updates recording succeeds.
1159 * \retval negative errno if updates recording fails.
1161 int check_and_prepare_update_record(const struct lu_env *env,
1162 struct thandle_update_records *tur)
1164 struct llog_update_record *lur;
1167 if (tur->tur_update_records == NULL) {
1168 rc = tur_update_records_create(tur);
1173 if (tur->tur_update_params == NULL) {
1174 rc = tur_update_params_create(tur);
1179 lur = tur->tur_update_records;
1180 lur->lur_update_rec.ur_update_count = 0;
1181 lur->lur_update_rec.ur_param_count = 0;
1182 lur->lur_update_rec.ur_master_transno = 0;
1183 lur->lur_update_rec.ur_batchid = 0;
1184 lur->lur_update_rec.ur_flags = 0;
1185 lur->lur_hdr.lrh_len = LLOG_MIN_CHUNK_SIZE;
1187 tur->tur_update_param_count = 0;
1192 static void update_key_fini(const struct lu_context *ctx,
1193 struct lu_context_key *key, void *data)
1195 struct update_thread_info *info = data;
1196 struct thandle_exec_args *args = &info->uti_tea;
1199 for (i = 0; i < args->ta_alloc_args; i++) {
1200 if (args->ta_args[i] != NULL)
1201 OBD_FREE_PTR(args->ta_args[i]);
1204 if (args->ta_args != NULL)
1205 OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
1206 args->ta_alloc_args);
1208 if (info->uti_tur.tur_update_records != NULL)
1209 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
1210 info->uti_tur.tur_update_records_buf_size);
1211 if (info->uti_tur.tur_update_params != NULL)
1212 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
1213 info->uti_tur.tur_update_params_buf_size);
1218 /* context key constructor/destructor: update_key_init, update_key_fini */
1219 LU_KEY_INIT(update, struct update_thread_info);
1220 /* context key: update_thread_key */
1221 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
1222 LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
1223 EXPORT_SYMBOL(update_thread_key);
1224 LU_KEY_INIT_GENERIC(update);
1226 void update_info_init(void)
1228 update_key_init_generic(&update_thread_key, NULL);
1229 lu_context_key_register(&update_thread_key);
1232 void update_info_fini(void)
1234 lu_context_key_degister(&update_thread_key);