You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by skonto <gi...@git.apache.org> on 2016/01/18 18:29:17 UTC
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
GitHub user skonto opened a pull request:
https://github.com/apache/spark/pull/10808
[SPARK-11714][Mesos] Make Spark on Mesos honor port restrictions
- Adds support for port constraints on MesosExecutorBackend for fine-grained mode.
It keeps the convention for random port being the zero port. In such a case, a random port is chosen
from the available ranges (previously accepted from offers).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/skonto/spark honour_ports
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/10808.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #10808
----
commit aa6698c4ebee6e593cc9aa075993b490af80ca02
Author: Stavros Kontopoulos <st...@typesafe.com>
Date: 2016-01-06T01:47:50Z
honour ports
commit f19c71eff97a8b3cfcf4ad928fbcd99f26a98193
Author: Stavros Kontopoulos <st...@typesafe.com>
Date: 2016-01-18T16:26:47Z
Merge branch 'master' into honour_ports
Conflicts:
core/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendSuite.scala
merge
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51494045
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala ---
@@ -140,8 +141,13 @@ private[spark] class MesosSchedulerBackend(
val (resourcesAfterMem, usedMemResources) =
partitionResources(resourcesAfterCpu.asJava, "mem", calculateTotalMemory(sc))
+ val (resourcesAfterPort, usedPortResources) = resourcesAfterMem.partition {
+ r => ! ( r.getType == Value.Type.RANGES & r.getName == "ports" )
--- End diff --
style
```
val (resourcesAfterPort, usedPortResources) = resourcesAfterMem.partition { r =>
r.getType == Value.Type.RANGES && r.getName == "ports"
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172846113
a couple of style comments, otherwise this LGTM. @tnachen @andrewor14 can you please have a look?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-178645250
I tried to do a quick fix here without touching anything other than mesos fine grained mode, my opinion is that the code here needs refactoring in order to be easily extensible.
It is clear this part is deprecated but when i started working with it, it was not. However, i could work on the coarse grain more or less its the same concept.
The few issues could be fixed easily including the abstraction. The real issue is deprecation i think.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51582735
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+
+ def portRangeToList(ranges: String): List[(Long, Long)] = {
+ ranges.split(":").map { r => val ret = r.substring(1, r.length - 1).split(",")
+ (ret(0).toLong, ret(1).toLong)
+ }.toList
+ }
+
+ def startOnce(tryPort: Int): (Option[T], Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ (Some(service), port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ logWarning(s"Service$serviceString could not bind on port $tryPort. ")
+ (None, -1)
+ }
+ }
+
+ def retryPort(next: Int => Int, maxRetries: Int): (T, Int) = {
+
+ for (offset <- 0 to maxRetries) {
+ val tryPort = next(offset)
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ return (service, port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ if (offset >= maxRetries) {
+ val exceptionMessage =
+ s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
+ val exception = new BindException(exceptionMessage)
+ // restore original stack trace
+ exception.setStackTrace(e.getStackTrace)
+ throw exception
+ }
+ logWarning(s"Service$serviceString could not bind on port $tryPort.")
+ }
+ }
+ // Should never happen
+ throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ def startFromAvailable(rand: Boolean = false): (T, Int) = {
+
+ val ports = portRangeToList(getPortRangeRestriction(conf).get) // checked above for empty
+
+ val filteredPorts = ports.map(r => (r._1 to r._2).toList).reduce(_ ++ _).
+ distinct.filterNot(_ == startPort)
+ val maxRetries = Math.min(portMaxRetries(conf), filteredPorts.size)
+
+ val availPorts = {
+ if (rand) {
+ scala.util.Random.shuffle(filteredPorts.take(maxRetries)).toArray
+ } else {
+ filteredPorts.sorted.toArray
+ }
+ }
+
+ val tryPort = (x: Int) => availPorts(x).toInt
+
+ retryPort(tryPort, maxRetries)
+ }
+
require(startPort == 0 || (1024 <= startPort && startPort < 65536),
"startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port.")
- val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
- val maxRetries = portMaxRetries(conf)
- for (offset <- 0 to maxRetries) {
+ if (hasPortRangeRestriction(conf)) {
+
+ if (startPort != 0) {
+
+ val (service, port) = startOnce(startPort)
+
+ if (port != -1) {
+ (service.get, port)
+ } else {
+ // try other
+ startFromAvailable()
+ }
+ } else {
+ // try random
+ startFromAvailable(true)
+ }
+
+ } else {
+
+ val maxRetries = portMaxRetries(conf)
+
// Do not increment port if startPort is 0, which is treated as a special port
- val tryPort = if (startPort == 0) {
+ val tryPort = (x: Int) => if (startPort == 0) {
startPort
} else {
// If the new port wraps around, do not try a privilege port
- ((startPort + offset - 1024) % (65536 - 1024)) + 1024
- }
- try {
- val (service, port) = startService(tryPort)
- logInfo(s"Successfully started service$serviceString on port $port.")
- return (service, port)
- } catch {
- case e: Exception if isBindCollision(e) =>
- if (offset >= maxRetries) {
- val exceptionMessage =
- s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
- val exception = new BindException(exceptionMessage)
- // restore original stack trace
- exception.setStackTrace(e.getStackTrace)
- throw exception
- }
- logWarning(s"Service$serviceString could not bind on port $tryPort. " +
- s"Attempting port ${tryPort + 1}.")
+ ((startPort + x - 1024) % (65536 - 1024)) + 1024
}
+
+ retryPort(tryPort, maxRetries)
}
- // Should never happen
- throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ /**
+ * Checks if there is any port restriction (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return true if port range restriction holds false otherwise
+ */
+ def hasPortRangeRestriction(conf: SparkConf): Boolean = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges").isDefined
+ }
+
+ /**
+ * Gets the allowed port ranges to use (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return the string containing a list of port ranges in the format:
+ * (begin_port_number_1, end_port_number_1):...:(begin_port_number_n, end_port_number_n)
+ */
+ def getPortRangeRestriction(conf: SparkConf): Option[String] = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges")
+ }
+
+ /**
+ * Sets the allowed port ranges to use (currently used only with mesos).
+ * @param conf the SparkConfig to use
+ * @param ports a list of port ranges with format: (begin_port_number, end_port_number)
+ */
+ def setPortRangeRestriction(conf: SparkConf, ports: List[(Long, Long)]): Unit = {
+ // Caution: the property is for internal use only
+ conf.set("spark.mesos.executor.port.ranges", ports.mkString(":"))
--- End diff --
Ok i see. A proper abstraction for port related restriction would require more refactoring that is why i followed this approach anyway.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51495911
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
--- End diff --
this method is now very large and difficult to follow. Is all the logic required? If so we should make smaller helper methods with good documentation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #10808: [SPARK-11714][Mesos] Make Spark on Mesos honor port rest...
Posted by drcrallen <gi...@git.apache.org>.
Github user drcrallen commented on the issue:
https://github.com/apache/spark/pull/10808
In https://issues.apache.org/jira/browse/SPARK-15992 I tried to cleanup the offer consideration for the coarse executor a little more. The idea being that using the "improved" refactoring in 15992 would make it easier to add port considerations.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-173203578
**[Test build #49783 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49783/consoleFull)** for PR 10808 at commit [`c4a5115`](https://github.com/apache/spark/commit/c4a5115ea4bf3d322bd38a95cc497cc07c97c001).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172878959
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49685/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51589834
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala ---
@@ -340,4 +347,24 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
}
+ protected def getExecutorRelatedPort(sc: SparkContext, configProperty: String): Int = {
+ // either: spark.blockManager.port or spark.executor.port
--- End diff --
yes my bad...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-192465797
**[Test build #52485 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52485/consoleFull)** for PR 10808 at commit [`54b18a9`](https://github.com/apache/spark/commit/54b18a9d0d4011025e5e5ba347ad4acaa0a22c40).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #10808: [SPARK-11714][Mesos] Make Spark on Mesos honor port rest...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on the issue:
https://github.com/apache/spark/pull/10808
Fixed for coarse grained
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-173181318
**[Test build #49783 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49783/consoleFull)** for PR 10808 at commit [`c4a5115`](https://github.com/apache/spark/commit/c4a5115ea4bf3d322bd38a95cc497cc07c97c001).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-173203783
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49783/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51495310
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala ---
@@ -340,4 +347,24 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
}
+ protected def getExecutorRelatedPort(sc: SparkContext, configProperty: String): Int = {
+ // either: spark.blockManager.port or spark.executor.port
--- End diff --
I don't think this is a great abstraction; you're passing the whole `sc` in here but only to get its conf. I think the caller can just get the port themselves easily.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51495503
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala ---
@@ -340,4 +347,24 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
}
+ protected def getExecutorRelatedPort(sc: SparkContext, configProperty: String): Int = {
+ // either: spark.blockManager.port or spark.executor.port
+ sc.conf.getInt(s"$configProperty", 0)
+ }
+
+ protected def checkPorts(sc: SparkContext, ports: List[(Long, Long)]): Boolean = {
--- End diff --
you need to document what these tuples mean. This method needs a java doc
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-192519304
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52485/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51496118
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+
+ def portRangeToList(ranges: String): List[(Long, Long)] = {
+ ranges.split(":").map { r => val ret = r.substring(1, r.length - 1).split(",")
+ (ret(0).toLong, ret(1).toLong)
+ }.toList
+ }
+
+ def startOnce(tryPort: Int): (Option[T], Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ (Some(service), port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ logWarning(s"Service$serviceString could not bind on port $tryPort. ")
+ (None, -1)
+ }
+ }
+
+ def retryPort(next: Int => Int, maxRetries: Int): (T, Int) = {
+
+ for (offset <- 0 to maxRetries) {
+ val tryPort = next(offset)
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ return (service, port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ if (offset >= maxRetries) {
+ val exceptionMessage =
+ s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
+ val exception = new BindException(exceptionMessage)
+ // restore original stack trace
+ exception.setStackTrace(e.getStackTrace)
+ throw exception
+ }
+ logWarning(s"Service$serviceString could not bind on port $tryPort.")
+ }
+ }
+ // Should never happen
+ throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ def startFromAvailable(rand: Boolean = false): (T, Int) = {
+
+ val ports = portRangeToList(getPortRangeRestriction(conf).get) // checked above for empty
+
+ val filteredPorts = ports.map(r => (r._1 to r._2).toList).reduce(_ ++ _).
+ distinct.filterNot(_ == startPort)
+ val maxRetries = Math.min(portMaxRetries(conf), filteredPorts.size)
+
+ val availPorts = {
+ if (rand) {
+ scala.util.Random.shuffle(filteredPorts.take(maxRetries)).toArray
+ } else {
+ filteredPorts.sorted.toArray
+ }
+ }
+
+ val tryPort = (x: Int) => availPorts(x).toInt
+
+ retryPort(tryPort, maxRetries)
+ }
+
require(startPort == 0 || (1024 <= startPort && startPort < 65536),
"startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port.")
- val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
- val maxRetries = portMaxRetries(conf)
- for (offset <- 0 to maxRetries) {
+ if (hasPortRangeRestriction(conf)) {
+
+ if (startPort != 0) {
+
+ val (service, port) = startOnce(startPort)
+
+ if (port != -1) {
+ (service.get, port)
+ } else {
+ // try other
+ startFromAvailable()
+ }
+ } else {
+ // try random
+ startFromAvailable(true)
+ }
+
+ } else {
+
+ val maxRetries = portMaxRetries(conf)
+
// Do not increment port if startPort is 0, which is treated as a special port
- val tryPort = if (startPort == 0) {
+ val tryPort = (x: Int) => if (startPort == 0) {
startPort
} else {
// If the new port wraps around, do not try a privilege port
- ((startPort + offset - 1024) % (65536 - 1024)) + 1024
- }
- try {
- val (service, port) = startService(tryPort)
- logInfo(s"Successfully started service$serviceString on port $port.")
- return (service, port)
- } catch {
- case e: Exception if isBindCollision(e) =>
- if (offset >= maxRetries) {
- val exceptionMessage =
- s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
- val exception = new BindException(exceptionMessage)
- // restore original stack trace
- exception.setStackTrace(e.getStackTrace)
- throw exception
- }
- logWarning(s"Service$serviceString could not bind on port $tryPort. " +
- s"Attempting port ${tryPort + 1}.")
+ ((startPort + x - 1024) % (65536 - 1024)) + 1024
}
+
+ retryPort(tryPort, maxRetries)
}
- // Should never happen
- throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ /**
+ * Checks if there is any port restriction (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return true if port range restriction holds false otherwise
+ */
+ def hasPortRangeRestriction(conf: SparkConf): Boolean = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges").isDefined
+ }
+
+ /**
+ * Gets the allowed port ranges to use (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return the string containing a list of port ranges in the format:
+ * (begin_port_number_1, end_port_number_1):...:(begin_port_number_n, end_port_number_n)
+ */
+ def getPortRangeRestriction(conf: SparkConf): Option[String] = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges")
+ }
+
+ /**
+ * Sets the allowed port ranges to use (currently used only with mesos).
+ * @param conf the SparkConfig to use
+ * @param ports a list of port ranges with format: (begin_port_number, end_port_number)
+ */
+ def setPortRangeRestriction(conf: SparkConf, ports: List[(Long, Long)]): Unit = {
+ // Caution: the property is for internal use only
+ conf.set("spark.mesos.executor.port.ranges", ports.mkString(":"))
--- End diff --
this is kind of an anti-pattern; `conf` is supposed to be immutable once we create the `SparkContext`. If we set things after that some things might get the new value but others might not. It's a confusing pattern that we try to shy away from in general.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-175003023
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51492455
--- Diff: core/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala ---
@@ -18,6 +18,7 @@
package org.apache.spark.executor
import java.nio.ByteBuffer
+import java.util
--- End diff --
is this used?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172600103
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51583901
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+
+ def portRangeToList(ranges: String): List[(Long, Long)] = {
+ ranges.split(":").map { r => val ret = r.substring(1, r.length - 1).split(",")
+ (ret(0).toLong, ret(1).toLong)
+ }.toList
+ }
+
+ def startOnce(tryPort: Int): (Option[T], Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ (Some(service), port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ logWarning(s"Service$serviceString could not bind on port $tryPort. ")
+ (None, -1)
+ }
+ }
+
+ def retryPort(next: Int => Int, maxRetries: Int): (T, Int) = {
+
+ for (offset <- 0 to maxRetries) {
+ val tryPort = next(offset)
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ return (service, port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ if (offset >= maxRetries) {
+ val exceptionMessage =
+ s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
+ val exception = new BindException(exceptionMessage)
+ // restore original stack trace
+ exception.setStackTrace(e.getStackTrace)
+ throw exception
+ }
+ logWarning(s"Service$serviceString could not bind on port $tryPort.")
+ }
+ }
+ // Should never happen
+ throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ def startFromAvailable(rand: Boolean = false): (T, Int) = {
+
+ val ports = portRangeToList(getPortRangeRestriction(conf).get) // checked above for empty
+
+ val filteredPorts = ports.map(r => (r._1 to r._2).toList).reduce(_ ++ _).
+ distinct.filterNot(_ == startPort)
+ val maxRetries = Math.min(portMaxRetries(conf), filteredPorts.size)
+
+ val availPorts = {
+ if (rand) {
+ scala.util.Random.shuffle(filteredPorts.take(maxRetries)).toArray
+ } else {
+ filteredPorts.sorted.toArray
+ }
+ }
+
+ val tryPort = (x: Int) => availPorts(x).toInt
+
+ retryPort(tryPort, maxRetries)
+ }
+
require(startPort == 0 || (1024 <= startPort && startPort < 65536),
"startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port.")
- val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
- val maxRetries = portMaxRetries(conf)
- for (offset <- 0 to maxRetries) {
+ if (hasPortRangeRestriction(conf)) {
+
+ if (startPort != 0) {
+
+ val (service, port) = startOnce(startPort)
+
+ if (port != -1) {
+ (service.get, port)
+ } else {
+ // try other
+ startFromAvailable()
+ }
+ } else {
+ // try random
+ startFromAvailable(true)
+ }
+
+ } else {
+
+ val maxRetries = portMaxRetries(conf)
+
// Do not increment port if startPort is 0, which is treated as a special port
- val tryPort = if (startPort == 0) {
+ val tryPort = (x: Int) => if (startPort == 0) {
startPort
} else {
// If the new port wraps around, do not try a privilege port
- ((startPort + offset - 1024) % (65536 - 1024)) + 1024
- }
- try {
- val (service, port) = startService(tryPort)
- logInfo(s"Successfully started service$serviceString on port $port.")
- return (service, port)
- } catch {
- case e: Exception if isBindCollision(e) =>
- if (offset >= maxRetries) {
- val exceptionMessage =
- s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
- val exception = new BindException(exceptionMessage)
- // restore original stack trace
- exception.setStackTrace(e.getStackTrace)
- throw exception
- }
- logWarning(s"Service$serviceString could not bind on port $tryPort. " +
- s"Attempting port ${tryPort + 1}.")
+ ((startPort + x - 1024) % (65536 - 1024)) + 1024
}
+
+ retryPort(tryPort, maxRetries)
}
- // Should never happen
- throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ /**
+ * Checks if there is any port restriction (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return true if port range restriction holds false otherwise
+ */
+ def hasPortRangeRestriction(conf: SparkConf): Boolean = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges").isDefined
+ }
+
+ /**
+ * Gets the allowed port ranges to use (currently used only with mesos)
--- End diff --
The port restriction idea could be useful in other environments for ops use cases, the reason i put it there is because another method in Utils use it. In other words, depending on the environment, now only mesos though, we need to make startService code port restriction aware or move that logic one layer up. In the latter case a larger refactoring is needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-178248032
@dragos @skonto I haven't gone through the entire patch in great detail yet but I have 3 high level concerns:
1. This patch lacks comments / proper abstractions and makes the code difficult to follow
2. Currently the only consumer of port restriction if Mesos fine-grained mode, which isn't even the default. I would leave as much logic outside of core's `Utils` as possible. If we do decide to support it for other modes then we should file a separate issue so we can review that change in isolation.
3. Fine-grained mode will be deprecated, if not removed, soon. I'm not sure if this change is super worth it given its complexity. If it's a small, isolated change then maybe it makes sense to add it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51658411
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala ---
@@ -340,4 +347,24 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
}
+ protected def getExecutorRelatedPort(sc: SparkContext, configProperty: String): Int = {
+ // either: spark.blockManager.port or spark.executor.port
+ sc.conf.getInt(s"$configProperty", 0)
+ }
+
+ protected def checkPorts(sc: SparkContext, ports: List[(Long, Long)]): Boolean = {
--- End diff --
yeah, the existing code is not perfect, especially Mesos integration code which really lacks documentation. In general I would err on adding more comments than not adding them at all, since the goal of adding comments is to make it easy for someone new to follow. Even just a sentence on what the return value means is better than nothing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51492511
--- Diff: core/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala ---
@@ -81,6 +88,16 @@ private[spark] class MesosExecutorBackend
env)
}
+ private def getRangesFromResources(executorInfo: ExecutorInfo): List[(Long, Long)] = {
+ val ranges = executorInfo.getResourcesList.asScala
+ .filter(r => r.getType == Value.Type.RANGES & r.getName == "ports")
+ .toList.map(r => r.getRanges.getRangeList)
+
+ // a final list of all ranges in format (Long , Long)
+ ranges.flatMap { ranges => ranges.asScala.map { r => (r.getBegin, r.getEnd) }}
--- End diff --
need to indent this block 2 more spaces
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51496158
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+
+ def portRangeToList(ranges: String): List[(Long, Long)] = {
+ ranges.split(":").map { r => val ret = r.substring(1, r.length - 1).split(",")
+ (ret(0).toLong, ret(1).toLong)
+ }.toList
+ }
+
+ def startOnce(tryPort: Int): (Option[T], Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ (Some(service), port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ logWarning(s"Service$serviceString could not bind on port $tryPort. ")
+ (None, -1)
+ }
+ }
+
+ def retryPort(next: Int => Int, maxRetries: Int): (T, Int) = {
+
+ for (offset <- 0 to maxRetries) {
+ val tryPort = next(offset)
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ return (service, port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ if (offset >= maxRetries) {
+ val exceptionMessage =
+ s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
+ val exception = new BindException(exceptionMessage)
+ // restore original stack trace
+ exception.setStackTrace(e.getStackTrace)
+ throw exception
+ }
+ logWarning(s"Service$serviceString could not bind on port $tryPort.")
+ }
+ }
+ // Should never happen
+ throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ def startFromAvailable(rand: Boolean = false): (T, Int) = {
+
+ val ports = portRangeToList(getPortRangeRestriction(conf).get) // checked above for empty
+
+ val filteredPorts = ports.map(r => (r._1 to r._2).toList).reduce(_ ++ _).
+ distinct.filterNot(_ == startPort)
+ val maxRetries = Math.min(portMaxRetries(conf), filteredPorts.size)
+
+ val availPorts = {
+ if (rand) {
+ scala.util.Random.shuffle(filteredPorts.take(maxRetries)).toArray
+ } else {
+ filteredPorts.sorted.toArray
+ }
+ }
+
+ val tryPort = (x: Int) => availPorts(x).toInt
+
+ retryPort(tryPort, maxRetries)
+ }
+
require(startPort == 0 || (1024 <= startPort && startPort < 65536),
"startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port.")
- val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
- val maxRetries = portMaxRetries(conf)
- for (offset <- 0 to maxRetries) {
+ if (hasPortRangeRestriction(conf)) {
+
+ if (startPort != 0) {
+
+ val (service, port) = startOnce(startPort)
+
+ if (port != -1) {
+ (service.get, port)
+ } else {
+ // try other
+ startFromAvailable()
+ }
+ } else {
+ // try random
+ startFromAvailable(true)
+ }
+
+ } else {
+
+ val maxRetries = portMaxRetries(conf)
+
// Do not increment port if startPort is 0, which is treated as a special port
- val tryPort = if (startPort == 0) {
+ val tryPort = (x: Int) => if (startPort == 0) {
startPort
} else {
// If the new port wraps around, do not try a privilege port
- ((startPort + offset - 1024) % (65536 - 1024)) + 1024
- }
- try {
- val (service, port) = startService(tryPort)
- logInfo(s"Successfully started service$serviceString on port $port.")
- return (service, port)
- } catch {
- case e: Exception if isBindCollision(e) =>
- if (offset >= maxRetries) {
- val exceptionMessage =
- s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
- val exception = new BindException(exceptionMessage)
- // restore original stack trace
- exception.setStackTrace(e.getStackTrace)
- throw exception
- }
- logWarning(s"Service$serviceString could not bind on port $tryPort. " +
- s"Attempting port ${tryPort + 1}.")
+ ((startPort + x - 1024) % (65536 - 1024)) + 1024
}
+
+ retryPort(tryPort, maxRetries)
}
- // Should never happen
- throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ /**
+ * Checks if there is any port restriction (currently used only with mesos)
+ * @param conf the SparkConfig to use
+ * @return true if port range restriction holds false otherwise
+ */
+ def hasPortRangeRestriction(conf: SparkConf): Boolean = {
+ // Caution: the property is for internal use only
+ conf.getOption("spark.mesos.executor.port.ranges").isDefined
+ }
+
+ /**
+ * Gets the allowed port ranges to use (currently used only with mesos)
--- End diff --
if these are only used in Mesos they don't belong here. We should put them in `MesosSchedulerUtils` or something
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-174962724
**[Test build #50100 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/50100/consoleFull)** for PR 10808 at commit [`54b18a9`](https://github.com/apache/spark/commit/54b18a9d0d4011025e5e5ba347ad4acaa0a22c40).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51658530
--- Diff: core/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala ---
@@ -81,6 +88,16 @@ private[spark] class MesosExecutorBackend
env)
}
+ private def getRangesFromResources(executorInfo: ExecutorInfo): List[(Long, Long)] = {
+ val ranges = executorInfo.getResourcesList.asScala
+ .filter(r => r.getType == Value.Type.RANGES & r.getName == "ports")
+ .toList.map(r => r.getRanges.getRangeList)
+
+ // a final list of all ranges in format (Long , Long)
+ ranges.flatMap { ranges => ranges.asScala.map { r => (r.getBegin, r.getEnd) }}
--- End diff --
scalastyle is incomplete. I think it gets us 60% of the way there but there are still many things it doesn't catch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172845387
ok to test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-192519302
Build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51587253
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,138 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
--- End diff --
If i didnt do it this way then this method would have been similar but not the same with the pre-existing method in that file. Then it would have been be a problem of redundancy. The logic is fairly complex and it is the minimum i could use to cover all cases.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #10808: [SPARK-11714][Mesos] Make Spark on Mesos honor po...
Posted by skonto <gi...@git.apache.org>.
Github user skonto closed the pull request at:
https://github.com/apache/spark/pull/10808
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172848612
**[Test build #49685 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49685/consoleFull)** for PR 10808 at commit [`9991c3f`](https://github.com/apache/spark/commit/9991c3fbdcd9e1daa7e34740a0fd8bc2c0cc72cd).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172878954
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r50108893
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -1967,39 +1967,139 @@ private[spark] object Utils extends Logging {
conf: SparkConf,
serviceName: String = ""): (T, Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+
+ def portRangeToList(ranges: String): List[(Long, Long)] = {
+ ranges.split(":").map { r => val ret = r.substring(1, r.length - 1).split(",")
+ (ret(0).toLong, ret(1).toLong)
+ }.toList
+ }
+
+ def startOnce(tryPort: Int): (Option[T], Int) = {
+ val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ (Some(service), port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ logWarning(s"Service$serviceString could not bind on port $tryPort. ")
+ (None, -1)
+ }
+ }
+
+ def retryPort(next: Int => Int, maxRetries: Int): (T, Int) = {
+
+ for (offset <- 0 to maxRetries) {
+ val tryPort = next(offset)
+ try {
+ val (service, port) = startService(tryPort)
+ logInfo(s"Successfully started service$serviceString on port $port.")
+ return (service, port)
+ } catch {
+ case e: Exception if isBindCollision(e) =>
+ if (offset >= maxRetries) {
+ val exceptionMessage =
+ s"${e.getMessage}: Service$serviceString failed after $maxRetries retries!"
+ val exception = new BindException(exceptionMessage)
+ // restore original stack trace
+ exception.setStackTrace(e.getStackTrace)
+ throw exception
+ }
+ logWarning(s"Service$serviceString could not bind on port $tryPort.")
+ }
+ }
+ // Should never happen
+ throw new SparkException(s"Failed to start service$serviceString on port $startPort")
+ }
+
+ def startFromAvailable(rand: Boolean = false): (T, Int) = {
+
+ val ports = portRangeToList(getPortRangeRestriction(conf).get) // checked above for empty
+
+ val filteredPorts = ports.map(r => (r._1 to r._2).toList).reduce(_ ++ _).
+ distinct.filterNot(_ == startPort)
+ val maxRetries = Math.min(portMaxRetries(conf), filteredPorts.size)
+
+ val availPorts = {
+ if (rand) {
+ scala.util.Random.shuffle(filteredPorts.take(maxRetries)).toArray
+ } else {
+ filteredPorts.sorted.toArray
+ }
+ }
+
+ val tryPort = (x: Int) => availPorts(x).toInt
+
+ retryPort(tryPort, maxRetries)
+ }
+
require(startPort == 0 || (1024 <= startPort && startPort < 65536),
"startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port.")
- val serviceString = if (serviceName.isEmpty) "" else s" '$serviceName'"
- val maxRetries = portMaxRetries(conf)
- for (offset <- 0 to maxRetries) {
+ if (hasPortRangeRestriction(conf)) {
+
+ if (startPort != 0) {
+
+ val (service, port) = startOnce(startPort)
+
+ if (port != -1) {
+ (service.get, port)
+ } else {
+ // try other
+ startFromAvailable()
+ }
+ }
+ else {
--- End diff --
`else` should go on the line above, as in Java.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51586019
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala ---
@@ -140,8 +141,13 @@ private[spark] class MesosSchedulerBackend(
val (resourcesAfterMem, usedMemResources) =
partitionResources(resourcesAfterCpu.asJava, "mem", calculateTotalMemory(sc))
+ val (resourcesAfterPort, usedPortResources) = resourcesAfterMem.partition {
+ r => ! ( r.getType == Value.Type.RANGES & r.getName == "ports" )
--- End diff --
It is easy to miss that :( didnt get any warning or something... my bad..
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-173203780
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-175003026
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/50100/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-192518790
**[Test build #52485 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52485/consoleFull)** for PR 10808 at commit [`54b18a9`](https://github.com/apache/spark/commit/54b18a9d0d4011025e5e5ba347ad4acaa0a22c40).
* This patch **fails PySpark unit tests**.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-175002458
**[Test build #50100 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/50100/consoleFull)** for PR 10808 at commit [`54b18a9`](https://github.com/apache/spark/commit/54b18a9d0d4011025e5e5ba347ad4acaa0a22c40).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10808#issuecomment-172878661
**[Test build #49685 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49685/consoleFull)** for PR 10808 at commit [`9991c3f`](https://github.com/apache/spark/commit/9991c3fbdcd9e1daa7e34740a0fd8bc2c0cc72cd).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51586260
--- Diff: core/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala ---
@@ -81,6 +88,16 @@ private[spark] class MesosExecutorBackend
env)
}
+ private def getRangesFromResources(executorInfo: ExecutorInfo): List[(Long, Long)] = {
+ val ranges = executorInfo.getResourcesList.asScala
+ .filter(r => r.getType == Value.Type.RANGES & r.getName == "ports")
+ .toList.map(r => r.getRanges.getRangeList)
+
+ // a final list of all ranges in format (Long , Long)
+ ranges.flatMap { ranges => ranges.asScala.map { r => (r.getBegin, r.getEnd) }}
--- End diff --
shouldnt be this caught by scalastyle maybe not? i was feeling confident when it passed with zero errors/warnings...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by skonto <gi...@git.apache.org>.
Github user skonto commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r51588223
--- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala ---
@@ -340,4 +347,24 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
}
+ protected def getExecutorRelatedPort(sc: SparkContext, configProperty: String): Int = {
+ // either: spark.blockManager.port or spark.executor.port
+ sc.conf.getInt(s"$configProperty", 0)
+ }
+
+ protected def checkPorts(sc: SparkContext, ports: List[(Long, Long)]): Boolean = {
--- End diff --
Judging from pre-existing methods not all of them have a proper javadoc with parameters etc so it seemed to me a bit ad hoc the whole documenting process. I am confused.
Compare def startServiceOnPort doc versus any other method in Utils. Also i dont see much in the scala style doc. Maybe i am wrong...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11714][Mesos] Make Spark on Mesos honor...
Posted by dragos <gi...@git.apache.org>.
Github user dragos commented on a diff in the pull request:
https://github.com/apache/spark/pull/10808#discussion_r50108822
--- Diff: core/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala ---
@@ -31,6 +32,8 @@ import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.scheduler.cluster.mesos.MesosTaskLaunchData
import org.apache.spark.util.Utils
+
+
--- End diff --
Nit: no need for extra empty lines
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #10808: [SPARK-11714][Mesos] Make Spark on Mesos honor port rest...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/10808
@skonto you can close this PR now
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org