You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

938 lines
33 KiB

  1. /* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
  2. // vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
  3. #ident "$Id$"
  4. #ident "Copyright (c) 2007-2012 Tokutek Inc. All rights reserved."
  5. #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."
  6. #include "ft.h"
  7. #include "ft-internal.h"
  8. #include "ft-cachetable-wrappers.h"
  9. #include "log-internal.h"
  10. #include <ft/log_header.h>
  11. #include <memory.h>
  12. #include <toku_assert.h>
  13. #include <portability/toku_atomic.h>
  14. void
  15. toku_ft_suppress_rollbacks(FT h, TOKUTXN txn) {
  16. TXNID txnid = toku_txn_get_txnid(txn);
  17. assert(h->txnid_that_created_or_locked_when_empty == TXNID_NONE ||
  18. h->txnid_that_created_or_locked_when_empty == txnid);
  19. h->txnid_that_created_or_locked_when_empty = txnid;
  20. }
  21. void
  22. toku_reset_root_xid_that_created(FT ft, TXNID new_root_xid_that_created) {
  23. // Reset the root_xid_that_created field to the given value.
  24. // This redefines which xid created the dictionary.
  25. // hold lock around setting and clearing of dirty bit
  26. // (see cooperative use of dirty bit in ft_begin_checkpoint())
  27. toku_ft_lock (ft);
  28. ft->h->root_xid_that_created = new_root_xid_that_created;
  29. ft->h->dirty = 1;
  30. toku_ft_unlock (ft);
  31. }
  32. static void
  33. ft_destroy(FT ft) {
  34. //header and checkpoint_header have same Blocktable pointer
  35. //cannot destroy since it is still in use by CURRENT
  36. assert(ft->h->type == FT_CURRENT);
  37. toku_blocktable_destroy(&ft->blocktable);
  38. if (ft->descriptor.dbt.data) toku_free(ft->descriptor.dbt.data);
  39. if (ft->cmp_descriptor.dbt.data) toku_free(ft->cmp_descriptor.dbt.data);
  40. toku_ft_destroy_reflock(ft);
  41. toku_free(ft->h);
  42. }
  43. // Make a copy of the header for the purpose of a checkpoint
  44. // Not reentrant for a single FT.
  45. // See ft_checkpoint for explanation of why
  46. // FT lock must be held.
  47. static void
  48. ft_copy_for_checkpoint_unlocked(FT ft, LSN checkpoint_lsn) {
  49. assert(ft->h->type == FT_CURRENT);
  50. assert(ft->checkpoint_header == NULL);
  51. FT_HEADER XMEMDUP(ch, ft->h);
  52. ch->type = FT_CHECKPOINT_INPROGRESS; //Different type
  53. //printf("checkpoint_lsn=%" PRIu64 "\n", checkpoint_lsn.lsn);
  54. ch->checkpoint_lsn = checkpoint_lsn;
  55. //ch->blocktable is SHARED between the two headers
  56. ft->checkpoint_header = ch;
  57. }
  58. void
  59. toku_ft_free (FT ft) {
  60. ft_destroy(ft);
  61. toku_free(ft);
  62. }
  63. void
  64. toku_ft_init_reflock(FT ft) {
  65. toku_mutex_init(&ft->ft_ref_lock, NULL);
  66. }
  67. void
  68. toku_ft_destroy_reflock(FT ft) {
  69. toku_mutex_destroy(&ft->ft_ref_lock);
  70. }
  71. void
  72. toku_ft_grab_reflock(FT ft) {
  73. toku_mutex_lock(&ft->ft_ref_lock);
  74. }
  75. void
  76. toku_ft_release_reflock(FT ft) {
  77. toku_mutex_unlock(&ft->ft_ref_lock);
  78. }
  79. /////////////////////////////////////////////////////////////////////////
  80. // Start of Functions that are callbacks to the cachefule
  81. //
  82. // maps to cf->log_fassociate_during_checkpoint
  83. static void
  84. ft_log_fassociate_during_checkpoint (CACHEFILE cf, void *header_v) {
  85. FT ft = (FT) header_v;
  86. char* fname_in_env = toku_cachefile_fname_in_env(cf);
  87. BYTESTRING bs = { .len = (uint32_t) strlen(fname_in_env), // don't include the NUL
  88. .data = fname_in_env };
  89. TOKULOGGER logger = toku_cachefile_logger(cf);
  90. FILENUM filenum = toku_cachefile_filenum(cf);
  91. bool unlink_on_close = toku_cachefile_is_unlink_on_close(cf);
  92. toku_log_fassociate(logger, NULL, 0, filenum, ft->h->flags, bs, unlink_on_close);
  93. }
  94. // maps to cf->log_suppress_rollback_during_checkpoint
  95. static void
  96. ft_log_suppress_rollback_during_checkpoint (CACHEFILE cf, void *header_v) {
  97. FT h = (FT) header_v;
  98. TXNID xid = h->txnid_that_created_or_locked_when_empty;
  99. if (xid != TXNID_NONE) {
  100. //Only log if useful.
  101. TOKULOGGER logger = toku_cachefile_logger(cf);
  102. FILENUM filenum = toku_cachefile_filenum (cf);
  103. // We don't have access to the txn here, but the txn is
  104. // necessarily already marked as non-readonly. Use NULL.
  105. TOKUTXN txn = NULL;
  106. toku_log_suppress_rollback(logger, NULL, 0, txn, filenum, xid);
  107. }
  108. }
  109. // Maps to cf->begin_checkpoint_userdata
  110. // Create checkpoint-in-progress versions of header and translation (btt) (and fifo for now...).
  111. // Has access to fd (it is protected).
  112. //
  113. // Not reentrant for a single FT (see ft_checkpoint)
  114. static void ft_begin_checkpoint (LSN checkpoint_lsn, void *header_v) {
  115. FT ft = (FT) header_v;
  116. // hold lock around copying and clearing of dirty bit
  117. toku_ft_lock (ft);
  118. assert(ft->h->type == FT_CURRENT);
  119. assert(ft->checkpoint_header == NULL);
  120. ft_copy_for_checkpoint_unlocked(ft, checkpoint_lsn);
  121. ft->h->dirty = 0; // this is only place this bit is cleared (in currentheader)
  122. toku_block_translation_note_start_checkpoint_unlocked(ft->blocktable);
  123. toku_ft_unlock (ft);
  124. }
  125. // #4922: Hack to remove data corruption race condition.
  126. // Reading (and upgrading) a node up to version 19 causes this.
  127. // We COULD skip this if we know that no nodes remained (as of last checkpoint)
  128. // that are below version 19.
  129. // If there are no nodes < version 19 this is harmless (field is unused).
  130. // If there are, this will make certain the value is at least as low as necessary,
  131. // and not much lower. (Too low is good, too high can cause data corruption).
  132. // TODO(yoni): If we ever stop supporting upgrades of nodes < version 19 we can delete this.
  133. // TODO(yoni): If we know no nodes are left to upgrade, we can skip this. (Probably not worth doing).
  134. static void
  135. ft_hack_highest_unused_msn_for_upgrade_for_checkpoint(FT ft) {
  136. if (ft->h->layout_version_original < FT_LAYOUT_VERSION_19) {
  137. ft->checkpoint_header->highest_unused_msn_for_upgrade = ft->h->highest_unused_msn_for_upgrade;
  138. }
  139. }
  140. // maps to cf->checkpoint_userdata
  141. // Write checkpoint-in-progress versions of header and translation to disk (really to OS internal buffer).
  142. // Copy current header's version of checkpoint_staging stat64info to checkpoint header.
  143. // Must have access to fd (protected).
  144. // Requires: all pending bits are clear. This implies that no thread will modify the checkpoint_staging
  145. // version of the stat64info.
  146. //
  147. // No locks are taken for checkpoint_count/lsn because this is single threaded. Can be called by:
  148. // - ft_close
  149. // - end_checkpoint
  150. // checkpoints hold references to FTs and so they cannot be closed during a checkpoint.
  151. // ft_close is not reentrant for a single FT
  152. // end_checkpoint is not reentrant period
  153. static void ft_checkpoint (CACHEFILE cf, int fd, void *header_v) {
  154. FT ft = (FT) header_v;
  155. FT_HEADER ch = ft->checkpoint_header;
  156. //printf("%s:%d allocated_limit=%lu writing queue to %lu\n", __FILE__, __LINE__,
  157. // block_allocator_allocated_limit(h->block_allocator), h->unused_blocks.b*h->nodesize);
  158. assert(ch);
  159. assert(ch->type == FT_CHECKPOINT_INPROGRESS);
  160. if (ch->dirty) { // this is only place this bit is tested (in checkpoint_header)
  161. TOKULOGGER logger = toku_cachefile_logger(cf);
  162. if (logger) {
  163. toku_logger_fsync_if_lsn_not_fsynced(logger, ch->checkpoint_lsn);
  164. }
  165. uint64_t now = (uint64_t) time(NULL);
  166. ft->h->time_of_last_modification = now;
  167. ch->time_of_last_modification = now;
  168. ch->checkpoint_count++;
  169. ft_hack_highest_unused_msn_for_upgrade_for_checkpoint(ft);
  170. // write translation and header to disk (or at least to OS internal buffer)
  171. toku_serialize_ft_to(fd, ch, ft->blocktable, ft->cf);
  172. ch->dirty = 0; // this is only place this bit is cleared (in checkpoint_header)
  173. // fsync the cachefile
  174. toku_cachefile_fsync(cf);
  175. ft->h->checkpoint_count++; // checkpoint succeeded, next checkpoint will save to alternate header location
  176. ft->h->checkpoint_lsn = ch->checkpoint_lsn; //Header updated.
  177. }
  178. else {
  179. toku_block_translation_note_skipped_checkpoint(ft->blocktable);
  180. }
  181. }
  182. // maps to cf->end_checkpoint_userdata
  183. // free unused disk space
  184. // (i.e. tell BlockAllocator to liberate blocks used by previous checkpoint).
  185. // Must have access to fd (protected)
  186. static void ft_end_checkpoint (CACHEFILE UU(cachefile), int fd, void *header_v) {
  187. FT ft = (FT) header_v;
  188. assert(ft->h->type == FT_CURRENT);
  189. toku_block_translation_note_end_checkpoint(ft->blocktable, fd);
  190. if (ft->checkpoint_header) {
  191. toku_free(ft->checkpoint_header);
  192. ft->checkpoint_header = NULL;
  193. }
  194. }
  195. // maps to cf->close_userdata
  196. // Has access to fd (it is protected).
  197. static void ft_close(CACHEFILE cachefile, int fd, void *header_v, bool oplsn_valid, LSN oplsn) {
  198. FT ft = (FT) header_v;
  199. assert(ft->h->type == FT_CURRENT);
  200. // We already have exclusive access to this field already, so skip the locking.
  201. // This should already never fail.
  202. invariant(!toku_ft_needed_unlocked(ft));
  203. assert(ft->cf == cachefile);
  204. TOKULOGGER logger = toku_cachefile_logger(cachefile);
  205. LSN lsn = ZERO_LSN;
  206. //Get LSN
  207. if (oplsn_valid) {
  208. //Use recovery-specified lsn
  209. lsn = oplsn;
  210. //Recovery cannot reduce lsn of a header.
  211. if (lsn.lsn < ft->h->checkpoint_lsn.lsn) {
  212. lsn = ft->h->checkpoint_lsn;
  213. }
  214. }
  215. else {
  216. //Get LSN from logger
  217. lsn = ZERO_LSN; // if there is no logger, we use zero for the lsn
  218. if (logger) {
  219. char* fname_in_env = toku_cachefile_fname_in_env(cachefile);
  220. assert(fname_in_env);
  221. BYTESTRING bs = {.len=(uint32_t) strlen(fname_in_env), .data=fname_in_env};
  222. toku_log_fclose(logger, &lsn, ft->h->dirty, bs, toku_cachefile_filenum(cachefile)); // flush the log on close (if new header is being written), otherwise it might not make it out.
  223. }
  224. }
  225. if (ft->h->dirty) { // this is the only place this bit is tested (in currentheader)
  226. if (logger) { //Rollback cachefile MUST NOT BE CLOSED DIRTY
  227. //It can be checkpointed only via 'checkpoint'
  228. assert(logger->rollback_cachefile != cachefile);
  229. }
  230. ft_begin_checkpoint(lsn, header_v);
  231. ft_checkpoint(cachefile, fd, ft);
  232. ft_end_checkpoint(cachefile, fd, header_v);
  233. assert(!ft->h->dirty); // dirty bit should be cleared by begin_checkpoint and never set again (because we're closing the dictionary)
  234. }
  235. toku_ft_free(ft);
  236. }
  237. // maps to cf->note_pin_by_checkpoint
  238. //Must be protected by ydb lock.
  239. //Is only called by checkpoint begin, which holds it
  240. static void ft_note_pin_by_checkpoint (CACHEFILE UU(cachefile), void *header_v) {
  241. // Note: open_close lock is held by checkpoint begin
  242. FT ft = (FT) header_v;
  243. toku_ft_grab_reflock(ft);
  244. assert(!ft->pinned_by_checkpoint);
  245. assert(toku_ft_needed_unlocked(ft));
  246. ft->pinned_by_checkpoint = true;
  247. toku_ft_release_reflock(ft);
  248. }
  249. // Requires: the reflock is held.
  250. static void unpin_by_checkpoint_callback(FT ft, void *extra) {
  251. invariant(extra == NULL);
  252. invariant(ft->pinned_by_checkpoint);
  253. ft->pinned_by_checkpoint = false;
  254. }
  255. // maps to cf->note_unpin_by_checkpoint
  256. //Must be protected by ydb lock.
  257. //Called by end_checkpoint, which grabs ydb lock around note_unpin
  258. static void ft_note_unpin_by_checkpoint (CACHEFILE UU(cachefile), void *header_v) {
  259. FT ft = (FT) header_v;
  260. toku_ft_remove_reference(ft, false, ZERO_LSN, unpin_by_checkpoint_callback, NULL);
  261. }
  262. //
  263. // End of Functions that are callbacks to the cachefile
  264. /////////////////////////////////////////////////////////////////////////
  265. void toku_node_save_ct_pair(CACHEKEY UU(key), void *value_data, PAIR p) {
  266. FTNODE CAST_FROM_VOIDP(node, value_data);
  267. node->ct_pair = p;
  268. }
  269. static void setup_initial_ft_root_node(FT ft, BLOCKNUM blocknum) {
  270. FTNODE XCALLOC(node);
  271. toku_initialize_empty_ftnode(node, blocknum, 0, 1, ft->h->layout_version, ft->h->flags);
  272. BP_STATE(node,0) = PT_AVAIL;
  273. uint32_t fullhash = toku_cachetable_hash(ft->cf, blocknum);
  274. node->fullhash = fullhash;
  275. toku_cachetable_put(ft->cf, blocknum, fullhash,
  276. node, make_ftnode_pair_attr(node),
  277. get_write_callbacks_for_node(ft),
  278. toku_node_save_ct_pair);
  279. toku_unpin_ftnode(ft, node);
  280. }
  281. static void ft_init(FT ft, FT_OPTIONS options, CACHEFILE cf) {
  282. // fake, prevent unnecessary upgrade logic
  283. ft->layout_version_read_from_disk = FT_LAYOUT_VERSION;
  284. ft->checkpoint_header = NULL;
  285. toku_list_init(&ft->live_ft_handles);
  286. ft->compare_fun = options->compare_fun;
  287. ft->update_fun = options->update_fun;
  288. if (ft->cf != NULL) {
  289. assert(ft->cf == cf);
  290. }
  291. ft->cf = cf;
  292. ft->in_memory_stats = ZEROSTATS;
  293. setup_initial_ft_root_node(ft, ft->h->root_blocknum);
  294. toku_cachefile_set_userdata(ft->cf,
  295. ft,
  296. ft_log_fassociate_during_checkpoint,
  297. ft_log_suppress_rollback_during_checkpoint,
  298. ft_close,
  299. ft_checkpoint,
  300. ft_begin_checkpoint,
  301. ft_end_checkpoint,
  302. ft_note_pin_by_checkpoint,
  303. ft_note_unpin_by_checkpoint);
  304. toku_block_verify_no_free_blocknums(ft->blocktable);
  305. }
  306. static FT_HEADER
  307. ft_header_create(FT_OPTIONS options, BLOCKNUM root_blocknum, TXNID root_xid_that_created)
  308. {
  309. uint64_t now = (uint64_t) time(NULL);
  310. struct ft_header h = {
  311. .type = FT_CURRENT,
  312. .dirty = 0,
  313. .checkpoint_count = 0,
  314. .checkpoint_lsn = ZERO_LSN,
  315. .layout_version = FT_LAYOUT_VERSION,
  316. .layout_version_original = FT_LAYOUT_VERSION,
  317. .build_id = BUILD_ID,
  318. .build_id_original = BUILD_ID,
  319. .time_of_creation = now,
  320. .root_xid_that_created = root_xid_that_created,
  321. .time_of_last_modification = now,
  322. .time_of_last_verification = 0,
  323. .root_blocknum = root_blocknum,
  324. .flags = options->flags,
  325. .nodesize = options->nodesize,
  326. .basementnodesize = options->basementnodesize,
  327. .compression_method = options->compression_method,
  328. .highest_unused_msn_for_upgrade = { .msn = (MIN_MSN.msn - 1) },
  329. .max_msn_in_ft = ZERO_MSN,
  330. .time_of_last_optimize_begin = 0,
  331. .time_of_last_optimize_end = 0,
  332. .count_of_optimize_in_progress = 0,
  333. .count_of_optimize_in_progress_read_from_disk = 0,
  334. .msn_at_start_of_last_completed_optimize = ZERO_MSN,
  335. .on_disk_stats = ZEROSTATS
  336. };
  337. return (FT_HEADER) toku_xmemdup(&h, sizeof h);
  338. }
  339. // allocate and initialize a fractal tree.
  340. void toku_ft_create(FT *ftp, FT_OPTIONS options, CACHEFILE cf, TOKUTXN txn) {
  341. invariant(ftp);
  342. FT XCALLOC(ft);
  343. memset(&ft->descriptor, 0, sizeof(ft->descriptor));
  344. memset(&ft->cmp_descriptor, 0, sizeof(ft->cmp_descriptor));
  345. ft->h = ft_header_create(options, make_blocknum(0), (txn ? txn->ancestor_txnid64 : TXNID_NONE));
  346. toku_ft_init_reflock(ft);
  347. // Assign blocknum for root block, also dirty the header
  348. toku_blocktable_create_new(&ft->blocktable);
  349. toku_allocate_blocknum(ft->blocktable, &ft->h->root_blocknum, ft);
  350. ft_init(ft, options, cf);
  351. *ftp = ft;
  352. }
  353. // TODO: (Zardosht) get rid of brt parameter
  354. int toku_read_ft_and_store_in_cachefile (FT_HANDLE brt, CACHEFILE cf, LSN max_acceptable_lsn, FT *header, bool* was_open)
  355. // If the cachefile already has the header, then just get it.
  356. // If the cachefile has not been initialized, then don't modify anything.
  357. // max_acceptable_lsn is the latest acceptable checkpointed version of the file.
  358. {
  359. {
  360. FT h;
  361. if ((h = (FT) toku_cachefile_get_userdata(cf))!=0) {
  362. *header = h;
  363. *was_open = true;
  364. assert(brt->options.update_fun == h->update_fun);
  365. assert(brt->options.compare_fun == h->compare_fun);
  366. return 0;
  367. }
  368. }
  369. *was_open = false;
  370. FT h;
  371. int r;
  372. {
  373. int fd = toku_cachefile_get_fd(cf);
  374. r = toku_deserialize_ft_from(fd, max_acceptable_lsn, &h);
  375. if (r == TOKUDB_BAD_CHECKSUM) {
  376. fprintf(stderr, "Checksum failure while reading header in file %s.\n", toku_cachefile_fname_in_env(cf));
  377. assert(false); // make absolutely sure we crash before doing anything else
  378. }
  379. }
  380. if (r!=0) return r;
  381. h->cf = cf;
  382. h->compare_fun = brt->options.compare_fun;
  383. h->update_fun = brt->options.update_fun;
  384. toku_cachefile_set_userdata(cf,
  385. (void*)h,
  386. ft_log_fassociate_during_checkpoint,
  387. ft_log_suppress_rollback_during_checkpoint,
  388. ft_close,
  389. ft_checkpoint,
  390. ft_begin_checkpoint,
  391. ft_end_checkpoint,
  392. ft_note_pin_by_checkpoint,
  393. ft_note_unpin_by_checkpoint);
  394. *header = h;
  395. return 0;
  396. }
  397. void
  398. toku_ft_note_ft_handle_open(FT ft, FT_HANDLE live) {
  399. toku_ft_grab_reflock(ft);
  400. live->ft = ft;
  401. toku_list_push(&ft->live_ft_handles, &live->live_ft_handle_link);
  402. toku_ft_release_reflock(ft);
  403. }
  404. // the reference count for a ft is the number of txn's that
  405. // touched it plus the number of open handles plus one if
  406. // pinned by a checkpoint.
  407. static int
  408. ft_get_reference_count(FT ft) {
  409. uint32_t pinned_by_checkpoint = ft->pinned_by_checkpoint ? 1 : 0;
  410. int num_handles = toku_list_num_elements_est(&ft->live_ft_handles);
  411. return pinned_by_checkpoint + ft->num_txns + num_handles;
  412. }
  413. // a ft is needed in memory iff its reference count is non-zero
  414. bool
  415. toku_ft_needed_unlocked(FT ft) {
  416. return ft_get_reference_count(ft) != 0;
  417. }
  418. // get the reference count and return true if it was 1
  419. bool
  420. toku_ft_has_one_reference_unlocked(FT ft) {
  421. return ft_get_reference_count(ft) == 1;
  422. }
  423. // evict a ft from memory by closing its cachefile. any future work
  424. // will have to read in the ft in a new cachefile and new FT object.
  425. void toku_ft_evict_from_memory(FT ft, bool oplsn_valid, LSN oplsn) {
  426. assert(ft->cf);
  427. toku_cachefile_close(&ft->cf, oplsn_valid, oplsn);
  428. }
  429. // Verifies there exists exactly one ft handle and returns it.
  430. FT_HANDLE toku_ft_get_only_existing_ft_handle(FT h) {
  431. FT_HANDLE ft_handle_ret = NULL;
  432. toku_ft_grab_reflock(h);
  433. assert(toku_list_num_elements_est(&h->live_ft_handles) == 1);
  434. ft_handle_ret = toku_list_struct(toku_list_head(&h->live_ft_handles), struct ft_handle, live_ft_handle_link);
  435. toku_ft_release_reflock(h);
  436. return ft_handle_ret;
  437. }
  438. // Purpose: set fields in brt_header to capture accountability info for start of HOT optimize.
  439. // Note: HOT accountability variables in header are modified only while holding header lock.
  440. // (Header lock is really needed for touching the dirty bit, but it's useful and
  441. // convenient here for keeping the HOT variables threadsafe.)
  442. void
  443. toku_ft_note_hot_begin(FT_HANDLE brt) {
  444. FT ft = brt->ft;
  445. time_t now = time(NULL);
  446. // hold lock around setting and clearing of dirty bit
  447. // (see cooperative use of dirty bit in ft_begin_checkpoint())
  448. toku_ft_lock(ft);
  449. ft->h->time_of_last_optimize_begin = now;
  450. ft->h->count_of_optimize_in_progress++;
  451. ft->h->dirty = 1;
  452. toku_ft_unlock(ft);
  453. }
  454. // Purpose: set fields in brt_header to capture accountability info for end of HOT optimize.
  455. // Note: See note for toku_ft_note_hot_begin().
  456. void
  457. toku_ft_note_hot_complete(FT_HANDLE brt, bool success, MSN msn_at_start_of_hot) {
  458. FT ft = brt->ft;
  459. time_t now = time(NULL);
  460. toku_ft_lock(ft);
  461. ft->h->count_of_optimize_in_progress--;
  462. if (success) {
  463. ft->h->time_of_last_optimize_end = now;
  464. ft->h->msn_at_start_of_last_completed_optimize = msn_at_start_of_hot;
  465. // If we just successfully completed an optimization and no other thread is performing
  466. // an optimization, then the number of optimizations in progress is zero.
  467. // If there was a crash during a HOT optimization, this is how count_of_optimize_in_progress
  468. // would be reset to zero on the disk after recovery from that crash.
  469. if (ft->h->count_of_optimize_in_progress == ft->h->count_of_optimize_in_progress_read_from_disk)
  470. ft->h->count_of_optimize_in_progress = 0;
  471. }
  472. ft->h->dirty = 1;
  473. toku_ft_unlock(ft);
  474. }
  475. void
  476. toku_ft_init(FT ft,
  477. BLOCKNUM root_blocknum_on_disk,
  478. LSN checkpoint_lsn,
  479. TXNID root_xid_that_created,
  480. uint32_t target_nodesize,
  481. uint32_t target_basementnodesize,
  482. enum toku_compression_method compression_method)
  483. {
  484. memset(ft, 0, sizeof *ft);
  485. struct ft_options options = {
  486. .nodesize = target_nodesize,
  487. .basementnodesize = target_basementnodesize,
  488. .compression_method = compression_method,
  489. .flags = 0,
  490. .compare_fun = NULL,
  491. .update_fun = NULL
  492. };
  493. ft->h = ft_header_create(&options, root_blocknum_on_disk, root_xid_that_created);
  494. ft->h->checkpoint_count = 1;
  495. ft->h->checkpoint_lsn = checkpoint_lsn;
  496. }
  497. // Open a brt for use by redirect. The new brt must have the same dict_id as the old_ft passed in. (FILENUM is assigned by the ft_handle_open() function.)
  498. static int
  499. ft_handle_open_for_redirect(FT_HANDLE *new_ftp, const char *fname_in_env, TOKUTXN txn, FT old_h) {
  500. FT_HANDLE t;
  501. assert(old_h->dict_id.dictid != DICTIONARY_ID_NONE.dictid);
  502. toku_ft_handle_create(&t);
  503. toku_ft_set_bt_compare(t, old_h->compare_fun);
  504. toku_ft_set_update(t, old_h->update_fun);
  505. toku_ft_handle_set_nodesize(t, old_h->h->nodesize);
  506. toku_ft_handle_set_basementnodesize(t, old_h->h->basementnodesize);
  507. toku_ft_handle_set_compression_method(t, old_h->h->compression_method);
  508. CACHETABLE ct = toku_cachefile_get_cachetable(old_h->cf);
  509. int r = toku_ft_handle_open_with_dict_id(t, fname_in_env, 0, 0, ct, txn, old_h->dict_id);
  510. assert_zero(r);
  511. assert(t->ft->dict_id.dictid == old_h->dict_id.dictid);
  512. *new_ftp = t;
  513. return r;
  514. }
  515. // This function performs most of the work to redirect a dictionary to different file.
  516. // It is called for redirect and to abort a redirect. (This function is almost its own inverse.)
  517. static int
  518. dictionary_redirect_internal(const char *dst_fname_in_env, FT src_h, TOKUTXN txn, FT *dst_hp) {
  519. int r;
  520. FILENUM src_filenum = toku_cachefile_filenum(src_h->cf);
  521. FILENUM dst_filenum = FILENUM_NONE;
  522. FT dst_h = NULL;
  523. struct toku_list *list;
  524. // open a dummy brt based off of
  525. // dst_fname_in_env to get the header
  526. // then we will change all the brt's to have
  527. // their headers point to dst_h instead of src_h
  528. FT_HANDLE tmp_dst_ft = NULL;
  529. r = ft_handle_open_for_redirect(&tmp_dst_ft, dst_fname_in_env, txn, src_h);
  530. assert_zero(r);
  531. dst_h = tmp_dst_ft->ft;
  532. // some sanity checks on dst_filenum
  533. dst_filenum = toku_cachefile_filenum(dst_h->cf);
  534. assert(dst_filenum.fileid!=FILENUM_NONE.fileid);
  535. assert(dst_filenum.fileid!=src_filenum.fileid); //Cannot be same file.
  536. // for each live brt, brt->ft is currently src_h
  537. // we want to change it to dummy_dst
  538. toku_ft_grab_reflock(src_h);
  539. while (!toku_list_empty(&src_h->live_ft_handles)) {
  540. list = src_h->live_ft_handles.next;
  541. FT_HANDLE src_handle = NULL;
  542. src_handle = toku_list_struct(list, struct ft_handle, live_ft_handle_link);
  543. toku_list_remove(&src_handle->live_ft_handle_link);
  544. toku_ft_note_ft_handle_open(dst_h, src_handle);
  545. if (src_handle->redirect_callback) {
  546. src_handle->redirect_callback(src_handle, src_handle->redirect_callback_extra);
  547. }
  548. }
  549. assert(dst_h);
  550. // making sure that we are not leaking src_h
  551. assert(toku_ft_needed_unlocked(src_h));
  552. toku_ft_release_reflock(src_h);
  553. toku_ft_handle_close(tmp_dst_ft);
  554. *dst_hp = dst_h;
  555. return r;
  556. }
  557. //This is the 'abort redirect' function. The redirect of old_h to new_h was done
  558. //and now must be undone, so here we redirect new_h back to old_h.
  559. int
  560. toku_dictionary_redirect_abort(FT old_h, FT new_h, TOKUTXN txn) {
  561. char *old_fname_in_env = toku_cachefile_fname_in_env(old_h->cf);
  562. int r;
  563. {
  564. FILENUM old_filenum = toku_cachefile_filenum(old_h->cf);
  565. FILENUM new_filenum = toku_cachefile_filenum(new_h->cf);
  566. assert(old_filenum.fileid!=new_filenum.fileid); //Cannot be same file.
  567. //No living brts in old header.
  568. toku_ft_grab_reflock(old_h);
  569. assert(toku_list_empty(&old_h->live_ft_handles));
  570. toku_ft_release_reflock(old_h);
  571. }
  572. FT dst_h;
  573. // redirect back from new_h to old_h
  574. r = dictionary_redirect_internal(old_fname_in_env, new_h, txn, &dst_h);
  575. assert_zero(r);
  576. assert(dst_h == old_h);
  577. return r;
  578. }
  579. /****
  580. * on redirect or abort:
  581. * if redirect txn_note_doing_work(txn)
  582. * if redirect connect src brt to txn (txn modified this brt)
  583. * for each src brt
  584. * open brt to dst file (create new brt struct)
  585. * if redirect connect dst brt to txn
  586. * redirect db to new brt
  587. * redirect cursors to new brt
  588. * close all src brts
  589. * if redirect make rollback log entry
  590. *
  591. * on commit:
  592. * nothing to do
  593. *
  594. *****/
  595. int
  596. toku_dictionary_redirect (const char *dst_fname_in_env, FT_HANDLE old_ft_h, TOKUTXN txn) {
  597. // Input args:
  598. // new file name for dictionary (relative to env)
  599. // old_ft_h is a live brt of open handle ({DB, BRT} pair) that currently refers to old dictionary file.
  600. // (old_ft_h may be one of many handles to the dictionary.)
  601. // txn that created the loader
  602. // Requires:
  603. // multi operation lock is held.
  604. // The brt is open. (which implies there can be no zombies.)
  605. // The new file must be a valid dictionary.
  606. // The block size and flags in the new file must match the existing BRT.
  607. // The new file must already have its descriptor in it (and it must match the existing descriptor).
  608. // Effect:
  609. // Open new FTs (and related header and cachefile) to the new dictionary file with a new FILENUM.
  610. // Redirect all DBs that point to brts that point to the old file to point to brts that point to the new file.
  611. // Copy the dictionary id (dict_id) from the header of the original file to the header of the new file.
  612. // Create a rollback log entry.
  613. // The original BRT, header, cachefile and file remain unchanged. They will be cleaned up on commmit.
  614. // If the txn aborts, then this operation will be undone
  615. int r;
  616. FT old_ft = old_ft_h->ft;
  617. // dst file should not be open. (implies that dst and src are different because src must be open.)
  618. {
  619. CACHETABLE ct = toku_cachefile_get_cachetable(old_ft->cf);
  620. CACHEFILE cf;
  621. r = toku_cachefile_of_iname_in_env(ct, dst_fname_in_env, &cf);
  622. if (r==0) {
  623. r = EINVAL;
  624. goto cleanup;
  625. }
  626. assert(r==ENOENT);
  627. r = 0;
  628. }
  629. if (txn) {
  630. toku_txn_maybe_note_ft(txn, old_ft); // mark old ft as touched by this txn
  631. }
  632. FT new_ft;
  633. r = dictionary_redirect_internal(dst_fname_in_env, old_ft, txn, &new_ft);
  634. assert_zero(r);
  635. // make rollback log entry
  636. if (txn) {
  637. toku_txn_maybe_note_ft(txn, new_ft); // mark new ft as touched by this txn
  638. FILENUM old_filenum = toku_cachefile_filenum(old_ft->cf);
  639. FILENUM new_filenum = toku_cachefile_filenum(new_ft->cf);
  640. toku_logger_save_rollback_dictionary_redirect(txn, old_filenum, new_filenum);
  641. TXNID xid = toku_txn_get_txnid(txn);
  642. toku_ft_suppress_rollbacks(new_ft, txn);
  643. toku_log_suppress_rollback(txn->logger, NULL, 0, txn, new_filenum, xid);
  644. }
  645. cleanup:
  646. return r;
  647. }
  648. // Insert reference to transaction into ft
  649. void
  650. toku_ft_add_txn_ref(FT ft) {
  651. toku_ft_grab_reflock(ft);
  652. ++ft->num_txns;
  653. toku_ft_release_reflock(ft);
  654. }
  655. static void
  656. remove_txn_ref_callback(FT ft, void *UU(context)) {
  657. invariant(ft->num_txns > 0);
  658. --ft->num_txns;
  659. }
  660. void
  661. toku_ft_remove_txn_ref(FT ft) {
  662. toku_ft_remove_reference(ft, false, ZERO_LSN, remove_txn_ref_callback, NULL);
  663. }
  664. void toku_calculate_root_offset_pointer (
  665. FT ft,
  666. CACHEKEY* root_key,
  667. uint32_t *roothash
  668. )
  669. {
  670. *roothash = toku_cachetable_hash(ft->cf, ft->h->root_blocknum);
  671. *root_key = ft->h->root_blocknum;
  672. }
  673. void toku_ft_set_new_root_blocknum(
  674. FT ft,
  675. CACHEKEY new_root_key
  676. )
  677. {
  678. ft->h->root_blocknum = new_root_key;
  679. }
  680. LSN toku_ft_checkpoint_lsn(FT ft) {
  681. return ft->h->checkpoint_lsn;
  682. }
  683. void
  684. toku_ft_stat64 (FT ft, struct ftstat64_s *s) {
  685. s->fsize = toku_cachefile_size(ft->cf);
  686. // just use the in memory stats from the header
  687. // prevent appearance of negative numbers for numrows, numbytes
  688. int64_t n = ft->in_memory_stats.numrows;
  689. if (n < 0) {
  690. n = 0;
  691. }
  692. s->nkeys = s->ndata = n;
  693. n = ft->in_memory_stats.numbytes;
  694. if (n < 0) {
  695. n = 0;
  696. }
  697. s->dsize = n;
  698. s->create_time_sec = ft->h->time_of_creation;
  699. s->modify_time_sec = ft->h->time_of_last_modification;
  700. s->verify_time_sec = ft->h->time_of_last_verification;
  701. }
  702. void
  703. toku_ft_update_descriptor(FT ft, DESCRIPTOR d)
  704. // Effect: Changes the descriptor in a tree (log the change, make sure it makes it to disk eventually).
  705. // requires: the ft is fully user-opened with a valid cachefile.
  706. // descriptor updates cannot happen in parallel for an FT
  707. // (ydb layer uses a row lock to enforce this)
  708. {
  709. assert(ft->cf);
  710. int fd = toku_cachefile_get_fd(ft->cf);
  711. toku_ft_update_descriptor_with_fd(ft, d, fd);
  712. }
  713. // upadate the descriptor for an ft and serialize it using
  714. // the given descriptor instead of reading the descriptor
  715. // from the ft's cachefile. we do this so serialize code can
  716. // update a descriptor before the ft is fully opened and has
  717. // a valid cachefile.
  718. void
  719. toku_ft_update_descriptor_with_fd(FT ft, DESCRIPTOR d, int fd) {
  720. // the checksum is four bytes, so that's where the magic number comes from
  721. // make space for the new descriptor and write it out to disk
  722. DISKOFF offset, size;
  723. size = toku_serialize_descriptor_size(d) + 4;
  724. toku_realloc_descriptor_on_disk(ft->blocktable, size, &offset, ft, fd);
  725. toku_serialize_descriptor_contents_to_fd(fd, d, offset);
  726. // cleanup the old descriptor and set the in-memory descriptor to the new one
  727. if (ft->descriptor.dbt.data) {
  728. toku_free(ft->descriptor.dbt.data);
  729. }
  730. ft->descriptor.dbt.size = d->dbt.size;
  731. ft->descriptor.dbt.data = toku_memdup(d->dbt.data, d->dbt.size);
  732. }
  733. void
  734. toku_ft_update_cmp_descriptor(FT ft) {
  735. if (ft->cmp_descriptor.dbt.data != NULL) {
  736. toku_free(ft->cmp_descriptor.dbt.data);
  737. }
  738. ft->cmp_descriptor.dbt.size = ft->descriptor.dbt.size;
  739. ft->cmp_descriptor.dbt.data = toku_xmemdup(
  740. ft->descriptor.dbt.data,
  741. ft->descriptor.dbt.size
  742. );
  743. }
  744. DESCRIPTOR
  745. toku_ft_get_descriptor(FT_HANDLE ft_handle) {
  746. return &ft_handle->ft->descriptor;
  747. }
  748. DESCRIPTOR
  749. toku_ft_get_cmp_descriptor(FT_HANDLE ft_handle) {
  750. return &ft_handle->ft->cmp_descriptor;
  751. }
  752. void
  753. toku_ft_update_stats(STAT64INFO headerstats, STAT64INFO_S delta) {
  754. (void) toku_sync_fetch_and_add(&(headerstats->numrows), delta.numrows);
  755. (void) toku_sync_fetch_and_add(&(headerstats->numbytes), delta.numbytes);
  756. }
  757. void
  758. toku_ft_decrease_stats(STAT64INFO headerstats, STAT64INFO_S delta) {
  759. (void) toku_sync_fetch_and_sub(&(headerstats->numrows), delta.numrows);
  760. (void) toku_sync_fetch_and_sub(&(headerstats->numbytes), delta.numbytes);
  761. }
  762. void
  763. toku_ft_remove_reference(FT ft, bool oplsn_valid, LSN oplsn, remove_ft_ref_callback remove_ref, void *extra) {
  764. toku_ft_grab_reflock(ft);
  765. if (toku_ft_has_one_reference_unlocked(ft)) {
  766. toku_ft_release_reflock(ft);
  767. toku_ft_open_close_lock();
  768. toku_ft_grab_reflock(ft);
  769. remove_ref(ft, extra);
  770. bool needed = toku_ft_needed_unlocked(ft);
  771. toku_ft_release_reflock(ft);
  772. // if we're running during recovery, we must close the underlying ft.
  773. // we know we're running in recovery if we were passed a valid lsn.
  774. if (oplsn_valid) {
  775. assert(!needed);
  776. }
  777. if (!needed) {
  778. // close header
  779. toku_ft_evict_from_memory(ft, oplsn_valid, oplsn);
  780. }
  781. toku_ft_open_close_unlock();
  782. }
  783. else {
  784. remove_ref(ft, extra);
  785. toku_ft_release_reflock(ft);
  786. }
  787. }
  788. void toku_ft_set_nodesize(FT ft, unsigned int nodesize) {
  789. toku_ft_lock(ft);
  790. ft->h->nodesize = nodesize;
  791. ft->h->dirty = 1;
  792. toku_ft_unlock(ft);
  793. }
  794. void toku_ft_get_nodesize(FT ft, unsigned int *nodesize) {
  795. toku_ft_lock(ft);
  796. *nodesize = ft->h->nodesize;
  797. toku_ft_unlock(ft);
  798. }
  799. void toku_ft_set_basementnodesize(FT ft, unsigned int basementnodesize) {
  800. toku_ft_lock(ft);
  801. ft->h->basementnodesize = basementnodesize;
  802. ft->h->dirty = 1;
  803. toku_ft_unlock(ft);
  804. }
  805. void toku_ft_get_basementnodesize(FT ft, unsigned int *basementnodesize) {
  806. toku_ft_lock(ft);
  807. *basementnodesize = ft->h->basementnodesize;
  808. toku_ft_unlock(ft);
  809. }
  810. void toku_ft_set_compression_method(FT ft, enum toku_compression_method method) {
  811. toku_ft_lock(ft);
  812. ft->h->compression_method = method;
  813. ft->h->dirty = 1;
  814. toku_ft_unlock(ft);
  815. }
  816. void toku_ft_get_compression_method(FT ft, enum toku_compression_method *methodp) {
  817. toku_ft_lock(ft);
  818. *methodp = ft->h->compression_method;
  819. toku_ft_unlock(ft);
  820. }
  821. // mark the ft as a blackhole. any message injections will be a no op.
  822. void toku_ft_set_blackhole(FT_HANDLE ft_handle) {
  823. ft_handle->ft->blackhole = true;
  824. }