Github user arunmahadevan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21385#discussion_r190316408
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/UnsafeRowReceiver.scala
---
@@ -41,11 +50,15 @@ private[shuffle] case class ReceiverEpochMarker()
extends UnsafeRowReceiverMessa
*/
private[shuffle] class UnsafeRowReceiver(
queueSize: Int,
+ numShuffleWriters: Int,
+ checkpointIntervalMs: Long,
override val rpcEnv: RpcEnv)
extends ThreadSafeRpcEndpoint with ContinuousShuffleReader with
Logging {
// Note that this queue will be drained from the main task thread and
populated in the RPC
// response thread.
- private val queue = new
ArrayBlockingQueue[UnsafeRowReceiverMessage](queueSize)
+ private val queues = Array.fill(numShuffleWriters) {
--- End diff --
Queue per writer makes it easier to handle the alignment. However, if the
queues do not get filled evenly, it makes it much more difficult to set a
single size across all the queues. In a single queue this would not be much of
an issue since the queues are shared by all writers.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
