forked from percona/PerconaFT
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathule.cc
2579 lines (2337 loc) · 89.2 KB
/
ule.cc
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
#ident "$Id$"
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
/*
COPYING CONDITIONS NOTICE:
This program is free software; you can redistribute it and/or modify
it under the terms of version 2 of the GNU General Public License as
published by the Free Software Foundation, and provided that the
following conditions are met:
* Redistributions of source code must retain this COPYING
CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
PATENT MARKING NOTICE (below), and the PATENT RIGHTS
GRANT (below).
* Redistributions in binary form must reproduce this COPYING
CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
PATENT MARKING NOTICE (below), and the PATENT RIGHTS
GRANT (below) in the documentation and/or other materials
provided with the distribution.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
02110-1301, USA.
COPYRIGHT NOTICE:
TokuFT, Tokutek Fractal Tree Indexing Library.
Copyright (C) 2007-2013 Tokutek, Inc.
DISCLAIMER:
This program is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
General Public License for more details.
UNIVERSITY PATENT NOTICE:
The technology is licensed by the Massachusetts Institute of
Technology, Rutgers State University of New Jersey, and the Research
Foundation of State University of New York at Stony Brook under
United States of America Serial No. 11/760379 and to the patents
and/or patent applications resulting from it.
PATENT MARKING NOTICE:
This software is covered by US Patent No. 8,185,551.
This software is covered by US Patent No. 8,489,638.
PATENT RIGHTS GRANT:
"THIS IMPLEMENTATION" means the copyrightable works distributed by
Tokutek as part of the Fractal Tree project.
"PATENT CLAIMS" means the claims of patents that are owned or
licensable by Tokutek, both currently or in the future; and that in
the absence of this license would be infringed by THIS
IMPLEMENTATION or by using or running THIS IMPLEMENTATION.
"PATENT CHALLENGE" shall mean a challenge to the validity,
patentability, enforceability and/or non-infringement of any of the
PATENT CLAIMS or otherwise opposing any of the PATENT CLAIMS.
Tokutek hereby grants to you, for the term and geographical scope of
the PATENT CLAIMS, a non-exclusive, no-charge, royalty-free,
irrevocable (except as stated in this section) patent license to
make, have made, use, offer to sell, sell, import, transfer, and
otherwise run, modify, and propagate the contents of THIS
IMPLEMENTATION, where such license applies only to the PATENT
CLAIMS. This grant does not include claims that would be infringed
only as a consequence of further modifications of THIS
IMPLEMENTATION. If you or your agent or licensee institute or order
or agree to the institution of patent litigation against any entity
(including a cross-claim or counterclaim in a lawsuit) alleging that
THIS IMPLEMENTATION constitutes direct or contributory patent
infringement, or inducement of patent infringement, then any rights
granted to you under this License shall terminate as of the date
such litigation is filed. If you or your agent or exclusive
licensee institute or order or agree to the institution of a PATENT
CHALLENGE, then Tokutek may terminate any rights granted to you
under this License.
*/
#ident "Copyright (c) 2007-2013 Tokutek Inc. All rights reserved."
// Purpose of this file is to handle all modifications and queries to the database
// at the level of leafentry.
//
// ule = Unpacked Leaf Entry
//
// This design unpacks the leafentry into a convenient format, performs all work
// on the unpacked form, then repacks the leafentry into its compact format.
//
// See design documentation for nested transactions at
// TokuWiki/Imp/TransactionsOverview.
#include "portability/toku_portability.h"
#include "ft/ft-internal.h"
#include "ft/leafentry.h"
#include "ft/logger/logger.h"
#include "ft/msg.h"
#include "ft/txn/txn.h"
#include "ft/txn/txn_manager.h"
#include "ft/ule.h"
#include "ft/ule-internal.h"
#include "ft/txn/xids.h"
#include "util/bytestring.h"
#include "util/omt.h"
#include "util/partitioned_counter.h"
#include "util/scoped_malloc.h"
#include "util/status.h"
#define ULE_DEBUG 0
static uint32_t ule_get_innermost_numbytes(ULE ule, uint32_t keylen);
///////////////////////////////////////////////////////////////////////////////////
// Engine status
//
// Status is intended for display to humans to help understand system behavior.
// It does not need to be perfectly thread-safe.
static LE_STATUS_S le_status;
#define STATUS_INIT(k,c,t,l,inc) TOKUFT_STATUS_INIT(le_status, k, c, t, "le: " l, inc)
void toku_ule_status_init(void) {
// Note, this function initializes the keyname, type, and legend fields.
// Value fields are initialized to zero by compiler.
STATUS_INIT(LE_MAX_COMMITTED_XR, nullptr, UINT64, "max committed xr", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_MAX_PROVISIONAL_XR, nullptr, UINT64, "max provisional xr", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_EXPANDED, nullptr, UINT64, "expanded", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_MAX_MEMSIZE, nullptr, UINT64, "max memsize", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_APPLY_GC_BYTES_IN, nullptr, PARCOUNT, "size of leafentries before garbage collection (during message application)", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_APPLY_GC_BYTES_OUT, nullptr, PARCOUNT, "size of leafentries after garbage collection (during message application)", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_NORMAL_GC_BYTES_IN, nullptr, PARCOUNT, "size of leafentries before garbage collection (outside message application)", TOKU_ENGINE_STATUS);
STATUS_INIT(LE_NORMAL_GC_BYTES_OUT,nullptr, PARCOUNT, "size of leafentries after garbage collection (outside message application)", TOKU_ENGINE_STATUS);
le_status.initialized = true;
}
#undef STATUS_INIT
void toku_ule_status_destroy(void) {
for (int i = 0; i < LE_STATUS_NUM_ROWS; ++i) {
if (le_status.status[i].type == PARCOUNT) {
destroy_partitioned_counter(le_status.status[i].value.parcount);
}
}
}
void toku_le_get_status(LE_STATUS statp) {
*statp = le_status;
}
#define STATUS_VALUE(x) le_status.status[x].value.num
#define STATUS_INC(x, d) \
do { \
if (le_status.status[x].type == PARCOUNT) { \
increment_partitioned_counter(le_status.status[x].value.parcount, d); \
} else { \
toku_sync_fetch_and_add(&le_status.status[x].value.num, d); \
} \
} while (0)
///////////////////////////////////////////////////////////////////////////////////
// Accessor functions used by outside world (e.g. indexer)
//
ULEHANDLE
toku_ule_create(LEAFENTRY le) {
ULE XMALLOC(ule_p);
le_unpack(ule_p, le);
return (ULEHANDLE) ule_p;
}
void toku_ule_free(ULEHANDLE ule_p) {
ule_cleanup((ULE) ule_p);
toku_free(ule_p);
}
///////////////////////////////////////////////////////////////////////////////////
//
// Question: Can any software outside this file modify or read a leafentry?
// If so, is it worthwhile to put it all here?
//
// There are two entries, one each for modification and query:
// toku_le_apply_msg() performs all inserts/deletes/aborts
//
//
//
//
//This is what we use to initialize Xuxrs[0] in a new unpacked leafentry.
const UXR_S committed_delete = {
.type = XR_DELETE,
.vallen = 0,
.valp = NULL,
.xid = 0
}; // static allocation of uxr with type set to committed delete and xid = 0
#define INSERT_LENGTH(len) ((1U << 31) | len)
#define DELETE_LENGTH(len) (0)
#define GET_LENGTH(len) (len & ((1U << 31)-1))
#define IS_INSERT(len) (len & (1U << 31))
#define IS_VALID_LEN(len) (len < (1U<<31))
// Local functions:
static void msg_init_empty_ule(ULE ule);
static void msg_modify_ule(ULE ule, const ft_msg &msg);
static void ule_init_empty_ule(ULE ule);
static void ule_do_implicit_promotions(ULE ule, XIDS xids);
static void ule_try_promote_provisional_outermost(ULE ule, TXNID oldest_possible_live_xid);
static void ule_promote_provisional_innermost_to_index(ULE ule, uint32_t index);
static void ule_promote_provisional_innermost_to_committed(ULE ule);
static void ule_apply_insert(ULE ule, XIDS xids, uint32_t vallen, void * valp);
static void ule_apply_delete(ULE ule, XIDS xids);
static void ule_prepare_for_new_uxr(ULE ule, XIDS xids);
static void ule_apply_abort(ULE ule, XIDS xids);
static void ule_apply_broadcast_commit_all (ULE ule);
static void ule_apply_commit(ULE ule, XIDS xids);
static void ule_push_insert_uxr(ULE ule, bool is_committed, TXNID xid, uint32_t vallen, void * valp);
static void ule_push_delete_uxr(ULE ule, bool is_committed, TXNID xid);
static void ule_push_placeholder_uxr(ULE ule, TXNID xid);
static UXR ule_get_innermost_uxr(ULE ule);
static UXR ule_get_first_empty_uxr(ULE ule);
static void ule_remove_innermost_uxr(ULE ule);
static TXNID ule_get_innermost_xid(ULE ule);
static TXNID ule_get_xid(ULE ule, uint32_t index);
static void ule_remove_innermost_placeholders(ULE ule);
static void ule_add_placeholders(ULE ule, XIDS xids);
static void ule_optimize(ULE ule, XIDS xids);
static inline bool uxr_type_is_insert(uint8_t type);
static inline bool uxr_type_is_delete(uint8_t type);
static inline bool uxr_type_is_placeholder(uint8_t type);
static inline size_t uxr_pack_txnid(UXR uxr, uint8_t *p);
static inline size_t uxr_pack_type_and_length(UXR uxr, uint8_t *p);
static inline size_t uxr_pack_length_and_bit(UXR uxr, uint8_t *p);
static inline size_t uxr_pack_data(UXR uxr, uint8_t *p);
static inline size_t uxr_unpack_txnid(UXR uxr, uint8_t *p);
static inline size_t uxr_unpack_type_and_length(UXR uxr, uint8_t *p);
static inline size_t uxr_unpack_length_and_bit(UXR uxr, uint8_t *p);
static inline size_t uxr_unpack_data(UXR uxr, uint8_t *p);
static void get_space_for_le(
bn_data* data_buffer,
uint32_t idx,
void* keyp,
uint32_t keylen,
uint32_t old_keylen,
uint32_t old_le_size,
size_t size,
LEAFENTRY* new_le_space,
void **const maybe_free
)
{
if (data_buffer == nullptr) {
CAST_FROM_VOIDP(*new_le_space, toku_xmalloc(size));
}
else {
// this means we are overwriting something
if (old_le_size > 0) {
data_buffer->get_space_for_overwrite(idx, keyp, keylen, old_keylen, old_le_size, size, new_le_space, maybe_free);
}
// this means we are inserting something new
else {
data_buffer->get_space_for_insert(idx, keyp, keylen, size, new_le_space, maybe_free);
}
}
}
/////////////////////////////////////////////////////////////////////
// Garbage collection related functions
//
static TXNID
get_next_older_txnid(TXNID xc, const xid_omt_t &omt) {
int r;
TXNID xid;
r = omt.find<TXNID, toku_find_xid_by_xid>(xc, -1, &xid, nullptr);
if (r==0) {
invariant(xid < xc); //sanity check
}
else {
invariant(r==DB_NOTFOUND);
xid = TXNID_NONE;
}
return xid;
}
//
// This function returns true if live transaction TL1 is allowed to read a value committed by
// transaction xc, false otherwise.
//
static bool
xid_reads_committed_xid(TXNID tl1, TXNID xc, const xid_omt_t &snapshot_txnids, const rx_omt_t &referenced_xids) {
bool rval;
if (tl1 < xc) rval = false; //cannot read a newer txn
else {
TXNID x = toku_get_youngest_live_list_txnid_for(xc, snapshot_txnids, referenced_xids);
if (x == TXNID_NONE) rval = true; //Not in ANY live list, tl1 can read it.
else rval = tl1 > x; //Newer than the 'newest one that has it in live list'
// we know tl1 > xc
// we know x > xc
// if tl1 == x, then we do not read, because tl1 is in xc's live list
// if x is older than tl1, that means that xc < x < tl1
// and if xc is in x's live list, it CANNOT be in tl1's live list
}
return rval;
}
//
// This function does some simple garbage collection given a TXNID known
// to be the oldest referenced xid, that is, the oldest xid in any live list.
// We find the youngest entry in the stack with an xid less
// than oldest_referenced_xid. All elements below this entry are garbage,
// so we get rid of them.
//
static void
ule_simple_garbage_collection(ULE ule, txn_gc_info *gc_info) {
if (ule->num_cuxrs == 1) {
return;
}
uint32_t curr_index = 0;
if (gc_info->mvcc_needed) {
// starting at the top of the committed stack, find the first
// uxr with a txnid that is less than oldest_referenced_xid
for (uint32_t i = 0; i < ule->num_cuxrs; i++) {
curr_index = ule->num_cuxrs - i - 1;
if (ule->uxrs[curr_index].xid < gc_info->oldest_referenced_xid_for_simple_gc) {
break;
}
}
} else {
// if mvcc is not needed, we can need the top committed
// value and nothing else
curr_index = ule->num_cuxrs - 1;
}
// curr_index is now set to the youngest uxr older than oldest_referenced_xid
// so if it's not the bottom of the stack..
if (curr_index != 0) {
// ..then we need to get rid of the entries below curr_index
uint32_t num_entries = ule->num_cuxrs + ule->num_puxrs - curr_index;
memmove(&ule->uxrs[0], &ule->uxrs[curr_index], num_entries * sizeof(ule->uxrs[0]));
ule->uxrs[0].xid = TXNID_NONE; // New 'bottom of stack' loses its TXNID
ule->num_cuxrs -= curr_index;
}
}
// TODO: Clean this up
extern bool garbage_collection_debug;
static void
ule_garbage_collect(ULE ule, const xid_omt_t &snapshot_xids, const rx_omt_t &referenced_xids, const xid_omt_t &live_root_txns) {
if (ule->num_cuxrs == 1) {
return;
}
toku::scoped_calloc necessary_buf(ule->num_cuxrs * sizeof(bool));
bool *necessary = reinterpret_cast<bool *>(necessary_buf.get());
uint32_t curr_committed_entry;
curr_committed_entry = ule->num_cuxrs - 1;
while (true) {
// mark the curr_committed_entry as necessary
necessary[curr_committed_entry] = true;
if (curr_committed_entry == 0) break; //nothing left
// find the youngest live transaction that reads something
// below curr_committed_entry, if it exists
TXNID tl1;
TXNID xc = ule->uxrs[curr_committed_entry].xid;
//
// If we find that the committed transaction is in the live list,
// then xc is really in the process of being committed. It has not
// been fully committed. As a result, our assumption that transactions
// newer than what is currently in these OMTs will read the top of the stack
// is not necessarily accurate. Transactions may read what is just below xc.
// As a result, we must mark what is just below xc as necessary and move on.
// This issue was found while testing flusher threads, and was fixed for #3979
//
bool is_xc_live = toku_is_txn_in_live_root_txn_list(live_root_txns, xc);
if (is_xc_live) {
curr_committed_entry--;
continue;
}
tl1 = toku_get_youngest_live_list_txnid_for(xc, snapshot_xids, referenced_xids);
// if tl1 == xc, that means xc should be live and show up in live_root_txns, which we check above.
invariant(tl1 != xc);
if (tl1 == TXNID_NONE) {
// set tl1 to youngest live transaction older than ule->uxrs[curr_committed_entry]->xid
tl1 = get_next_older_txnid(xc, snapshot_xids);
if (tl1 == TXNID_NONE) {
// remainder is garbage, we're done
break;
}
}
if (garbage_collection_debug) {
int r = snapshot_xids.find_zero<TXNID, toku_find_xid_by_xid>(tl1, nullptr, nullptr);
invariant_zero(r); // make sure that the txn you are claiming is live is actually live
}
//
// tl1 should now be set
//
curr_committed_entry--;
while (curr_committed_entry > 0) {
xc = ule->uxrs[curr_committed_entry].xid;
if (xid_reads_committed_xid(tl1, xc, snapshot_xids, referenced_xids)) {
break;
}
curr_committed_entry--;
}
}
uint32_t first_free = 0;
for (uint32_t i = 0; i < ule->num_cuxrs; i++) {
// Shift values to 'delete' garbage values.
if (necessary[i]) {
ule->uxrs[first_free] = ule->uxrs[i];
first_free++;
}
}
uint32_t saved = first_free;
invariant(saved <= ule->num_cuxrs);
invariant(saved >= 1);
ule->uxrs[0].xid = TXNID_NONE; //New 'bottom of stack' loses its TXNID
if (first_free != ule->num_cuxrs) {
// Shift provisional values
memmove(&ule->uxrs[first_free], &ule->uxrs[ule->num_cuxrs], ule->num_puxrs * sizeof(ule->uxrs[0]));
}
ule->num_cuxrs = saved;
}
static size_t ule_packed_memsize(ULE ule) {
// Returns: The size 'ule' would be when packed into a leafentry, or 0 if the
// topmost committed value is a delete.
if (ule->num_cuxrs == 1 && ule->num_puxrs == 0) {
UXR uxr = ule_get_innermost_uxr(ule);
if (uxr_is_delete(uxr)) {
return 0;
}
}
return le_memsize_from_ule(ule);
}
// Heuristics to control when we decide to initialize
// txn manager state (possibly expensive) and run gc.
enum {
ULE_MIN_STACK_SIZE_TO_FORCE_GC = 5,
ULE_MIN_MEMSIZE_TO_FORCE_GC = 1024 * 1024
};
/////////////////////////////////////////////////////////////////////////////////
// This is the big enchilada. (Bring Tums.) Note that this level of abstraction
// has no knowledge of the inner structure of either leafentry or msg. It makes
// calls into the next lower layer (msg_xxx) which handles messages.
//
// NOTE: This is the only function (at least in this body of code) that modifies
// a leafentry.
// NOTE: It is the responsibility of the caller to make sure that the key is set
// in the FT_MSG, as it will be used to store the data in the data_buffer
//
// Return 0 on success.
// If the leafentry is destroyed it sets *new_leafentry_p to NULL.
// Otehrwise the new_leafentry_p points at the new leaf entry.
// As of October 2011, this function always returns 0.
void
toku_le_apply_msg(const ft_msg &msg,
LEAFENTRY old_leafentry, // NULL if there was no stored data.
bn_data* data_buffer, // bn_data storing leafentry, if NULL, means there is no bn_data
uint32_t idx, // index in data_buffer where leafentry is stored (and should be replaced
uint32_t old_keylen, // length of the any key in data_buffer
txn_gc_info *gc_info,
LEAFENTRY *new_leafentry_p,
int64_t * numbytes_delta_p) { // change in total size of key and val, not including any overhead
invariant_notnull(gc_info);
paranoid_invariant_notnull(new_leafentry_p);
ULE_S ule;
int64_t oldnumbytes = 0;
int64_t newnumbytes = 0;
uint64_t oldmemsize = 0;
uint32_t keylen = msg.kdbt()->size;
if (old_leafentry == NULL) {
msg_init_empty_ule(&ule);
} else {
oldmemsize = leafentry_memsize(old_leafentry);
le_unpack(&ule, old_leafentry); // otherwise unpack leafentry
oldnumbytes = ule_get_innermost_numbytes(&ule, keylen);
}
msg_modify_ule(&ule, msg); // modify unpacked leafentry
// - we may be able to immediately promote the newly-apllied outermost provisonal uxr
// - either way, run simple gc first, and then full gc if there are still some committed uxrs.
ule_try_promote_provisional_outermost(&ule, gc_info->oldest_referenced_xid_for_implicit_promotion);
ule_simple_garbage_collection(&ule, gc_info);
txn_manager_state *txn_state_for_gc = gc_info->txn_state_for_gc;
size_t size_before_gc = 0;
if (ule.num_cuxrs > 1 && txn_state_for_gc != nullptr && // there is garbage to clean, and our caller gave us state..
// ..and either the state is pre-initialized, or the committed stack is large enough
(txn_state_for_gc->initialized || ule.num_cuxrs >= ULE_MIN_STACK_SIZE_TO_FORCE_GC ||
// ..or the ule's raw memsize is sufficiently large
(size_before_gc = ule_packed_memsize(&ule)) >= ULE_MIN_MEMSIZE_TO_FORCE_GC)) {
// ..then it's worth running gc, possibly initializing the txn manager state, if it isn't already
if (!txn_state_for_gc->initialized) {
txn_state_for_gc->init();
}
size_before_gc = size_before_gc != 0 ? size_before_gc : // it's already been calculated above
ule_packed_memsize(&ule);
ule_garbage_collect(&ule,
txn_state_for_gc->snapshot_xids,
txn_state_for_gc->referenced_xids,
txn_state_for_gc->live_root_txns
);
size_t size_after_gc = ule_packed_memsize(&ule);
STATUS_INC(LE_APPLY_GC_BYTES_IN, size_before_gc);
STATUS_INC(LE_APPLY_GC_BYTES_OUT, size_after_gc);
}
void *maybe_free = nullptr;
int r = le_pack(
&ule, // create packed leafentry
data_buffer,
idx,
msg.kdbt()->data, // contract of this function is caller has this set, always
keylen, // contract of this function is caller has this set, always
old_keylen,
oldmemsize,
new_leafentry_p,
&maybe_free
);
invariant_zero(r);
if (*new_leafentry_p) {
newnumbytes = ule_get_innermost_numbytes(&ule, keylen);
}
*numbytes_delta_p = newnumbytes - oldnumbytes;
ule_cleanup(&ule);
if (maybe_free != nullptr) {
toku_free(maybe_free);
}
}
bool toku_le_worth_running_garbage_collection(LEAFENTRY le, txn_gc_info *gc_info) {
// Effect: Quickly determines if it's worth trying to run garbage collection on a leafentry
// Return: True if it makes sense to try garbage collection, false otherwise.
// Rationale: Garbage collection is likely to clean up under two circumstances:
// 1.) There are multiple committed entries. Some may never be read by new txns.
// 2.) There is only one committed entry, but the outermost provisional entry
// is older than the oldest known referenced xid, so it must have commited.
// Therefor we can promote it to committed and get rid of the old commited entry.
if (le->type != LE_MVCC) {
return false;
}
if (le->u.mvcc.num_cxrs > 1) {
return true;
} else {
paranoid_invariant(le->u.mvcc.num_cxrs == 1);
}
return le->u.mvcc.num_pxrs > 0 &&
le_outermost_uncommitted_xid(le) < gc_info->oldest_referenced_xid_for_implicit_promotion;
}
// Garbage collect one leaf entry, using the given OMT's.
// Parameters:
// -- old_leaf_entry : the leaf we intend to clean up through garbage
// collecting.
// -- new_leaf_entry (OUTPUT) : a pointer to the leaf entry after
// garbage collection.
// -- new_leaf_entry_memory_size : after this call, our leaf entry
// should be empty or smaller. This number represents that and is
// used in a previous call to truncate the existing size.
// -- omt : the memory where our leaf entry resides.
// -- mp : our memory pool.
// -- maybe_free (OUTPUT) : in a previous call, we may be able to free
// the memory completely, if we removed the leaf entry.
// -- snapshot_xids : we use these in memory transaction ids to
// determine what to garbage collect.
// -- referenced_xids : list of in memory active transactions.
// NOTE: it is not a good idea to garbage collect a leaf
// entry with only one committed value.
void
toku_le_garbage_collect(LEAFENTRY old_leaf_entry,
bn_data* data_buffer,
uint32_t idx,
void* keyp,
uint32_t keylen,
txn_gc_info *gc_info,
LEAFENTRY *new_leaf_entry,
int64_t * numbytes_delta_p) {
// We shouldn't want to run gc without having provided a snapshot of the txn system.
invariant_notnull(gc_info);
invariant_notnull(gc_info->txn_state_for_gc);
paranoid_invariant_notnull(new_leaf_entry);
ULE_S ule;
int64_t oldnumbytes = 0;
int64_t newnumbytes = 0;
le_unpack(&ule, old_leaf_entry);
oldnumbytes = ule_get_innermost_numbytes(&ule, keylen);
uint32_t old_mem_size = leafentry_memsize(old_leaf_entry);
// Before running garbage collection, try to promote the outermost provisional
// entries to committed if its xid is older than the oldest possible live xid.
//
// The oldest known refeferenced xid is a lower bound on the oldest possible
// live xid, so we use that. It's usually close enough to get rid of most
// garbage in leafentries.
ule_try_promote_provisional_outermost(&ule, gc_info->oldest_referenced_xid_for_implicit_promotion);
// No need to run simple gc here if we're going straight for full gc.
if (ule.num_cuxrs > 1) {
size_t size_before_gc = ule_packed_memsize(&ule);
ule_garbage_collect(&ule,
gc_info->txn_state_for_gc->snapshot_xids,
gc_info->txn_state_for_gc->referenced_xids,
gc_info->txn_state_for_gc->live_root_txns);
size_t size_after_gc = ule_packed_memsize(&ule);
STATUS_INC(LE_APPLY_GC_BYTES_IN, size_before_gc);
STATUS_INC(LE_APPLY_GC_BYTES_OUT, size_after_gc);
}
void *maybe_free = nullptr;
int r = le_pack(
&ule,
data_buffer,
idx,
keyp,
keylen,
keylen, // old_keylen, same because the key isn't going to change for gc
old_mem_size,
new_leaf_entry,
&maybe_free
);
invariant_zero(r);
if (*new_leaf_entry) {
newnumbytes = ule_get_innermost_numbytes(&ule, keylen);
}
*numbytes_delta_p = newnumbytes - oldnumbytes;
ule_cleanup(&ule);
if (maybe_free != nullptr) {
toku_free(maybe_free);
}
}
/////////////////////////////////////////////////////////////////////////////////
// This layer of abstraction (msg_xxx)
// knows the accessors of msg, but not of leafentry or unpacked leaf entry.
// It makes calls into the lower layer (le_xxx) which handles leafentries.
// Purpose is to init the ule with given key and no transaction records
//
static void
msg_init_empty_ule(ULE ule) {
ule_init_empty_ule(ule);
}
static bool do_implicit_promotion(enum ft_msg_type type, XIDS xids, ULE ule) {
if (type == FT_OPTIMIZE || type == FT_OPTIMIZE_FOR_UPGRADE) {
return false;
}
// as part of FT-603, commit messages may now hit a leafentry
// after the relevant lock tree locks are released. That is, other messages
// may hit the leafentry before this commit message, and as a result,
// implicit promotion done by that previous message will have done the
// the necessary work.
//
// So, we don't want to blindly implicitly promote a possibly live transaction
// because a commit message is hitting this leafentry late.
// Take the following example:
// - Transaction A does a provisional write.
// - Transaction A commits, but the commit message
// is not yet sent.
// - Transaction B does a write, which causes the leafentry to
// commit A via implicit promotion. Now B is on top of the stack
// with a provisional entry
// - Transaction A's commit message hits this leafentry. We don't
// want to use implicit promotion to commit B. That would be a bug.
// Therefore, if we have different transaction stacks in the ule and xids,
// we don't want implicit promotion.
//
// On the other hand, if the root TXNID of the ule's provisional stack
// matches that of the xids, then we want to run implicit promotion,
// because ule_apply_commit depends on it. We don't need to worry about
// messages coming out of order, because transactions are associated
// with a single thread, and a new child cannot begin work until a previous
// child has finished committing.
if (type == FT_COMMIT_ANY || type == FT_COMMIT_BROADCAST_TXN) {
TXNID current_msg_xid = toku_xids_get_xid(xids, 0);
invariant(current_msg_xid != TXNID_NONE);
TXNID current_ule_xid = ule_get_xid(ule, 0);
if (current_ule_xid != current_msg_xid) {
return false;
}
}
return true;
}
// Purpose is to modify the unpacked leafentry in our private workspace.
//
static void
msg_modify_ule(ULE ule, const ft_msg &msg) {
XIDS xids = msg.xids();
invariant(toku_xids_get_num_xids(xids) < MAX_TRANSACTION_RECORDS);
enum ft_msg_type type = msg.type();
if (do_implicit_promotion(type, xids, ule)) {
ule_do_implicit_promotions(ule, xids);
}
switch (type) {
case FT_INSERT_NO_OVERWRITE: {
UXR old_innermost_uxr = ule_get_innermost_uxr(ule);
//If something exists, quit (no overwrite).
if (uxr_is_insert(old_innermost_uxr)) break;
//else it is just an insert, so
//fall through to FT_INSERT on purpose.
}
case FT_INSERT: {
uint32_t vallen = msg.vdbt()->size;
invariant(IS_VALID_LEN(vallen));
void * valp = msg.vdbt()->data;
ule_apply_insert(ule, xids, vallen, valp);
break;
}
case FT_DELETE_ANY:
ule_apply_delete(ule, xids);
break;
case FT_ABORT_ANY:
case FT_ABORT_BROADCAST_TXN:
ule_apply_abort(ule, xids);
break;
case FT_COMMIT_BROADCAST_ALL:
ule_apply_broadcast_commit_all(ule);
break;
case FT_COMMIT_ANY:
case FT_COMMIT_BROADCAST_TXN:
ule_apply_commit(ule, xids);
break;
case FT_OPTIMIZE:
case FT_OPTIMIZE_FOR_UPGRADE:
ule_optimize(ule, xids);
break;
case FT_UPDATE:
case FT_UPDATE_BROADCAST_ALL:
assert(false); // These messages don't get this far. Instead they get translated (in setval_fun in do_update) into FT_INSERT messages.
break;
default:
assert(false); /* illegal ft msg type */
break;
}
}
void test_msg_modify_ule(ULE ule, const ft_msg &msg){
msg_modify_ule(ule,msg);
}
static void ule_optimize(ULE ule, XIDS xids) {
if (ule->num_puxrs) {
TXNID uncommitted = ule->uxrs[ule->num_cuxrs].xid; // outermost uncommitted
TXNID oldest_living_xid = TXNID_NONE;
uint32_t num_xids = toku_xids_get_num_xids(xids);
if (num_xids > 0) {
invariant(num_xids==1);
oldest_living_xid = toku_xids_get_xid(xids, 0);
}
if (oldest_living_xid == TXNID_NONE || uncommitted < oldest_living_xid) {
ule_promote_provisional_innermost_to_committed(ule);
}
}
}
/////////////////////////////////////////////////////////////////////////////////
// This layer of abstraction (le_xxx) understands the structure of the leafentry
// and of the unpacked leafentry. It is the only layer that understands the
// structure of leafentry. It has no knowledge of any other data structures.
//
//
// required for every le_unpack that is done
//
void
ule_cleanup(ULE ule) {
invariant(ule->uxrs);
if (ule->uxrs != ule->uxrs_static) {
toku_free(ule->uxrs);
ule->uxrs = NULL;
}
}
// populate an unpacked leafentry using pointers into the given leafentry.
// thus, the memory referenced by 'le' must live as long as the ULE.
void
le_unpack(ULE ule, LEAFENTRY le) {
uint8_t type = le->type;
uint8_t *p;
uint32_t i;
switch (type) {
case LE_CLEAN: {
ule->uxrs = ule->uxrs_static; //Static version is always enough.
ule->num_cuxrs = 1;
ule->num_puxrs = 0;
UXR uxr = ule->uxrs;
uxr->type = XR_INSERT;
uxr->vallen = toku_dtoh32(le->u.clean.vallen);
uxr->valp = le->u.clean.val;
uxr->xid = TXNID_NONE;
//Set p to immediately after leafentry
p = le->u.clean.val + uxr->vallen;
break;
}
case LE_MVCC:
ule->num_cuxrs = toku_dtoh32(le->u.mvcc.num_cxrs);
invariant(ule->num_cuxrs);
ule->num_puxrs = le->u.mvcc.num_pxrs;
//Dynamic memory
if (ule->num_cuxrs < MAX_TRANSACTION_RECORDS) {
ule->uxrs = ule->uxrs_static;
}
else {
XMALLOC_N(ule->num_cuxrs + 1 + MAX_TRANSACTION_RECORDS, ule->uxrs);
}
p = le->u.mvcc.xrs;
//unpack interesting TXNIDs inner to outer.
if (ule->num_puxrs!=0) {
UXR outermost = ule->uxrs + ule->num_cuxrs;
p += uxr_unpack_txnid(outermost, p);
}
//unpack other TXNIDS (not for ule->uxrs[0])
ule->uxrs[0].xid = TXNID_NONE; //0 for super-root is implicit
for (i = 0; i < ule->num_cuxrs - 1; i++) {
p += uxr_unpack_txnid(ule->uxrs + ule->num_cuxrs - 1 - i, p);
}
//unpack interesting lengths inner to outer.
if (ule->num_puxrs!=0) {
UXR innermost = ule->uxrs + ule->num_cuxrs + ule->num_puxrs - 1;
p += uxr_unpack_length_and_bit(innermost, p);
}
for (i = 0; i < ule->num_cuxrs; i++) {
p += uxr_unpack_length_and_bit(ule->uxrs + ule->num_cuxrs - 1 - i, p);
}
//unpack interesting values inner to outer
if (ule->num_puxrs!=0) {
UXR innermost = ule->uxrs + ule->num_cuxrs + ule->num_puxrs - 1;
p += uxr_unpack_data(innermost, p);
}
for (i = 0; i < ule->num_cuxrs; i++) {
p += uxr_unpack_data(ule->uxrs + ule->num_cuxrs - 1 - i, p);
}
//unpack provisional xrs outer to inner
if (ule->num_puxrs > 1) {
{
//unpack length, bit, data for outermost uncommitted
UXR outermost = ule->uxrs + ule->num_cuxrs;
p += uxr_unpack_type_and_length(outermost, p);
p += uxr_unpack_data(outermost, p);
}
//unpack txnid, length, bit, data for non-outermost, non-innermost
for (i = ule->num_cuxrs + 1; i < ule->num_cuxrs + ule->num_puxrs - 1; i++) {
UXR uxr = ule->uxrs + i;
p += uxr_unpack_txnid(uxr, p);
p += uxr_unpack_type_and_length(uxr, p);
p += uxr_unpack_data(uxr, p);
}
{
//Just unpack txnid for innermost
UXR innermost = ule->uxrs + ule->num_cuxrs + ule->num_puxrs - 1;
p += uxr_unpack_txnid(innermost, p);
}
}
break;
default:
invariant(false);
}
#if ULE_DEBUG
size_t memsize = le_memsize_from_ule(ule);
assert(p == ((uint8_t*)le) + memsize);
#endif
}
static inline size_t
uxr_pack_txnid(UXR uxr, uint8_t *p) {
*(TXNID*)p = toku_htod64(uxr->xid);
return sizeof(TXNID);
}
static inline size_t
uxr_pack_type_and_length(UXR uxr, uint8_t *p) {
size_t rval = 1;
*p = uxr->type;
if (uxr_is_insert(uxr)) {
*(uint32_t*)(p+1) = toku_htod32(uxr->vallen);
rval += sizeof(uint32_t);
}
return rval;
}
static inline size_t
uxr_pack_length_and_bit(UXR uxr, uint8_t *p) {
uint32_t length_and_bit;
if (uxr_is_insert(uxr)) {
length_and_bit = INSERT_LENGTH(uxr->vallen);
}
else {
length_and_bit = DELETE_LENGTH(uxr->vallen);
}
*(uint32_t*)p = toku_htod32(length_and_bit);
return sizeof(uint32_t);
}
static inline size_t
uxr_pack_data(UXR uxr, uint8_t *p) {
if (uxr_is_insert(uxr)) {
memcpy(p, uxr->valp, uxr->vallen);
return uxr->vallen;
}
return 0;
}
static inline size_t
uxr_unpack_txnid(UXR uxr, uint8_t *p) {
uxr->xid = toku_dtoh64(*(TXNID*)p);
return sizeof(TXNID);
}
static inline size_t
uxr_unpack_type_and_length(UXR uxr, uint8_t *p) {
size_t rval = 1;
uxr->type = *p;
if (uxr_is_insert(uxr)) {
uxr->vallen = toku_dtoh32(*(uint32_t*)(p+1));
rval += sizeof(uint32_t);
}
return rval;
}
static inline size_t
uxr_unpack_length_and_bit(UXR uxr, uint8_t *p) {
uint32_t length_and_bit = toku_dtoh32(*(uint32_t*)p);
if (IS_INSERT(length_and_bit)) {
uxr->type = XR_INSERT;
uxr->vallen = GET_LENGTH(length_and_bit);
}
else {
uxr->type = XR_DELETE;
uxr->vallen = 0;
}
return sizeof(uint32_t);
}
static inline size_t
uxr_unpack_data(UXR uxr, uint8_t *p) {
if (uxr_is_insert(uxr)) {
uxr->valp = p;
return uxr->vallen;
}
return 0;
}
// executed too often to be worth making threadsafe
static inline void
update_le_status(ULE ule, size_t memsize) {
if (ule->num_cuxrs > STATUS_VALUE(LE_MAX_COMMITTED_XR))
STATUS_VALUE(LE_MAX_COMMITTED_XR) = ule->num_cuxrs;
if (ule->num_puxrs > STATUS_VALUE(LE_MAX_PROVISIONAL_XR))
STATUS_VALUE(LE_MAX_PROVISIONAL_XR) = ule->num_puxrs;
if (ule->num_cuxrs > MAX_TRANSACTION_RECORDS)
STATUS_VALUE(LE_EXPANDED)++;
if (memsize > STATUS_VALUE(LE_MAX_MEMSIZE))
STATUS_VALUE(LE_MAX_MEMSIZE) = memsize;
}
// Purpose is to return a newly allocated leaf entry in packed format, or
// return null if leaf entry should be destroyed (if no transaction records
// are for inserts).
// Transaction records in packed le are stored inner to outer (first xr is innermost),
// with some information extracted out of the transaction records into the header.
// Transaction records in ule are stored outer to inner (uxr[0] is outermost).