RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-01-16 07:52:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #37699 from dillaman/wip-47880

Browse Source 
journal: possible race condition between flush and append callback

Reviewed-by: Mykola Golub <mgolub@suse.com>
This commit is contained in:
Mykola Golub 2020-10-18 15:43:51 +03:00 committed by GitHub
commit b94369f7c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/journal/ObjectRecorder.cc
View File

@ -97,7 +97,7 @@ void ObjectRecorder::flush(Context *on_safe) {
// if currently handling flush notifications, wait so that
// we notify in the correct order (since lock is dropped on
// callback)
while (m_in_flight_callbacks) {
while (m_in_flight_callbacks > 0) {
m_in_flight_callbacks_cond.wait(locker);
}
@ -138,7 +138,7 @@ void ObjectRecorder::flush(const ceph::ref_t<FutureImpl>& future) {
}
if (!m_object_closed && !m_overflowed && send_appends(true, future)) {
m_in_flight_callbacks = true;
++m_in_flight_callbacks;
notify_handler_unlock(locker, true);
}
}
@ -169,7 +169,7 @@ bool ObjectRecorder::close() {
ceph_assert(!m_object_closed);
m_object_closed = true;
if (!m_in_flight_tids.empty() || m_in_flight_callbacks) {
if (!m_in_flight_tids.empty() || m_in_flight_callbacks > 0) {
m_object_closed_notify = true;
return false;
}
@ -181,7 +181,7 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
ldout(m_cct, 20) << "tid=" << tid << ", r=" << r << dendl;
std::unique_lock locker{*m_lock};
m_in_flight_callbacks = true;
++m_in_flight_callbacks;
auto tid_iter = m_in_flight_tids.find(tid);
ceph_assert(tid_iter != m_in_flight_tids.end());
@ -235,7 +235,9 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
ldout(m_cct, 20) << "pending tids=" << m_in_flight_tids << dendl;
// all remaining unsent appends should be redirected to new object
// notify of overflow if one just occurred or indicate that all in-flight
// appends have completed on a closed object (or wake up stalled flush
// requests that was waiting for this strand to complete).
notify_handler_unlock(locker, notify_overflowed);
}
@ -379,14 +381,16 @@ bool ObjectRecorder::send_appends(bool force, ceph::ref_t<FutureImpl> flush_futu
void ObjectRecorder::wake_up_flushes() {
ceph_assert(ceph_mutex_is_locked(*m_lock));
m_in_flight_callbacks = false;
m_in_flight_callbacks_cond.notify_all();
--m_in_flight_callbacks;
if (m_in_flight_callbacks == 0) {
m_in_flight_callbacks_cond.notify_all();
}
}
void ObjectRecorder::notify_handler_unlock(
std::unique_lock<ceph::mutex>& locker, bool notify_overflowed) {
ceph_assert(ceph_mutex_is_locked(*m_lock));
ceph_assert(m_in_flight_callbacks);
ceph_assert(m_in_flight_callbacks > 0);
if (!m_object_closed && notify_overflowed) {
// TODO need to delay completion until after aio_notify completes

2
src/journal/ObjectRecorder.h
View File

@ -143,7 +143,7 @@ private:
bufferlist m_prefetch_bl;
bool m_in_flight_callbacks = false;
uint32_t m_in_flight_callbacks = 0;
ceph::condition_variable m_in_flight_callbacks_cond;
uint64_t m_in_flight_bytes = 0;