Ngone51 commented on a change in pull request #28117: [SPARK-31344][CORE]
Polish implementation of barrier() and allGather()
URL: https://github.com/apache/spark/pull/28117#discussion_r403437779
##########
File path: core/src/main/scala/org/apache/spark/BarrierCoordinator.scala
##########
@@ -256,62 +223,30 @@ private[spark] class BarrierCoordinator(
private[spark] sealed trait BarrierCoordinatorMessage extends Serializable
-private[spark] sealed trait RequestToSync extends BarrierCoordinatorMessage {
- def numTasks: Int
- def stageId: Int
- def stageAttemptId: Int
- def taskAttemptId: Long
- def barrierEpoch: Int
- def partitionId: Int
- def requestMethod: RequestMethod.Value
-}
-
/**
- * A global sync request message from BarrierTaskContext, by `barrier()` call.
Each request is
+ * A global sync request message from BarrierTaskContext. Each request is
* identified by stageId + stageAttemptId + barrierEpoch.
*
* @param numTasks The number of global sync requests the BarrierCoordinator
shall receive
* @param stageId ID of current stage
* @param stageAttemptId ID of current stage attempt
* @param taskAttemptId Unique ID of current task
- * @param barrierEpoch ID of the `barrier()` call, a task may consist multiple
`barrier()` calls
+ * @param barrierEpoch ID of a runBarrier() call, a task may consist multiple
runBarrier() calls
* @param partitionId ID of the current partition the task is assigned to
+ * @param message Message sent from the BarrierTaskContext
* @param requestMethod The BarrierTaskContext method that was called to
trigger BarrierCoordinator
*/
-private[spark] case class BarrierRequestToSync(
+private[spark] case class RequestToSync(
numTasks: Int,
stageId: Int,
stageAttemptId: Int,
taskAttemptId: Long,
barrierEpoch: Int,
partitionId: Int,
- requestMethod: RequestMethod.Value
-) extends RequestToSync
-
-/**
- * A global sync request message from BarrierTaskContext, by `allGather()`
call. Each request is
- * identified by stageId + stageAttemptId + barrierEpoch.
- *
- * @param numTasks The number of global sync requests the BarrierCoordinator
shall receive
- * @param stageId ID of current stage
- * @param stageAttemptId ID of current stage attempt
- * @param taskAttemptId Unique ID of current task
- * @param barrierEpoch ID of the `barrier()` call, a task may consist multiple
`barrier()` calls
- * @param partitionId ID of the current partition the task is assigned to
- * @param requestMethod The BarrierTaskContext method that was called to
trigger BarrierCoordinator
- * @param allGatherMessage Message sent from the BarrierTaskContext if
requestMethod is ALL_GATHER
- */
-private[spark] case class AllGatherRequestToSync(
- numTasks: Int,
- stageId: Int,
- stageAttemptId: Int,
- taskAttemptId: Long,
- barrierEpoch: Int,
- partitionId: Int,
- requestMethod: RequestMethod.Value,
- allGatherMessage: String
-) extends RequestToSync
+ message: String,
+ requestMethod: RequestMethod.Value) extends BarrierCoordinatorMessage
private[spark] object RequestMethod extends Enumeration {
val BARRIER, ALL_GATHER = Value
}
+
Review comment:
TODO: remove redundant empty line.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
