-
Notifications
You must be signed in to change notification settings - Fork 149
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Addressed review comments from Kamil
- Handled the relay log recovery. - Changed thd_enter_apply_monitor() to thd_enter_async_monitor(). - Changed thd_leave_apply_monitor() to thd_leave_async_monitor() - Dont't track the last_left and last_entered - The new logic now uses scheduled_seqnos queue to track the the lowest seqno the server is executing. - The new logic also uses the skipped_seqno to track the skipped seqnos. - Added unittests - Added copyright to the newly introduced files.
- Loading branch information
1 parent
25c09e8
commit 30112dc
Showing
12 changed files
with
348 additions
and
105 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,38 +1,94 @@ | ||
| /* Copyright (c) 2021 Percona LLC and/or its affiliates. All rights reserved. | ||
|
|
||
| This program is free software; you can redistribute it and/or | ||
| modify it under the terms of the GNU General Public License | ||
| as published by the Free Software Foundation; version 2 of | ||
| the License. | ||
|
|
||
| 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. | ||
|
|
||
| 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 St, Fifth Floor, Boston, MA 02110-1301 USA */ | ||
|
|
||
| #ifdef WITH_WSREP | ||
| #include "sql/wsrep_async_monitor.h" | ||
| #include <algorithm> | ||
| #include <cassert> | ||
|
|
||
| // Method for main thread to add scheduled seqnos | ||
| void Wsrep_async_monitor::schedule(seqno_t seqno) { | ||
| std::unique_lock<std::mutex> lock(m_mutex); | ||
| scheduled_seqnos.push(seqno); | ||
| } | ||
|
|
||
| // Method for both DDL and DML to enter the monitor | ||
| void Wsrep_async_monitor::enter(seqno_t seqno) { | ||
| std::unique_lock<std::mutex> lock(m_mutex); | ||
|
|
||
| // Wait for its turn before entering | ||
| m_cond.wait(lock, [this, seqno]() { return seqno == m_last_left + 1; }); | ||
| m_last_entered = (seqno > m_last_entered) ? seqno : m_last_entered; | ||
| fprintf(stderr, "Entered the monitor with seqno: %llu\n", seqno); | ||
| // Wait until this transaction is at the head of the scheduled queue | ||
| m_cond.wait(lock, [this, seqno] { | ||
| // Remove skipped transactions | ||
| while (!scheduled_seqnos.empty() && | ||
| skipped_seqnos.count(scheduled_seqnos.front()) > 0) { | ||
| scheduled_seqnos.pop(); | ||
| } | ||
| return !scheduled_seqnos.empty() && scheduled_seqnos.front() == seqno; | ||
| }); | ||
| } | ||
|
|
||
| // Method to be called after DDL/DML processing is complete | ||
| void Wsrep_async_monitor::leave(seqno_t seqno) { | ||
| // Wait for its turn before leaving | ||
| std::unique_lock<std::mutex> lock(m_mutex); | ||
| m_cond.wait(lock, [this, seqno]() { return seqno == m_last_left + 1; }); | ||
| m_last_left = seqno; | ||
|
|
||
| fprintf(stderr, "Left the monitor seqno: %llu\n", seqno); | ||
| // Notify all waiting threads | ||
| // Check if the sequence number matches the front of the queue. | ||
| // In a correctly functioning monitor this should not happen | ||
| // as each transaction should exit in the order it was scheduled | ||
| // and processed. | ||
| if (!scheduled_seqnos.empty() && scheduled_seqnos.front() == seqno) { | ||
| // Remove the seqno from the scheduled queue now that it has completed | ||
| scheduled_seqnos.pop(); | ||
| } else { | ||
| // std::cout << "Error: Mismatch in sequence numbers. Expected " | ||
| // << (scheduled_seqnos.empty() | ||
| // ? "none" | ||
| // : std::to_string(scheduled_seqnos.front())) | ||
| // << " but got " << seqno << "." << std::endl; | ||
| assert(false && "Sequence number mismatch in leave()"); | ||
| exit(1); | ||
| } | ||
|
|
||
| // Notify waiting threads in case the next scheduled sequence can enter | ||
| m_cond.notify_all(); | ||
| } | ||
|
|
||
| void Wsrep_async_monitor::reset(seqno_t seqno) { | ||
| // Method to skip a transaction that will not call enter() and exit() | ||
| void Wsrep_async_monitor::skip(unsigned long seqno) { | ||
| std::unique_lock<std::mutex> lock(m_mutex); | ||
|
|
||
| // We can reset only if the last entered and last left | ||
| // are same, meaning that there is no one who is inside | ||
| // the monitor | ||
| assert(m_last_entered == m_last_left); | ||
| m_last_entered = seqno; | ||
| m_last_left = seqno; | ||
| // Check if the seqno is already marked as skipped | ||
| if (skipped_seqnos.count(seqno) > 0) { | ||
| return; // Already skipped, so do nothing | ||
| } | ||
|
|
||
| // Mark the seqno as skipped | ||
| skipped_seqnos.insert(seqno); | ||
|
|
||
| // Remove it from the scheduled queue if it is at the front | ||
| if (!scheduled_seqnos.empty() && scheduled_seqnos.front() == seqno) { | ||
| scheduled_seqnos.pop(); | ||
| } | ||
|
|
||
| // Notify in case other transactions are waiting to enter | ||
| m_cond.notify_all(); | ||
| } | ||
|
|
||
| // Method to return if the monitor is empty, used by the unittests | ||
| bool Wsrep_async_monitor::is_empty() { | ||
| std::unique_lock<std::mutex> lock(m_mutex); | ||
| return scheduled_seqnos.empty(); | ||
| } | ||
| #endif /* WITH_WSREP */ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.