SpillOutputStream Class Reference

SpillOutputStream implements the ByteOutputStream interface by starting with writes to a cache scratch page. More...

#include <SpillOutputStream.h>

Inheritance diagram for SpillOutputStream:

List of all members.

Public Member Functions
virtual	~SpillOutputStream ()
SharedByteInputStream	getInputStream (SeekPosition seekPosition=SEEK_STREAM_BEGIN)
	Obtains a ByteInputStream suitable for accessing the contents of this SpillOutputStream.
SharedSegment	getSegment ()
	Obtains a reference to the underlying segment if this stream has spilled.
SharedSegOutputStream	getSegOutputStream ()
	Obtains a reference to the underlying SegOutputStream if this stream has spilled.
virtual void	setWriteLatency (WriteLatency writeLatency)
	Changes the write latency.
void	writeBytes (void const *pData, uint cbRequested)
	Writes bytes to the stream.
PBuffer	getWritePointer (uint cbRequested, uint *pcbActual=NULL)
	Copyless alternative for writing bytes to the stream.
void	consumeWritePointer (uint cbUsed)
	Advances stream position after a call to getWritePointer.
void	hardPageBreak ()
	Marks the current buffer as complete regardless of how much data it contains.
template<class T>
void	writeValue (T const &value)
	Writes a fixed-size type to the stream.
FileSize	getOffset () const
	Returns: current offset from beginning of stream
bool	isClosed () const
	Returns: whether the object has been closed
void	close ()
	Closes this object, releasing any unallocated resources.
Static Public Member Functions
static SharedSpillOutputStream	newSpillOutputStream (SharedSegmentFactory pSegmentFactory, SharedCacheAccessor pCacheAccessor, std::string spillFileName)
	Creates a new SpillOutputStream.
Protected Member Functions
void	setBuffer (PBuffer pBuffer, uint cbBuffer)
	Sets the current buffer to be written.
uint	getBytesAvailable () const
	Returns: number of bytes remaining to be written in current buffer
Protected Attributes
WriteLatency	writeLatency
	Current write latency mode.
FileSize	cbOffset
	Byte position in stream.
bool	needsClose
Private Member Functions
virtual void	flushBuffer (uint cbRequested)
	Must be implemented by derived class to flush buffered data.
virtual void	closeImpl ()
	Must be implemented by derived class to release any resources.
void	spill ()
void	updatePage ()
	SpillOutputStream (SharedSegmentFactory, SharedCacheAccessor, std::string)
Private Attributes
SharedSegmentFactory	pSegmentFactory
	Factory to use for creating spill segment if necessary.
SharedCacheAccessor	pCacheAccessor
	CacheAccessor to use for accessing spill segment if necessary.
SharedSegOutputStream	pSegOutputStream
	Spill segment output stream.
uint	cbBuffer
	Total number of bytes locked in current page.
SegmentAccessor	scratchAccessor
	Accessor for scratch segment.
SegPageLock	scratchPageLock
	Page lock on scratch page for short streams.
std::string	spillFileName
	Filename to assign to spill segment.

---

Detailed Description

SpillOutputStream implements the ByteOutputStream interface by starting with writes to a cache scratch page.

If this overflows, the contents are spilled to a new SegOutputStream to which all further writes are directed.

Definition at line 40 of file SpillOutputStream.h.

---

Constructor & Destructor Documentation

SpillOutputStream::SpillOutputStream	(	SharedSegmentFactory	,
		SharedCacheAccessor	,
		std::string
	)			[explicit, private]

Definition at line 45 of file SpillOutputStream.cpp.

References SegPageLock::accessSegment(), SegPageLock::allocatePage(), cbBuffer, SegPageLock::getPage(), CachePage::getWritableData(), pCacheAccessor, SegmentAccessor::pSegment, pSegmentFactory, scratchAccessor, scratchPageLock, and ByteOutputStream::setBuffer().

Referenced by newSpillOutputStream().

    : pSegmentFactory(pSegmentFactoryInit),
      pCacheAccessor(pCacheAccessorInit),
      spillFileName(spillFileNameInit)
{
    // REVIEW:  this causes pCacheAccessor to lose its chance to intercept any
    // of the scratch calls
    scratchAccessor = pSegmentFactory->newScratchSegment(
        pCacheAccessor->getCache(),
        1);
    scratchPageLock.accessSegment(scratchAccessor);
    scratchPageLock.allocatePage();
    cbBuffer = scratchAccessor.pSegment->getUsablePageSize();
    setBuffer(
        scratchPageLock.getPage().getWritableData(),
        cbBuffer);
}

