Copilot commented on code in PR #12370:
URL: https://github.com/apache/gluten/pull/12370#discussion_r3537631788
##########
docs/velox-configuration.md:
##########
@@ -35,6 +35,8 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.floatingPointMode
| 🔄 Dynamic | loose | Config used to control the tolerance
of floating point operations alignment with Spark. When the mode is set to
strict, flushing is disabled for sum(float/double)and avg(float/double). When
set to loose, flushing will be enabled.
|
| spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation
| 🔄 Dynamic | true | Enable flushable aggregation. If true,
Gluten will try converting regular aggregation into Velox's flushable
aggregation when applicable. A flushable aggregation could emit intermediate
result at anytime when memory is full / data reduction ratio is low.
|
| spark.gluten.sql.columnar.backend.velox.footerEstimatedSize
| âš“ Static | 32KB | Set the footer estimated size for
velox file scan, refer to Velox's footer-estimated-size
|
+| spark.gluten.sql.columnar.backend.velox.gpuShuffleReader.enableAsync
| âš“ Static | true | Experimental: Enable GPU async
shuffle reader. When true, the gpu shuffle reader will use a thread pool to
read and deserialize the input streams. When false, the shuffle reader will
execute in the current thread.
|
+| spark.gluten.sql.columnar.backend.velox.gpuShuffleReader.threadPoolSize
| âš“ Static | 1 | The number of threads used by GPU
async shuffle reader for decompressing and deserializing input streams.
|
Review Comment:
The documented config keys and defaults for GPU async shuffle reader don't
match the actual Spark / Velox config entries. Code uses
`gpuAsyncShuffleReader.enabled` (dynamic, default false) and
`gpuAsyncShuffleReader.threadPoolSize` (static, default 1), plus
`gpuAsyncShuffleReader.maxPrefetchBytes` (dynamic, default 1GB). As written,
users will set the wrong keys and get unexpected behavior.
##########
cpp/velox/shuffle/VeloxShuffleReader.cc:
##########
@@ -1020,18 +1061,21 @@ void
VeloxShuffleReaderDeserializerFactory::initFromSchema() {
}
}
-VeloxShuffleReader::VeloxShuffleReader(std::unique_ptr<VeloxShuffleReaderDeserializerFactory>
factory)
- : factory_(std::move(factory)) {}
-
std::shared_ptr<ResultIterator> VeloxShuffleReader::read(const
std::shared_ptr<StreamReader>& streamReader) {
- return
std::make_shared<ResultIterator>(factory_->createDeserializer(streamReader));
+ // TODO: Support reader priority for async reader.
+ createDeserializer(streamReader);
+ return std::make_shared<ResultIterator>(deserializer_->deserializeStreams());
}
int64_t VeloxShuffleReader::getDecompressTime() const {
- return factory_->getDecompressTime();
+ return decompressTime_;
}
int64_t VeloxShuffleReader::getDeserializeTime() const {
- return factory_->getDeserializeTime();
+ return deserializeTime_;
+}
+
+void VeloxShuffleReader::stop() {
+ deserializer_->stop();
}
Review Comment:
`VeloxShuffleReader::stop()` unconditionally dereferences `deserializer_`.
If `stop()` is called before `read()` (or after a failed
`createDeserializer()`), this will crash. Make `stop()` a no-op when the
deserializer hasn't been created yet.
##########
cpp/velox/shuffle/ReaderThreadPool.cc:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/ReaderThreadPool.h"
+#include <glog/logging.h>
+
+namespace gluten {
+
+ReaderThreadPool::ReaderThreadPool(size_t numThreads) :
numThreads_(numThreads) {
+ workers_.reserve(numThreads);
+ for (size_t i = 0; i < numThreads; ++i) {
+ workers_.emplace_back([this]() { workerThread(); });
+ }
+ LOG(WARNING) << "Created ReaderThreadPool with " << numThreads << "
threads.";
+}
+
+ReaderThreadPool::~ReaderThreadPool() {
+ shutdown();
+}
+
+void ReaderThreadPool::submitBatch(std::vector<Task> tasks, int32_t priority) {
+ std::lock_guard<std::mutex> lock(taskQueueMtx_);
+ if (stop_.load(std::memory_order_acquire)) {
+ return;
+ }
+ for (auto& task : tasks) {
+ tasks_.push({std::move(task), priority});
+ }
+}
+
+void ReaderThreadPool::start() {
+ // Wake up all worker threads to start processing.
+ wakeUpCV_.notify_all();
+ LOG(WARNING) << "Started ReaderThreadPool execution.";
Review Comment:
Starting the ReaderThreadPool is part of normal operation, but it's logged
at WARNING level. This is likely to create unnecessary noise in production
logs; INFO (or VLOG) would be more appropriate.
##########
cpp/velox/shuffle/ReaderThreadPool.cc:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/ReaderThreadPool.h"
+#include <glog/logging.h>
+
+namespace gluten {
+
+ReaderThreadPool::ReaderThreadPool(size_t numThreads) :
numThreads_(numThreads) {
+ workers_.reserve(numThreads);
+ for (size_t i = 0; i < numThreads; ++i) {
+ workers_.emplace_back([this]() { workerThread(); });
+ }
+ LOG(WARNING) << "Created ReaderThreadPool with " << numThreads << "
threads.";
Review Comment:
Creating the ReaderThreadPool is part of normal operation, but it's logged
at WARNING level. This is likely to create unnecessary noise in production
logs; INFO (or VLOG) would be more appropriate.
##########
cpp/velox/shuffle/ReaderThreadPool.cc:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/ReaderThreadPool.h"
+#include <glog/logging.h>
+
+namespace gluten {
+
+ReaderThreadPool::ReaderThreadPool(size_t numThreads) :
numThreads_(numThreads) {
+ workers_.reserve(numThreads);
+ for (size_t i = 0; i < numThreads; ++i) {
+ workers_.emplace_back([this]() { workerThread(); });
+ }
+ LOG(WARNING) << "Created ReaderThreadPool with " << numThreads << "
threads.";
+}
+
+ReaderThreadPool::~ReaderThreadPool() {
+ shutdown();
+}
+
+void ReaderThreadPool::submitBatch(std::vector<Task> tasks, int32_t priority) {
+ std::lock_guard<std::mutex> lock(taskQueueMtx_);
+ if (stop_.load(std::memory_order_acquire)) {
+ return;
+ }
+ for (auto& task : tasks) {
+ tasks_.push({std::move(task), priority});
+ }
+}
+
+void ReaderThreadPool::start() {
+ // Wake up all worker threads to start processing.
+ wakeUpCV_.notify_all();
+ LOG(WARNING) << "Started ReaderThreadPool execution.";
+}
+
+void ReaderThreadPool::shutdown() {
+ if (!isShutdown()) {
+ stop_.store(true, std::memory_order_release);
+ wakeUpCV_.notify_all();
+
+ // Wait for all worker threads to finish their current tasks and join.
+ for (auto& worker : workers_) {
+ if (worker.joinable()) {
+ worker.join();
+ }
+ }
+ }
+}
+
+void ReaderThreadPool::workerThread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(taskQueueMtx_);
+
+ wakeUpCV_.wait(lock, [this]() { return
stop_.load(std::memory_order_acquire) || !tasks_.empty(); });
+
+ if (stop_.load(std::memory_order_acquire)) {
+ // Discard remaining tasks and exit the thread.
+ return;
+ }
+ }
+
+ while (true) {
+ Task task;
+ {
+ std::lock_guard<std::mutex> lock(taskQueueMtx_);
+ if (tasks_.empty()) {
+ break;
+ }
+ auto& prioritizedTask = tasks_.top();
+ LOG(INFO) << "Worker thread " << std::this_thread::get_id() << " is
executing a task with priority "
+ << prioritizedTask.priority;
Review Comment:
Logging once per executed task at INFO level inside the worker loop can be
extremely noisy and can materially affect performance under load. Consider
downgrading this to a debug-only log (e.g., `DLOG` / `VLOG`).
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
