git commit: TS-1330: Fix logging core at checkout_write()

[yu...@apache.org]

Updated Branches:
  refs/heads/t [created] c1c3db9d7


TS-1330: Fix logging core at checkout_write()

== Analysis ==
1. One thread(A) got an LogBuffer object in LogObject::_checkout_write() and ran
   into LogBuffer::checkout_write(), before it updated the LogBuffer::m_state,
   it was scheduled out by the kernel.

2. Other threads got the same LogBuffer object, but they all finished
   _checkout_write() codes and made the LogBuffer object become full, finally
   one of these thread flushed the LogBuffer object into write_list queue.

3. The LogBuffer object's m_references == 0 at this time.

4. The flushing thread would call LogBufferManager::flush_buffers() to loop
   write_list. As the LogBuffer's m_references == 0, so it would be flushed
   and deleted.

5. Now, thread(A) would be scheduled in by the kernel, but the LogBuffer object
   has been deleted, so the m_buffer == NULL, that is why entry_header is 0x0.

== Solution ==
1. Remove flushing code from LogObject::log(), we should flush LogBuffer into
   queue only when new_buffer is created, so that we can promise that LogBuffer
   must be flushed only once.

2. The flushing thread should both check m_references and m_state.s.num_writers.

3. When allocate a new_buffer, should avoid memory leak.

Signed-off-by: Yunkai Zhang <qi...@taobao.com>


Project: http://git-wip-us.apache.org/repos/asf/trafficserver/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafficserver/commit/c1c3db9d
Tree: http://git-wip-us.apache.org/repos/asf/trafficserver/tree/c1c3db9d
Diff: http://git-wip-us.apache.org/repos/asf/trafficserver/diff/c1c3db9d

Branch: refs/heads/t
Commit: c1c3db9d7b524176e611a6aba3066d1a732b4bd2
Parents: cf62135
Author: Yunkai Zhang <qi...@taobao.com>
Authored: Thu Aug 1 00:31:21 2013 +0800
Committer: Yunkai Zhang <qi...@taobao.com>
Committed: Fri Aug 2 11:25:25 2013 +0800

----------------------------------------------------------------------
 proxy/logging/LogBuffer.h  |  3 +--
 proxy/logging/LogObject.cc | 26 +++++++++++++-------------
 2 files changed, 14 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafficserver/blob/c1c3db9d/proxy/logging/LogBuffer.h
----------------------------------------------------------------------
diff --git a/proxy/logging/LogBuffer.h b/proxy/logging/LogBuffer.h
index 31ee951..191b3a5 100644
--- a/proxy/logging/LogBuffer.h
+++ b/proxy/logging/LogBuffer.h
@@ -197,8 +197,6 @@ private:
   size_t m_buf_align;           // the buffer alignment
   size_t m_write_align;         // the write alignment mask
 
-  volatile LB_State m_state;    // buffer state
-
   int m_max_entries;            // max number of entries allowed
   long m_expiration_time;       // buffer expiration time
 
@@ -207,6 +205,7 @@ private:
 
   uint32_t m_id;                // unique buffer id (for debugging)
 public:
+  volatile LB_State m_state;    // buffer state
   volatile int m_references;    // oustanding checkout_write references.
 private:
 

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/c1c3db9d/proxy/logging/LogObject.cc
----------------------------------------------------------------------
diff --git a/proxy/logging/LogObject.cc b/proxy/logging/LogObject.cc
index fd12e64..6af216e 100644
--- a/proxy/logging/LogObject.cc
+++ b/proxy/logging/LogObject.cc
@@ -44,7 +44,8 @@ LogBufferManager::flush_buffers(LogBufferSink *sink) {
   SList(LogBuffer, write_link) q(write_list.popall()), new_q;
   LogBuffer *b = NULL;
   while ((b = q.pop())) {
-    if (b->m_references) {  // Still has outstanding references.
+    if (b->m_references || b->m_state.s.num_writers) {
+      // Still has outstanding references.
       write_list.push(b);
     } else if (_num_flush_buffers > FLUSH_ARRAY_SIZE) {
       delete b;
@@ -433,6 +434,14 @@ LogObject::_checkout_write(size_t * write_offset, size_t bytes_needed) {
       head_p old_h;
       do {
         INK_QUEUE_LD(old_h, m_log_buffer);
+        if (FREELIST_POINTER(old_h) != FREELIST_POINTER(h)) {
+          ink_atomic_increment(&buffer->m_references, -1);
+
+          // another thread should be taking care of creating a new
+          // buffer, so delete new_buffer and try again
+          delete new_buffer;
+          break;
+        }
         head_p tmp_h;
         SET_FREELIST_POINTER_VERSION(tmp_h, new_buffer, 0);
 #if TS_HAS_128BIT_CAS
@@ -441,14 +450,13 @@ LogObject::_checkout_write(size_t * write_offset, size_t bytes_needed) {
        result = ink_atomic_cas((int64_t *) &m_log_buffer.data, old_h.data, tmp_h.data);
 #endif
       } while (!result);
-      if (FREELIST_POINTER(old_h) == FREELIST_POINTER(h))
+      if (FREELIST_POINTER(old_h) == FREELIST_POINTER(h)) {
         ink_atomic_increment(&buffer->m_references, FREELIST_VERSION(old_h) - 1);
 
-      if (result_code == LogBuffer::LB_FULL_NO_WRITERS) {
-        // there are no writers, move the old buffer to the flush list
         Debug("log-logbuffer", "adding buffer %d to flush list after checkout", buffer->get_id());
         m_buffer_manager.add_to_flush_queue(buffer);
         ink_cond_signal(&Log::flush_cond);
+
       }
       decremented = true;
       break;
@@ -604,15 +612,7 @@ LogObject::log(LogAccess * lad, char *text_entry)
     ink_strlcpy(&(*buffer)[offset], text_entry, bytes_needed);
   }
 
-  LogBuffer::LB_ResultCode result_code = buffer->checkin_write(offset);
-
-  if (result_code == LogBuffer::LB_ALL_WRITERS_DONE) {
-    // all checkins completed, put this buffer in the flush list
-    Debug("log-logbuffer", "adding buffer %d to flush list after checkin", buffer->get_id());
-
-    m_buffer_manager.add_to_flush_queue(buffer);
-//    ink_cond_signal (&Log::flush_cond);
-  }
+  buffer->checkin_write(offset);
 
   LOG_INCREMENT_DYN_STAT(log_stat_event_log_access_stat);