Extract class AggregateWriteBuffer from Stripe

include/iocore/cache/AggregateWriteBuffer.h

@@ -0,0 +1,118 @@
+/** @file
+
+  A brief file description
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+ */
+
+#pragma once
+
+#include "iocore/eventsystem/Continuation.h"
+
+#include "tscore/ink_memory.h"
+#include "tscore/List.h"
+
+#include <cstring>
+
+#define AGG_SIZE       (4 * 1024 * 1024) // 4MB
+#define AGG_HIGH_WATER (AGG_SIZE / 2)    // 2MB
+
+struct CacheVC;
+
+class AggregateWriteBuffer
+{
+public:
+  AggregateWriteBuffer()
+  {
+    this->_buffer = static_cast<char *>(ats_memalign(ats_pagesize(), AGG_SIZE));
+    memset(this->_buffer, 0, AGG_SIZE);
+  }
+
+  ~AggregateWriteBuffer() { ats_free(this->_buffer); }
+
+  AggregateWriteBuffer(AggregateWriteBuffer const &)            = delete;
+  AggregateWriteBuffer &operator=(AggregateWriteBuffer const &) = delete;
+
+  // move semantics not supported yet to keep things simple
+  AggregateWriteBuffer(AggregateWriteBuffer &&other)            = delete;
+  AggregateWriteBuffer &operator=(AggregateWriteBuffer &&other) = delete;
+
+  Queue<CacheVC, Continuation::Link_link> &get_pending_writers();
+  char *get_buffer();
+  int get_buffer_pos() const;
+  void add_buffer_pos(int size);
+  void seek(int offset);
+  void reset_buffer_pos();
+  int get_bytes_pending_aggregation() const;
+  void add_bytes_pending_aggregation(int size);
+
+private:
+  Queue<CacheVC, Continuation::Link_link> _pending_writers;
+  char *_buffer                  = nullptr;
+  int _bytes_pending_aggregation = 0;
+  int _buffer_pos                = 0;
+};
+
+inline Queue<CacheVC, Continuation::Link_link> &
+AggregateWriteBuffer::get_pending_writers()
+{
+  return this->_pending_writers;
+}
+
+inline char *
+AggregateWriteBuffer::get_buffer()
+{
+  return this->_buffer;
+}
+
+inline int
+AggregateWriteBuffer::get_buffer_pos() const
+{
+  return this->_buffer_pos;
+}
+
+inline void
+AggregateWriteBuffer::add_buffer_pos(int size)
+{
+  this->_buffer_pos += size;
+}
+
+inline void
+AggregateWriteBuffer::seek(int offset)
+{
+  this->_buffer_pos = offset;
+}
+
+inline void
+AggregateWriteBuffer::reset_buffer_pos()
+{
+  this->seek(0);
+}
+
+inline int
+AggregateWriteBuffer::get_bytes_pending_aggregation() const
+{
+  return this->_bytes_pending_aggregation;
+}
+
+inline void
+AggregateWriteBuffer::add_bytes_pending_aggregation(int size)
+{
+  this->_bytes_pending_aggregation += size;
+}

include/iocore/cache/CacheVC.h

@@ -48,7 +48,7 @@
 
 #include <cstdint>
 
-struct Stripe;
+class Stripe;
 class HttpConfigAccessor;
 
 struct CacheVC : public CacheVConnection {

src/iocore/cache/CacheDir.cc

@@ -996,21 +996,21 @@ sync_cache_dir_on_shutdown()
     // check if we have data in the agg buffer
     // dont worry about the cachevc s in the agg queue
     // directories have not been inserted for these writes
-    if (vol->agg_buf_pos) {
+    if (vol->get_agg_buf_pos()) {
       Dbg(dbg_ctl_cache_dir_sync, "Dir %s: flushing agg buffer first", vol->hash_text.get());
 
       // set write limit
-      vol->header->agg_pos = vol->header->write_pos + vol->agg_buf_pos;
+      vol->header->agg_pos = vol->header->write_pos + vol->get_agg_buf_pos();
 
-      int r = pwrite(vol->fd, vol->agg_buffer, vol->agg_buf_pos, vol->header->write_pos);
-      if (r != vol->agg_buf_pos) {
+      int r = pwrite(vol->fd, vol->get_agg_buffer(), vol->get_agg_buf_pos(), vol->header->write_pos);
+      if (r != vol->get_agg_buf_pos()) {
         ink_assert(!"flushing agg buffer failed");
         continue;
       }
       vol->header->last_write_pos  = vol->header->write_pos;
-      vol->header->write_pos      += vol->agg_buf_pos;
+      vol->header->write_pos      += vol->get_agg_buf_pos();
       ink_assert(vol->header->write_pos == vol->header->agg_pos);
-      vol->agg_buf_pos = 0;
+      vol->reset_agg_buf_pos();
       vol->header->write_serial++;
     }
 
@@ -1136,7 +1136,7 @@ CacheSync::mainEvent(int event, Event *e)
         Dbg(dbg_ctl_cache_dir_sync, "Dir %s not dirty", vol->hash_text.get());
         goto Ldone;
       }
-      if (vol->is_io_in_progress() || vol->agg_buf_pos) {
+      if (vol->is_io_in_progress() || vol->get_agg_buf_pos()) {
         Dbg(dbg_ctl_cache_dir_sync, "Dir %s: waiting for agg buffer", vol->hash_text.get());
         vol->dir_sync_waiting = true;
         if (!vol->is_io_in_progress()) {

src/iocore/cache/CacheVC.cc

@@ -490,9 +490,9 @@ CacheVC::handleRead(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
   if (dir_agg_buf_valid(vol, &dir)) {
     int agg_offset = vol->vol_offset(&dir) - vol->header->write_pos;
     buf            = new_IOBufferData(iobuffer_size_to_index(io.aiocb.aio_nbytes, MAX_BUFFER_SIZE_INDEX), MEMALIGNED);
-    ink_assert((agg_offset + io.aiocb.aio_nbytes) <= (unsigned)vol->agg_buf_pos);
+    ink_assert((agg_offset + io.aiocb.aio_nbytes) <= (unsigned)vol->get_agg_buf_pos());
     char *doc = buf->data();
-    char *agg = vol->agg_buffer + agg_offset;
+    char *agg = vol->get_agg_buffer() + agg_offset;
     memcpy(doc, agg, io.aiocb.aio_nbytes);
     io.aio_result = io.aiocb.aio_nbytes;
     SET_HANDLER(&CacheVC::handleReadDone);