dongjoon-hyun commented on code in PR #58619:
URL: https://github.com/apache/spark/pull/58619#discussion_r4076606921
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -622,7 +622,13 @@ private[deploy] class Worker(
val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Use the app id itself as the executor's local dir, so the
external shuffle
+ // service can require registered localDirs to be scoped to
the registering
+ // application
(spark.shuffle.service.requireAppScopedLocalDirs).
+ val appDir = new File(dir, appId)
Review Comment:
`Utils.getOrCreateLocalRootDirs` is cached JVM-wide, so in `local-cluster`
mode all Workers in the same JVM resolve to the same `<root>/spark-*/<appId>`
directory. Unlike the previous `Utils.createDirectory(dir, namePrefix =
"executor")`, which always created a fresh directory,
`Utils.createDirectory(appDir)` silently accepts an existing one. As a result,
the Workers share one directory and whichever Worker runs
`maybeCleanupApplication` first deletes it for the others' executors.
I guess the `chmod700` change in this PR is a workaround for the same
sharing. Could we avoid the sharing instead, e.g.
`<root>/spark-*/<workerId>/<appId>`? The `appId` segment check would still pass.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -908,6 +908,25 @@ package object config {
.stringConf
.createWithDefault("spark_shuffle")
+ private[spark] val SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS =
Review Comment:
The `appId` in `RegisterExecutor` comes from the client, and `checkAuth`
enforces it only when `spark.authenticate=true`. Also, standalone executors run
as the Worker's OS user, so they can create a directory with any name under the
roots. Could you mention in the doc (here and `docs/spark-standalone.md`) that
this check is meaningful only with `spark.authenticate` enabled?
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -23,7 +23,8 @@ import java.lang.reflect.Field
import java.net.{BindException, ServerSocket, URI}
import java.nio.{ByteBuffer, ByteOrder}
import java.nio.charset.StandardCharsets.UTF_8
-import java.nio.file.Files
+import java.nio.file.{FileSystems, Files}
Review Comment:
This fails `Scala linter` in CI (`Files should come before FileSystems.`).
```suggestion
import java.nio.file.{Files, FileSystems}
```
##########
core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala:
##########
@@ -443,4 +443,28 @@ class WorkerSuite extends SparkFunSuite with Matchers with
BeforeAndAfter with P
assert(getHeartbeatTask(worker) == heartbeatTask)
assert(getWorkDirCleanupTask(worker) == workDirCleanupTask)
}
+
+ test("app cleanup removes the per-application executor local dir") {
Review Comment:
This test builds the layout by hand and puts it into `appDirectories`
directly, so it only covers the existing cleanup logic, not the new
`LaunchExecutor` code. As mentioned before, an end-to-end test with
`local-cluster` and `spark.shuffle.service.requireAppScopedLocalDirs=true`
would be helpful, ideally with 2+ Workers to cover the shared directory case.
##########
core/src/main/scala/org/apache/spark/util/Utils.scala:
##########
@@ -226,16 +227,29 @@ private[spark] object Utils
/**
* JDK equivalent of `chmod 700 file`.
*
+ * On POSIX filesystems this is a single `chmod` call, which is atomic:
concurrent calls on
+ * the same path (e.g. Workers sharing a local root dir in local-cluster
mode) cannot
+ * interleave into a broken permission set, unlike the `File.set*` fallback
below, each call
+ * of which is a non-atomic stat-then-chmod.
+ *
* @param file the file whose permissions will be modified
* @return true if the permissions were successfully changed, false
otherwise.
*/
def chmod700(file: File): Boolean = {
Review Comment:
`chmod700` is a shared utility (`HistoryServerDiskManager`, `KVUtils`,
`getOrCreateLocalRootDirsImpl`), so this looks out of scope for this PR. If the
Worker-level directory sharing is avoided, is this still needed? If so, could
you make it a separate JIRA and PR?
##########
core/src/test/scala/org/apache/spark/util/UtilsSuite.scala:
##########
@@ -572,6 +573,30 @@ class UtilsSuite extends SparkFunSuite with
ResetSystemProperties {
assert(!Utils.createDirectory(scenario8))
}
+ test("chmod700 leaves rwx------ under concurrent calls on the same path") {
+ // Workers in local-cluster mode share local root dirs and can chmod the
same
+ // per-application directory concurrently; the stat-then-chmod File.set*
sequence
+ // interleaves into broken permission sets, so chmod700 must be atomic.
+
assume(FileSystems.getDefault.supportedFileAttributeViews().contains("posix"))
+ withTempDir { dir =>
+ (1 to 10).foreach { _ =>
+ val threads = (1 to 4).map { _ =>
+ new Thread(() => (1 to 20).foreach(_ => Utils.chmod700(dir)))
+ }
+ threads.foreach(_.start())
+ threads.foreach(_.join())
+ assert(
+ Files.getPosixFilePermissions(dir.toPath) ===
+ PosixFilePermissions.fromString("rwx------"))
+ }
+ }
+ }
+
+ test("chmod700 returns false instead of throwing for a nonexistent path") {
+ val missing = new File(System.getProperty("java.io.tmpdir"), "chmod700-" +
System.nanoTime())
Review Comment:
nit. `withTempDir { dir => ... new File(dir, "missing") ... }` would be
cleaner than using `java.io.tmpdir` directly.
--
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]
