[GitHub] [spark] Ngone51 commented on a change in pull request #28117: [SPARK-31344][CORE] Polish implementation of barrier() and allGather()

[GitBox <gi...@apache.org>]

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:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org