1 /***************************************************************************** 2 3 Copyright (c) 2018, 2020, Oracle and/or its affiliates. All Rights Reserved. 4 5 This program is free software; you can redistribute it and/or modify it under 6 the terms of the GNU General Public License, version 2.0, as published by the 7 Free Software Foundation. 8 9 This program is also distributed with certain software (including but not 10 limited to OpenSSL) that is licensed under separate terms, as designated in a 11 particular file or component or in included license documentation. The authors 12 of MySQL hereby grant you an additional permission to link the program and 13 your derivative works with the separately licensed software that they have 14 included with MySQL. 15 16 This program is distributed in the hope that it will be useful, but WITHOUT 17 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS 18 FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0, 19 for more details. 20 21 You should have received a copy of the GNU General Public License along with 22 this program; if not, write to the Free Software Foundation, Inc., 23 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA 24 25 *****************************************************************************/ 26 27 /** @file include/clone0repl.h 28 GTID persistence interface 29 30 *******************************************************/ 31 32 #ifndef CLONE_REPL_INCLUDE 33 #define CLONE_REPL_INCLUDE 34 35 #include <vector> 36 #include "clone0monitor.h" 37 #include "os0thread-create.h" 38 #include "sql/rpl_gtid.h" 39 #include "srv0srv.h" 40 #include "srv0start.h" 41 #include "trx0sys.h" 42 43 class Clone_persist_gtid; 44 45 /** Serialized GTID information size */ 46 static const size_t GTID_INFO_SIZE = 64; 47 48 /** GTID format version. */ 49 static const uint32_t GTID_VERSION = 1; 50 51 /** Serialized GTID */ 52 using Gtid_info = std::array<unsigned char, GTID_INFO_SIZE>; 53 54 /** List of GTIDs */ 55 using Gitd_info_list = std::vector<Gtid_info>; 56 57 /** GTID descriptor with version information. */ 58 struct Gtid_desc { 59 /** If GTID descriptor is set. */ 60 bool m_is_set; 61 /** Serialized GTID information. */ 62 Gtid_info m_info; 63 /* GTID version. */ 64 uint32_t m_version; 65 }; 66 67 /** Persist GTID along with transaction commit */ 68 class Clone_persist_gtid { 69 public: 70 /** Constructor: start gtid thread */ Clone_persist_gtid()71 Clone_persist_gtid() { 72 m_event = os_event_create(); 73 /* No background is created yet. */ 74 m_thread_active.store(false); 75 m_gtid_trx_no.store(0); 76 m_flush_number.store(0); 77 m_explicit_request.store(false); 78 m_active_number.store(m_flush_number.load() + 1); 79 /* We accept GTID even before the background service is started. This 80 is needed because we add GTIDs from undo log during recovery. */ 81 m_active.store(true); 82 m_num_gtid_mem.store(0); 83 m_flush_in_progress.store(false); 84 m_close_thread.store(false); 85 } 86 87 /** Destructor: stop gtid thread */ ~Clone_persist_gtid()88 ~Clone_persist_gtid() { 89 ut_ad(!m_thread_active.load()); 90 stop(); 91 os_event_destroy(m_event); 92 } 93 94 /** Start GTID persistence and background thread. 95 @return true, if successful. */ 96 bool start(); 97 98 /* Stop GTID persistence. */ 99 void stop(); 100 101 /* Wait for immediate flush. 102 @param[in] wait wait for completion 103 @param[in] compress_gtid request GTID compression. 104 @param[in] early_timeout don't wait long if flush is blocked. 105 @param[in] cbk alert callback for long wait. */ 106 void wait_flush(bool wait, bool compress_gtid, bool early_timeout, 107 Clone_Alert_Func cbk); 108 109 /** Flush immediately if transaction is operating on GTID table. 110 @param[in,out] thd current session thread */ 111 void flush_if_implicit_gtid(THD *thd); 112 113 /**@return true, if GTID persistence is active. */ is_active()114 bool is_active() const { return (m_active.load()); } 115 116 /**@return true, if GTID thread is active. */ is_thread_active()117 bool is_thread_active() const { return (m_thread_active.load()); } 118 119 /** Get oldest transaction number for which GTID is not persisted to table. 120 Transactions committed after this point should not be purged. 121 @return oldest transaction number. */ get_oldest_trx_no()122 trx_id_t get_oldest_trx_no() { 123 trx_id_t ret_no = m_gtid_trx_no.load(); 124 /* Should never be zero. It can be set to max only before 125 GTID persister is active and no GTID is persisted. */ 126 ut_ad(ret_no > 0 || srv_force_recovery >= SRV_FORCE_NO_UNDO_LOG_SCAN); 127 if (ret_no == TRX_ID_MAX) { 128 ut_ad(!is_thread_active()); 129 ut_ad(m_num_gtid_mem.load() == 0); 130 } else if (m_num_gtid_mem.load() == 0) { 131 /* For all transactions that are committed before this function is called 132 have their GTID flushed if flush is not in progress. "flush not in 133 progress" is sufficient but not necessary condition here. This is mainly 134 for cases when there is no GTID and purge doesn't need to wait. */ 135 if (!m_flush_in_progress.load()) { 136 ret_no = TRX_ID_MAX; 137 } 138 } 139 return (ret_no); 140 } 141 142 /** Set oldest transaction number for which GTID is not persisted to table. 143 This is set during recovery from persisted value. 144 @param[in] max_trx_no transaction number */ set_oldest_trx_no_recovery(trx_id_t max_trx_no)145 void set_oldest_trx_no_recovery(trx_id_t max_trx_no) { 146 ib::info(ER_IB_CLONE_GTID_PERSIST) 147 << "GTID recovery trx_no: " << max_trx_no; 148 /* Zero is special value. It is from old database without GTID 149 persistence. */ 150 if (max_trx_no == 0) { 151 max_trx_no = TRX_ID_MAX; 152 } 153 m_gtid_trx_no.store(max_trx_no); 154 } 155 156 /** Get transaction GTID information. 157 @param[in,out] trx innodb transaction 158 @param[out] gtid_desc descriptor with serialized GTID */ 159 void get_gtid_info(trx_t *trx, Gtid_desc >id_desc); 160 161 /** Set transaction flag to persist GTID and check if space need to be 162 allocated for GTID. 163 @param[in,out] trx current innodb transaction 164 @param[in] prepare if operation is Prepare 165 @param[in] rollback if operation is Rollback 166 @return true, if undo space needs to be allocated. */ 167 bool trx_check_set(trx_t *trx, bool prepare, bool rollback); 168 169 /** Check if current transaction has GTID. 170 @param[in] trx innodb transaction 171 @param[in,out] thd session THD 172 @param[out] passed_check true if transaction is good for GTID 173 @return true, if transaction has valid GTID. */ 174 bool has_gtid(trx_t *trx, THD *&thd, bool &passed_check); 175 176 /** Check if GTID persistence is set 177 @param[in] trx current innnodb transaction 178 @return true, iff GTID persistence is set. */ 179 bool persists_gtid(const trx_t *trx); 180 181 /** Set or reset GTID persist flag in THD. 182 @param[in,out] trx current innnodb transaction 183 @param[in] set true, if need to set */ 184 void set_persist_gtid(trx_t *trx, bool set); 185 186 /** Add GTID to in memory list. 187 @param[in] gtid_desc Descriptor with serialized GTID */ 188 void add(const Gtid_desc >id_desc); 189 190 /** Write GTIDs periodically to disk table. */ 191 void periodic_write(); 192 193 /** Write GTIDs of non Innodb transactions to table. */ 194 int write_other_gtids(); 195 196 /** Disable copy construction */ 197 Clone_persist_gtid(Clone_persist_gtid const &) = delete; 198 199 /** Disable assignment */ 200 Clone_persist_gtid &operator=(Clone_persist_gtid const &) = delete; 201 202 private: 203 /** Check if GTID needs to persist at XA prepare. 204 @param[in] thd session THD 205 @param[in,out] trx current innnodb transaction 206 @param[in] found_gtid session is owning GTID 207 @param[in,out] alloc in:transaction checks are passed 208 out:GTID space need to be allocated 209 @return true, if GTID needs to be persisted */ 210 bool check_gtid_prepare(THD *thd, trx_t *trx, bool found_gtid, bool &alloc); 211 212 /** Check if GTID needs to persist at commit. 213 @param[in] thd session THD 214 @param[in] found_gtid session is owning GTID 215 @return true, if GTID needs to be persisted */ 216 bool check_gtid_commit(THD *thd, bool found_gtid); 217 218 /** Check if GTID needs to persist at rollback. 219 @param[in] thd session THD 220 @param[in,out] trx current innnodb transaction 221 @param[in] found_gtid session is owning GTID 222 @return true, if GTID needs to be persisted */ 223 bool check_gtid_rollback(THD *thd, trx_t *trx, bool found_gtid); 224 225 /** Wait for gtid thread to start, finish or flush. 226 @param[in] start if waiting for start 227 @param[in] flush wait for immediate flush 228 @param[in] flush_number wait flush to reach this number 229 @param[in] compress wait also for compression 230 @param[in] early_timeout don't wait long if flush is blocked 231 @param[in] cbk alert callback for long wait 232 @return true if successful. */ 233 bool wait_thread(bool start, bool flush, uint64_t flush_number, bool compress, 234 bool early_timeout, Clone_Alert_Func cbk); 235 236 /** @return current active GTID list */ get_active_list()237 Gitd_info_list &get_active_list() { 238 ut_ad(trx_sys_mutex_own()); 239 return (get_list(m_active_number)); 240 } 241 242 /** @return GTID list by number. 243 @param[in] list_number list number 244 @return GTID list reference. */ get_list(uint64_t list_number)245 Gitd_info_list &get_list(uint64_t list_number) { 246 int list_index = (list_number & static_cast<uint64_t>(1)); 247 return (m_gtids[list_index]); 248 } 249 250 /** Check if we need to skip write or compression based on debug variables. 251 @param[in] compression check for compression 252 @return true, if we should skip. */ 253 bool debug_skip_write(bool compression); 254 255 /** Request immediate flush of all GTIDs accumulated. 256 @param[in] compress request compression of GTID table 257 @return flush list number to track and wait for flush to complete. */ request_immediate_flush(bool compress)258 uint64_t request_immediate_flush(bool compress) { 259 trx_sys_mutex_enter(); 260 /* We want to flush all GTIDs. */ 261 uint64_t request_number = m_active_number.load(); 262 /* If no GTIDs added to active, wait for previous index. */ 263 if (m_num_gtid_mem.load() == 0) { 264 ut_a(request_number > 0); 265 --request_number; 266 } 267 m_flush_request_number = request_number; 268 trx_sys_mutex_exit(); 269 270 if (compress) { 271 m_explicit_request.store(true); 272 } 273 return (request_number); 274 } 275 276 /** Check if flush has finished up to a list number. 277 @param[in] request_number flush request number 278 @return true, if it is already flushed. */ check_flushed(uint64_t request_number)279 bool check_flushed(uint64_t request_number) const { 280 return (m_flush_number >= request_number); 281 } 282 283 /** @return true, iff background needs to flush immediately. */ flush_immediate()284 bool flush_immediate() const { 285 return (m_flush_number < m_flush_request_number || m_explicit_request); 286 } 287 288 /** Check if GTID compression is necessary based on threshold. 289 @return true, if GTID table needs to be compressed. */ 290 bool check_compress(); 291 292 /** Switch active GTID list. */ switch_active_list()293 uint64_t switch_active_list() { 294 /* Switch active list under transaction system mutex. */ 295 ut_ad(trx_sys_mutex_own()); 296 uint64_t flush_number = m_active_number; 297 ++m_active_number; 298 m_compression_gtid_counter += m_num_gtid_mem; 299 m_num_gtid_mem.store(0); 300 #ifdef UNIV_DEBUG 301 /* The new active list must have no elements. */ 302 auto &active_list = get_active_list(); 303 ut_ad(active_list.size() == 0); 304 #endif 305 return (flush_number); 306 } 307 308 /** Persist GTID to gtid_executed table. 309 @param[in] flush_list_number list number to flush 310 @param[in,out] table_gtid_set GTIDs in table during recovery 311 @param[in,out] sid_map SID map for GTIDs 312 @return mysql error code. */ 313 int write_to_table(uint64_t flush_list_number, Gtid_set &table_gtid_set, 314 Sid_map &sid_map); 315 316 /** Update transaction number upto which GTIDs are flushed to table. 317 @param[in] new_gtid_trx_no GTID transaction number */ 318 void update_gtid_trx_no(trx_id_t new_gtid_trx_no); 319 320 /** Write all GTIDs to table and update GTID transaction number. 321 @param[in,out] thd current session thread */ 322 void flush_gtids(THD *thd); 323 324 private: 325 /** Time threshold to trigger persisting GTID. Insert GTID once per 1k 326 transactions or every 100 millisecond. */ 327 const static uint32_t s_time_threshold_ms = 100; 328 329 /** Threshold for the count for compressing GTID. */ 330 const static uint32_t s_compression_threshold = 50; 331 332 /** Number of transaction/GTID threshold for writing to disk table. */ 333 const static int s_gtid_threshold = 1024; 334 335 /** Two lists of GTID. One of them is active where running transactions 336 add their GTIDs. Other list is used to persist them to table from time 337 to time. */ 338 Gitd_info_list m_gtids[2]; 339 340 /** Number of the current GTID list. Increased when list is switched */ 341 std::atomic<uint64_t> m_active_number; 342 343 /** Number up to which GTIDs are flushed. Increased when list is flushed.*/ 344 std::atomic<uint64_t> m_flush_number; 345 346 /** If explicit request to flush is made. */ 347 std::atomic<bool> m_explicit_request; 348 349 /** Number for which last flush request was made. */ 350 uint64_t m_flush_request_number{0}; 351 352 /** Event for GTID background thread. */ 353 os_event_t m_event; 354 355 /** Counter to keep track of the number of writes till it reaches 356 compression threshold. */ 357 uint32_t m_compression_counter{0}; 358 359 /** Counter to keep number of GTIDs flushed before compression. */ 360 uint32_t m_compression_gtid_counter{0}; 361 362 /* Oldest transaction number for which GTID is not persisted. */ 363 std::atomic<uint64_t> m_gtid_trx_no; 364 365 /** Number of GTID accumulated in memory */ 366 std::atomic<int> m_num_gtid_mem; 367 368 /** Flush of GTID is in progress. */ 369 std::atomic<bool> m_flush_in_progress; 370 371 /** Set to true, when the background thread is asked to exit. */ 372 std::atomic<bool> m_close_thread; 373 374 /** TRUE, if background thread is active.*/ 375 std::atomic<bool> m_thread_active; 376 377 /** TRUE, if GTID persistence is active.*/ 378 std::atomic<bool> m_active; 379 380 unsigned long m_thread_id{0}; 381 }; 382 383 #endif /* CLONE_REPL_INCLUDE */ 384