Unlike other composers, this one doesn't need merge sort. More...

Detailed Description

Unlike other composers, this one doesn't need merge sort.

It's dumb simple.

Definition at line 52 of file sequential_composer_impl.cpp.

Collaboration diagram for foedus::storage::sequential::StreamStatus:

Public Member Functions
ErrorCode	init (snapshot::SortedBuffer *stream)

ErrorCode	next ()

ErrorCode	wind_stream ()

void	read_entry ()

const SequentialAppendLogType *	get_entry () const

Public Attributes
snapshot::SortedBuffer *	stream_

const char *	buffer_

uint64_t	buffer_size_

uint64_t	cur_absolute_pos_

uint64_t	cur_relative_pos_

uint64_t	end_absolute_pos_

uint32_t	cur_length_

const void *	cur_payload_

xct::XctId	cur_owner_id_

bool	ended_

Member Function Documentation

const SequentialAppendLogType* foedus::storage::sequential::StreamStatus::get_entry ( ) const

inline

Definition at line 107 of file sequential_composer_impl.cpp.

References buffer_, and cur_relative_pos_.

Referenced by foedus::storage::sequential::SequentialComposer::compose(), and read_entry().

                                                    {
     return reinterpret_cast<const SequentialAppendLogType*>(buffer_ + cur_relative_pos_);
   }

Here is the caller graph for this function:

ErrorCode foedus::storage::sequential::StreamStatus::init ( snapshot::SortedBuffer * stream )

inline

Definition at line 53 of file sequential_composer_impl.cpp.

References foedus::snapshot::SortedBuffer::assert_checks(), ASSERT_ND, buffer_, buffer_size_, CHECK_ERROR_CODE, cur_absolute_pos_, cur_length_, cur_relative_pos_, end_absolute_pos_, ended_, foedus::snapshot::SortedBuffer::get_buffer(), foedus::snapshot::SortedBuffer::get_buffer_size(), foedus::snapshot::SortedBuffer::get_cur_block_abosulte_begin(), foedus::snapshot::SortedBuffer::get_cur_block_abosulte_end(), foedus::snapshot::SortedBuffer::get_offset(), foedus::kErrorCodeOk, read_entry(), stream_, and wind_stream().

Referenced by foedus::storage::sequential::SequentialComposer::compose().

                                                {
     stream_ = stream;
     stream_->assert_checks();
     buffer_ = stream->get_buffer();
     buffer_size_ = stream->get_buffer_size();
     cur_absolute_pos_ = stream->get_cur_block_abosulte_begin();
     ASSERT_ND(cur_absolute_pos_ >= stream->get_offset());
     cur_relative_pos_ = cur_absolute_pos_ - stream->get_offset();
     end_absolute_pos_ = stream->get_cur_block_abosulte_end();
     ended_ = false;
     read_entry();
     if (cur_absolute_pos_ >= end_absolute_pos_) {
       ended_ = true;
     } else if (cur_relative_pos_ + cur_length_ > buffer_size_) {
       LOG(INFO) << "interesting, super unlucky case. wind immediately";
       CHECK_ERROR_CODE(wind_stream());
       read_entry();
       ASSERT_ND(cur_relative_pos_ + cur_length_ <= buffer_size_);
     }
     return kErrorCodeOk;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorCode foedus::storage::sequential::StreamStatus::next ( )

inline

Definition at line 74 of file sequential_composer_impl.cpp.

References ASSERT_ND, buffer_size_, CHECK_ERROR_CODE, cur_absolute_pos_, cur_length_, cur_relative_pos_, end_absolute_pos_, ended_, foedus::kErrorCodeOk, read_entry(), UNLIKELY, and wind_stream().

Referenced by foedus::storage::sequential::SequentialComposer::compose().

                    {
     ASSERT_ND(!ended_);
     ASSERT_ND(cur_relative_pos_ + cur_length_ <= buffer_size_);
     cur_absolute_pos_ += cur_length_;
     cur_relative_pos_ += cur_length_;
     if (UNLIKELY(cur_absolute_pos_ >= end_absolute_pos_)) {
       ended_ = true;
       return kErrorCodeOk;
     } else if (UNLIKELY(cur_relative_pos_ >= buffer_size_)) {
       CHECK_ERROR_CODE(wind_stream());
     }
     read_entry();
     if (UNLIKELY(cur_relative_pos_ + cur_length_ > buffer_size_)) {
       CHECK_ERROR_CODE(wind_stream());
       read_entry();
     }
 
     ASSERT_ND(cur_relative_pos_ + cur_length_ <= buffer_size_);
     return kErrorCodeOk;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::storage::sequential::StreamStatus::read_entry ( )

inline

Definition at line 99 of file sequential_composer_impl.cpp.

References ASSERT_ND, cur_length_, cur_owner_id_, cur_payload_, get_entry(), foedus::log::LogHeader::get_type(), foedus::log::BaseLogType::header_, foedus::log::kLogCodeSequentialAppend, foedus::log::LogHeader::log_length_, foedus::storage::sequential::SequentialAppendLogType::payload_, and foedus::log::LogHeader::xct_id_.

Referenced by init(), and next().

                     {
     const SequentialAppendLogType* entry = get_entry();
     ASSERT_ND(entry->header_.get_type() == log::kLogCodeSequentialAppend);
     ASSERT_ND(entry->header_.log_length_ > 0);
     cur_length_ = entry->header_.log_length_;
     cur_payload_ = entry->payload_;
     cur_owner_id_ = entry->header_.xct_id_;
   }

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorCode foedus::storage::sequential::StreamStatus::wind_stream ( )

inline

Definition at line 94 of file sequential_composer_impl.cpp.

References CHECK_ERROR_CODE, cur_absolute_pos_, cur_relative_pos_, foedus::kErrorCodeOk, stream_, foedus::snapshot::SortedBuffer::to_relative_pos(), and foedus::snapshot::SortedBuffer::wind().

Referenced by init(), and next().

                           {
     CHECK_ERROR_CODE(stream_->wind(cur_absolute_pos_));
     cur_relative_pos_ = stream_->to_relative_pos(cur_absolute_pos_);
     return kErrorCodeOk;
   }