SpillOutputStream::~SpillOutputStream

(

)

[virtual]

Definition at line 66 of file SpillOutputStream.cpp.

{
}

---

Member Function Documentation

void SpillOutputStream::flushBuffer

(

uint

cbRequested

)

[private, virtual]

Must be implemented by derived class to flush buffered data.

Parameters:

cbRequested

if non-zero, the derived class should allocate a new buffer with at least the requested size and call setBuffer

Implements ByteOutputStream.

Definition at line 70 of file SpillOutputStream.cpp.

References cbBuffer, ByteOutputStream::getBytesAvailable(), SegPageLock::isLocked(), pSegOutputStream, scratchPageLock, ByteOutputStream::setBuffer(), and spill().

{
    if (scratchPageLock.isLocked()) {
        assert(!pSegOutputStream);
        // grow from short to long
        spill();
    } else {
        assert(pSegOutputStream);
        assert(!scratchPageLock.isLocked());
        // already long
        assert(cbBuffer >= getBytesAvailable());
        pSegOutputStream->consumeWritePointer(cbBuffer - getBytesAvailable());
    }
    assert(pSegOutputStream);
    if (cbRequested) {
        PBuffer pBuffer =
            pSegOutputStream->getWritePointer(cbRequested,&cbBuffer);
        setBuffer(pBuffer,cbBuffer);
    } else {
        pSegOutputStream->hardPageBreak();
        cbBuffer = 0;
    }
}

void SpillOutputStream::closeImpl

(

)

[private, virtual]

Must be implemented by derived class to release any resources.

Reimplemented from ByteOutputStream.

Definition at line 94 of file SpillOutputStream.cpp.

References cbBuffer, ByteOutputStream::getBytesAvailable(), SegPageLock::isLocked(), pSegOutputStream, scratchPageLock, and SegPageLock::unlock().

{
    if (scratchPageLock.isLocked()) {
        // discard contents
        scratchPageLock.unlock();
    } else {
        assert(pSegOutputStream);
        assert(!scratchPageLock.isLocked());
        // flush long log
        pSegOutputStream->consumeWritePointer(cbBuffer - getBytesAvailable());
        cbBuffer = 0;
        pSegOutputStream.reset();
    }
}

void SpillOutputStream::spill

(

)

[private]

Definition at line 123 of file SpillOutputStream.cpp.

References cbBuffer, DeviceMode::createNew, DeviceMode::direct, ByteOutputStream::getBytesAvailable(), SegPageLock::getCacheAccessor(), SegPageLock::getPage(), CachePage::getReadableData(), SegOutputStream::newSegOutputStream(), pCacheAccessor, pSegmentFactory, pSegOutputStream, scratchPageLock, spillFileName, SegPageLock::unlock(), and ByteOutputStream::writeLatency.

Referenced by flushBuffer().

{
    DeviceMode devMode = DeviceMode::createNew;
    // TODO:  make this a parameter; for now it's always direct since this is
    // only used for log streams
    devMode.direct = true;
    SharedSegment pLongLogSegment =
        pSegmentFactory->newTempDeviceSegment(
            scratchPageLock.getCacheAccessor()->getCache(),
            devMode,
            spillFileName);
    SegmentAccessor segmentAccessor(pLongLogSegment,pCacheAccessor);
    pSegOutputStream = SegOutputStream::newSegOutputStream(segmentAccessor);
    pSegOutputStream->setWriteLatency(writeLatency);
    pSegOutputStream->writeBytes(
        scratchPageLock.getPage().getReadableData(),
        cbBuffer - getBytesAvailable());
    scratchPageLock.unlock();
}

void SpillOutputStream::updatePage

(

)

[private]

Definition at line 173 of file SpillOutputStream.cpp.

References cbBuffer, ByteOutputStream::getBytesAvailable(), and pSegOutputStream.

Referenced by getInputStream().

{
    if (!cbBuffer) {
        return;
    }
    assert(cbBuffer > getBytesAvailable());
    uint cbConsumed = cbBuffer - getBytesAvailable();
    pSegOutputStream->consumeWritePointer(cbConsumed);
    cbBuffer -= cbConsumed;
    pSegOutputStream->updatePage();
}

SharedSpillOutputStream SpillOutputStream::newSpillOutputStream	(	SharedSegmentFactory	pSegmentFactory,
		SharedCacheAccessor	pCacheAccessor,
		std::string	spillFileName
	)			[static]

