CASSANDRA-21134: Address review feedback for background write Direct I/O

- Reuse ChecksumWriter for O_DIRECT compressed writes via DirectChecksumWriter
- Centralize Direct I/O write paths and drop redundant background-write support validation
- Rely on startup checks for engagement; remove per-operation log and its reflective test seam
- Exercise isDirectIOSupported path checks in DatabaseDescriptorTest

Author: samueldlightfoot

conf/cassandra.yaml

@@ -697,8 +697,8 @@ commitlog_disk_access_mode: legacy
 # (compaction, streaming, cleanup, repair, etc.). The allowed values are:
 # - standard: use buffered I/O (default)
 # - direct: use direct I/O, bypassing the OS page cache
-# Note: Only applies to compressed tables. Uncompressed tables always use buffered I/O.
-# Note: Memtable flushes always use buffered I/O regardless of this setting, as flushed
+# Only applies to compressed tables. Uncompressed tables always use buffered I/O.
+# Memtable flushes always use buffered I/O regardless of this setting, as flushed
 # data benefits from page cache for subsequent reads.
 # background_write_disk_access_mode: standard
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -3429,11 +3429,31 @@ public static DataStorageSpec.IntKibibytesBound getDirectWriteBufferSize()
         return conf.direct_write_buffer_size;
     }
 
+    /**
+     * The directories opened with O_DIRECT for writing. Startup checks that must reason about O_DIRECT
+     * write targets (e.g. the kernel-bug 1057843 check) derive their path set from here.
+     */
+    public static Set<Path> getDirectIOWritePaths()
+    {
+        Set<Path> paths = new HashSet<>();
+
+        if (getCommitLogWriteDiskAccessMode() == DiskAccessMode.direct)
+            paths.add(new File(getCommitLogLocation()).toPath());
+
+        if (getBackgroundWriteDiskAccessMode() == DiskAccessMode.direct)
+            for (String dataDir : getAllDataFileLocations())
+                paths.add(new File(dataDir).toPath());
+
+        return paths;
+    }
+
     @VisibleForTesting
     public static void initializeBackgroundWriteDiskAccessMode()
     {
         DiskAccessMode providedMode = conf.background_write_disk_access_mode;
 
+        // 'auto' is a forward-looking placeholder: it resolves to standard until Direct I/O is a safe
+        // default, at which point it will resolve to direct (mirroring commitlog_disk_access_mode).
         if (providedMode == DiskAccessMode.auto)
         {
             providedMode = DiskAccessMode.standard;
@@ -3448,38 +3468,12 @@ public static void initializeBackgroundWriteDiskAccessMode()
                 throw new ConfigurationException("direct_write_buffer_size must be > 0 when background_write_disk_access_mode is 'direct'. " +
                                                  "Got: " + conf.direct_write_buffer_size, false);
 
+            // Create the data directories up front (as we do for the commit log) so the kernel-bug 1057843
+            // startup check can stat each O_DIRECT write target. Direct I/O support is validated separately
+            // by the directio_support startup check.
             if (!toolInitialized)
-            {
-                List<String> unsupportedLocations = new ArrayList<>();
-
-                for (String dataDir : conf.data_file_directories)
-                {
-                    try
-                    {
-                        File dataDirFile = new File(dataDir);
-                        PathUtils.createDirectoriesIfNotExists(dataDirFile.toPath());
-
-                        if (!FileUtils.isDirectIOSupported(dataDirFile))
-                        {
-                            unsupportedLocations.add(dataDir);
-                        }
-                    }
-                    catch (RuntimeException e)
-                    {
-                        logger.warn("Unable to determine Direct IO support for data directory {}: {}", dataDir, e.getMessage());
-                        unsupportedLocations.add(dataDir + " (check failed: " + e.getMessage() + ")");
-                    }
-                }
-
-                if (!unsupportedLocations.isEmpty())
-                {
-                    throw new ConfigurationException(
-                        String.format("background_write_disk_access_mode is set to 'direct', but the following data directories " +
-                                      "do not support Direct I/O: %s. Either change background_write_disk_access_mode to 'standard' " +
-                                      "in cassandra.yaml, or ensure all data directories are on filesystems that support Direct I/O.",
-                                      unsupportedLocations), false);
-                }
-            }
+                for (String dataDir : getAllDataFileLocations())
+                    PathUtils.createDirectoriesIfNotExists(new File(dataDir).toPath());
         }
         else if (providedMode != DiskAccessMode.standard)
         {

src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java

@@ -47,7 +47,7 @@
 
 public class CompressedSequentialWriter extends SequentialWriter
 {
-    private final ChecksumWriter crcMetadata;
+    protected final ChecksumWriter crcMetadata;
 
     // holds offset in the file where current chunk should be written
     // changed only by flush() method where data buffer gets compressed and stored to the file
@@ -153,7 +153,16 @@ public CompressedSequentialWriter(File file,
         metadataWriter = CompressionMetadata.Writer.open(parameters, offsetsFile, compressionDictionary);
 
         this.sstableMetadataCollector = sstableMetadataCollector;
-        crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(this.channel)));
+        crcMetadata = createChecksumWriter();
+    }
+
+    /**
+     * Creates the {@link ChecksumWriter} for the chunk and full-file checksums. Invoked from the constructor,
+     * so overrides must not read subclass fields.
+     */
+    protected ChecksumWriter createChecksumWriter()
+    {
+        return new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel)));
     }
 
     @Override

