Skip to content

[SPARK-19182][DStream] Optimize the lock in StreamingJobProgressListener to not block UI when generating Streaming jobs #16601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
uncleGen wants to merge 3 commits into from
Closed

[SPARK-19182][DStream] Optimize the lock in StreamingJobProgressListener to not block UI when generating Streaming jobs #16601

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
46036bf
Optimize the lock in StreamingJobProgressListener to not block UI whe…
uncleGen Jan 16, 2017
eaa7b15
update
uncleGen Jan 17, 2017
e51623c
update
uncleGen Jan 18, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Optimize the lock in StreamingJobProgressListener to not block UI whe… 
…n generating Streaming jobs
  • Loading branch information
@uncleGen
uncleGen committed Jan 16, 2017
commit 46036bf683632e03f970de20f7bcd17b5369d5dc
10 changes: 4 additions & 6 deletions streaming/src/main/scala/org/apache/spark/streaming/DStreamGraph.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,12 +112,10 @@ final private[streaming] class DStreamGraph extends Serializable with Logging {

def generateJobs(time: Time): Seq[Job] = {
logDebug("Generating jobs for time " + time)
val jobs = this.synchronized {
outputStreams.flatMap { outputStream =>
val jobOption = outputStream.generateJob(time)
jobOption.foreach(_.setCallSite(outputStream.creationSite))
jobOption
}
val jobs = getOutputStreams().flatMap { outputStream =>
Copy link
Member

@zsxwing zsxwing Jan 17, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

synchronized is to make sure writeObject never write some intermediate states of DStreamGraph.

Copy link
Contributor Author

@uncleGen uncleGen Jan 17, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I have put the question to be too simple

val jobOption = outputStream.generateJob(time)
jobOption.foreach(_.setCallSite(outputStream.creationSite))
jobOption
}
logDebug("Generated " + jobs.length + " jobs for time " + time)
jobs
Expand Down