journal: delay object overflow event until in-flight appends settled

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

diff --git a/src/journal/ObjectRecorder.cc b/src/journal/ObjectRecorder.cc
index f9a41100111f286b8d7b7632baf1ed1d723559f3..f7f1966224b9734934fc63fa259ce60dacab7180 100644 (file)
--- a/src/journal/ObjectRecorder.cc
+++ b/src/journal/ObjectRecorder.cc
@@ -39,6 +39,7 @@ ObjectRecorder::ObjectRecorder(librados::IoCtx &ioctx, const std::string &oid,
 ObjectRecorder::~ObjectRecorder() {
   assert(m_append_task == NULL);
   assert(m_append_buffers.empty());
+  assert(m_in_flight_tids.empty());
   assert(m_in_flight_appends.empty());
 }
 
@@ -142,6 +143,7 @@ void ObjectRecorder::claim_append_buffers(AppendBuffers *append_buffers) {
   ldout(m_cct, 20) << __func__ << ": " << m_oid << dendl;
 
   Mutex::Locker locker(m_lock);
+  assert(m_in_flight_tids.empty());
   assert(m_in_flight_appends.empty());
   assert(m_object_closed || m_overflowed);
   append_buffers->splice(append_buffers->end(), m_append_buffers,
@@ -229,18 +231,28 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
   AppendBuffers append_buffers;
   {
     Mutex::Locker locker(m_lock);
+    auto tid_iter = m_in_flight_tids.find(tid);
+    assert(tid_iter != m_in_flight_tids.end());
+    m_in_flight_tids.erase(tid_iter);
+
     InFlightAppends::iterator iter = m_in_flight_appends.find(tid);
-    if (iter == m_in_flight_appends.end()) {
-      // must have seen an overflow on a previous append op
-      assert(m_overflowed);
-      return;
-    } else if (r == -EOVERFLOW) {
-      m_overflowed = true;
-      append_overflowed(tid);
+    if (r == -EOVERFLOW || m_overflowed) {
+      if (iter != m_in_flight_appends.end()) {
+        m_overflowed = true;
+        append_overflowed(tid);
+      } else {
+        // must have seen an overflow on a previous append op
+        assert(r == -EOVERFLOW && m_overflowed);
+      }
+
+      // notify of overflow once all in-flight ops are complete
+      if (m_in_flight_tids.empty()) {
+        notify_overflow();
+      }
       return;
     }
 
-    assert(!m_overflowed || r != 0);
+    assert(iter != m_in_flight_appends.end());
     append_buffers.swap(iter->second);
     assert(!append_buffers.empty());
 
@@ -291,7 +303,6 @@ void ObjectRecorder::append_overflowed(uint64_t tid) {
                                 m_append_buffers.begin(),
                                 m_append_buffers.end());
   restart_append_buffers.swap(m_append_buffers);
-  notify_overflow();
 }
 
 void ObjectRecorder::send_appends(AppendBuffers *append_buffers) {
@@ -315,6 +326,7 @@ void ObjectRecorder::send_appends(AppendBuffers *append_buffers) {
     op.set_op_flags2(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
     m_size += it->second.length();
   }
+  m_in_flight_tids.insert(append_tid);
   m_in_flight_appends[append_tid].swap(*append_buffers);
 
   librados::AioCompletion *rados_completion =

diff --git a/src/journal/ObjectRecorder.h b/src/journal/ObjectRecorder.h
index 2cc85415e5eb5c4f410f532923b12e8e69a11d11..378ab64df8c660355325076240f22f598ed18a99 100644 (file)
--- a/src/journal/ObjectRecorder.h
+++ b/src/journal/ObjectRecorder.h
@@ -11,6 +11,8 @@
 #include "common/RefCountedObj.h"
 #include "journal/FutureImpl.h"
 #include <list>
+#include <map>
+#include <set>
 #include <boost/intrusive_ptr.hpp>
 #include <boost/noncopyable.hpp>
 #include "include/assert.h"
@@ -63,6 +65,7 @@ public:
   }
 
 private:
+  typedef std::set<uint64_t> InFlightTids;
   typedef std::map<uint64_t, AppendBuffers> InFlightAppends;
 
   struct FlushHandler : public FutureImpl::FlushHandler {
@@ -125,6 +128,7 @@ private:
   uint64_t m_append_tid;
   uint32_t m_pending_bytes;
 
+  InFlightTids m_in_flight_tids;
   InFlightAppends m_in_flight_appends;
   uint64_t m_size;
   bool m_overflowed;