src/java/org/apache/cassandra/io/compress/DirectCompressedSequentialWriter.java

@@ -19,9 +19,8 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.zip.CRC32;
+import java.util.function.IntConsumer;
 
 import javax.annotation.Nullable;
 
@@ -37,9 +36,9 @@
 import org.apache.cassandra.db.compression.CompressionDictionaryManager;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
+import org.apache.cassandra.io.util.ChecksumWriter;
 import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.io.util.File;
-import org.apache.cassandra.io.util.FileOutputStreamPlus;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.io.util.SequentialWriterOption;
@@ -75,14 +74,9 @@ public class DirectCompressedSequentialWriter extends CompressedSequentialWriter
     // fit contiguously. flushCompleteBlocks aligns down to blockSize, leaving up to
     // (blockSize - 1) bytes carried over via compact(); the floor accounts for that.
     private ByteBuffer writeBuffer;
-    private int writeBufferPosition = 0;
     private long actualDataSize = 0;
 
     private final int blockSize;
-    // ChecksumWriter writes CRCs directly to the channel, bypassing writeBuffer; track checksums ourselves.
-    private final CRC32 fullFileChecksum = new CRC32();
-    private final CRC32 chunkChecksum = new CRC32();
-    private final ByteBuffer crcBuffer = ByteBuffer.allocate(CRC_LENGTH);
 
     public DirectCompressedSequentialWriter(File file,
                                             File offsetsFile,
@@ -130,6 +124,14 @@ public DirectCompressedSequentialWriter(File file,
         }
     }
 
+    // Invoked from super()'s constructor before writeBuffer exists; safe because we only capture the
+    // writeCrcToAlignedBuffer reference, which isn't called until the first flush.
+    @Override
+    protected ChecksumWriter createChecksumWriter()
+    {
+        return new DirectChecksumWriter(this::writeCrcToAlignedBuffer);
+    }
+
     // Parent reads fchannel.position(), which lags by writeBuffer contents under O_DIRECT.
     // getEstimatedOnDiskBytesWritten is intentionally NOT overridden: parent returns chunkOffset,
     // which already represents the eventual on-disk size — correct under DIO.
@@ -155,16 +157,11 @@ protected void writeChunk(ByteBuffer toWrite)
 
         int chunkLength = toWrite.remaining();
 
-        chunkChecksum.reset();
-        chunkChecksum.update(toWrite);
-        int crcValue = (int) chunkChecksum.getValue();
-        toWrite.rewind();
-
         writeToAlignedBuffer(toWrite);
-        writeCrcToAlignedBuffer(crcValue);
 
