| 111 | } |
| 112 | |
| 113 | void AsyncLogger::RunThread() { |
| 114 | MutexLock l(lock_); |
| 115 | while (state_ == RUNNING || active_buf_->needs_flush_or_write()) { |
| 116 | while (!active_buf_->needs_flush_or_write() && state_ == RUNNING) { |
| 117 | if (!wake_flusher_cond_.WaitFor(MonoDelta::FromSeconds(FLAGS_logbufsecs))) { |
| 118 | // In case of wait timeout, force it to flush regardless whether there is anything enqueued. |
| 119 | active_buf_->flush = true; |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | active_buf_.swap(flushing_buf_); |
| 124 | // If the buffer that we are about to flush was full, then |
| 125 | // we may have other threads which were blocked that we now |
| 126 | // need to wake up. |
| 127 | if (BufferFull(*flushing_buf_)) { |
| 128 | free_buffer_cond_.Broadcast(); |
| 129 | } |
| 130 | l.Unlock(); |
| 131 | |
| 132 | for (const auto& msg : flushing_buf_->messages) { |
| 133 | wrapped_->Write(false, msg.ts, msg.message.data(), msg.message.size()); |
| 134 | } |
| 135 | if (flushing_buf_->flush) { |
| 136 | wrapped_->Flush(); |
| 137 | } |
| 138 | flushing_buf_->clear(); |
| 139 | |
| 140 | l.Lock(); |
| 141 | flush_count_++; |
| 142 | flush_complete_cond_.Broadcast(); |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | bool AsyncLogger::BufferFull(const Buffer& buf) const { |
| 147 | // We evenly divide our total buffer space between the two buffers. |