Vamsi-klu commented on code in PR #19082:
URL: https://github.com/apache/pinot/pull/19082#discussion_r3725703413
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ public static void moveLocalTarFileToRemote(File
localMetadataTarFile, URI outpu
* If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
* log a warning and continue. We assume that source and destination
directories are on the same filesystem,
* so that move() can be used.
+ * Uses {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @param fs filesystem used for both source and destination
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
*/
public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite)
- throws IOException, URISyntaxException {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @param fs filesystem used for both source and destination (must be safe
for concurrent move of distinct paths)
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @param parallelism number of concurrent move workers; values <= 1 run
serially
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
+ */
+ public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite, int parallelism)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = listSourceFiles(fs, sourceDir);
+ if (sourceFileUris.isEmpty()) {
+ return;
+ }
+ int effectiveParallelism = Math.max(1, Math.min(parallelism,
sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
+ if (effectiveParallelism == 1) {
+ for (URI sourceFileUri : sourceFileUris) {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ }
+ return;
+ }
+
+ ExecutorService executor =
Executors.newFixedThreadPool(effectiveParallelism, r -> {
+ Thread t = new Thread(r, "pinot-staging-copy");
+ t.setDaemon(true);
+ return t;
+ });
+ try {
+ List<Future<Void>> futures = new ArrayList<>(sourceFileUris.size());
+ for (URI sourceFileUri : sourceFileUris) {
+ futures.add(executor.submit(() -> {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ return null;
+ }));
+ }
+ IOException firstFailure = null;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } catch (Exception e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (firstFailure == null) {
+ firstFailure = cause instanceof IOException ? (IOException) cause
+ : new IOException("Failed to move files from " + sourceDir + "
to " + destDir, cause);
+ } else {
+ firstFailure.addSuppressed(cause);
+ }
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Resolve staging-copy parallelism from job {@code
executionFrameworkSpec.extraConfigs}.
+ * Missing/invalid/non-positive values fall back to {@link
#DEFAULT_STAGING_COPY_PARALLELISM}.
+ * Values above {@link #MAX_STAGING_COPY_PARALLELISM} are capped.
+ */
+ public static int getStagingCopyParallelism(Map<String, String>
extraConfigs) {
+ if (extraConfigs == null) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ String value = extraConfigs.get(STAGING_COPY_PARALLELISM);
+ if (value == null || value.isEmpty()) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ try {
+ int parallelism = Integer.parseInt(value.trim());
+ if (parallelism < 1) {
+ LOGGER.warn("Invalid {}={}, using default {}",
STAGING_COPY_PARALLELISM, value,
+ DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ if (parallelism > MAX_STAGING_COPY_PARALLELISM) {
+ LOGGER.warn("Capping {}={} to max {}", STAGING_COPY_PARALLELISM,
parallelism, MAX_STAGING_COPY_PARALLELISM);
+ return MAX_STAGING_COPY_PARALLELISM;
+ }
+ return parallelism;
+ } catch (NumberFormatException e) {
+ LOGGER.warn("Invalid {}={}, using default {}", STAGING_COPY_PARALLELISM,
value, DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ }
+
+ private static List<URI> listSourceFiles(PinotFS fs, URI sourceDir)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = new ArrayList<>();
for (String sourcePath : fs.listFiles(sourceDir, true)) {
URI sourceFileUri = SegmentGenerationUtils.getFileURI(sourcePath,
sourceDir);
- String sourceFilename =
SegmentGenerationUtils.getFileName(sourceFileUri);
- URI destFileUri =
- SegmentGenerationUtils.getRelativeOutputPath(sourceDir,
sourceFileUri, destDir).resolve(sourceFilename);
-
- if (!overwrite && fs.exists(destFileUri)) {
- LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
- } else {
- fs.move(sourceFileUri, destFileUri, true);
+ if (fs.isDirectory(sourceFileUri)) {
+ continue;
}
+ sourceFileUris.add(sourceFileUri);
+ }
+ return sourceFileUris;
+ }
+
+ private static void moveOneFile(PinotFS fs, URI sourceDir, URI
sourceFileUri, URI destDir, boolean overwrite)
+ throws IOException, URISyntaxException {
+ String sourceFilename = SegmentGenerationUtils.getFileName(sourceFileUri);
+ URI destFileUri =
+ SegmentGenerationUtils.getRelativeOutputPath(sourceDir, sourceFileUri,
destDir).resolve(sourceFilename);
+ if (!overwrite && fs.exists(destFileUri)) {
+ LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
+ } else {
+ fs.move(sourceFileUri, destFileUri, true);
}
Review Comment:
Good catch. To be precise, this exists-then-move(..., true) pattern is
carried over unchanged from the serial moveFiles on master, so the
check-then-act window is pre-existing, but parallelism does widen it and there
is no reason to keep it. BasePinotFS.move already enforces the flag (when the
destination exists and overwrite is false it logs a warning and returns false),
so I will change moveOneFile to pass the actual overwrite flag through to
fs.move and keep the exists() check only as a cheap short-circuit for the
warning log. Will push that change.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ public static void moveLocalTarFileToRemote(File
localMetadataTarFile, URI outpu
* If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
* log a warning and continue. We assume that source and destination
directories are on the same filesystem,
* so that move() can be used.
+ * Uses {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @param fs filesystem used for both source and destination
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
*/
public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite)
- throws IOException, URISyntaxException {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @param fs filesystem used for both source and destination (must be safe
for concurrent move of distinct paths)
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @param parallelism number of concurrent move workers; values <= 1 run
serially
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
+ */
+ public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite, int parallelism)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = listSourceFiles(fs, sourceDir);
+ if (sourceFileUris.isEmpty()) {
+ return;
+ }
+ int effectiveParallelism = Math.max(1, Math.min(parallelism,
sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
+ if (effectiveParallelism == 1) {
+ for (URI sourceFileUri : sourceFileUris) {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ }
+ return;
+ }
+
+ ExecutorService executor =
Executors.newFixedThreadPool(effectiveParallelism, r -> {
+ Thread t = new Thread(r, "pinot-staging-copy");
+ t.setDaemon(true);
+ return t;
+ });
+ try {
+ List<Future<Void>> futures = new ArrayList<>(sourceFileUris.size());
+ for (URI sourceFileUri : sourceFileUris) {
+ futures.add(executor.submit(() -> {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ return null;
+ }));
+ }
+ IOException firstFailure = null;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } catch (Exception e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (firstFailure == null) {
+ firstFailure = cause instanceof IOException ? (IOException) cause
+ : new IOException("Failed to move files from " + sourceDir + "
to " + destDir, cause);
+ } else {
+ firstFailure.addSuppressed(cause);
+ }
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Resolve staging-copy parallelism from job {@code
executionFrameworkSpec.extraConfigs}.
+ * Missing/invalid/non-positive values fall back to {@link
#DEFAULT_STAGING_COPY_PARALLELISM}.
+ * Values above {@link #MAX_STAGING_COPY_PARALLELISM} are capped.
+ */
+ public static int getStagingCopyParallelism(Map<String, String>
extraConfigs) {
+ if (extraConfigs == null) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ String value = extraConfigs.get(STAGING_COPY_PARALLELISM);
+ if (value == null || value.isEmpty()) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ try {
+ int parallelism = Integer.parseInt(value.trim());
+ if (parallelism < 1) {
+ LOGGER.warn("Invalid {}={}, using default {}",
STAGING_COPY_PARALLELISM, value,
+ DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ if (parallelism > MAX_STAGING_COPY_PARALLELISM) {
+ LOGGER.warn("Capping {}={} to max {}", STAGING_COPY_PARALLELISM,
parallelism, MAX_STAGING_COPY_PARALLELISM);
+ return MAX_STAGING_COPY_PARALLELISM;
+ }
+ return parallelism;
+ } catch (NumberFormatException e) {
+ LOGGER.warn("Invalid {}={}, using default {}", STAGING_COPY_PARALLELISM,
value, DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ }
+
+ private static List<URI> listSourceFiles(PinotFS fs, URI sourceDir)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = new ArrayList<>();
for (String sourcePath : fs.listFiles(sourceDir, true)) {
URI sourceFileUri = SegmentGenerationUtils.getFileURI(sourcePath,
sourceDir);
- String sourceFilename =
SegmentGenerationUtils.getFileName(sourceFileUri);
- URI destFileUri =
- SegmentGenerationUtils.getRelativeOutputPath(sourceDir,
sourceFileUri, destDir).resolve(sourceFilename);
-
- if (!overwrite && fs.exists(destFileUri)) {
- LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
- } else {
- fs.move(sourceFileUri, destFileUri, true);
+ if (fs.isDirectory(sourceFileUri)) {
+ continue;
}
+ sourceFileUris.add(sourceFileUri);
+ }
+ return sourceFileUris;
Review Comment:
Agreed, that is one extra remote stat per file on cloud filesystems.
listFilesWithMetadata is overridden by all the shipped remote implementations
(S3, GCS, ADLS, HDFS) plus LocalPinotFS, and the default implementation throws
UnsupportedOperationException, so the fallback is straightforward. I will
update listSourceFiles to try fs.listFilesWithMetadata(sourceDir, true) and
filter on FileMetadata.isDirectory(), catching UnsupportedOperationException to
fall back to the current listFiles plus isDirectory path for custom PinotFS
plugins. Will push that change.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ public static void moveLocalTarFileToRemote(File
localMetadataTarFile, URI outpu
* If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
* log a warning and continue. We assume that source and destination
directories are on the same filesystem,
* so that move() can be used.
+ * Uses {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @param fs filesystem used for both source and destination
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
*/
public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite)
- throws IOException, URISyntaxException {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @param fs filesystem used for both source and destination (must be safe
for concurrent move of distinct paths)
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @param parallelism number of concurrent move workers; values <= 1 run
serially
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
+ */
+ public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite, int parallelism)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = listSourceFiles(fs, sourceDir);
+ if (sourceFileUris.isEmpty()) {
+ return;
+ }
+ int effectiveParallelism = Math.max(1, Math.min(parallelism,
sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
Review Comment:
Fair point. The config path is already capped at
MAX_STAGING_COPY_PARALLELISM in getStagingCopyParallelism, but a direct caller
of the moveFiles overload bypasses that, and with enough source files the
thread count would equal the file count. I will clamp effectiveParallelism to
MAX_STAGING_COPY_PARALLELISM inside moveFiles as well so the cap holds
regardless of the caller. Will push that change.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ public static void moveLocalTarFileToRemote(File
localMetadataTarFile, URI outpu
* If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
* log a warning and continue. We assume that source and destination
directories are on the same filesystem,
* so that move() can be used.
+ * Uses {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @param fs filesystem used for both source and destination
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
*/
public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite)
- throws IOException, URISyntaxException {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @param fs filesystem used for both source and destination (must be safe
for concurrent move of distinct paths)
+ * @param sourceDir source directory URI
+ * @param destDir destination directory URI
+ * @param overwrite whether to overwrite existing destination files
+ * @param parallelism number of concurrent move workers; values <= 1 run
serially
+ * @throws IOException on listing or move failure
+ * @throws URISyntaxException on URI construction failure
+ */
+ public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite, int parallelism)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = listSourceFiles(fs, sourceDir);
+ if (sourceFileUris.isEmpty()) {
+ return;
+ }
+ int effectiveParallelism = Math.max(1, Math.min(parallelism,
sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
+ if (effectiveParallelism == 1) {
+ for (URI sourceFileUri : sourceFileUris) {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ }
+ return;
+ }
+
+ ExecutorService executor =
Executors.newFixedThreadPool(effectiveParallelism, r -> {
+ Thread t = new Thread(r, "pinot-staging-copy");
+ t.setDaemon(true);
+ return t;
+ });
+ try {
+ List<Future<Void>> futures = new ArrayList<>(sourceFileUris.size());
+ for (URI sourceFileUri : sourceFileUris) {
+ futures.add(executor.submit(() -> {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ return null;
+ }));
+ }
+ IOException firstFailure = null;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } catch (Exception e) {
Review Comment:
Confirmed, on InterruptedException the loop cancels only the current future
and then blocks on get() for the remaining ones, since the interrupt status was
consumed. I will change the handler to cancel all outstanding futures, break
out of the wait loop immediately, restore the interrupt status, and throw the
IOException wrapping the InterruptedException. The finally block's shutdownNow
stays as a backstop. Will push that change.
--
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]
