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, 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_update.h>
46 #include <obd_class.h>
47 #include "tgt_internal.h"
49 #define UPDATE_RECORDS_BUFFER_SIZE 8192
50 #define UPDATE_PARAMS_BUFFER_SIZE 8192
54 * Dump all of updates in the update_records, mostly for debugging purpose.
56 * \param[in] records update records to be dumpped
57 * \param[in] mask debug level mask
58 * \param[in] dump_params if dump all of updates the updates.
61 void update_records_dump(const struct update_records *records,
62 unsigned int mask, bool dump_updates)
64 const struct update_ops *ops;
65 const struct update_op *op = NULL;
66 struct update_params *params;
69 ops = &records->ur_ops;
70 params = update_records_get_params(records);
72 CDEBUG(mask, "master transno = "LPU64" batchid = "LPU64" flags = %x"
73 " ops = %d params = %d\n", records->ur_master_transno,
74 records->ur_batchid, records->ur_flags, records->ur_update_count,
75 records->ur_param_count);
77 if (records->ur_update_count == 0)
83 op = &ops->uops_op[0];
84 for (i = 0; i < records->ur_update_count; i++) {
87 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
88 PFID(&op->uop_fid), update_op_str(op->uop_type),
91 for (j = 0; j < op->uop_param_count; j++) {
92 struct object_update_param *param;
94 param = update_params_get_param(params,
95 (unsigned int)op->uop_params_off[j],
96 records->ur_param_count);
98 LASSERT(param != NULL);
99 CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
100 param, j, op->uop_params_off[j], param->oup_len);
103 op = update_op_next_op(op);
108 * Pack parameters to update records
110 * Find and insert parameter to update records, if the parameter
111 * already exists in \a params, then just return the offset of this
112 * parameter, otherwise insert the parameter and return its offset
114 * \param[in] params update params in which to insert parameter
115 * \param[in] new_param parameters to be inserted.
116 * \param[in] new_param_size the size of \a new_param
118 * \retval index inside \a params if parameter insertion
120 * \retval negative errno if it fails.
122 static unsigned int update_records_param_pack(struct update_params *params,
123 const void *new_param,
124 size_t new_param_size,
125 unsigned int *param_count)
127 struct object_update_param *param;
130 for (i = 0; i < *param_count; i++) {
131 struct object_update_param *param;
133 param = update_params_get_param(params, i, *param_count);
134 if ((new_param == NULL && param->oup_len == new_param_size) ||
135 (param->oup_len == new_param_size &&
136 memcmp(param->oup_buf, new_param, new_param_size) == 0))
137 /* Found the parameter and return its index */
141 param = (struct object_update_param *)((char *)params +
142 update_params_size(params, *param_count));
144 param->oup_len = new_param_size;
145 if (new_param != NULL)
146 memcpy(param->oup_buf, new_param, new_param_size);
148 *param_count = *param_count + 1;
150 return *param_count - 1;
154 * Pack update to update records
156 * Pack the update and its parameters to the update records. First it will
157 * insert parameters, get the offset of these parameter, then fill the
158 * update with these offset. If insertion exceed the maximum size of
159 * current update records, it will return -E2BIG here, and the caller might
160 * extend the update_record size \see lod_updates_pack.
162 * \param[in] env execution environment
163 * \param[in] fid FID of the update.
164 * \param[in] op_type operation type of the update
165 * \param[in] ops ur_ops in update records
166 * \param[in|out] op_count pointer to the count of ops
167 * \param[in|out] max_op_size maximum size of the update
168 * \param[in] params ur_params in update records
169 * \param[in|out] param_count pointer to the count of params
170 * \param[in|out] max_param_size maximum size of the parameter
171 * \param[in] param_bufs buffers of parameters
172 * \param[in] params_buf_count the count of the parameter buffers
173 * \param[in] param_size sizes of parameters
175 * \retval 0 if packing succeeds
176 * \retval negative errno if packing fails
178 static int update_records_update_pack(const struct lu_env *env,
179 const struct lu_fid *fid,
180 enum update_type op_type,
181 struct update_ops *ops,
182 unsigned int *op_count,
184 struct update_params *params,
185 unsigned int *param_count,
186 size_t *max_param_size,
187 unsigned int param_bufs_count,
188 const void **param_bufs,
191 struct update_op *op;
192 size_t total_param_sizes = 0;
196 /* Check whether the packing exceeding the maximum update size */
197 if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
198 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
199 *max_op_size, update_op_size(param_bufs_count));
200 *max_op_size = update_op_size(param_bufs_count);
204 for (i = 0; i < param_bufs_count; i++)
206 cfs_size_round(sizeof(struct object_update_param) +
209 /* Check whether the packing exceeding the maximum parameter size */
210 if (unlikely(*max_param_size < total_param_sizes)) {
211 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
212 *max_param_size, total_param_sizes);
214 *max_param_size = total_param_sizes;
218 op = update_ops_get_op(ops, *op_count, *op_count);
220 op->uop_type = op_type;
221 op->uop_param_count = param_bufs_count;
222 for (i = 0; i < param_bufs_count; i++) {
223 index = update_records_param_pack(params, param_bufs[i],
224 param_sizes[i], param_count);
228 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
229 update_op_str(op_type), i, index, param_sizes[i]);
231 op->uop_params_off[i] = index;
233 CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
234 *op_count, PFID(fid), update_op_str(op_type), *param_count);
236 *op_count = *op_count + 1;
244 * Pack create update into update records.
246 * \param[in] env execution environment
247 * \param[in] ops ur_ops in update records
248 * \param[in|out] op_count pointer to the count of ops
249 * \param[in|out] max_op_size maximum size of the update
250 * \param[in] params ur_params in update records
251 * \param[in|out] param_count pointer to the count of params
252 * \param[in|out] max_param_size maximum size of the parameter
253 * \param[in] fid FID of the object to be created
254 * \param[in] attr attribute of the object to be created
255 * \param[in] hint creation hint
256 * \param[in] dof creation format information
258 * \retval 0 if packing succeeds.
259 * \retval negative errno if packing fails.
261 int update_records_create_pack(const struct lu_env *env,
262 struct update_ops *ops,
263 unsigned int *op_count,
264 size_t *max_ops_size,
265 struct update_params *params,
266 unsigned int *param_count,
267 size_t *max_param_size,
268 const struct lu_fid *fid,
269 const struct lu_attr *attr,
270 const struct dt_allocation_hint *hint,
271 struct dt_object_format *dof)
276 const struct lu_fid *parent_fid = NULL;
277 struct lu_fid tmp_fid;
282 obdo = &update_env_info(env)->uti_obdo;
284 obdo_from_la(obdo, attr, attr->la_valid);
285 lustre_set_wire_obdo(NULL, obdo, obdo);
286 bufs[buf_count] = obdo;
287 sizes[buf_count] = sizeof(*obdo);
291 if (hint != NULL && hint->dah_parent != NULL) {
292 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
293 fid_cpu_to_le(&tmp_fid, parent_fid);
294 bufs[buf_count] = &tmp_fid;
295 sizes[buf_count] = sizeof(tmp_fid);
299 rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
300 max_ops_size, params, param_count,
301 max_param_size, buf_count, bufs, sizes);
304 EXPORT_SYMBOL(update_records_create_pack);
307 * Pack attr set update
309 * Pack attr_set update into update records.
311 * \param[in] env execution environment
312 * \param[in] ops ur_ops in update records
313 * \param[in|out] op_count pointer to the count of ops
314 * \param[in|out] max_op_size maximum size of the update
315 * \param[in] params ur_params in update records
316 * \param[in|out] param_count pointer to the count of params
317 * \param[in|out] max_param_size maximum size of the parameter
318 * \param[in] fid FID of the object to set attr
319 * \param[in] attr attribute of attr set
321 * \retval 0 if packing succeeds.
322 * \retval negative errno if packing fails.
324 int update_records_attr_set_pack(const struct lu_env *env,
325 struct update_ops *ops,
326 unsigned int *op_count,
327 size_t *max_ops_size,
328 struct update_params *params,
329 unsigned int *param_count,
330 size_t *max_param_size,
331 const struct lu_fid *fid,
332 const struct lu_attr *attr)
334 struct obdo *obdo = &update_env_info(env)->uti_obdo;
335 size_t size = sizeof(*obdo);
338 obdo_from_la(obdo, attr, attr->la_valid);
339 lustre_set_wire_obdo(NULL, obdo, obdo);
340 return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
341 max_ops_size, params, param_count,
343 (const void **)&obdo, &size);
345 EXPORT_SYMBOL(update_records_attr_set_pack);
348 * Pack ref add update
350 * Pack ref add update into update records.
352 * \param[in] env execution environment
353 * \param[in] ops ur_ops in update records
354 * \param[in|out] op_count pointer to the count of ops
355 * \param[in|out] max_op_size maximum size of the update
356 * \param[in] params ur_params in update records
357 * \param[in|out] param_count pointer to the count of params
358 * \param[in|out] max_param_size maximum size of the parameter
359 * \param[in] fid FID of the object to add reference
361 * \retval 0 if packing succeeds.
362 * \retval negative errno if packing fails.
364 int update_records_ref_add_pack(const struct lu_env *env,
365 struct update_ops *ops,
366 unsigned int *op_count,
367 size_t *max_ops_size,
368 struct update_params *params,
369 unsigned int *param_count,
370 size_t *max_param_size,
371 const struct lu_fid *fid)
373 return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
374 max_ops_size, params, param_count,
375 max_param_size, 0, NULL, NULL);
377 EXPORT_SYMBOL(update_records_ref_add_pack);
380 * Pack ref del update
382 * Pack ref del update into update records.
384 * \param[in] env execution environment
385 * \param[in] ops ur_ops in update records
386 * \param[in|out] op_count pointer to the count of ops
387 * \param[in|out] max_op_size maximum size of the update
388 * \param[in] params ur_params in update records
389 * \param[in] param_count pointer to the count of params
390 * \param[in|out] max_param_size maximum size of the parameter
391 * \param[in] fid FID of the object to delete reference
393 * \retval 0 if packing succeeds.
394 * \retval negative errno if packing fails.
396 int update_records_ref_del_pack(const struct lu_env *env,
397 struct update_ops *ops,
398 unsigned int *op_count,
399 size_t *max_ops_size,
400 struct update_params *params,
401 unsigned int *param_count,
402 size_t *max_param_size,
403 const struct lu_fid *fid)
405 return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
406 max_ops_size, params, param_count,
407 max_param_size, 0, NULL, NULL);
409 EXPORT_SYMBOL(update_records_ref_del_pack);
412 * Pack object destroy update
414 * Pack object destroy update into update records.
416 * \param[in] env execution environment
417 * \param[in] ops ur_ops in update records
418 * \param[in|out] op_count pointer to the count of ops
419 * \param[in|out] max_op_size maximum size of the update
420 * \param[in] params ur_params in update records
421 * \param[in|out] param_count pointer to the count of params
422 * \param[in|out] max_param_size maximum size of the parameter
423 * \param[in] fid FID of the object to delete reference
425 * \retval 0 if packing succeeds.
426 * \retval negative errno if packing fails.
428 int update_records_object_destroy_pack(const struct lu_env *env,
429 struct update_ops *ops,
430 unsigned int *op_count,
431 size_t *max_ops_size,
432 struct update_params *params,
433 unsigned int *param_count,
434 size_t *max_param_size,
435 const struct lu_fid *fid)
437 return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
438 max_ops_size, params, param_count,
439 max_param_size, 0, NULL, NULL);
441 EXPORT_SYMBOL(update_records_object_destroy_pack);
444 * Pack index insert update
446 * Pack index insert update into update records.
448 * \param[in] env execution environment
449 * \param[in] ops ur_ops in update records
450 * \param[in] op_count pointer to the count of ops
451 * \param[in|out] max_op_size maximum size of the update
452 * \param[in] params ur_params in update records
453 * \param[in] param_count pointer to the count of params
454 * \param[in|out] max_param_size maximum size of the parameter
455 * \param[in] fid FID of the object to insert index
456 * \param[in] rec record of insertion
457 * \param[in] key key of insertion
459 * \retval 0 if packing succeeds.
460 * \retval negative errno if packing fails.
462 int update_records_index_insert_pack(const struct lu_env *env,
463 struct update_ops *ops,
464 unsigned int *op_count,
465 size_t *max_ops_size,
466 struct update_params *params,
467 unsigned int *param_count,
468 size_t *max_param_size,
469 const struct lu_fid *fid,
470 const struct dt_rec *rec,
471 const struct dt_key *key)
473 struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
474 struct lu_fid rec_fid;
475 __u32 type = cpu_to_le32(rec1->rec_type);
476 size_t sizes[3] = { strlen((const char *)key) + 1,
479 const void *bufs[3] = { key,
483 fid_cpu_to_le(&rec_fid, rec1->rec_fid);
485 return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
486 op_count, max_ops_size, params,
487 param_count, max_param_size,
490 EXPORT_SYMBOL(update_records_index_insert_pack);
493 * Pack index delete update
495 * Pack index delete update into update records.
497 * \param[in] env execution environment
498 * \param[in] ops ur_ops in update records
499 * \param[in|out] op_count pointer to the count of ops
500 * \param[in|out] max_op_size maximum size of the update
501 * \param[in] params ur_params in update records
502 * \param[in|ount] param_count pointer to the count of params
503 * \param[in|out] max_param_size maximum size of the parameter
504 * \param[in] fid FID of the object to delete index
505 * \param[in] key key of deletion
507 * \retval 0 if packing succeeds.
508 * \retval negative errno if packing fails.
510 int update_records_index_delete_pack(const struct lu_env *env,
511 struct update_ops *ops,
512 unsigned int *op_count,
513 size_t *max_ops_size,
514 struct update_params *params,
515 unsigned int *param_count,
516 size_t *max_param_size,
517 const struct lu_fid *fid,
518 const struct dt_key *key)
520 size_t size = strlen((const char *)key) + 1;
522 return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
523 op_count, max_ops_size, params,
524 param_count, max_param_size,
525 1, (const void **)&key, &size);
527 EXPORT_SYMBOL(update_records_index_delete_pack);
530 * Pack xattr set update
532 * Pack xattr set 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|out] 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 set xattr
542 * \param[in] buf xattr to be set
543 * \param[in] name name of the xattr
544 * \param[in] flag flag for setting xattr
546 * \retval 0 if packing succeeds.
547 * \retval negative errno if packing fails.
549 int update_records_xattr_set_pack(const struct lu_env *env,
550 struct update_ops *ops,
551 unsigned int *op_count,
552 size_t *max_ops_size,
553 struct update_params *params,
554 unsigned int *param_count,
555 size_t *max_param_size,
556 const struct lu_fid *fid,
557 const struct lu_buf *buf, const char *name,
560 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
561 const void *bufs[3] = {name, buf->lb_buf, &flag};
563 flag = cpu_to_le32(flag);
565 return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
566 op_count, max_ops_size, params,
567 param_count, max_param_size,
570 EXPORT_SYMBOL(update_records_xattr_set_pack);
573 * Pack xattr delete update
575 * Pack xattr delete update into update records.
577 * \param[in] env execution environment
578 * \param[in] ops ur_ops in update records
579 * \param[in|out] op_count pointer to the count of ops
580 * \param[in|out] max_op_size maximum size of the update
581 * \param[in] params ur_params in update records
582 * \param[in|out] param_count pointer to the count of params
583 * \param[in|out] max_param_size maximum size of the parameter
584 * \param[in] fid FID of the object to delete xattr
585 * \param[in] name name of the xattr
587 * \retval 0 if packing succeeds.
588 * \retval negative errno if packing fails.
590 int update_records_xattr_del_pack(const struct lu_env *env,
591 struct update_ops *ops,
592 unsigned int *op_count,
593 size_t *max_ops_size,
594 struct update_params *params,
595 unsigned int *param_count,
596 size_t *max_param_size,
597 const struct lu_fid *fid,
600 size_t size = strlen(name) + 1;
602 return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
603 op_count, max_ops_size, params,
604 param_count, max_param_size,
605 1, (const void **)&name, &size);
607 EXPORT_SYMBOL(update_records_xattr_del_pack);
612 * Pack write update into update records.
614 * \param[in] env execution environment
615 * \param[in] ops ur_ops in update records
616 * \param[in|out] op_count pointer to the count of ops
617 * \param[in|out] max_op_size maximum size of the update
618 * \param[in] params ur_params in update records
619 * \param[in|out] param_count pointer to the count of params
620 * \param[in|out] max_param_size maximum size of the parameter
621 * \param[in] fid FID of the object to write into
622 * \param[in] buf buffer to write which includes an embedded size field
623 * \param[in] pos offet in the object to start writing at
625 * \retval 0 if packing succeeds.
626 * \retval negative errno if packing fails.
628 int update_records_write_pack(const struct lu_env *env,
629 struct update_ops *ops,
630 unsigned int *op_count,
631 size_t *max_ops_size,
632 struct update_params *params,
633 unsigned int *param_count,
634 size_t *max_param_size,
635 const struct lu_fid *fid,
636 const struct lu_buf *buf,
639 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
640 const void *bufs[2] = {buf->lb_buf, &pos};
642 pos = cpu_to_le64(pos);
644 return update_records_update_pack(env, fid, OUT_WRITE, ops,
645 op_count, max_ops_size, params,
646 param_count, max_param_size,
649 EXPORT_SYMBOL(update_records_write_pack);
654 * Pack punch update into update records.
656 * \param[in] env execution environment
657 * \param[in] ops ur_ops in update records
658 * \param[in|out] op_count pointer to the count of ops
659 * \param[in|out] max_op_size maximum size of the update
660 * \param[in] params ur_params in update records
661 * \param[in|out] param_count pointer to the count of params
662 * \param[in|out] max_param_size maximum size of the parameter
663 * \param[in] fid FID of the object to write into
664 * \param[in] start start offset of punch
665 * \param[in] end end offet of punch
667 * \retval 0 if packing succeeds.
668 * \retval negative errno if packing fails.
670 int update_records_punch_pack(const struct lu_env *env,
671 struct update_ops *ops,
672 unsigned int *op_count,
673 size_t *max_ops_size,
674 struct update_params *params,
675 unsigned int *param_count,
676 size_t *max_param_size,
677 const struct lu_fid *fid,
678 __u64 start, __u64 end)
680 size_t sizes[2] = {sizeof(start), sizeof(end)};
681 const void *bufs[2] = {&start, &end};
683 start = cpu_to_le64(start);
684 end = cpu_to_le64(end);
686 return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
687 max_ops_size, params, param_count,
688 max_param_size, 2, bufs, sizes);
690 EXPORT_SYMBOL(update_records_punch_pack);
693 * Create update records in thandle_update_records
695 * Allocate update_records for thandle_update_records, the initial size
698 * \param[in] tur thandle_update_records where update_records will be
700 * \retval 0 if allocation succeeds.
701 * \retval negative errno if allocation fails.
703 static int tur_update_records_create(struct thandle_update_records *tur)
705 if (tur->tur_update_records != NULL)
708 OBD_ALLOC_LARGE(tur->tur_update_records,
709 UPDATE_RECORDS_BUFFER_SIZE);
711 if (tur->tur_update_records == NULL)
714 tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
720 * Extend update records
722 * Extend update_records to the new size in thandle_update_records.
724 * \param[in] tur thandle_update_records where update_records will be
726 * \retval 0 if extension succeeds.
727 * \retval negative errno if extension fails.
729 int tur_update_records_extend(struct thandle_update_records *tur,
732 struct llog_update_record *record;
734 OBD_ALLOC_LARGE(record, new_size);
738 if (tur->tur_update_records != NULL) {
739 memcpy(record, tur->tur_update_records,
740 tur->tur_update_records_buf_size);
741 OBD_FREE_LARGE(tur->tur_update_records,
742 tur->tur_update_records_buf_size);
745 tur->tur_update_records = record;
746 tur->tur_update_records_buf_size = new_size;
750 EXPORT_SYMBOL(tur_update_records_extend);
753 * Extend update records
755 * Extend update records in thandle to make sure it is able to hold
756 * the update with certain update_op and params size.
758 * \param [in] tur thandle_update_records to be extend
759 * \param [in] new_op_size update_op size of the update record
760 * \param [in] new_param_size params size of the update record
762 * \retval 0 if the update_records is being extended.
763 * \retval negative errno if the update_records is not being
766 int tur_update_extend(struct thandle_update_records *tur,
767 size_t new_op_size, size_t new_param_size)
775 record_size = llog_update_record_size(tur->tur_update_records);
776 /* extend update records buffer */
777 if (new_op_size > (tur->tur_update_records_buf_size - record_size -
778 sizeof(*tur->tur_update_records))) {
779 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
780 rc = tur_update_records_extend(tur,
781 tur->tur_update_records_buf_size +
787 /* extend parameters buffer */
788 params_size = update_params_size(tur->tur_update_params,
789 tur->tur_update_param_count);
790 if (new_param_size > (tur->tur_update_params_buf_size -
792 extend_size = round_up(new_param_size,
793 UPDATE_PARAMS_BUFFER_SIZE);
794 rc = tur_update_params_extend(tur,
795 tur->tur_update_params_buf_size +
803 EXPORT_SYMBOL(tur_update_extend);
806 * Create update params in thandle_update_records
808 * Allocate update_params for thandle_update_records, the initial size
811 * \param[in] tur thandle_update_records where update_params will be
813 * \retval 0 if allocation succeeds.
814 * \retval negative errno if allocation fails.
816 static int tur_update_params_create(struct thandle_update_records *tur)
818 if (tur->tur_update_params != NULL)
821 OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
822 if (tur->tur_update_params == NULL)
825 tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
830 * Extend update params
832 * Extend update_params to the new size in thandle_update_records.
834 * \param[in] tur thandle_update_records where update_params will be
836 * \retval 0 if extension succeeds.
837 * \retval negative errno if extension fails.
839 int tur_update_params_extend(struct thandle_update_records *tur,
842 struct update_params *params;
844 OBD_ALLOC_LARGE(params, new_size);
848 if (tur->tur_update_params != NULL) {
849 memcpy(params, tur->tur_update_params,
850 tur->tur_update_params_buf_size);
851 OBD_FREE_LARGE(tur->tur_update_params,
852 tur->tur_update_params_buf_size);
855 tur->tur_update_params = params;
856 tur->tur_update_params_buf_size = new_size;
860 EXPORT_SYMBOL(tur_update_params_extend);
863 * Check and prepare whether it needs to record update.
865 * Checks if the transaction needs to record updates, and if it
866 * does, then initialize the update record buffer in the transaction.
868 * \param[in] env execution environment
869 * \param[in] th transaction handle
871 * \retval 0 if updates recording succeeds.
872 * \retval negative errno if updates recording fails.
874 int check_and_prepare_update_record(const struct lu_env *env,
875 struct thandle_update_records *tur)
877 struct llog_update_record *lur;
880 if (tur->tur_update_records == NULL) {
881 rc = tur_update_records_create(tur);
886 if (tur->tur_update_params == NULL) {
887 rc = tur_update_params_create(tur);
892 lur = tur->tur_update_records;
893 lur->lur_update_rec.ur_update_count = 0;
894 lur->lur_update_rec.ur_param_count = 0;
895 lur->lur_update_rec.ur_master_transno = 0;
896 lur->lur_update_rec.ur_batchid = 0;
897 lur->lur_update_rec.ur_flags = 0;
898 lur->lur_hdr.lrh_len = LLOG_MIN_CHUNK_SIZE;
900 tur->tur_update_param_count = 0;
905 static void update_key_fini(const struct lu_context *ctx,
906 struct lu_context_key *key, void *data)
908 struct update_thread_info *info = data;
909 struct thandle_exec_args *args = &info->uti_tea;
912 for (i = 0; i < args->ta_alloc_args; i++) {
913 if (args->ta_args[i] != NULL)
914 OBD_FREE_PTR(args->ta_args[i]);
917 if (args->ta_args != NULL)
918 OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
919 args->ta_alloc_args);
921 if (info->uti_tur.tur_update_records != NULL)
922 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
923 info->uti_tur.tur_update_records_buf_size);
924 if (info->uti_tur.tur_update_params != NULL)
925 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
926 info->uti_tur.tur_update_params_buf_size);
931 /* context key constructor/destructor: update_key_init, update_key_fini */
932 LU_KEY_INIT(update, struct update_thread_info);
933 /* context key: update_thread_key */
934 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
935 LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
936 EXPORT_SYMBOL(update_thread_key);
937 LU_KEY_INIT_GENERIC(update);
939 void update_info_init(void)
941 update_key_init_generic(&update_thread_key, NULL);
942 lu_context_key_register(&update_thread_key);
945 void update_info_fini(void)
947 lu_context_key_degister(&update_thread_key);