szaszm commented on code in PR #1038:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1038#discussion_r925740372
##########
libminifi/include/utils/FlowFileQueue.h:
##########
@@ -29,22 +35,78 @@ namespace minifi {
namespace utils {
class FlowFileQueue {
+ friend struct ::FlowFileQueueTestAccessor;
+ using TimePoint = std::chrono::steady_clock::time_point;
+
public:
using value_type = std::shared_ptr<core::FlowFile>;
+ explicit FlowFileQueue(std::shared_ptr<SwapManager> swap_manager = {});
+
value_type pop();
- void push(const value_type& element);
- void push(value_type&& element);
+ std::optional<value_type> tryPop();
+ std::optional<value_type> tryPop(std::chrono::milliseconds timeout);
+ void push(value_type element);
bool isWorkAvailable() const;
bool empty() const;
size_t size() const;
+ void setMinSize(size_t min_size);
+ void setTargetSize(size_t target_size);
+ void setMaxSize(size_t max_size);
+ void clear();
private:
+ std::optional<value_type>
tryPopImpl(std::optional<std::chrono::milliseconds> timeout);
+
+ void initiateLoadIfNeeded();
+
+ struct LoadTask {
+ TimePoint min;
+ TimePoint max;
+ std::future<std::vector<std::shared_ptr<core::FlowFile>>> items;
+ size_t count;
+ // flow files that have been pushed into the queue while a
+ // load was pending
+ std::vector<value_type> intermediate_items;
+
+ LoadTask(TimePoint min, TimePoint max,
std::future<std::vector<std::shared_ptr<core::FlowFile>>> items, size_t count)
+ : min(min), max(max), items(std::move(items)), count(count) {}
+
+ size_t size() const {
+ return count + intermediate_items.size();
+ }
+ };
+
+ bool processLoadTaskWait(std::optional<std::chrono::milliseconds> timeout);
+
struct FlowFilePenaltyExpirationComparator {
- bool operator()(const value_type& left, const value_type& right);
+ bool operator()(const value_type& left, const value_type& right) const;
+ };
+
+ struct SwappedFlowFileComparator {
+ bool operator()(const SwappedFlowFile& left, const SwappedFlowFile& right)
const;
};
- std::priority_queue<value_type, std::vector<value_type>,
FlowFilePenaltyExpirationComparator> queue_;
+ size_t shouldSwapOutCount() const;
+
+ size_t shouldSwapInCount() const;
+
+ std::shared_ptr<SwapManager> swap_manager_;
Review Comment:
Does this need to own the flow file repo?
##########
libminifi/src/utils/FlowFileQueue.cpp:
##########
@@ -16,59 +16,232 @@
*/
#include "utils/FlowFileQueue.h"
+#include "core/logging/LoggerConfiguration.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
-bool FlowFileQueue::FlowFilePenaltyExpirationComparator::operator()(const
value_type& left, const value_type& right) {
- // this is operator< implemented using > so that top() is the element with
the smallest key (earliest expiration)
- // rather than the element with the largest key, which is the default for
std::priority_queue
- return left->getPenaltyExpiration() > right->getPenaltyExpiration();
+bool FlowFileQueue::FlowFilePenaltyExpirationComparator::operator()(const
value_type& left, const value_type& right) const {
+ // a flow file with earlier expiration compares less
+ return left->getPenaltyExpiration() < right->getPenaltyExpiration();
}
+bool FlowFileQueue::SwappedFlowFileComparator::operator()(const
SwappedFlowFile& left, const SwappedFlowFile& right) const {
+ // a swapped flow file with earlier expiration compares less
+ return left.to_be_processed_after < right.to_be_processed_after;
+}
+
+FlowFileQueue::FlowFileQueue(std::shared_ptr<SwapManager> swap_manager)
+ : swap_manager_(std::move(swap_manager)),
+ logger_(core::logging::LoggerFactory<FlowFileQueue>::getLogger()) {}
+
FlowFileQueue::value_type FlowFileQueue::pop() {
- if (empty()) {
- throw std::logic_error{"pop() called on an empty FlowFileQueue"};
- }
+ return tryPopImpl({}).value();
+}
- value_type next_flow_file = queue_.top();
- queue_.pop();
- return next_flow_file;
+std::optional<FlowFileQueue::value_type> FlowFileQueue::tryPop() {
+ return tryPopImpl(std::chrono::milliseconds{0});
}
-void FlowFileQueue::push(const value_type& element) {
- if (!element->isPenalized()) {
- element->penalize(std::chrono::milliseconds{0});
+std::optional<FlowFileQueue::value_type>
FlowFileQueue::tryPop(std::chrono::milliseconds timeout) {
+ return tryPopImpl(timeout);
+}
+
+std::optional<FlowFileQueue::value_type>
FlowFileQueue::tryPopImpl(std::optional<std::chrono::milliseconds> timeout) {
+ std::optional<std::shared_ptr<core::FlowFile>> result;
+ if (!queue_.empty()) {
+ result = queue_.popMin();
+ if (processLoadTaskWait(std::chrono::milliseconds{0})) {
+ initiateLoadIfNeeded();
+ }
+ return result;
+ }
+ if (load_task_) {
+ logger_->log_debug("Head is empty checking already running load task");
+ if (!processLoadTaskWait(timeout)) {
+ return std::nullopt;
+ }
+ if (!queue_.empty()) {
+ // load provided items
+ result = queue_.popMin();
+ initiateLoadIfNeeded();
+ return result;
+ }
}
+ // no pending load_task_ and no items in the queue_
+ initiateLoadIfNeeded();
+ return std::nullopt;
+}
- queue_.push(element);
+bool
FlowFileQueue::processLoadTaskWait(std::optional<std::chrono::milliseconds>
timeout) {
+ if (!load_task_) {
+ return true;
+ }
+ std::future_status status = std::future_status::ready;
+ if (timeout) {
+ status = load_task_.value().items.wait_for(timeout.value());
+ }
+ if (status == std::future_status::timeout) {
+ logger_->log_debug("Load task is not yet completed");
+ return false;
+ }
+ if (status != std::future_status::ready) {
+ throw std::logic_error("Unknown future status deferred future?");
+ }
Review Comment:
I would add a precondition for this, instead of throwing. It's not a runtime
issue, but a programming error.
##########
extensions/rocksdb-repos/FlowFileLoader.h:
##########
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <future>
+#include <list>
+#include <vector>
+#include <memory>
+
+#include "database/RocksDatabase.h"
+#include "FlowFile.h"
+#include "utils/gsl.h"
+#include "core/ContentRepository.h"
+#include "SwapManager.h"
+#include "utils/ThreadPool.h"
+#include "core/logging/Logger.h"
+
+namespace org::apache::nifi::minifi {
+
+class FlowFileLoader {
+ using FlowFilePtr = std::shared_ptr<core::FlowFile>;
+ using FlowFilePtrVec = std::vector<FlowFilePtr>;
+
+ static constexpr size_t thread_count_ = 2;
+
+ public:
+ FlowFileLoader();
+
+ ~FlowFileLoader();
+
+ void initialize(gsl::not_null<minifi::internal::RocksDatabase*> db,
std::shared_ptr<core::ContentRepository> content_repo);
Review Comment:
I would prefer late initialization with the constructor over two phase
initialization. It just cuts down the number of valid states here, and
simplifies logic.
Otherwise, `load` and `loadImpl` should declare preconditions that `db_` and
`content_repo` are valid. They have the precondition of `thread_pool_` running
either way, please add a declaration for that.
##########
extensions/rocksdb-repos/FlowFileLoader.h:
##########
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <future>
+#include <list>
+#include <vector>
+#include <memory>
+
+#include "database/RocksDatabase.h"
+#include "FlowFile.h"
+#include "utils/gsl.h"
+#include "core/ContentRepository.h"
+#include "SwapManager.h"
+#include "utils/ThreadPool.h"
+#include "core/logging/Logger.h"
+
+namespace org::apache::nifi::minifi {
+
+class FlowFileLoader {
+ using FlowFilePtr = std::shared_ptr<core::FlowFile>;
+ using FlowFilePtrVec = std::vector<FlowFilePtr>;
+
+ static constexpr size_t thread_count_ = 2;
Review Comment:
Is it necessary to use multiple threads for this purpose? Is it any faster?
##########
extensions/rocksdb-repos/FlowFileLoader.h:
##########
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <future>
+#include <list>
+#include <vector>
+#include <memory>
+
+#include "database/RocksDatabase.h"
+#include "FlowFile.h"
+#include "utils/gsl.h"
+#include "core/ContentRepository.h"
+#include "SwapManager.h"
+#include "utils/ThreadPool.h"
+#include "core/logging/Logger.h"
+
+namespace org::apache::nifi::minifi {
+
+class FlowFileLoader {
+ using FlowFilePtr = std::shared_ptr<core::FlowFile>;
+ using FlowFilePtrVec = std::vector<FlowFilePtr>;
+
+ static constexpr size_t thread_count_ = 2;
+
+ public:
+ FlowFileLoader();
+
+ ~FlowFileLoader();
+
+ void initialize(gsl::not_null<minifi::internal::RocksDatabase*> db,
std::shared_ptr<core::ContentRepository> content_repo);
+
+ void start();
+
+ void stop();
+
+ std::future<FlowFilePtrVec> load(std::vector<SwappedFlowFile> flow_files);
+
+ private:
+ utils::TaskRescheduleInfo loadImpl(const std::vector<SwappedFlowFile>&
flow_files, const std::shared_ptr<std::promise<FlowFilePtrVec>>& output);
+
+ utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_{thread_count_,
false, nullptr, "FlowFileLoaderThreadPool"};
+
+ minifi::internal::RocksDatabase* db_{nullptr};
+
+ std::shared_ptr<core::ContentRepository> content_repo_;
Review Comment:
This ownership may not be needed
##########
extensions/rocksdb-repos/FlowFileRepository.h:
##########
@@ -203,6 +205,24 @@ class FlowFileRepository : public core::Repository {
running_ = true;
thread_ = std::thread(&FlowFileRepository::run, this);
logger_->log_debug("%s Repository Monitor Thread Start", getName());
+ swap_loader.start();
+ }
+
+ void stop() override {
+ swap_loader.stop();
+ core::Repository::stop();
+ }
+
+ void store(std::vector<std::shared_ptr<core::FlowFile>> flow_files) override
{
+ for (auto& flow_file : flow_files) {
+ if (!flow_file->isStored()) {
+ throw Exception(FLOW_EXCEPTION, "A flow file that is being swapped out
is not stored in the flow repository");
+ }
+ }
Review Comment:
Consider replacing this with `gsl_Assert`/`gsl_AssertAudit`
##########
libminifi/include/utils/FlowFileQueue.h:
##########
@@ -29,22 +35,78 @@ namespace minifi {
namespace utils {
class FlowFileQueue {
+ friend struct ::FlowFileQueueTestAccessor;
+ using TimePoint = std::chrono::steady_clock::time_point;
+
public:
using value_type = std::shared_ptr<core::FlowFile>;
+ explicit FlowFileQueue(std::shared_ptr<SwapManager> swap_manager = {});
+
value_type pop();
- void push(const value_type& element);
- void push(value_type&& element);
+ std::optional<value_type> tryPop();
+ std::optional<value_type> tryPop(std::chrono::milliseconds timeout);
+ void push(value_type element);
bool isWorkAvailable() const;
bool empty() const;
size_t size() const;
+ void setMinSize(size_t min_size);
+ void setTargetSize(size_t target_size);
+ void setMaxSize(size_t max_size);
+ void clear();
private:
+ std::optional<value_type>
tryPopImpl(std::optional<std::chrono::milliseconds> timeout);
+
+ void initiateLoadIfNeeded();
+
+ struct LoadTask {
+ TimePoint min;
+ TimePoint max;
+ std::future<std::vector<std::shared_ptr<core::FlowFile>>> items;
+ size_t count;
+ // flow files that have been pushed into the queue while a
+ // load was pending
+ std::vector<value_type> intermediate_items;
+
+ LoadTask(TimePoint min, TimePoint max,
std::future<std::vector<std::shared_ptr<core::FlowFile>>> items, size_t count)
+ : min(min), max(max), items(std::move(items)), count(count) {}
+
+ size_t size() const {
+ return count + intermediate_items.size();
+ }
+ };
+
+ bool processLoadTaskWait(std::optional<std::chrono::milliseconds> timeout);
+
struct FlowFilePenaltyExpirationComparator {
- bool operator()(const value_type& left, const value_type& right);
+ bool operator()(const value_type& left, const value_type& right) const;
+ };
+
+ struct SwappedFlowFileComparator {
+ bool operator()(const SwappedFlowFile& left, const SwappedFlowFile& right)
const;
};
- std::priority_queue<value_type, std::vector<value_type>,
FlowFilePenaltyExpirationComparator> queue_;
+ size_t shouldSwapOutCount() const;
+
+ size_t shouldSwapInCount() const;
+
+ std::shared_ptr<SwapManager> swap_manager_;
+ // a load is initiated if the queue_ shrinks below this threshold
+ std::atomic<size_t> min_size_{0};
+ // a given operation (load/store) will try to approach this size
+ std::atomic<size_t> target_size_{0};
+ // a store is initiated if the queue_ grows beyond this threshold
+ std::atomic<size_t> max_size_{0};
+
+ MinMaxHeap<SwappedFlowFile, SwappedFlowFileComparator> swapped_flow_files_;
+ // the pending swap-in operation (if any)
+ std::optional<LoadTask> load_task_;
+ MinMaxHeap<value_type, FlowFilePenaltyExpirationComparator> queue_;
+
+ std::shared_ptr<timeutils::SteadyClock>
clock_{std::make_shared<timeutils::SteadyClock>()};
Review Comment:
Maybe this could be `unique_ptr`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
