[SPARK-28574][CORE] Allow to config different sizes for event queues #25307

yunzoud · 2019-07-31T04:32:17Z

What changes were proposed in this pull request?

Add configuration spark.scheduler.listenerbus.eventqueue.${name}.capacity to allow configuration of different event queue size.

How was this patch tested?

Unit test in core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

jiangxb1987 · 2019-07-31T05:18:05Z

add to whitelist

jiangxb1987 · 2019-07-31T05:20:02Z

ok to test

jiangxb1987

Looks good only some nits

jiangxb1987 · 2019-07-31T05:28:58Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

+  test("event queue size can be configued through spark conf") {
+    val conf = new SparkConf(false)
+      .set(LISTENER_BUS_EVENT_QUEUE_CAPACITY, 5)
+      .set("spark.scheduler.listenerbus.eventqueue.shared.capacity", "1")


nit: I would use s${SHARED_QUEUE}.capacity

jiangxb1987 · 2019-07-31T05:29:37Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

+    val conf = new SparkConf(false)
+      .set(LISTENER_BUS_EVENT_QUEUE_CAPACITY, 5)
+      .set("spark.scheduler.listenerbus.eventqueue.shared.capacity", "1")
+      .set("spark.scheduler.listenerbus.eventqueue.eventLog.capacity", "2")


similarly, I would use s${EVENT_LOG_QUEUE}.capacity

jiangxb1987 · 2019-07-31T05:31:42Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

+    val counter2 = new BasicJobCounter()
+    val counter3 = new BasicJobCounter()
+
+    bus.addToSharedQueue(counter1)


Maybe add comment to explain this is just to trigger add a new Queue.

added comments

jiangxb1987 · 2019-07-31T05:33:16Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+  // if no such conf is specified, use the value specified in
+  // LISTENER_BUS_EVENT_QUEUE_CAPACITY
+  protected def capacity: Int = conf.getInt(
+    s"spark.scheduler.listenerbus.eventqueue.${name}.capacity",


We shall assert the capacity is > 0, and add test to cover the case.

add assertion

jiangxb1987 · 2019-07-31T05:38:19Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala


+  // For testing only.
+  private[scheduler] def getQueueCapacity(name: String): Int = {
+    queues.asScala.find(_.name == name) match {


maybe queues.asScala.find(_.name == name).map(_.capacity).getOrElse(-1) ?

yaooqinn · 2019-07-31T09:54:06Z

the $name is from private class, considering add some instructions for each in the doc configurations.md?

zsxwing · 2019-07-31T17:48:21Z

add to whitelist

zsxwing · 2019-07-31T17:48:35Z

ok to test

SparkQA · 2019-07-31T20:30:09Z

Test build #108483 has finished for PR 25307 at commit 36995f1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-07-31T20:55:38Z

Test build #108485 has finished for PR 25307 at commit 4ab040c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing

Looks good overall. Left some nits.

zsxwing · 2019-08-01T17:43:38Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

  }

+  // For testing only.
+  private[scheduler] def getQueueCapacity(name: String): Int = {


nit: could you change the return type to Option[Int] and use None to indicate an unknown queue rather than -1?

zsxwing · 2019-08-01T17:45:05Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

+      .set(s"spark.scheduler.listenerbus.eventqueue.${SHARED_QUEUE}.capacity", "1")
+      .set(s"spark.scheduler.listenerbus.eventqueue.${EVENT_LOG_QUEUE}.capacity", "2")
+
+    val bus = new LiveListenerBus(conf)


nit: bus.stop() is missing. It should be called in finally to stop the internal thread.

The bus is actually never started, so no need to stop.

zsxwing · 2019-08-01T21:58:46Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+  private[scheduler] def capacity: Int = {
+    val queuesize = conf.getInt(s"spark.scheduler.listenerbus.eventqueue.${name}.capacity",
+                                conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY))
+    assert(queuesize > 0, s"capacity for event queue $name must be greater than 0," +


nit: missing space after ,

zsxwing · 2019-08-01T22:01:22Z

the $name is from private class, considering add some instructions for each in the doc configurations.md?

Since names of the queues are private, it's fine to not document them to keep them internal right now.

zsxwing · 2019-08-02T17:12:35Z

LGTM pending tests

SparkQA · 2019-08-02T18:54:13Z

Test build #108574 has finished for PR 25307 at commit 261beb0.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-08-02T19:07:58Z

Test build #108576 has finished for PR 25307 at commit 1c6f69a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-08-02T19:50:34Z

retest this please

jiangxb1987 · 2019-08-02T20:09:02Z

LGTM

SparkQA · 2019-08-02T22:21:28Z

Test build #108579 has finished for PR 25307 at commit 1c6f69a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-08-02T22:27:55Z

Thanks! Merging to master.

gatorsmile · 2019-11-26T07:15:21Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+  // LISTENER_BUS_EVENT_QUEUE_CAPACITY
+  private[scheduler] def capacity: Int = {
+    val queuesize = conf.getInt(s"spark.scheduler.listenerbus.eventqueue.${name}.capacity",
+                                conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY))


Nit: indent.

gatorsmile · 2019-11-26T07:17:01Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

-    conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY))
+  // The capacity can be configured by spark.scheduler.listenerbus.eventqueue.${name}.capacity,
+  // if no such conf is specified, use the value specified in
+  // LISTENER_BUS_EVENT_QUEUE_CAPACITY


We need to update the conf description of LISTENER_BUS_EVENT_QUEUE_CAPACITY.

gatorsmile · 2019-11-26T07:17:47Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+  // if no such conf is specified, use the value specified in
+  // LISTENER_BUS_EVENT_QUEUE_CAPACITY
+  private[scheduler] def capacity: Int = {
+    val queuesize = conf.getInt(s"spark.scheduler.listenerbus.eventqueue.${name}.capacity",


Instead of hard-coded here, can we define it in core/src/main/scala/org/apache/spark/internal/config/package.scala ?

gatorsmile · 2019-11-26T07:18:19Z

cc @jiangxb1987 @Ngone51

yunzoud added 2 commits July 29, 2019 10:50

configurable event queue size

5542a7a

add protect field

36995f1

yunzoud changed the title ~~[SPARK-28574] Allow to config different sizes for event queues~~ [SPARK-28574][CORE] Allow to config different sizes for event queues Jul 31, 2019

jiangxb1987 reviewed Jul 31, 2019

View reviewed changes

dongjoon-hyun added the SPARK CORE label Jul 31, 2019

address feedback

4ab040c

zsxwing requested changes Aug 1, 2019

View reviewed changes

zsxwing reviewed Aug 1, 2019

View reviewed changes

yunzoud added 2 commits August 2, 2019 09:55

address feedback

261beb0

simply code

1c6f69a

zsxwing approved these changes Aug 2, 2019

View reviewed changes

asfgit closed this in c212c9d Aug 2, 2019

jiangxb1987 mentioned this pull request Nov 20, 2019

[SPARK-29902][Doc][Minor]Add listener event queue capacity configuration to documentation #26529

Closed

gatorsmile reviewed Nov 26, 2019

View reviewed changes

[SPARK-28574][CORE] Allow to config different sizes for event queues #25307

[SPARK-28574][CORE] Allow to config different sizes for event queues #25307

Uh oh!

Conversation

yunzoud commented Jul 31, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

jiangxb1987 commented Jul 31, 2019

Uh oh!

jiangxb1987 commented Jul 31, 2019

Uh oh!

jiangxb1987 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yaooqinn commented Jul 31, 2019

Uh oh!

zsxwing commented Jul 31, 2019

Uh oh!

zsxwing commented Jul 31, 2019

Uh oh!

SparkQA commented Jul 31, 2019

Uh oh!

SparkQA commented Jul 31, 2019

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Aug 1, 2019

Uh oh!

zsxwing commented Aug 2, 2019

Uh oh!

SparkQA commented Aug 2, 2019

Uh oh!

SparkQA commented Aug 2, 2019

Uh oh!

zsxwing commented Aug 2, 2019

Uh oh!

jiangxb1987 commented Aug 2, 2019

Uh oh!

SparkQA commented Aug 2, 2019

Uh oh!

zsxwing commented Aug 2, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!