content: /*
   * 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.
   */
  package org.apache.cassandra.hints;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.Queue;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentLinkedQueue;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.zip.CRC32;
  
  import org.apache.cassandra.io.util.DataOutputBuffer;
  import org.apache.cassandra.io.util.DataOutputBufferFixed;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.utils.AbstractIterator;
  import org.apache.cassandra.utils.concurrent.OpOrder;
  import org.apache.cassandra.utils.memory.MemoryUtil;
  
  import static org.apache.cassandra.utils.FBUtilities.updateChecksum;
  import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
  
  /**
   * A shared buffer that temporarily holds the serialized hints before they are flushed to disk.
   *
   * Consists of :
   * - a ByteBuffer holding the serialized hints (length, length checksum and total checksum included)
   * - a pointer to the current allocation offset
   * - an {@link OpOrder} appendOrder for {@link HintsWriteExecutor} to wait on for all writes completion
   * - a map of (host id -> offset queue) for the hints written
   *
   * It's possible to write a single hint for two or more hosts at the same time, in which case the same offset will be put
   * into two or more offset queues.
   */
  final class HintsBuffer
  {
      // hint entry overhead in bytes (int length, int length checksum, int body checksum)
      static final int ENTRY_OVERHEAD_SIZE = 12;
  
      private final ByteBuffer slab; // the underlying backing ByteBuffer for all the serialized hints
      private final AtomicLong position; // the position in the slab that we currently allocate from
  
      private final ConcurrentMap<UUID, Queue<Integer>> offsets;
      private final OpOrder appendOrder;
  
      private HintsBuffer(ByteBuffer slab)
      {
          this.slab = slab;
  
          position = new AtomicLong();
          offsets = new ConcurrentHashMap<>();
          appendOrder = new OpOrder();
      }
  
      static HintsBuffer create(int slabSize)
      {
          return new HintsBuffer(ByteBuffer.allocateDirect(slabSize));
      }
  
      boolean isClosed()
      {
          return position.get() < 0;
      }
  
      int capacity()
      {
          return slab.capacity();
      }
  
      int remaining()
      {
          long pos = position.get();
          return (int) (pos < 0 ? 0 : Math.max(0, capacity() - pos));
      }
  
      HintsBuffer recycle()
      {
          slab.clear();
          return new HintsBuffer(slab);
      }
  
      void free()
      {
          MemoryUtil.clean(slab);
      }
  
      /**
       * Wait for any appends started before this method was called.
       */
      void waitForModifications()
      {
          appendOrder.awaitNewBarrier(); // issue a barrier and wait for it
      }
  
      Set<UUID> hostIds()
      {
          return offsets.keySet();
      }
  
      /**
       * Coverts the queue of offsets for the selected host id into an iterator of hints encoded as ByteBuffers.
       */
      Iterator<ByteBuffer> consumingHintsIterator(UUID hostId)
      {
          final Queue<Integer> bufferOffsets = offsets.get(hostId);
  
          if (bufferOffsets == null)
              return Collections.emptyIterator();
  
          return new AbstractIterator<>()
          {
              private final ByteBuffer flyweight = slab.duplicate();
  
              protected ByteBuffer computeNext()
              {
                  Integer offset = bufferOffsets.poll();
  
                  if (offset == null)
                      return endOfData();
  
                  int totalSize = slab.getInt(offset) + ENTRY_OVERHEAD_SIZE;
  
                  return flyweight.clear().position(offset).limit(offset + totalSize);
              }
          };
      }
  
      Allocation allocate(int hintSize)
      {
          int totalSize = hintSize + ENTRY_OVERHEAD_SIZE;
  
          if (totalSize > slab.capacity() / 2)
          {
              throw new IllegalArgumentException(String.format("Hint of %s bytes is too large - the maximum size is %s",
                                                               hintSize,
                                                               slab.capacity() / 2));
          }
  
          OpOrder.Group opGroup = appendOrder.start(); // will eventually be closed by the receiver of the allocation
          try
          {
              return allocate(totalSize, opGroup);
          }
          catch (Throwable t)
          {
              opGroup.close();
              throw t;
          }
      }
  
      private Allocation allocate(int totalSize, OpOrder.Group opGroup)
      {
          int offset = allocateBytes(totalSize);
          if (offset < 0)
          {
              opGroup.close();
              return null;
          }
          return new Allocation(offset, totalSize, opGroup);
      }
  
      /**
       * Allocate bytes in the segment, or return -1 if not enough space. Method ensures that marker bytes
       * for each allocation (i.e. offset of its end) is written as a 32 bit integer at its beginning, and
       * that these marker bytes are always written sequentially. In other words, if allocation A has a lower
       * starting offset than allocation B, A's marker will always be written before the offset for B is returned.
       *
       * `allocateOffset` consists of two integers:
       *    64                 32                0
       *    | (i32) inProgress | (i32) writtenTo |
       *
       *  If inProgress bytes are not zeroes, they contain an unwritten offset. Before allocating any bytes,
       *  inProgresss bytes need to be written at the writtenTo location in the target buffer.
       */
      private int allocateBytes(int totalSize)
      {
          long prev = position.getAndAdd(totalSize);
  
          if (prev < 0) // the slab has been 'closed'
              return -1;
  
          if ((prev + totalSize) > slab.capacity())
          {
              position.set(Long.MIN_VALUE); // mark the slab as no longer allocating if we've exceeded its capacity
              return -1;
          }
  
          return (int)prev;
      }
  
      private void put(UUID hostId, int offset)
      {
          // we intentionally don't just return offsets.computeIfAbsent() because it's expensive compared to simple get(),
          // and the method is on a really hot path
          Queue<Integer> queue = offsets.get(hostId);
          if (queue == null)
              queue = offsets.computeIfAbsent(hostId, (id) -> new ConcurrentLinkedQueue<>());
          queue.offer(offset);
      }
  
      /**
       * A placeholder for hint serialization. Should always be used in a try-with-resources block.
       */
      final class Allocation implements AutoCloseable
      {
          private final Integer offset;
          private final int totalSize;
          private final OpOrder.Group opGroup;
  
          Allocation(int offset, int totalSize, OpOrder.Group opGroup)
          {
              this.offset = offset;
              this.totalSize = totalSize;
              this.opGroup = opGroup;
          }
  
          void write(Iterable<UUID> hostIds, Hint hint)
          {
              write(hint);
              for (UUID hostId : hostIds)
                  put(hostId, offset);
          }
  
          public void close()
          {
              opGroup.close();
          }
  
          private void write(Hint hint)
          {
              ByteBuffer buffer = slab.duplicate().position(offset).limit(offset + totalSize);
              CRC32 crc = new CRC32();
              int hintSize = totalSize - ENTRY_OVERHEAD_SIZE;
              try (DataOutputBuffer dop = new DataOutputBufferFixed(buffer))
              {
                  dop.writeInt(hintSize);
                  updateChecksumInt(crc, hintSize);
                  dop.writeInt((int) crc.getValue());
  
                  Hint.serializer.serialize(hint, dop, MessagingService.current_version);
                  updateChecksum(crc, buffer, buffer.position() - hintSize, hintSize);
                  dop.writeInt((int) crc.getValue());
              }
              catch (IOException e)
              {
                  throw new AssertionError(); // cannot happen
              }
          }
      }
  }
  