Creates a new SpillOutputStream.

Parameters:

	pSegmentFactory	the SegmentFactory to use if the output stream spills
	pCacheAccessor	the CacheAccessor to use
	spillFileName	filename to assign to spill segment

Definition at line 35 of file SpillOutputStream.cpp.

References SpillOutputStream().

Referenced by LogicalTxn::LogicalTxn(), and SegStreamTest::testWriteSpillAndRead().

{
    return SharedSpillOutputStream(
        new SpillOutputStream(pSegmentFactory,pCacheAccessor,spillFileName),
        ClosableObjectDestructor());
}

SharedByteInputStream SpillOutputStream::getInputStream

(

SeekPosition

seekPosition = SEEK_STREAM_BEGIN

)

Obtains a ByteInputStream suitable for accessing the contents of this SpillOutputStream.

If spill has already occurred, then this is a SegInputStream, otherwise a ByteArrayInputStream.

Parameters:

seekPosition

if SEEK_STREAM_BEGIN (the default), the input stream is initially positioned before the first byte of stream data; otherwise, after the last byte

Returns:

new ByteInputStream

Definition at line 143 of file SpillOutputStream.cpp.

References ByteStream::getOffset(), SegPageLock::getPage(), CachePage::getReadableData(), SegPageLock::isLocked(), ByteArrayInputStream::newByteArrayInputStream(), SegInputStream::newSegInputStream(), pCacheAccessor, pSegOutputStream, scratchPageLock, SEEK_STREAM_END, and updatePage().

{
    if (scratchPageLock.isLocked()) {
        SharedByteInputStream pInputStream =
            ByteArrayInputStream::newByteArrayInputStream(
                scratchPageLock.getPage().getReadableData(),
                getOffset());
        if (seekPosition == SEEK_STREAM_END) {
            pInputStream->seekForward(getOffset());
        }
        return pInputStream;
    } else {
        assert(pSegOutputStream);
        updatePage();
        SharedSegment pSegment = pSegOutputStream->getSegment();
        SegStreamPosition endPos;
        if (seekPosition == SEEK_STREAM_END) {
            pSegOutputStream->getSegPos(endPos);
        }
        SegmentAccessor segmentAccessor(pSegment,pCacheAccessor);
        SharedSegInputStream pInputStream =
            SegInputStream::newSegInputStream(segmentAccessor);
        if (seekPosition == SEEK_STREAM_END) {
            pInputStream->seekSegPos(endPos);
        }
        return pInputStream;
    }
}

SharedSegment SpillOutputStream::getSegment

(

)

Obtains a reference to the underlying segment if this stream has spilled.

Returns:

the segment, or a singular pointer if the stream has not spilled

Definition at line 185 of file SpillOutputStream.cpp.

References pSegOutputStream.

{
    if (pSegOutputStream) {
        return pSegOutputStream->getSegment();
    } else {
        return SharedSegment();
    }
}

SharedSegOutputStream SpillOutputStream::getSegOutputStream

(

)

Obtains a reference to the underlying SegOutputStream if this stream has spilled.

Returns:

the stream, or a singular pointer if the stream has not spilled

Definition at line 194 of file SpillOutputStream.cpp.

References pSegOutputStream.

{
    return pSegOutputStream;
}

void SpillOutputStream::setWriteLatency

(

WriteLatency

writeLatency

)

[virtual]

Changes the write latency.

May not be meaningful for all stream implementations.

Parameters:

writeLatency

new WriteLatency setting

Reimplemented from ByteOutputStream.

Definition at line 109 of file SpillOutputStream.cpp.

References pSegOutputStream, and ByteOutputStream::setWriteLatency().

{
    ByteOutputStream::setWriteLatency(writeLatency);
    if (pSegOutputStream) {
        pSegOutputStream->setWriteLatency(writeLatency);
    }
}

void ByteOutputStream::setBuffer	(	PBuffer	pBuffer,
		uint	cbBuffer
	)			[inline, protected, inherited]

Sets the current buffer to be written.

Parameters:

	pBuffer	receives start address of new buffer
	cbBuffer	number of bytes in buffer

Definition at line 162 of file ByteOutputStream.h.

References ByteOutputStream::cbWritable, and ByteOutputStream::pNextByte.

Referenced by ByteArrayOutputStream::ByteArrayOutputStream(), ByteArrayOutputStream::clear(), flushBuffer(), SegOutputStream::flushBuffer(), and SpillOutputStream().

