natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space 

patch by David Capwell; reviewed by Yifan Cai for CASSANDRA-20577
This commit is contained in:
David Capwell 2025-05-02 14:23:17 -07:00
parent 80224a9767
commit 4f25e58f26
3 changed files with 44 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0.5
* zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space (CASSANDRA-20577)
* CQLSSTableWriter supports setting the format (BTI or Big) (CASSANDRA-20609)
* Don't allocate in ThreadLocalReadAheadBuffer#close() (CASSANDRA-20551)
* Ensure RowFilter#isMutableIntersection() properly evaluates numeric ranges on a single column (CASSANDRA-20566)

47
src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java
View File

@ -38,9 +38,12 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.io.util.SequentialWriterOption;
import org.apache.cassandra.net.AsyncStreamingInputPlus;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.ByteBufferUtil;
import static java.lang.String.format;
import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory;
@ -50,7 +53,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
private static final Logger logger = LoggerFactory.getLogger(SSTableZeroCopyWriter.class);
private volatile SSTableReader finalReader;
private final Map<String, SequentialWriter> componentWriters; // indexed by component name
private final Map<String, ZeroCopySequentialWriter> componentWriters; // indexed by component name
public SSTableZeroCopyWriter(Builder<?, ?> builder,
LifecycleNewTracker lifecycleNewTracker,
@ -89,12 +92,12 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
throw new UnsupportedOperationException();
}
private SequentialWriter makeWriter(Descriptor descriptor, Component component)
private ZeroCopySequentialWriter makeWriter(Descriptor descriptor, Component component)
{
return new SequentialWriter(descriptor.fileFor(component), ioOptions.writerOptions, false);
return new ZeroCopySequentialWriter(descriptor.fileFor(component), ioOptions.writerOptions, false);
}
private void write(DataInputPlus in, long size, SequentialWriter out) throws FSWriteError
private void write(DataInputPlus in, long size, ZeroCopySequentialWriter out) throws FSWriteError
{
final int BUFFER_SIZE = 1 << 20;
long bytesRead = 0;
@ -128,7 +131,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
{
setOpenResult(openResult);
for (SequentialWriter writer : componentWriters.values())
for (ZeroCopySequentialWriter writer : componentWriters.values())
writer.finish();
return finished();
@ -170,7 +173,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
@Override
public Throwable commit(Throwable accumulate)
{
for (SequentialWriter writer : componentWriters.values())
for (ZeroCopySequentialWriter writer : componentWriters.values())
accumulate = writer.commit(accumulate);
return accumulate;
}
@ -178,7 +181,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
@Override
public Throwable abort(Throwable accumulate)
{
for (SequentialWriter writer : componentWriters.values())
for (ZeroCopySequentialWriter writer : componentWriters.values())
accumulate = writer.abort(accumulate);
return accumulate;
}
@ -186,29 +189,30 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
@Override
public void prepareToCommit()
{
for (SequentialWriter writer : componentWriters.values())
for (ZeroCopySequentialWriter writer : componentWriters.values())
writer.prepareToCommit();
}
@Override
public void close()
{
for (SequentialWriter writer : componentWriters.values())
for (ZeroCopySequentialWriter writer : componentWriters.values())
writer.close();
}
public void writeComponent(Component component, DataInputPlus in, long size) throws ClosedChannelException
{
SequentialWriter writer = componentWriters.get(component.name);
ZeroCopySequentialWriter writer = componentWriters.get(component.name);
logger.info("Writing component {} to {} length {}", component, writer.getPath(), prettyPrintMemory(size));
if (in instanceof AsyncStreamingInputPlus)
write((AsyncStreamingInputPlus) in, size, writer);
else
// this code path is not valid for production and only exists to simplify unit tests
write(in, size, writer);
}
private void write(AsyncStreamingInputPlus in, long size, SequentialWriter writer) throws ClosedChannelException
private void write(AsyncStreamingInputPlus in, long size, ZeroCopySequentialWriter writer) throws ClosedChannelException
{
logger.info("Block Writing component to {} length {}", writer.getPath(), prettyPrintMemory(size));
@ -233,4 +237,25 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
throw new FSWriteError(e, writer.getPath());
}
}
private static class ZeroCopySequentialWriter extends SequentialWriter
{
private ZeroCopySequentialWriter(File file, SequentialWriterOption option, boolean strictFlushing)
{
super(file, ByteBufferUtil.EMPTY_BYTE_BUFFER, option, strictFlushing);
}
/**
* In production, we do not expect this method to be called, as only writeDirectlyToChannel should be invoked for zero-copy.
* <p>
* This method only exists for tests.
*/
@Override
public void write(byte[] b, int off, int len) throws IOException
{
if (this.buffer == ByteBufferUtil.EMPTY_BYTE_BUFFER)
this.buffer = option.allocateBuffer();
super.write(b, off, len);
}
}
}

9
src/java/org/apache/cassandra/io/util/SequentialWriter.java
View File

@ -56,7 +56,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
// whether to do trickling fsync() to avoid sudden bursts of dirty buffer flushing by kernel causing read
// latency spikes
private final SequentialWriterOption option;
protected final SequentialWriterOption option;
private int bytesSinceTrickleFsync = 0;
protected long lastFlushOffset;
@ -163,7 +163,12 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
*/
public SequentialWriter(File file, SequentialWriterOption option, boolean strictFlushing)
{
super(openChannel(file), option.allocateBuffer());
this(file, option.allocateBuffer(), option, strictFlushing);
}
protected SequentialWriter(File file, ByteBuffer buffer, SequentialWriterOption option, boolean strictFlushing)
{
super(openChannel(file), buffer);
this.strictFlushing = strictFlushing;
this.fchannel = (FileChannel)channel;