Skip to content

Commit

Permalink
sender log
Browse files Browse the repository at this point in the history
Signed-off-by: guo-shaoge <[email protected]>
  • Loading branch information
guo-shaoge committed Oct 19, 2024
1 parent dbd6afe commit b97d824
Showing 1 changed file with 17 additions and 0 deletions.
17 changes: 17 additions & 0 deletions dbms/src/Common/LooseBoundedMPMCQueue.h
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,20 @@ class LooseBoundedMPMCQueue
, push_callback(std::move(push_callback_))
{}

String getAddr() const
{
std::stringstream ss;
ss << this;
return ss.str();
}

void registerPipeReadTask(TaskPtr && task)
{
{
const auto pipe_task_cnt = pipe_reader_cv.getTaskCnt();
std::lock_guard lock(mu);
LOG_DEBUG(Logger::get(), "gjt debug registerPipeReadTask queue size: {}, status: {}, pipe_task_cnt: {}, workqueue: {}",
queue.size(), status == MPMCQueueStatus::NORMAL, pipe_task_cnt, getAddr());
if (queue.empty() && status == MPMCQueueStatus::NORMAL)
{
pipe_reader_cv.registerTask(std::move(task));
Expand All @@ -66,7 +76,10 @@ class LooseBoundedMPMCQueue
void registerPipeWriteTask(TaskPtr && task)
{
{
const auto pipe_task_cnt = pipe_writer_cv.getTaskCnt();
std::lock_guard lock(mu);
LOG_DEBUG(Logger::get(), "gjt debug registerPipeWriteTask isFullWithoutLock: {}, status: {}, pipe_task_cnt: {}, workqueue: {}",
isFullWithoutLock(), status == MPMCQueueStatus::NORMAL, pipe_task_cnt, getAddr());
if (isFullWithoutLock() && status == MPMCQueueStatus::NORMAL)
{
pipe_writer_cv.registerTask(std::move(task));
Expand Down Expand Up @@ -259,12 +272,16 @@ class LooseBoundedMPMCQueue
private:
void notifyOneReader()
{
LOG_DEBUG(Logger::get(), "gjt debug notifyOneReader: {}",
getAddr());
reader_cv.notify_one();
pipe_reader_cv.notifyOne();
}

void notifyOneWriter()
{
LOG_DEBUG(Logger::get(), "gjt debug notifyOneWriter: {}",
getAddr());
writer_cv.notify_one();
pipe_writer_cv.notifyOne();
}
Expand Down

0 comments on commit b97d824

Please sign in to comment.