{
    pNextByte = pBuffer;
    cbWritable = cbBuffer;
}

uint ByteOutputStream::getBytesAvailable

(

)

const [inline, protected, inherited]

Returns:

number of bytes remaining to be written in current buffer

Definition at line 168 of file ByteOutputStream.h.

References ByteOutputStream::cbWritable.

Referenced by closeImpl(), flushBuffer(), SegOutputStream::getBytesWrittenThisPage(), SegOutputStream::getSegPos(), spill(), and updatePage().

{
    return cbWritable;
}

void ByteOutputStream::writeBytes	(	void const *	pData,
		uint	cbRequested
	)			[inherited]

Writes bytes to the stream.

Parameters:

	pData	source buffer containing bytes to be written
	cbRequested	number of bytes to write

Definition at line 35 of file ByteOutputStream.cpp.

References ByteStream::cbOffset, ByteOutputStream::cbWritable, ByteOutputStream::flushBuffer(), and ByteOutputStream::pNextByte.

Referenced by FtrsTableWriter::execute().

{
    cbOffset += cb;
    if (!cbWritable) {
        flushBuffer(1);
    }
    for (;;) {
        assert(cbWritable);
        if (cb <= cbWritable) {
            memcpy(pNextByte,pData,cb);
            cbWritable -= cb;
            pNextByte += cb;
            return;
        }
        memcpy(pNextByte,pData,cbWritable);
        pData = static_cast<char const *>(pData) + cbWritable;
        cb -= cbWritable;
        cbWritable = 0;
        flushBuffer(1);
    }
}

PBuffer ByteOutputStream::getWritePointer	(	uint	cbRequested,
		uint *	pcbActual = NULL
	)			[inline, inherited]

Copyless alternative for writing bytes to the stream.

Provides direct access to the stream's internal buffer, but doesn't move the stream position (see consumeWritePointer).

Parameters:

	cbRequested	number of contiguous bytes to access; if fewer bytes are currently available in the buffer, the buffer is flushed and a new buffer is returned
	pcbActual	if non-NULL, receives actual number of contiguous writable bytes, which will always be greater than or equal to cbRequested

Returns:

pointer to cbActual bytes of writable buffer space

Definition at line 141 of file ByteOutputStream.h.

References ByteOutputStream::cbWritable, ByteOutputStream::flushBuffer(), and ByteOutputStream::pNextByte.

Referenced by BTreeWriter::deleteCurrent(), BTreeWriter::insertTupleFromBuffer(), and SegOutputStream::SegOutputStream().

{
    if (cbWritable < cbRequested) {
        flushBuffer(cbRequested);
        assert(cbWritable >= cbRequested);
    }
    if (pcbActual) {
        *pcbActual = cbWritable;
    }
    return pNextByte;
}

void ByteOutputStream::consumeWritePointer

(

uint

cbUsed

)

[inline, inherited]

Advances stream position after a call to getWritePointer.

Parameters:

cbUsed

number of bytes to advance; must be less than or equal to the value of cbActual returned by the last call to getWritePointer

Definition at line 154 of file ByteOutputStream.h.

References ByteStream::cbOffset, ByteOutputStream::cbWritable, and ByteOutputStream::pNextByte.

Referenced by BTreeWriter::deleteCurrent(), and BTreeWriter::insertTupleFromBuffer().

{
    assert(cbUsed <= cbWritable);
    cbWritable -= cbUsed;
    pNextByte += cbUsed;
    cbOffset += cbUsed;
}

void ByteOutputStream::hardPageBreak

(

)

[inherited]

Marks the current buffer as complete regardless of how much data it contains.

The exact semantics are dependent on the buffering implementation.

Definition at line 62 of file ByteOutputStream.cpp.

References ByteOutputStream::cbWritable, ByteOutputStream::flushBuffer(), and ByteOutputStream::pNextByte.

{
    flushBuffer(0);
    cbWritable = 0;
    pNextByte = NULL;
}

template<class T>

void ByteOutputStream::writeValue

(

T const &

value

)

[inline, inherited]

Writes a fixed-size type to the stream.

Parameters:

value

value to read; type must be memcpy-safe

Definition at line 135 of file ByteOutputStream.h.

Referenced by FtrsTableWriter::describeIndex(), LogicalTxnTest::describeParticipant(), FtrsTableWriter::describeParticipant(), BTreeWriter::describeParticipant(), DatabaseTest::executeIncrementAction(), TupleProjection::writePersistent(), and TupleDescriptor::writePersistent().

    {
        writeBytes(&value,sizeof(value));
    }

