give rabit tracker a port number
- Dominant language
- C++
- Stars
- 28.8k
- Forks
- 8.9k
- Avg merge
- 1d 12h
- Merged PRs (30d)
- 54
Description
HI, All :
when I use xgboost4j-spark on k8s and istio ,the training will hang for tracker waiting for worker .
I found the reason. xgboost use AKKA as protocol between tracker and worker. but if I run Tracker in pod with Istio proxy. it will reject inbound communication from worker. so tracker will wait until time_out.
I found solution for AKKA with Istio below:
https://doc.akka.io/docs/akka-management/current/bootstrap/istio.html#allowing-inbound-communication
but it's require Tracker port is a fixed port , but xgboost given a random port for Now.
I read the source code
in code of [RabitTracker.scala](https://github.com/dmlc/xgboost/blob/299e5000a422a939f6ecf7970a09c590fee40711/jvm-packages/xgboost4j/src/main/scala/ml/dmlc/xgboost4j/scala/rabit/RabitTracker.scala) I found code below:
```
private[scala] class RabitTracker(numWorkers: Int, port: Option[Int] = None,
maxPortTrials: Int = 1000)
```
and tracker start with code
```
private def start(timeout: Duration): Boolean = {
val hostAddress = Option(TrackerProperties.getInstance().getHostIp)
.map(InetAddress.getByName).getOrElse(InetAddress.getLocalHost)
handler ? RabitTrackerHandler.StartTracker(
new InetSocketAddress(hostAddress, port.getOrElse(0)), maxPortTrials, timeout)
// block by waiting for the actor to bind to a port
Try(Await.result(handler ? RabitTrackerHandler.RequestBoundFuture, askTimeout.duration)
.asInstanceOf[Future[Map[String, String]]]) match {
case Success(futurePortBound) =>
// The success of the Future is contingent on binding to an InetSocketAddress.
val isBound = Try(Await.ready(futurePortBound, tcpBindingTimeout)).isSuccess
if (isBound) {
workerEnvs = Await.result(futurePortBound, 0 nano)
}
isBound
case Failure(ex: Throwable) =>
false
}
}
```
RabitTrackerHandler found a random port for tracker if port is None
```
case msg: StartTracker =>
maxPortTrials = msg.maxPortTrials
workerConnectionTimeout = msg.connectionTimeout
// if the port number is missing, try binding to a random ephemeral port.
if (msg.addr.getPort == 0) {
tcpManager ! Tcp.Bind(self,
new InetSocketAddress(msg.addr.getAddress, new Random().nextInt(61000 - 32768) + 32768),
backlog = 256)
} else {
tcpManager ! Tcp.Bind(self, msg.addr, backlog = 256)
}
sender() ! true
```
but in code of [xgboost.scala](https://github.com/dmlc/xgboost/blob/299e5000a422a939f6ecf7970a09c590fee40711/jvm-packages/xgboost4j-spark/src/main/scala/ml/dmlc/xgboost4j/scala/spark/XGBoost.scala) I
we didn't set port when create Tracker instance
code as below:
```
/** visiable for testing */
private[scala] def getTracker(nWorkers: Int, trackerConf: TrackerConf): IRabitTracker = {
val tracker: IRabitTracker = trackerConf.trackerImpl match {
case "scala" => new RabitTracker(nWorkers)
case "python" => new PyRabitTracker(nWorkers, trackerConf.hostIp, trackerConf.pythonExec)
case _ => new PyRabitTracker(nWorkers)
}
tracker
}
```
so if add an new option in TrackerConf
```
case class TrackerConf(workerConnectionTimeout: Long, trackerImpl: String,
hostIp: String = "", pythonExec: String = "" ,port:Option[Int]=None)
```
and given a port if any settings not None
```
/** visiable for testing */
private[scala] def getTracker(nWorkers: Int, trackerConf: TrackerConf): IRabitTracker = {
val tracker: IRabitTracker = trackerConf.trackerImpl match {
case "scala" => new RabitTracker(nWorkers, trackerConf.port)
case "python" => new PyRabitTracker(nWorkers, trackerConf.hostIp, trackerConf.pythonExec)
case _ => new PyRabitTracker(nWorkers)
}
tracker
}
```
we can support istio excludeInboundPort.
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.