diff options
author | Martin Neupauer <martin.neupauer@10gen.com> | 2018-06-20 13:33:34 -0400 |
---|---|---|
committer | Martin Neupauer <martin.neupauer@mongodb.com> | 2018-07-06 15:02:32 -0400 |
commit | fda766f6be1a20fa28ce361511bc62e5c995186b (patch) | |
tree | 4289bb82ff7ce5ca55bc9f33915ff3205431dedf /src/mongo/db/pipeline/document_source_exchange.h | |
parent | 5b2739dbff77811dbbfbccbc8a7ca8b973c8525f (diff) | |
download | mongo-fda766f6be1a20fa28ce361511bc62e5c995186b.tar.gz |
SERVER-35894 The initial implementation of the producer document source
for the exchange operator.
SERVER-35940 Remove IntrusiveCounter in favor of RefCountable.
Diffstat (limited to 'src/mongo/db/pipeline/document_source_exchange.h')
-rw-r--r-- | src/mongo/db/pipeline/document_source_exchange.h | 172 |
1 files changed, 172 insertions, 0 deletions
diff --git a/src/mongo/db/pipeline/document_source_exchange.h b/src/mongo/db/pipeline/document_source_exchange.h new file mode 100644 index 00000000000..3457cf6e4fa --- /dev/null +++ b/src/mongo/db/pipeline/document_source_exchange.h @@ -0,0 +1,172 @@ +/** + * Copyright (C) 2018 MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License, version 3, + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the GNU Affero General Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#pragma once + +#include <deque> +#include <vector> + +#include "mongo/db/pipeline/document_source.h" +#include "mongo/db/pipeline/document_source_exchange_gen.h" +#include "mongo/stdx/condition_variable.h" +#include "mongo/stdx/mutex.h" + +namespace mongo { + +class Exchange : public RefCountable { + static constexpr size_t kInvalidThreadId{std::numeric_limits<size_t>::max()}; + + static std::vector<std::string> extractBoundaries( + const boost::optional<std::vector<BSONObj>>& obj); + +public: + explicit Exchange(const ExchangeSpec& spec); + DocumentSource::GetNextResult getNext(size_t consumerId); + + size_t getConsumers() const { + return _consumers.size(); + } + + void setSource(DocumentSource* source) { + pSource = source; + } + + const auto& getSpec() const { + return _spec; + } + +private: + size_t loadNextBatch(); + + size_t getTargetConsumer(const Document& input); + + class ExchangeBuffer { + public: + bool appendDocument(DocumentSource::GetNextResult input, size_t limit); + DocumentSource::GetNextResult getNext(); + bool isEmpty() const { + return _buffer.empty(); + } + + private: + size_t _bytesInBuffer{0}; + std::deque<DocumentSource::GetNextResult> _buffer; + }; + + // Keep a copy of the spec for serialization purposes. + const ExchangeSpec _spec; + + // A pattern for extracting a key from a document used by range and hash policies. + const BSONObj _keyPattern; + + // Range boundaries. + const std::vector<std::string> _boundaries; + + // A policy that tells how to distribute input documents to consumers. + const ExchangePolicyEnum _policy; + + // If set to true then a producer sends special 'high watermark' documents to consumers in order + // to prevent deadlocks. + const bool _orderPreserving; + + // A maximum size of buffer per consumer. + const size_t _maxBufferSize; + + // An input to the exchange operator + DocumentSource* pSource; + + // Synchronization. + stdx::mutex _mutex; + stdx::condition_variable _haveBufferSpace; + + // A thread that is currently loading the exchange buffers. + size_t _loadingThreadId{kInvalidThreadId}; + + size_t _roundRobinCounter{0}; + + std::vector<std::unique_ptr<ExchangeBuffer>> _consumers; +}; + +class DocumentSourceExchange final : public DocumentSource, public NeedsMergerDocumentSource { +public: + static boost::intrusive_ptr<DocumentSource> createFromBson( + BSONElement spec, const boost::intrusive_ptr<ExpressionContext>& pExpCtx); + + DocumentSourceExchange(const boost::intrusive_ptr<ExpressionContext>& expCtx, + const boost::intrusive_ptr<Exchange> exchange, + size_t consumerId); + + GetNextResult getNext() final; + + StageConstraints constraints(Pipeline::SplitState pipeState) const final { + return {StreamType::kStreaming, + PositionRequirement::kNone, + HostTypeRequirement::kNone, + DiskUseRequirement::kNoDiskUse, + FacetRequirement::kAllowed, + TransactionRequirement::kAllowed}; + } + + const char* getSourceName() const final; + + Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final; + + boost::intrusive_ptr<DocumentSource> getShardSource() final { + return this; + } + std::list<boost::intrusive_ptr<DocumentSource>> getMergeSources() final { + // TODO SERVER-35974 we have to revisit this when we implement consumers. + return {this}; + } + + /** + * Set the underlying source this source should use to get Documents from. Must not throw + * exceptions. + */ + void setSource(DocumentSource* source) final { + DocumentSource::setSource(source); + _exchange->setSource(source); + } + + GetNextResult getNext(size_t consumerId); + + size_t getConsumers() const { + return _exchange->getConsumers(); + } + + auto getExchange() const { + return _exchange; + } + +private: + boost::intrusive_ptr<Exchange> _exchange; + + const size_t _consumerId; +}; + +} // namespace mongo |