From 1edf828f54e33f693a5d2181ddfb2d1bef356951 Mon Sep 17 00:00:00 2001 From: Eldar Akchurin Date: Fri, 5 Feb 2016 15:08:17 +0100 Subject: [PATCH] Introducing chunks --- .../ImageReader/ImageDataDeserializer.cpp | 129 +++++++++--------- .../ImageReader/ImageDataDeserializer.h | 10 +- .../Readers/ImageReader/ImageReader.vcxproj | 3 +- .../ImageReader/ImageReader.vcxproj.filters | 1 + Source/Readers/ImageReader/ImageSequence.h | 17 +++ .../Readers/ImageReader/ImageTransformers.cpp | 47 ++++--- .../Readers/ImageReader/ImageTransformers.h | 18 +-- Source/Readers/ReaderLib/BlockRandomizer.cpp | 29 ++-- Source/Readers/ReaderLib/BlockRandomizer.h | 2 +- Source/Readers/ReaderLib/DataDeserializer.h | 45 +++--- .../Readers/ReaderLib/DataDeserializerBase.h | 11 -- Source/Readers/ReaderLib/NoRandomizer.cpp | 41 +++++- Source/Readers/ReaderLib/NoRandomizer.h | 4 +- Source/Readers/ReaderLib/TransformerBase.h | 14 +- .../EndToEndTests/Image/AlexNet/testcases.yml | 4 +- 15 files changed, 209 insertions(+), 166 deletions(-) create mode 100644 Source/Readers/ImageReader/ImageSequence.h diff --git a/Source/Readers/ImageReader/ImageDataDeserializer.cpp b/Source/Readers/ImageReader/ImageDataDeserializer.cpp index c1106202b..284dbf740 100644 --- a/Source/Readers/ImageReader/ImageDataDeserializer.cpp +++ b/Source/Readers/ImageReader/ImageDataDeserializer.cpp @@ -11,6 +11,7 @@ #ifndef UNREFERENCED_PARAMETER #define UNREFERENCED_PARAMETER(P) (P) #endif +#include "ImageSequence.h" namespace Microsoft { namespace MSR { namespace CNTK { @@ -79,6 +80,68 @@ ImageDataDeserializer::ImageDataDeserializer(const ConfigParameters& config) CreateSequenceDescriptions(configHelper.GetMapPath(), labelDimension); } +class ImageDataDeserializer::ImageChunk : public Chunk, public std::enable_shared_from_this +{ + ImageSequenceDescription m_description; + ImageDataDeserializer& m_parent; + +public: + ImageChunk(ImageSequenceDescription& description, ImageDataDeserializer& parent) + : m_description(description), m_parent(parent) + { + } + + virtual std::vector GetSequence(const size_t& sequenceId) override + { + assert(sequenceId == m_description.m_id); + UNREFERENCED_PARAMETER(sequenceId); + const auto& imageSequence = m_description; + + auto image = std::make_shared(); + image->m_image = std::move(cv::imread(imageSequence.m_path, cv::IMREAD_COLOR)); + auto& cvImage = image->m_image; + + if (!cvImage.data) + { + RuntimeError("Cannot open file '%s'", imageSequence.m_path.c_str()); + } + + // Convert element type. + int dataType = m_parent.m_featureElementType == ElementType::tfloat ? CV_32F : CV_64F; + if (cvImage.type() != CV_MAKETYPE(dataType, cvImage.channels())) + { + cvImage.convertTo(cvImage, dataType); + } + + if (!cvImage.isContinuous()) + { + cvImage = cvImage.clone(); + } + assert(cvImage.isContinuous()); + + image->m_data = image->m_image.data; + ImageDimensions dimensions(cvImage.cols, cvImage.rows, cvImage.channels()); + image->m_sampleLayout = std::make_shared(dimensions.AsTensorShape(HWC)); + image->m_numberOfSamples = 1; + image->m_chunk = shared_from_this(); + + SparseSequenceDataPtr label = std::make_shared(); + label->m_chunk = shared_from_this(); + m_parent.m_labelGenerator->CreateLabelFor(imageSequence.m_classId, *label); + return std::vector { image, label }; + } + + ~ImageChunk() + { + } +}; + +ChunkPtr ImageDataDeserializer::GetChunk(size_t chunkId) +{ + auto sequenceDescription = m_imageSequences[chunkId]; + return std::make_shared(sequenceDescription, *this); +} + void ImageDataDeserializer::CreateSequenceDescriptions(std::string mapPath, size_t labelDimension) { UNREFERENCED_PARAMETER(labelDimension); @@ -127,72 +190,6 @@ std::vector ImageDataDeserializer::GetStreamDescriptions() return m_streams; } -std::vector> ImageDataDeserializer::GetSequencesById(const std::vector& ids) -{ - if (ids.empty()) - { - RuntimeError("Number of requested sequences cannot be zero."); - } - - m_currentImages.resize(ids.size()); - m_labels.resize(ids.size()); - - std::vector> result; - result.resize(ids.size()); - -#pragma omp parallel for ordered schedule(dynamic) - for (int i = 0; i < ids.size(); ++i) - { - if (ids[i] >= m_imageSequences.size()) - { - RuntimeError("Invalid sequence id is provided '%d', expected range [0..%d].", - static_cast(ids[i]), - static_cast(m_imageSequences.size()) - 1); - } - - const auto& imageSequence = m_imageSequences[ids[i]]; - - // Construct image - m_currentImages[i] = std::move(cv::imread(imageSequence.m_path, cv::IMREAD_COLOR)); - cv::Mat& cvImage = m_currentImages[i]; - - if (!cvImage.data) - { - RuntimeError("Cannot open file '%s'", imageSequence.m_path.c_str()); - } - - // Convert element type. - // TODO We should all native CV element types to be able to match the behavior of the old reader. - int dataType = m_featureElementType == ElementType::tfloat ? CV_32F : CV_64F; - if (cvImage.type() != CV_MAKETYPE(dataType, cvImage.channels())) - { - cvImage.convertTo(cvImage, dataType); - } - - if (!cvImage.isContinuous()) - { - cvImage = cvImage.clone(); - } - assert(cvImage.isContinuous()); - - ImageDimensions dimensions(cvImage.cols, cvImage.rows, cvImage.channels()); - auto image = std::make_shared(); - image->m_data = cvImage.data; - image->m_sampleLayout = std::make_shared(dimensions.AsTensorShape(HWC)); - image->m_numberOfSamples = 1; - - if (m_labels[i] == nullptr) - { - m_labels[i] = std::make_shared(); - } - - m_labelGenerator->CreateLabelFor(imageSequence.m_classId, *m_labels[i]); - result[i] = std::move(std::vector{image, m_labels[i]}); - } - - return result; -} - void ImageDataDeserializer::FillSequenceDescriptions(SequenceDescriptions& timeline) const { timeline.resize(m_imageSequences.size()); diff --git a/Source/Readers/ImageReader/ImageDataDeserializer.h b/Source/Readers/ImageReader/ImageDataDeserializer.h index b38d958ae..0ab0506f8 100644 --- a/Source/Readers/ImageReader/ImageDataDeserializer.h +++ b/Source/Readers/ImageReader/ImageDataDeserializer.h @@ -24,7 +24,7 @@ public: std::vector GetStreamDescriptions() const override; // Get sequences by specified ids. Order of returned sequences corresponds to the order of provided ids. - std::vector> GetSequencesById(const std::vector& ids) override; + virtual ChunkPtr GetChunk(size_t chunkId) override; protected: void FillSequenceDescriptions(SequenceDescriptions& timeline) const override; @@ -40,6 +40,8 @@ private: size_t m_classId; }; + class ImageChunk; + // A helper class for generation of type specific labels (currently float/double only). class LabelGenerator; typedef std::shared_ptr LabelGeneratorPtr; @@ -48,12 +50,6 @@ private: // Sequence descriptions for all input data. std::vector m_imageSequences; - // Buffer to store label data. - std::vector m_labels; - - // Buffer to store feature data. - std::vector m_currentImages; - // Element type of the feature/label stream (currently float/double only). ElementType m_featureElementType; }; diff --git a/Source/Readers/ImageReader/ImageReader.vcxproj b/Source/Readers/ImageReader/ImageReader.vcxproj index 3a0ea6192..5a8d4646e 100644 --- a/Source/Readers/ImageReader/ImageReader.vcxproj +++ b/Source/Readers/ImageReader/ImageReader.vcxproj @@ -113,6 +113,7 @@ + @@ -150,4 +151,4 @@ - + \ No newline at end of file diff --git a/Source/Readers/ImageReader/ImageReader.vcxproj.filters b/Source/Readers/ImageReader/ImageReader.vcxproj.filters index 948bf6554..0e2cea8ba 100644 --- a/Source/Readers/ImageReader/ImageReader.vcxproj.filters +++ b/Source/Readers/ImageReader/ImageReader.vcxproj.filters @@ -43,6 +43,7 @@ + diff --git a/Source/Readers/ImageReader/ImageSequence.h b/Source/Readers/ImageReader/ImageSequence.h new file mode 100644 index 000000000..a24e827c8 --- /dev/null +++ b/Source/Readers/ImageReader/ImageSequence.h @@ -0,0 +1,17 @@ +#pragma once + +#include +#include "DataDeserializer.h" + +namespace Microsoft { namespace MSR { namespace CNTK { + + // Used to keep track of the image. Used in the implementations of open cv transformers + // and deserializer. Accessed only using DenseSequenceData interface. + struct ImageSequenceData : DenseSequenceData + { + cv::Mat m_image; + + // In case we do not copy data - we preserve original sequence. + SequenceDataPtr m_original; + }; +}}} \ No newline at end of file diff --git a/Source/Readers/ImageReader/ImageTransformers.cpp b/Source/Readers/ImageReader/ImageTransformers.cpp index ae1d226c7..62b8b56a1 100644 --- a/Source/Readers/ImageReader/ImageTransformers.cpp +++ b/Source/Readers/ImageReader/ImageTransformers.cpp @@ -14,6 +14,7 @@ #include "ImageConfigHelper.h" #include "StringUtil.h" #include "ElementTypeUtils.h" +#include "ImageSequence.h" namespace Microsoft { namespace MSR { namespace CNTK { @@ -34,10 +35,11 @@ void ImageTransformerBase::Initialize(TransformerPtr next, } SequenceDataPtr -ImageTransformerBase::Apply(const DenseSequenceData &inputSequence, - const StreamDescription &inputStream, cv::Mat &buffer, +ImageTransformerBase::Apply(SequenceDataPtr sequence, + const StreamDescription &inputStream, const StreamDescription & /*outputStream*/) { + auto inputSequence = reinterpret_cast(*sequence.get()); ImageDimensions dimensions(*inputSequence.m_sampleLayout, HWC); int columns = static_cast(dimensions.m_width); int rows = static_cast(dimensions.m_height); @@ -57,20 +59,25 @@ ImageTransformerBase::Apply(const DenseSequenceData &inputSequence, RuntimeError("Unsupported type"); } + auto result = std::make_shared(); int type = CV_MAKETYPE(typeId, channels); - buffer = cv::Mat(rows, columns, type, inputSequence.m_data); + cv::Mat buffer = cv::Mat(rows, columns, type, inputSequence.m_data); this->Apply(buffer); if (!buffer.isContinuous()) { buffer = buffer.clone(); } + else + { + result->m_original = sequence; + } assert(buffer.isContinuous()); + result->m_image = buffer; + result->m_data = buffer.ptr(); + result->m_numberOfSamples = inputSequence.m_numberOfSamples; - auto result = std::make_shared(); ImageDimensions outputDimensions(buffer.cols, buffer.rows, buffer.channels()); result->m_sampleLayout = std::make_shared(outputDimensions.AsTensorShape(HWC)); - result->m_numberOfSamples = inputSequence.m_numberOfSamples; - result->m_data = buffer.ptr(); return result; } @@ -367,7 +374,7 @@ void MeanTransformer::Apply(cv::Mat &mat) void TransposeTransformer::Initialize(TransformerPtr next, const ConfigParameters &readerConfig) { - Base::Initialize(next, readerConfig); + TransformerBase::Initialize(next, readerConfig); // Currently we only support a single stream. ImageConfigHelper config(readerConfig); @@ -392,39 +399,46 @@ void TransposeTransformer::Initialize(TransformerPtr next, } SequenceDataPtr -TransposeTransformer::Apply(const DenseSequenceData &inputSequence, +TransposeTransformer::Apply(SequenceDataPtr inputSequence, const StreamDescription &inputStream, - vector &buffer, const StreamDescription &outputStream) { if (inputStream.m_elementType == ElementType::tdouble) { - return TypedApply(inputSequence, inputStream, buffer, outputStream); + return TypedApply(inputSequence, inputStream, outputStream); } if (inputStream.m_elementType == ElementType::tfloat) { - return TypedApply(inputSequence, inputStream, buffer, outputStream); + return TypedApply(inputSequence, inputStream, outputStream); } RuntimeError("Unsupported type"); } +struct OwnedDenseSequence : DenseSequenceData +{ + std::vector m_buffer; +}; + template SequenceDataPtr -TransposeTransformer::TypedApply(const DenseSequenceData &inputSequence, +TransposeTransformer::TypedApply(SequenceDataPtr sequence, const StreamDescription &inputStream, - vector &buffer, const StreamDescription &outputStream) { + assert(inputStream.m_storageType == StorageType::dense); + auto inputSequence = reinterpret_cast(*sequence.get()); assert(inputSequence.m_numberOfSamples == 1); assert(inputStream.m_sampleLayout->GetNumElements() == outputStream.m_sampleLayout->GetNumElements()); size_t count = inputStream.m_sampleLayout->GetNumElements() * GetSizeByType(inputStream.m_elementType); - buffer.resize(count); - TElement* typedBuffer = reinterpret_cast(&buffer[0]); + auto result = std::make_shared(); + result->m_buffer.resize(count); + + TElement* typedBuffer = reinterpret_cast(&result->m_buffer[0]); ImageDimensions dimensions(*inputStream.m_sampleLayout, ImageLayoutKind::HWC); size_t rowCount = dimensions.m_height * dimensions.m_width; @@ -441,9 +455,8 @@ TransposeTransformer::TypedApply(const DenseSequenceData &inputSequence, } } - auto result = std::make_shared(); result->m_sampleLayout = outputStream.m_sampleLayout; - result->m_data = &buffer[0]; + result->m_data = &result->m_buffer[0]; result->m_numberOfSamples = inputSequence.m_numberOfSamples; return result; } diff --git a/Source/Readers/ImageReader/ImageTransformers.h b/Source/Readers/ImageReader/ImageTransformers.h index ba81f0a4c..d70e4b67a 100644 --- a/Source/Readers/ImageReader/ImageTransformers.h +++ b/Source/Readers/ImageReader/ImageTransformers.h @@ -20,7 +20,7 @@ class ConfigParameters; // Base class for image transformations based on OpenCV // that helps to wrap the sequences into OpenCV::Mat class. -class ImageTransformerBase : public TransformerBase +class ImageTransformerBase : public TransformerBase { public: // Initializes the transformer. @@ -44,13 +44,13 @@ protected: return m_seed; } - using Base = TransformerBase; + using Base = TransformerBase; using UniRealT = std::uniform_real_distribution; using UniIntT = std::uniform_int_distribution; // Applies transformation to the sequence. - SequenceDataPtr Apply(const DenseSequenceData &inputSequence, - const StreamDescription &inputStream, cv::Mat &buffer, + SequenceDataPtr Apply(SequenceDataPtr inputSequence, + const StreamDescription &inputStream, const StreamDescription &outputStream) override; // The only function that should be redefined by the inherited classes. @@ -139,7 +139,7 @@ private: }; // Transpose transformation from HWC to CHW. -class TransposeTransformer : public TransformerBase> +class TransposeTransformer : public TransformerBase { public: virtual void Initialize(TransformerPtr next, @@ -156,18 +156,14 @@ protected: return m_outputStreams; } - SequenceDataPtr Apply(const DenseSequenceData &inputSequence, + SequenceDataPtr Apply(SequenceDataPtr inputSequence, const StreamDescription &inputStream, - vector &buffer, const StreamDescription &outputStream) override; private: - using Base = TransformerBase>; - template - SequenceDataPtr TypedApply(const DenseSequenceData &inputSequence, + SequenceDataPtr TypedApply(SequenceDataPtr inputSequence, const StreamDescription &inputStream, - vector &buffer, const StreamDescription &outputStream); std::vector m_outputStreams; diff --git a/Source/Readers/ReaderLib/BlockRandomizer.cpp b/Source/Readers/ReaderLib/BlockRandomizer.cpp index 476486bfb..9a55e1569 100644 --- a/Source/Readers/ReaderLib/BlockRandomizer.cpp +++ b/Source/Readers/ReaderLib/BlockRandomizer.cpp @@ -296,8 +296,6 @@ void BlockRandomizer::Initialize(TransformerPtr next, const ConfigParameters& re void BlockRandomizer::StartEpoch(const EpochConfiguration& config) { - m_deserializer->StartEpoch(config); - m_workerRank = config.m_workerRank; m_numberOfWorkers = config.m_numberOfWorkers; @@ -319,10 +317,10 @@ void BlockRandomizer::StartEpoch(const EpochConfiguration& config) RandomizeForGlobalSamplePosition(timeframe); }; -bool BlockRandomizer::GetNextSequenceIds(size_t sampleCount, std::vector& originalIds) +bool BlockRandomizer::GetNextSequenceIds(size_t sampleCount, SequenceDescriptions& sequenceDescriptions) { assert(m_frameMode); // TODO !m_frameMode not implemented yet - assert(originalIds.size() == 0); + assert(sequenceDescriptions.size() == 0); assert(sampleCount < m_numSamples); if (m_samplePositionInEpoch < m_epochSize) @@ -332,7 +330,7 @@ bool BlockRandomizer::GetNextSequenceIds(size_t sampleCount, std::vector assert(m_numberOfWorkers == 1); // TODO needs implementation while ((m_samplePositionInEpoch < m_epochSize) && - (originalIds.size() < sampleCount)) + (sequenceDescriptions.size() < sampleCount)) { RandomizeIfNewSweepIsEntered(); @@ -340,7 +338,7 @@ bool BlockRandomizer::GetNextSequenceIds(size_t sampleCount, std::vector if ((seqDesc.m_chunkId % m_numberOfWorkers) == m_workerRank) { // Got one, collect it - originalIds.push_back(seqDesc.m_id); + sequenceDescriptions.push_back(m_deserializer->GetSequenceDescriptions()[seqDesc.m_id]); } m_samplePositionInEpoch += seqDesc.m_numberOfSamples; @@ -362,7 +360,7 @@ bool BlockRandomizer::GetNextSequenceIds(size_t sampleCount, std::vector if (strideBegin <= i && i < strideEnd) { const auto& seqDesc = m_randomTimeline[m_sequencePositionInSweep]; - originalIds.push_back(seqDesc.m_id); + sequenceDescriptions.push_back(m_deserializer->GetSequenceDescriptions()[seqDesc.m_id]); } } assert(m_samplePositionInEpoch == nextSamplePositionInEpoch); @@ -376,14 +374,13 @@ Sequences BlockRandomizer::GetNextSequences(size_t sampleCount) { assert(m_samplePositionInEpoch != SIZE_MAX); // SetEpochConfiguration() must be called first - std::vector originalIds; Sequences result; - assert(m_frameMode); // TODO sequence mode not implemented yet - result.m_endOfEpoch = GetNextSequenceIds(sampleCount, originalIds); + SequenceDescriptions sequenceDescriptions; + result.m_endOfEpoch = GetNextSequenceIds(sampleCount, sequenceDescriptions); - if (originalIds.size() == 0) + if (sequenceDescriptions.size() == 0) { return result; } @@ -391,7 +388,15 @@ Sequences BlockRandomizer::GetNextSequences(size_t sampleCount) // TODO implement require and release chunks from the data deserializer, but only for this worker // (probably in GetNextSequenceIds()) - result.m_data = m_deserializer->GetSequencesById(originalIds); + result.m_data.resize(sequenceDescriptions.size()); + +#pragma omp parallel for ordered schedule(dynamic) + for (int i = 0; i < sequenceDescriptions.size(); ++i) + { + ChunkPtr chunk = m_deserializer->GetChunk(sequenceDescriptions[i]->m_chunkId); + result.m_data[i] = chunk->GetSequence(sequenceDescriptions[i]->m_id); + } + return result; }; diff --git a/Source/Readers/ReaderLib/BlockRandomizer.h b/Source/Readers/ReaderLib/BlockRandomizer.h index 6ad9ff69f..31c5abd06 100644 --- a/Source/Readers/ReaderLib/BlockRandomizer.h +++ b/Source/Readers/ReaderLib/BlockRandomizer.h @@ -99,6 +99,6 @@ private: void RandomizeIfNewSweepIsEntered(); - bool GetNextSequenceIds(size_t sampleCount, std::vector& ids); + bool GetNextSequenceIds(size_t sampleCount, SequenceDescriptions& originalIds); }; } } } diff --git a/Source/Readers/ReaderLib/DataDeserializer.h b/Source/Readers/ReaderLib/DataDeserializer.h index 2beb2c739..469ff415f 100644 --- a/Source/Readers/ReaderLib/DataDeserializer.h +++ b/Source/Readers/ReaderLib/DataDeserializer.h @@ -10,6 +10,25 @@ namespace Microsoft { namespace MSR { namespace CNTK { +struct SequenceDataBase; +typedef std::shared_ptr SequenceDataPtr; + +class Chunk +{ +public: + // Gets sequences by id. + virtual std::vector GetSequence(const size_t& sequenceId) = 0; + virtual ~Chunk() {}; + +protected: + Chunk() {} + +private: + Chunk(const Chunk&) = delete; + Chunk& operator=(const Chunk&) = delete; +}; +typedef std::shared_ptr ChunkPtr; + // Defines main properties of a sequence. // Sequence descriptions are used by the randomizer to establish a global timeline for complete input. // A sequence is defined as an ordered set of samples (size == 1 is used for sample training). @@ -31,7 +50,12 @@ typedef std::vector SequenceDescriptions; struct SequenceDataBase { SequenceDataBase() : m_data(nullptr) { } + virtual ~SequenceDataBase() + { + + } + ChunkPtr m_chunk; // A non-owned pointer. The actual size is provided for particular sequences, // i.e. see DenseSequenceData, or SparseSequenceData. void* m_data; @@ -79,25 +103,8 @@ public: // Retrieves description of all sequences this data deserializer can produce. virtual const SequenceDescriptions& GetSequenceDescriptions() const = 0; - // Sets epoch configuration. - virtual void StartEpoch(const EpochConfiguration& config) = 0; - - // Gets sequences by id. - // The return value can be used until the next call to GetSequencesById. - // All non-owned pointers returned are valid till the next call to this method. - virtual std::vector> GetSequencesById(const std::vector& ids) = 0; - - // Requires the chunk. Each sequence is assigned to the IO chunk by the data deserializer. - // This information is communicated thru GetSequenceDescriptions method. - // The randomizer guarantees that it accesses sequences only from a limited number of chunks. - // When randomizer requires a sequence from a particular chunk it notifies about this the data deserializer, - // so that the data deserializer can load/cache sequences more efficiently (loading complete chunks in memory). - virtual void RequireChunk(size_t chunkIndex) = 0; - - // Releases the chunk. - // When randomizer read all sequences from a particular chunk it notifies the data deserializer - // that the chunk can be freed. - virtual void ReleaseChunk(size_t chunkIndex) = 0; + // Gets a chunk. + virtual ChunkPtr GetChunk(size_t chunkId) = 0; virtual ~DataDeserializer() {}; }; diff --git a/Source/Readers/ReaderLib/DataDeserializerBase.h b/Source/Readers/ReaderLib/DataDeserializerBase.h index d2f537d47..11d6cb3c4 100644 --- a/Source/Readers/ReaderLib/DataDeserializerBase.h +++ b/Source/Readers/ReaderLib/DataDeserializerBase.h @@ -17,9 +17,6 @@ public: DataDeserializerBase() : m_sequencesInitialized(false) {} - // Sets configuration for the current epoch. - void StartEpoch(const EpochConfiguration& /*config*/) override {}; - // Provides description of all sequences the deserializer can produce. const SequenceDescriptions& GetSequenceDescriptions() const override { @@ -31,14 +28,6 @@ public: return m_sequences; } - // To be called by the randomizer for prefetching the next chunk. - // By default IO read-ahead is not implemented. - void RequireChunk(size_t /*chunkIndex*/) override{}; - - // To be called by the randomizer for releasing a prefetched chunk. - // By default IO read-ahead is not implemented. - void ReleaseChunk(size_t /*chunkIndex*/) override{}; - protected: // Fills the timeline with sequence descriptions. // Inherited classes should provide the complete Sequence descriptions for all input data. diff --git a/Source/Readers/ReaderLib/NoRandomizer.cpp b/Source/Readers/ReaderLib/NoRandomizer.cpp index 245855ec1..9c6c8830a 100644 --- a/Source/Readers/ReaderLib/NoRandomizer.cpp +++ b/Source/Readers/ReaderLib/NoRandomizer.cpp @@ -35,7 +35,6 @@ void NoRandomizer::Initialize(TransformerPtr, const ConfigParameters&) void NoRandomizer::StartEpoch(const EpochConfiguration& config) { - m_deserializer->StartEpoch(config); m_config = config; if (m_config.m_totalEpochSizeInSamples == requestDataSize) @@ -62,24 +61,54 @@ Sequences NoRandomizer::GetNextSequences(size_t sampleCount) size_t end = maxSampleCount * (m_config.m_workerRank + 1) / m_config.m_numberOfWorkers; size_t subsetSize = end - start; - std::vector originalIds; - originalIds.reserve(subsetSize); + std::vector chunkIds; + SequenceDescriptions sequences; + sequences.reserve(subsetSize); + size_t previousChunk = std::numeric_limits::max(); for (size_t i = start; i < end; ++i) { const auto& sequence = m_timeline[(m_sequencePosition + i) % m_timeline.size()]; assert(sequence->m_numberOfSamples == 1); - originalIds.push_back(sequence->m_id); + sequences.push_back(sequence); + + if (previousChunk != sequence->m_chunkId) + { + chunkIds.push_back(sequence->m_chunkId); + previousChunk = sequence->m_chunkId; + } } m_samplePositionInEpoch += maxSampleCount; m_sequencePosition = (m_sequencePosition + maxSampleCount) % m_timeline.size(); - if (originalIds.size() == 0) + if (sequences.size() == 0) { return result; } - result.m_data = m_deserializer->GetSequencesById(originalIds); + std::map chunks; + for (size_t id : chunkIds) + { + auto chunk = m_chunks.find(id); + if (chunk == m_chunks.end()) + { + chunks[id] = m_deserializer->GetChunk(id); + } + else + { + chunks[id] = chunk->second; + } + } + + m_chunks.swap(chunks); + + result.m_data.resize(sequences.size()); + +#pragma omp parallel for ordered schedule(dynamic) + for (int i = 0; i < sequences.size(); ++i) + { + result.m_data[i] = m_chunks[sequences[i]->m_chunkId]->GetSequence(sequences[i]->m_id); + } return result; } diff --git a/Source/Readers/ReaderLib/NoRandomizer.h b/Source/Readers/ReaderLib/NoRandomizer.h index 70d5ea300..57f119edb 100644 --- a/Source/Readers/ReaderLib/NoRandomizer.h +++ b/Source/Readers/ReaderLib/NoRandomizer.h @@ -6,7 +6,7 @@ #pragma once #include - +#include #include "Transformer.h" #include "DataDeserializer.h" @@ -41,6 +41,8 @@ private: EpochConfiguration m_config; size_t m_samplePositionInEpoch; size_t m_sequencePosition; + + std::map m_chunks; }; }}} diff --git a/Source/Readers/ReaderLib/TransformerBase.h b/Source/Readers/ReaderLib/TransformerBase.h index 90805896f..256b38088 100644 --- a/Source/Readers/ReaderLib/TransformerBase.h +++ b/Source/Readers/ReaderLib/TransformerBase.h @@ -12,8 +12,6 @@ namespace Microsoft { namespace MSR { namespace CNTK { -// Currently supports only dense data format. -template class TransformerBase : public Transformer { public: @@ -48,7 +46,6 @@ public: const auto &appliedStreamIds = GetAppliedStreamIds(); const auto &outputStreams = GetOutputStreams(); assert(m_inputStreams.size() == outputStreams.size()); - m_buffer.resize(samples.m_data.size()); #pragma omp parallel for ordered schedule(dynamic) for (int i = 0; i < samples.m_data.size(); ++i) @@ -56,15 +53,10 @@ public: auto &sample = samples.m_data[i]; assert(sample.size() == m_inputStreams.size()); - m_buffer[i].resize(appliedStreamIds.size()); for (int j = 0; j < appliedStreamIds.size(); ++j) { size_t id = appliedStreamIds[j]; - assert(m_inputStreams[id]->m_storageType == StorageType::dense); - const DenseSequenceData &sequence = - reinterpret_cast(*sample[id]); - sample[id] = Apply(sequence, *m_inputStreams[id], m_buffer[i][j], - *outputStreams[id]); + sample[id] = Apply(sample[id], *m_inputStreams[id], *outputStreams[id]); } } @@ -85,14 +77,12 @@ protected: private: // Applies transformation to the sequence. - virtual SequenceDataPtr Apply(const DenseSequenceData &inputSequence, + virtual SequenceDataPtr Apply(SequenceDataPtr inputSequence, const StreamDescription &inputStream, - TBufferElement &buffer, const StreamDescription &outputStream) = 0; TransformerPtr m_next; std::vector m_featureStreamIds; - std::vector> m_buffer; std::vector m_inputStreams; }; diff --git a/Tests/EndToEndTests/Image/AlexNet/testcases.yml b/Tests/EndToEndTests/Image/AlexNet/testcases.yml index 8cc7a4340..d6a9f315c 100644 --- a/Tests/EndToEndTests/Image/AlexNet/testcases.yml +++ b/Tests/EndToEndTests/Image/AlexNet/testcases.yml @@ -1,9 +1,9 @@ dataDir: ../Data tags: # running on every BVT job in 'I' (Image) leg: - - bvt-i (build_sku == 'gpu') and (device=='gpu') and (flavor=='release') and (os=='linux') + - bvt-i (build_sku == 'gpu') and (device=='gpu') and (flavor=='release') # running every Nightly job in 'I' leg - - nightly-i (build_sku == 'gpu') and (device=='gpu') and (flavor=='release') and (os=='linux') + - nightly-i (build_sku == 'gpu') and (device=='gpu') and (flavor=='release') testCases: CNTK Run must be completed: