This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new c0f1119899 [VL] Distinguish driver and executor defaults for the
broadcast hash-table build thread pool (#13368)
c0f1119899 is described below
commit c0f111989951083287ca61f2b62883c53333a96b
Author: Kaifei Yi <[email protected]>
AuthorDate: Thu Oct 29 15:28:27 2026 +0800
[VL] Distinguish driver and executor defaults for the broadcast hash-table
build thread pool (#13368)
---
.../backendsapi/velox/VeloxListenerApi.scala | 16 +++
.../org/apache/gluten/config/VeloxConfig.scala | 9 +-
.../joins/GlutenParallelBroadcastJoinSuite.scala | 134 ++++++++++++++++-----
cpp/velox/compute/VeloxBackend.cc | 14 ++-
cpp/velox/compute/VeloxBackend.h | 3 +-
cpp/velox/config/VeloxConfig.h | 8 +-
cpp/velox/jni/VeloxJniWrapper.cc | 18 +--
cpp/velox/tests/VeloxBackendTest.cc | 41 ++++++-
docs/backends/velox/velox-configuration.md | 2 +-
9 files changed, 190 insertions(+), 55 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
index 597882479d..a48ea8fcb6 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
@@ -312,6 +312,22 @@ object VeloxListenerApi {
var parsed: Map[String, String] =
GlutenConfigUtil.parseConfig(conf.getAll.toMap)
+ val broadcastBuildThreadsKey =
COLUMNAR_VELOX_BROADCAST_HASH_TABLE_BUILD_THREADS.key
+ if (!parsed.contains(broadcastBuildThreadsKey)) {
+ val defaultThreads = if (isDriver) {
+ if (inLocalMode(conf)) {
+ SparkResourceUtil.getExecutorCores(conf)
+ } else {
+ conf.getInt("spark.driver.cores", 1)
+ }
+ } else {
+ conf.getInt(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key, 1)
+ }
+ // Resolve defaults per process without writing the driver's value back
to SparkConf,
+ // which is also propagated to executors. Local mode shares the driver's
thread pool.
+ parsed += (broadcastBuildThreadsKey -> math.max(1,
defaultThreads).toString)
+ }
+
// Workaround for https://github.com/apache/gluten/issues/7837
if (isDriver && !inLocalMode(conf)) {
parsed += (COLUMNAR_VELOX_CACHE_ENABLED.key -> "false")
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index 3345d615d0..ff056dbabc 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -256,9 +256,12 @@ object VeloxConfig extends ConfigRegistry {
buildStaticConf("spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads")
.doc(
"Size of the dedicated CPU thread pool for the parallel broadcast
hash-table build. " +
- "By default, the thread pool is sized to the maximum task slots per
Spark executor " +
- "on executors, or to spark.driver.cores on the driver (used when " +
- "driver-side broadcast hash-table build is enabled), with a minimum
of one thread.")
+ "If unset, the size is resolved separately for each process at
backend initialization: " +
+ "spark.gluten.numTaskSlotsPerExecutor on executors, and
spark.driver.cores on the " +
+ "driver. In local mode, the shared pool uses the number of cores
specified by " +
+ "spark.master (for example, local[N] uses N). An explicit value
applies to both " +
+ "driver and executors. A value less than or equal to one uses the
serial build path " +
+ "without creating a dedicated thread pool.")
.intConf
.createOptional
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
index ce15734871..7440727cc7 100644
---
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
@@ -16,6 +16,7 @@
*/
package org.apache.spark.sql.execution.joins
+import org.apache.gluten.backendsapi.velox.VeloxListenerApi
import org.apache.gluten.execution.{BroadcastHashJoinExecTransformer,
VeloxWholeStageTransformerSuite}
import org.apache.spark.SparkConf
@@ -32,36 +33,113 @@ class GlutenParallelBroadcastJoinSuite extends
VeloxWholeStageTransformerSuite {
.set("spark.gluten.velox.broadcast.build.targetBytesPerThread", "8KB")
.set("spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads",
"4")
- for (buildOnDriver <- Seq(false, true)) {
- test(s"parallel broadcast hash table build with IO disabled,
driver=$buildOnDriver") {
- withSQLConf(
-
"spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild" ->
- buildOnDriver.toString,
- "spark.gluten.velox.minTableRowsForParallelJoinBuild" -> "1"
- ) {
- withTable("parallel_broadcast_build") {
- spark.range(4096)
- .selectExpr("id", "sha2(cast(id as string), 256) AS value")
- .write.saveAsTable("parallel_broadcast_build")
-
- val result = spark.sql("""
- |SELECT /*+ BROADCAST(b) */ p.id, b.value
- |FROM parallel_broadcast_build p JOIN
parallel_broadcast_build b
- |ON p.id = b.id
- |""".stripMargin)
- val expected =
spark.table("parallel_broadcast_build").collect().toSeq
- checkAnswer(result, expected)
-
- val plan = result.queryExecution.executedPlan
- assert(plan.collect { case j: BroadcastHashJoinExecTransformer => j
}.nonEmpty)
- val exchanges = plan.collect { case e: ColumnarBroadcastExchangeExec
=> e }
- assert(exchanges.size == 1)
- assert(exchanges.head.metrics("buildThreads").value > 1)
- if (buildOnDriver) {
- assert(exchanges.head.metrics("serializedHashTableSize").value > 0)
+ private val threadsKey =
+ "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads"
+ private val taskSlotsKey = "spark.gluten.numTaskSlotsPerExecutor"
+
+ test("broadcast build threads: resolve process defaults without modifying
SparkConf") {
+ val conf = new SparkConf(false)
+ .setMaster("yarn")
+ .set("spark.driver.cores", "6")
+ .set("spark.executor.cores", "16")
+ .set("spark.task.cpus", "2")
+ .set(taskSlotsKey, "8")
+ val original = conf.getAll.toMap
+
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey) ==
"6")
+ assert(conf.getAll.toMap == original)
+ assert(VeloxListenerApi.parseConf(conf, isDriver = false)(threadsKey) ==
"8")
+ assert(conf.getAll.toMap == original)
+ }
+
+ test("broadcast build threads: explicit thread count takes precedence on
both " +
+ "driver and executor") {
+ val conf = new SparkConf(false)
+ .set("spark.driver.cores", "6")
+ .set(taskSlotsKey, "8")
+ .set(threadsKey, "3")
+
+ Seq(true, false).foreach {
+ isDriver => assert(VeloxListenerApi.parseConf(conf,
isDriver)(threadsKey) == "3")
+ }
+ assert(conf.get(threadsKey) == "3")
+ }
+
+ test("broadcast build threads: missing resource settings and spark.master
default " +
+ "to one thread") {
+ Seq(true, false).foreach {
+ isDriver =>
+ assert(VeloxListenerApi.parseConf(new SparkConf(false),
isDriver)(threadsKey) == "1")
+ }
+ }
+
+ test(
+ "broadcast build threads: driver defaults to one core independently of
executor task slots") {
+ val conf = new SparkConf(false).set(taskSlotsKey, "8")
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey) ==
"1")
+ }
+
+ test("broadcast build threads: local pool uses master cores instead of task
slots") {
+ val masters = Seq(
+ "local" -> 1,
+ "local[6]" -> 6,
+ "local[6,2]" -> 6,
+ "local[*]" -> Runtime.getRuntime.availableProcessors(),
+ "local[*,2]" -> Runtime.getRuntime.availableProcessors()
+ )
+ masters.foreach {
+ case (master, cores) =>
+ val conf = new SparkConf(false)
+ .setMaster(master)
+ .set("spark.driver.cores", "3")
+ .set("spark.task.cpus", "2")
+ .set(taskSlotsKey, "2")
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey)
== cores.toString)
+ }
+ }
+
+ test("broadcast build threads: explicit thread count takes precedence in
local mode") {
+ val conf = new SparkConf(false).setMaster("local[6]").set(threadsKey, "3")
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey) ==
"3")
+ }
+
+ test("broadcast build threads: zero task slots still provide at least one
build thread") {
+ val conf = new SparkConf(false).set(taskSlotsKey, "0")
+ assert(VeloxListenerApi.parseConf(conf, isDriver = false)(threadsKey) ==
"1")
+ }
+
+ test(s"parallel broadcast hash table build with IO disabled") {
+ Seq(true, false).foreach {
+ buildOnDriver =>
+ withSQLConf(
+
"spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild" ->
+ buildOnDriver.toString,
+ "spark.gluten.velox.minTableRowsForParallelJoinBuild" -> "1"
+ ) {
+ withTable("parallel_broadcast_build") {
+ spark.range(4096)
+ .selectExpr("id", "sha2(cast(id as string), 256) AS value")
+ .write.saveAsTable("parallel_broadcast_build")
+
+ val result =
+ spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ p.id, b.value
+ |FROM parallel_broadcast_build p JOIN
parallel_broadcast_build b
+ |ON p.id = b.id
+ |""".stripMargin)
+ val expected =
spark.table("parallel_broadcast_build").collect().toSeq
+ checkAnswer(result, expected)
+
+ val plan = result.queryExecution.executedPlan
+ assert(plan.collect { case j: BroadcastHashJoinExecTransformer =>
j }.nonEmpty)
+ val exchanges = plan.collect { case e:
ColumnarBroadcastExchangeExec => e }
+ assert(exchanges.size == 1)
+ assert(exchanges.head.metrics("buildThreads").value > 1)
+ if (buildOnDriver) {
+ assert(exchanges.head.metrics("serializedHashTableSize").value >
0)
+ }
}
}
- }
}
}
}
diff --git a/cpp/velox/compute/VeloxBackend.cc
b/cpp/velox/compute/VeloxBackend.cc
index e9db41783c..52f48f38ef 100644
--- a/cpp/velox/compute/VeloxBackend.cc
+++ b/cpp/velox/compute/VeloxBackend.cc
@@ -261,9 +261,17 @@ void VeloxBackend::init(
std::make_unique<folly::CPUThreadPoolExecutor>(ioThreads,
folly::CPUThreadPoolExecutor::makeLifoSemQueue());
}
- const auto broadcastHashTableBuildThreads =
- std::max<int32_t>(1,
backendConf_->get<int32_t>(kBroadcastHashTableBuildThreads,
numTaskSlotsPerExecutor));
- broadcastHashTableBuildExecutor_ =
std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads);
+ const int32_t broadcastHashTableBuildThreads = std::max<int32_t>(1, [&]() {
+ if (!backendConf_->valueExists(kBroadcastHashTableBuildThreads)) {
+ LOG(WARNING) << kBroadcastHashTableBuildThreads << " is not set. Falling
back to " << kNumTaskSlotsPerExecutor
+ << " (" << numTaskSlotsPerExecutor << ").";
+ return numTaskSlotsPerExecutor;
+ }
+ return backendConf_->get<int32_t>(kBroadcastHashTableBuildThreads).value();
+ }());
+ if (broadcastHashTableBuildThreads > 1) {
+ broadcastHashTableBuildExecutor_ =
std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads);
+ }
initJolFilesystem();
diff --git a/cpp/velox/compute/VeloxBackend.h b/cpp/velox/compute/VeloxBackend.h
index a7592d718b..e6ac202ac1 100644
--- a/cpp/velox/compute/VeloxBackend.h
+++ b/cpp/velox/compute/VeloxBackend.h
@@ -55,7 +55,8 @@ class VeloxBackend {
ReaderThreadPool* getReaderThreadPool();
/// A dedicated CPU thread pool for the parallel broadcast hash-table build.
The build is
- /// CPU-bound, so it must not share ioExecutor() which is reserved for async
IO tasks.
+ /// CPU-bound, so it must not share ioExecutor() which is reserved for async
IO tasks. Returns
+ /// nullptr when the configured thread count uses the serial build path.
folly::Executor* broadcastHashTableBuildExecutor() const {
return broadcastHashTableBuildExecutor_.get();
}
diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h
index 8d9a1ed2e0..575f477f55 100644
--- a/cpp/velox/config/VeloxConfig.h
+++ b/cpp/velox/config/VeloxConfig.h
@@ -170,10 +170,10 @@ const std::string kVeloxAsyncTimeoutOnTaskStopping =
"spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping";
const int32_t kVeloxAsyncTimeoutOnTaskStoppingDefault = 30000; // 30s
-// By default, executors use the number of task slots per executor, with a
-// minimum of one thread. When driver-side broadcast hash-table build is
enabled
-// and Spark is not running in local mode, Scala-side logic in VeloxListenerApi
-// overrides the driver default to use spark.driver.cores instead.
+// VeloxListenerApi resolves Spark defaults per process: executor task slots,
+// driver cores, or local master cores. Native callers that omit this setting
+// fall back to the number of task slots per executor. Values less than or
equal to one use the serial build path
+// without creating a dedicated thread pool.
const std::string kBroadcastHashTableBuildThreads =
"spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads";
diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc
index 12e1872376..ab44a97445 100644
--- a/cpp/velox/jni/VeloxJniWrapper.cc
+++ b/cpp/velox/jni/VeloxJniWrapper.cc
@@ -1022,10 +1022,11 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
cb.push_back(ObjectStore::retrieve<ColumnarBatch>(handle));
}
- if (numThreads == 1) {
- // Use default global pool for driver-side build
- // The hash table will be serialized and broadcast, so it doesn't need
runtime's pool
- // Using runtime pool causes lifecycle management issues
+ auto* executor = VeloxBackend::get()->broadcastHashTableBuildExecutor();
+ if (numThreads == 1 || executor == nullptr) {
+ // Allocate the hash table from the backend-global memory pool because it
may outlive the runtime that initiated the
+ // build. Driver-side builds are serialized for broadcast, while
executor-side builds may be cached and reused
+ // across tasks.
auto memoryPool = defaultLeafVeloxMemoryPool();
auto builder = nativeHashTableBuild(
@@ -1067,8 +1068,6 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
// Use a dedicated CPU thread pool for the parallel hash-table build. The
build is CPU-bound,
// so it must not run on the io executor which is reserved for async IO
tasks.
- auto executor = VeloxBackend::get()->broadcastHashTableBuildExecutor();
-
std::vector<std::shared_ptr<gluten::HashTableBuilder>>
hashTableBuilders(numThreads);
std::vector<std::unique_ptr<facebook::velox::exec::BaseHashTable>>
otherTables(numThreads);
std::vector<folly::Future<folly::Unit>> futures;
@@ -1086,8 +1085,9 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
threadBatches.push_back(cb[i]);
}
- // Use default global pool for driver-side build
- // The hash table will be serialized and broadcast, so it doesn't need
runtime's pool
+ // Allocate the hash table from the backend-global memory pool because
it may outlive the runtime that initiated
+ // the build. Driver-side builds are serialized for broadcast, while
executor-side builds may be cached and reused
+ // across tasks.
auto threadMemoryPool = defaultLeafVeloxMemoryPool();
auto builder = nativeHashTableBuild(
@@ -1138,7 +1138,7 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
facebook::velox::exec::BaseHashTable::kNoSpillInputStartPartitionBit,
hashTableBuilders[0]->joinBuildVectorHasherMaxNumDistinct(),
hashTableBuilders[0]->dropDuplicates(),
- allowParallelJoinBuild ?
VeloxBackend::get()->broadcastHashTableBuildExecutor() : nullptr);
+ executor);
for (int i = 1; i < numThreads; ++i) {
if (hashTableBuilders[i]->joinHasNullKeys()) {
diff --git a/cpp/velox/tests/VeloxBackendTest.cc
b/cpp/velox/tests/VeloxBackendTest.cc
index 65bacf6f1f..5a6686ca61 100644
--- a/cpp/velox/tests/VeloxBackendTest.cc
+++ b/cpp/velox/tests/VeloxBackendTest.cc
@@ -18,31 +18,60 @@
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/futures/Future.h>
#include <gtest/gtest.h>
+#include <optional>
+#include <string>
#include <thread>
+#include <unordered_map>
#include "compute/VeloxBackend.h"
#include "config/VeloxConfig.h"
namespace gluten {
-class VeloxBackendTest : public ::testing::Test {
+struct BroadcastBuildThreadsTestParam {
+ int taskSlots;
+ std::optional<int> buildThreads;
+ std::optional<int> expectedThreads;
+};
+
+class VeloxBackendTest : public
::testing::TestWithParam<BroadcastBuildThreadsTestParam> {
protected:
- static void SetUpTestSuite() {
- VeloxBackend::create(AllocationListener::noop(),
{{kNumTaskSlotsPerExecutor, "0"}});
+ void SetUp() override {
+ const auto& param = GetParam();
+ std::unordered_map<std::string, std::string>
conf{{kNumTaskSlotsPerExecutor, std::to_string(param.taskSlots)}};
+ if (param.buildThreads.has_value()) {
+ conf.emplace(kBroadcastHashTableBuildThreads,
std::to_string(param.buildThreads.value()));
+ }
+ VeloxBackend::create(AllocationListener::noop(), conf);
}
- static void TearDownTestSuite() {
+ void TearDown() override {
VeloxBackend::get()->tearDown();
}
};
-TEST_F(VeloxBackendTest, broadcastHashTableBuildExecutorWithZeroTaskSlots) {
+TEST_P(VeloxBackendTest, broadcastHashTableBuildExecutorThreadCount) {
auto* backend = VeloxBackend::get();
auto* executor =
dynamic_cast<folly::CPUThreadPoolExecutor*>(backend->broadcastHashTableBuildExecutor());
+ if (!GetParam().expectedThreads.has_value()) {
+ EXPECT_EQ(executor, nullptr);
+ return;
+ }
ASSERT_NE(executor, nullptr);
- ASSERT_EQ(executor->numThreads(), 1);
+ ASSERT_EQ(executor->numThreads(), GetParam().expectedThreads.value());
const auto callerThread = std::this_thread::get_id();
EXPECT_NE(folly::via(executor, [] { return std::this_thread::get_id();
}).get(), callerThread);
}
+INSTANTIATE_TEST_SUITE_P(
+ BuildThreads,
+ VeloxBackendTest,
+ ::testing::Values(
+ BroadcastBuildThreadsTestParam{0, std::nullopt, std::nullopt},
+ BroadcastBuildThreadsTestParam{8, std::nullopt, 8},
+ BroadcastBuildThreadsTestParam{0, 0, std::nullopt},
+ BroadcastBuildThreadsTestParam{0, 1, std::nullopt},
+ BroadcastBuildThreadsTestParam{0, 4, 4},
+ BroadcastBuildThreadsTestParam{8, 4, 4}));
+
} // namespace gluten
diff --git a/docs/backends/velox/velox-configuration.md
b/docs/backends/velox/velox-configuration.md
index acbce9484b..c721ea9ccb 100644
--- a/docs/backends/velox/velox-configuration.md
+++ b/docs/backends/velox/velox-configuration.md
@@ -18,7 +18,7 @@ Type: ⚓ Static - can only be set before the Spark application
starts; 🔄 Dyn
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinPct
| 🔄 | 90 | If partial aggregation aggregationPct greater
than this value, partial aggregation may be early abandoned. Note: this option
only works when flushable partial aggregation is enabled. Ignored when
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
[...]
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinRows
| 🔄 | 100000 | If partial aggregation input rows number
greater than this value, partial aggregation may be early abandoned. Note:
this option only works when flushable partial aggregation is enabled. Ignored
when spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
[...]
| spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping
| ⚓ | 30000ms | Timeout in milliseconds when waiting for
runtime-scoped async work to finish during teardown.
[...]
-| spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads
| ⚓ | <undefined> | Size of the dedicated CPU thread pool for the
parallel broadcast hash-table build. By default, the thread pool is sized to
the maximum task slots per Spark executor on executors, or to
spark.driver.cores on the driver (used when driver-side broadcast hash-table
build is enabled), with a minimum of one thread.
[...]
+| spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads
| ⚓ | <undefined> | Size of the dedicated CPU thread pool for the
parallel broadcast hash-table build. If unset, the size is resolved separately
for each process at backend initialization:
spark.gluten.numTaskSlotsPerExecutor on executors, and spark.driver.cores on
the driver. In local mode, the shared pool uses the number of cores specified
by spark.master (for example, local[N] uses N). An expli [...]
|
spark.gluten.sql.columnar.backend.velox.broadcastNLJ.fullOuterRewriteThreshold
| 🔄 | 10MB | Maximum per-side plan size in bytes for rewriting
a full outer broadcast nested loop join into a left outer join and an existence
join followed by union. The rewrite is applied only when both sides have known
statistics and each side is at or below this threshold. Set to -1 to disable
the rewrite.
[...]
| spark.gluten.sql.columnar.backend.velox.cacheEnabled
| ⚓ | false | Enable Velox cache, default off. It's
recommended to enablesoft-affinity as well when enable velox cache.
[...]
| spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct
| ⚓ | 0 | Set prefetch cache min pct for velox file scan
[...]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