FileSize ByteStream::getOffset

(

)

const [inline, inherited]

Returns:

current offset from beginning of stream

Definition at line 110 of file ByteStream.h.

References ByteStream::cbOffset.

Referenced by getInputStream(), and ByteInputStream::mark().

{
    return cbOffset;
}

bool ClosableObject::isClosed

(

)

const [inline, inherited]

Returns:

whether the object has been closed

Definition at line 58 of file ClosableObject.h.

    {
        return !needsClose;
    }

void ClosableObject::close

(

)

[inherited]

Closes this object, releasing any unallocated resources.

Reimplemented in CollectExecStream, CorrelationJoinExecStream, LcsClusterAppendExecStream, and LcsClusterReplaceExecStream.

Definition at line 39 of file ClosableObject.cpp.

References ClosableObject::closeImpl(), and ClosableObject::needsClose.

Referenced by CacheImpl< PageT, VictimPolicyT >::allocatePages(), LcsRowScanBaseExecStream::closeImpl(), ExecStreamGraphImpl::closeImpl(), FlatFileBuffer::open(), ClosableObjectDestructor::operator()(), and Segment::~Segment().

{
    if (!needsClose) {
        return;
    }
    needsClose = false;
    closeImpl();
}

---

Member Data Documentation

SharedSegmentFactory SpillOutputStream::pSegmentFactory [private]

Factory to use for creating spill segment if necessary.

Definition at line 46 of file SpillOutputStream.h.

Referenced by spill(), and SpillOutputStream().

SharedCacheAccessor SpillOutputStream::pCacheAccessor [private]

CacheAccessor to use for accessing spill segment if necessary.

Definition at line 51 of file SpillOutputStream.h.

Referenced by getInputStream(), spill(), and SpillOutputStream().

SharedSegOutputStream SpillOutputStream::pSegOutputStream [private]

Spill segment output stream.

Definition at line 56 of file SpillOutputStream.h.

Referenced by closeImpl(), flushBuffer(), getInputStream(), getSegment(), getSegOutputStream(), setWriteLatency(), spill(), and updatePage().

uint SpillOutputStream::cbBuffer [private]

Total number of bytes locked in current page.

Definition at line 61 of file SpillOutputStream.h.

Referenced by closeImpl(), flushBuffer(), spill(), SpillOutputStream(), and updatePage().

SegmentAccessor SpillOutputStream::scratchAccessor [private]

Accessor for scratch segment.

Definition at line 66 of file SpillOutputStream.h.

Referenced by SpillOutputStream().

SegPageLock SpillOutputStream::scratchPageLock [private]

Page lock on scratch page for short streams.

Definition at line 71 of file SpillOutputStream.h.

Referenced by closeImpl(), flushBuffer(), getInputStream(), spill(), and SpillOutputStream().

std::string SpillOutputStream::spillFileName [private]

Filename to assign to spill segment.

Definition at line 76 of file SpillOutputStream.h.

Referenced by spill().

WriteLatency ByteOutputStream::writeLatency [protected, inherited]

Current write latency mode.

Definition at line 51 of file ByteOutputStream.h.

Referenced by SegOutputStream::flushBuffer(), SegOutputStream::SegOutputStream(), ByteOutputStream::setWriteLatency(), and spill().

FileSize ByteStream::cbOffset [protected, inherited]

Byte position in stream.

Definition at line 41 of file ByteStream.h.

Referenced by ByteStream::ByteStream(), ByteInputStream::consumeReadPointer(), ByteOutputStream::consumeWritePointer(), ByteStream::getOffset(), SegOutputStream::getSegPos(), SegInputStream::getSegPos(), ByteInputStream::readBytes(), ByteInputStream::reset(), ByteArrayInputStream::resetArray(), ByteInputStream::seekBackward(), SegInputStream::seekSegPos(), and ByteOutputStream::writeBytes().

bool ClosableObject::needsClose [protected, inherited]

Definition at line 44 of file ClosableObject.h.

Referenced by SegStreamAllocation::beginWrite(), ExecStreamGraphImpl::clear(), ClosableObject::ClosableObject(), ClosableObject::close(), FlatFileBuffer::open(), ExecStreamGraphImpl::open(), ExecStream::open(), and ClosableObject::~ClosableObject().

---

The documentation for this class was generated from the following files:

• /home/pub/open/dev/fennel/segment/SpillOutputStream.h
• /home/pub/open/dev/fennel/segment/SpillOutputStream.cpp

---