+        // writeToAlignedBuffer drained toWrite; rewind so appendDirect can re-read it for the CRC.
         toWrite.rewind();
-        updateFullChecksum(toWrite, crcValue);
+        crcMetadata.appendDirect(toWrite, true);
 
         actualDataSize = chunkOffset + chunkLength + CRC_LENGTH;
     }
@@ -174,51 +171,57 @@ private void writeToAlignedBuffer(ByteBuffer data)
         int dataLength = data.remaining();
 
         // Buffer is sized for worst-case chunk + CRC + blockSize, so a flush always frees enough room.
-        if (writeBufferPosition + dataLength > writeBuffer.capacity())
+        if (writeBuffer.position() + dataLength > writeBuffer.capacity())
             flushCompleteBlocks();
 
-        writeBuffer.position(writeBufferPosition);
         writeBuffer.put(data);
-        writeBufferPosition = writeBuffer.position();
     }
 
     private void writeCrcToAlignedBuffer(int crcValue)
     {
         // After flush, leftover < blockSize, so there's always room for the CRC trailer.
-        if (writeBufferPosition + CRC_LENGTH > writeBuffer.capacity())
+        if (writeBuffer.position() + CRC_LENGTH > writeBuffer.capacity())
             flushCompleteBlocks();
 
-        writeBuffer.position(writeBufferPosition);
         writeBuffer.putInt(crcValue);
-        writeBufferPosition = writeBuffer.position();
+    }
+
+    // FileChannel.write may write fewer bytes than requested, and a partial write would short the file
+    // and desync the callers' leftover/truncate bookkeeping.
+    private void writeAlignedBlocks(int flushLimit) throws IOException
+    {
+        writeBuffer.position(0);
+        writeBuffer.limit(flushLimit);
+        while (writeBuffer.hasRemaining())
+            fchannel.write(writeBuffer);
     }
 
     private void flushCompleteBlocks()
     {
+        // writeAlignedBlocks rewinds position() to 0 to drain, so snapshot the cursor first.
+        int logicalPos = writeBuffer.position();
+
         // Align down: O_DIRECT cannot write partial blocks
-        int flushLimit = writeBufferPosition & -blockSize;
+        int flushLimit = logicalPos & -blockSize;
 
         if (flushLimit == 0)
             return;
 
         try
         {
-            writeBuffer.position(0);
-            writeBuffer.limit(flushLimit);
-            fchannel.write(writeBuffer);
+            writeAlignedBlocks(flushLimit);
 
-            int leftover = writeBufferPosition - flushLimit;
+            int leftover = logicalPos - flushLimit;
             if (leftover > 0)
             {
-                writeBuffer.limit(writeBufferPosition);
+                writeBuffer.limit(logicalPos);
                 writeBuffer.position(flushLimit);
                 writeBuffer.compact();
             }
             else
             {
                 writeBuffer.clear();
             }
-            writeBufferPosition = leftover;
         }
         catch (IOException e)
         {
@@ -228,19 +231,18 @@ private void flushCompleteBlocks()
 
     private void flushFinalWithPadding()
     {
-        if (writeBufferPosition == 0)
+        int logicalPos = writeBuffer.position();
+
+        if (logicalPos == 0)
             return;
 
         try
         {
-            int flushLimit = BitUtil.align(writeBufferPosition, blockSize);
+            int flushLimit = BitUtil.align(logicalPos, blockSize);
 
-            writeBuffer.position(writeBufferPosition);
-            ByteBufferUtil.writeZeroes(writeBuffer, flushLimit - writeBufferPosition);
+            ByteBufferUtil.writeZeroes(writeBuffer, flushLimit - logicalPos);
 
-            writeBuffer.position(0);
-            writeBuffer.limit(flushLimit);
-            fchannel.write(writeBuffer);
+            writeAlignedBlocks(flushLimit);
 
             // O_DIRECT required padding; truncate back to actual data size.
             fchannel.truncate(actualDataSize);
@@ -251,34 +253,6 @@ private void flushFinalWithPadding()
         }
     }
 
-    private void updateFullChecksum(ByteBuffer data, int crcValue)
-    {
-        fullFileChecksum.update(data);
-
-        // Include CRC bytes in the full-file checksum to match ChecksumWriter.appendDirect(..., true).
-        crcBuffer.clear();
-        crcBuffer.putInt(crcValue);
-        crcBuffer.flip();
-        fullFileChecksum.update(crcBuffer);
-    }
-
-    @Override
-    protected void writeDigestFile()
-    {
-        digestFile.ifPresent(file -> {
-            try (FileOutputStreamPlus fos = new FileOutputStreamPlus(file))
-            {
-                fos.write(String.valueOf(fullFileChecksum.getValue()).getBytes(StandardCharsets.UTF_8));
-                fos.flush();
-                fos.getChannel().force(true);
-            }
-            catch (IOException e)
-            {
-                throw new FSWriteError(e, file);
-            }
-        });
-    }
-
     // Gated out for SCRUB in DataComponent.buildWriter; these throws are a canary if the gate is bypassed.
     @Override
     public DataPosition mark()
@@ -332,4 +306,31 @@ protected SequentialWriter.TransactionalProxy txnProxy()
     {
         return new DirectTransactionalProxy();
     }
+
+    /**
+     * Sends the per-chunk CRC int into the block-aligned writeBuffer rather than straight to the channel, so
+     * the O_DIRECT writer reuses ChecksumWriter's checksum bookkeeping instead of duplicating it. Only the CRC
+     * trailer flows through here; {@link #writeChunk} places the chunk data in the buffer.
+     */
+    private static final class DirectChecksumWriter extends ChecksumWriter
+    {
+        private final IntConsumer alignedSink;
+
+        DirectChecksumWriter(IntConsumer alignedSink)
+        {
+            this.alignedSink = alignedSink;
+        }
+
+        @Override
+        protected void writeIncrementalInt(int value)
+        {
+            alignedSink.accept(value);
+        }
+
+        @Override
+        public void writeChunkSize(int length)
+        {
+            throw new UnsupportedOperationException("writeChunkSize is unused on the compressed O_DIRECT path");
+        }
+    }
 }

src/java/org/apache/cassandra/io/sstable/format/DataComponent.java

@@ -19,11 +19,8 @@
 package org.apache.cassandra.io.sstable.format;
 
 import java.util.EnumMap;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.cassandra.config.Config.DiskAccessMode;
 import org.apache.cassandra.config.Config.FlushCompression;
@@ -49,12 +46,8 @@
 
 public class DataComponent
 {
-    private static final Logger logger = LoggerFactory.getLogger(DataComponent.class);
-
     private static final EnumMap<OperationType, DirectIoSupport> DIRECT_WRITE_SUPPORT = buildDirectWriteSupport();
 
-    private static final Set<OperationType> directWriteLogged = ConcurrentHashMap.newKeySet();
-
     private static EnumMap<OperationType, DirectIoSupport> buildDirectWriteSupport()
     {
         EnumMap<OperationType, DirectIoSupport> m = new EnumMap<>(OperationType.class);
@@ -97,8 +90,6 @@ public static SequentialWriter buildWriter(Descriptor descriptor,
             if (DatabaseDescriptor.getBackgroundWriteDiskAccessMode() == DiskAccessMode.direct
                 && isDirectWriteSupported(operationType))
             {
-                if (directWriteLogged.add(operationType))
-                    logger.info("Direct I/O writer activated for {}", operationType);
                 return new DirectCompressedSequentialWriter(descriptor.fileFor(Components.DATA),
                                                             descriptor.fileFor(Components.COMPRESSION_INFO),
                                                             descriptor.fileFor(Components.DIGEST),
@@ -168,7 +159,8 @@ private static CompressionParams buildCompressionParams(TableMetadata metadata,
         return compressionParams;
     }
 
-    private static boolean isDirectWriteSupported(OperationType operationType)
+    @VisibleForTesting
+    static boolean isDirectWriteSupported(OperationType operationType)
     {
         DirectIoSupport support = DIRECT_WRITE_SUPPORT.get(operationType);
         if (support == null)