{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/207348484","pull_request_review_id":142944923,"id":207348484,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIwNzM0ODQ4NA==","diff_hunk":"@@ -0,0 +1,239 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.spark\n+\n+import java.util.{Timer, TimerTask}\n+import java.util.concurrent.ConcurrentHashMap\n+import java.util.concurrent.atomic.AtomicInteger\n+\n+import scala.collection.mutable.ArrayBuffer\n+\n+import org.apache.spark.internal.Logging\n+import org.apache.spark.rpc.{RpcCallContext, RpcEnv, ThreadSafeRpcEndpoint}\n+import org.apache.spark.scheduler.{LiveListenerBus, SparkListener, SparkListenerStageCompleted}\n+\n+/**\n+ * A coordinator that handles all global sync requests from BarrierTaskContext. Each global sync\n+ * request is generated by `BarrierTaskContext.barrier()`, and identified by\n+ * stageId + stageAttemptId + barrierEpoch. Reply all the blocking global sync requests upon\n+ * received all the requests for a group of `barrier()` calls. If the coordinator doesn't collect\n+ * enough global sync requests within a configured time, fail all the requests due to timeout.\n+ */\n+private[spark] class BarrierCoordinator(\n+    timeout: Int,\n+    listenerBus: LiveListenerBus,\n+    override val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint with Logging {\n+\n+  private val timer = new Timer(\"BarrierCoordinator barrier epoch increment timer\")\n+\n+  private val listener = new SparkListener {\n+    override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {\n+      val stageInfo = stageCompleted.stageInfo\n+      // Remove internal data from a finished stage attempt.\n+      cleanupSyncRequests(stageInfo.stageId, stageInfo.attemptNumber)\n+      barrierEpochByStageIdAndAttempt.remove((stageInfo.stageId, stageInfo.attemptNumber))\n+      cancelTimerTask(stageInfo.stageId, stageInfo.attemptNumber)\n+    }\n+  }\n+\n+  // Epoch counter for each barrier (stage, attempt).\n+  private val barrierEpochByStageIdAndAttempt = new ConcurrentHashMap[(Int, Int), Int]\n+\n+  // Remember all the blocking global sync requests for each barrier (stage, attempt).\n+  private val syncRequestsByStageIdAndAttempt =\n+    new ConcurrentHashMap[(Int, Int), ArrayBuffer[RpcCallContext]]\n+\n+  // Remember all the TimerTasks for each barrier (stage, attempt).\n+  private val timerTaskByStageIdAndAttempt = new ConcurrentHashMap[(Int, Int), TimerTask]\n+\n+  // Number of tasks for each stage.\n+  private val numTasksByStage = new ConcurrentHashMap[Int, Int]\n+\n+  override def onStart(): Unit = {\n+    super.onStart()\n+    listenerBus.addToStatusQueue(listener)\n+  }\n+\n+  /**\n+   * Get the array of [[RpcCallContext]]s that correspond to a barrier sync request from a stage\n+   * attempt.\n+   */\n+  private def getOrInitSyncRequests(\n+      stageId: Int,\n+      stageAttemptId: Int,\n+      numTasks: Int): ArrayBuffer[RpcCallContext] = {\n+    val requests = syncRequestsByStageIdAndAttempt.putIfAbsent((stageId, stageAttemptId),","path":"core/src/main/scala/org/apache/spark/BarrierCoordinator.scala","position":80,"original_position":80,"commit_id":"77d082b1601d09d5e320d689b2b71a542332a01b","original_commit_id":"77d082b1601d09d5e320d689b2b71a542332a01b","user":{"login":"kiszk","id":1315079,"node_id":"MDQ6VXNlcjEzMTUwNzk=","avatar_url":"https://avatars2.githubusercontent.com/u/1315079?v=4","gravatar_id":"","url":"https://api.github.com/users/kiszk","html_url":"https://github.com/kiszk","followers_url":"https://api.github.com/users/kiszk/followers","following_url":"https://api.github.com/users/kiszk/following{/other_user}","gists_url":"https://api.github.com/users/kiszk/gists{/gist_id}","starred_url":"https://api.github.com/users/kiszk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kiszk/subscriptions","organizations_url":"https://api.github.com/users/kiszk/orgs","repos_url":"https://api.github.com/users/kiszk/repos","events_url":"https://api.github.com/users/kiszk/events{/privacy}","received_events_url":"https://api.github.com/users/kiszk/received_events","type":"User","site_admin":false},"body":"I think that this code always allocate `ArrayBuffer()` regardless of existence of a value.\r\nIf allocating an array is acceptable, the following code looks simpler.\r\n```\r\nval value = new ArrayBuffer[RpcCallContext](numTasks);\r\nval requests = syncRequestsByStageIdAndAttempt.putIfAbsent((stageId, stageAttemptId), value)\r\nif (requests != null) {\r\n  requests\r\n} else {\r\n  value\r\n}\r\n```\r\n\r\nIf allocating an array is not acceptable, we may need to use `computeIfAbsent` like this\r\n```\r\nprivate val allocateArrayBuffer = new JFunction[Int, ArrayBuffer[RpcCallContext]] {\r\n  override def apply(numTasks: Int): ArrayBuffer[RpcCallContext]= new ArrayBuffer[RpcCallContext](numTasks)\r\n}\r\n\r\nprivate def getOrInitSyncRequests(...): ArrayBuffer[RpcCallContext] = {\r\n  syncRequestsByStageIdAndAttempt.putIfAbsent((stageId, stageAttemptId), allocateArrayBuffer)\r\n}\r\n```\r\n","created_at":"2018-08-02T19:38:03Z","updated_at":"2018-08-02T19:38:04Z","html_url":"https://github.com/apache/spark/pull/21898#discussion_r207348484","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/21898","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/207348484"},"html":{"href":"https://github.com/apache/spark/pull/21898#discussion_r207348484"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/21898"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/21898","id":204504131,"node_id":"MDExOlB1bGxSZXF1ZXN0MjA0NTA0MTMx","html_url":"https://github.com/apache/spark/pull/21898","diff_url":"https://github.com/apache/spark/pull/21898.diff","patch_url":"https://github.com/apache/spark/pull/21898.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/21898","number":21898,"state":"open","locked":false,"title":"[SPARK-24817][Core] Implement BarrierTaskContext.barrier()","user":{"login":"jiangxb1987","id":4784782,"node_id":"MDQ6VXNlcjQ3ODQ3ODI=","avatar_url":"https://avatars1.githubusercontent.com/u/4784782?v=4","gravatar_id":"","url":"https://api.github.com/users/jiangxb1987","html_url":"https://github.com/jiangxb1987","followers_url":"https://api.github.com/users/jiangxb1987/followers","following_url":"https://api.github.com/users/jiangxb1987/following{/other_user}","gists_url":"https://api.github.com/users/jiangxb1987/gists{/gist_id}","starred_url":"https://api.github.com/users/jiangxb1987/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jiangxb1987/subscriptions","organizations_url":"https://api.github.com/users/jiangxb1987/orgs","repos_url":"https://api.github.com/users/jiangxb1987/repos","events_url":"https://api.github.com/users/jiangxb1987/events{/privacy}","received_events_url":"https://api.github.com/users/jiangxb1987/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nImplement BarrierTaskContext.barrier(), to support global sync between all the tasks in a barrier stage.\r\nThe function set a global barrier and waits until all tasks in this stage hit this barrier. Similar to MPI_Barrier function in MPI, the barrier() function call blocks until all tasks in the same stage have reached this routine. The global sync shall finish immediately once all tasks in the same barrier stage reaches the same barrier.\r\n\r\nThis PR implements BarrierTaskContext.barrier() based on netty-based RPC client, introduces new `BarrierCoordinator` and new `BarrierCoordinatorMessage`, and new config to handle timeout issue.\r\n\r\n## How was this patch tested?\r\nAdd `BarrierTaskContextSuite` to test `BarrierTaskContext.barrier()`","created_at":"2018-07-27T18:16:39Z","updated_at":"2018-08-02T19:38:04Z","closed_at":null,"merged_at":null,"merge_commit_sha":"b463532b460b0ad6a22dfa2c88279155af96121d","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/21898/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/21898/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/21898/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/77d082b1601d09d5e320d689b2b71a542332a01b","head":{"label":"jiangxb1987:taskcontext.barrier","ref":"taskcontext.barrier","sha":"77d082b1601d09d5e320d689b2b71a542332a01b","user":{"login":"jiangxb1987","id":4784782,"node_id":"MDQ6VXNlcjQ3ODQ3ODI=","avatar_url":"https://avatars1.githubusercontent.com/u/4784782?v=4","gravatar_id":"","url":"https://api.github.com/users/jiangxb1987","html_url":"https://github.com/jiangxb1987","followers_url":"https://api.github.com/users/jiangxb1987/followers","following_url":"https://api.github.com/users/jiangxb1987/following{/other_user}","gists_url":"https://api.github.com/users/jiangxb1987/gists{/gist_id}","starred_url":"https://api.github.com/users/jiangxb1987/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jiangxb1987/subscriptions","organizations_url":"https://api.github.com/users/jiangxb1987/orgs","repos_url":"https://api.github.com/users/jiangxb1987/repos","events_url":"https://api.github.com/users/jiangxb1987/events{/privacy}","received_events_url":"https://api.github.com/users/jiangxb1987/received_events","type":"User","site_admin":false},"repo":{"id":35214686,"node_id":"MDEwOlJlcG9zaXRvcnkzNTIxNDY4Ng==","name":"spark","full_name":"jiangxb1987/spark","owner":{"login":"jiangxb1987","id":4784782,"node_id":"MDQ6VXNlcjQ3ODQ3ODI=","avatar_url":"https://avatars1.githubusercontent.com/u/4784782?v=4","gravatar_id":"","url":"https://api.github.com/users/jiangxb1987","html_url":"https://github.com/jiangxb1987","followers_url":"https://api.github.com/users/jiangxb1987/followers","following_url":"https://api.github.com/users/jiangxb1987/following{/other_user}","gists_url":"https://api.github.com/users/jiangxb1987/gists{/gist_id}","starred_url":"https://api.github.com/users/jiangxb1987/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jiangxb1987/subscriptions","organizations_url":"https://api.github.com/users/jiangxb1987/orgs","repos_url":"https://api.github.com/users/jiangxb1987/repos","events_url":"https://api.github.com/users/jiangxb1987/events{/privacy}","received_events_url":"https://api.github.com/users/jiangxb1987/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jiangxb1987/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/jiangxb1987/spark","forks_url":"https://api.github.com/repos/jiangxb1987/spark/forks","keys_url":"https://api.github.com/repos/jiangxb1987/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jiangxb1987/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jiangxb1987/spark/teams","hooks_url":"https://api.github.com/repos/jiangxb1987/spark/hooks","issue_events_url":"https://api.github.com/repos/jiangxb1987/spark/issues/events{/number}","events_url":"https://api.github.com/repos/jiangxb1987/spark/events","assignees_url":"https://api.github.com/repos/jiangxb1987/spark/assignees{/user}","branches_url":"https://api.github.com/repos/jiangxb1987/spark/branches{/branch}","tags_url":"https://api.github.com/repos/jiangxb1987/spark/tags","blobs_url":"https://api.github.com/repos/jiangxb1987/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jiangxb1987/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jiangxb1987/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/jiangxb1987/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jiangxb1987/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/jiangxb1987/spark/languages","stargazers_url":"https://api.github.com/repos/jiangxb1987/spark/stargazers","contributors_url":"https://api.github.com/repos/jiangxb1987/spark/contributors","subscribers_url":"https://api.github.com/repos/jiangxb1987/spark/subscribers","subscription_url":"https://api.github.com/repos/jiangxb1987/spark/subscription","commits_url":"https://api.github.com/repos/jiangxb1987/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/jiangxb1987/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/jiangxb1987/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/jiangxb1987/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/jiangxb1987/spark/contents/{+path}","compare_url":"https://api.github.com/repos/jiangxb1987/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jiangxb1987/spark/merges","archive_url":"https://api.github.com/repos/jiangxb1987/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jiangxb1987/spark/downloads","issues_url":"https://api.github.com/repos/jiangxb1987/spark/issues{/number}","pulls_url":"https://api.github.com/repos/jiangxb1987/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/jiangxb1987/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/jiangxb1987/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jiangxb1987/spark/labels{/name}","releases_url":"https://api.github.com/repos/jiangxb1987/spark/releases{/id}","deployments_url":"https://api.github.com/repos/jiangxb1987/spark/deployments","created_at":"2015-05-07T10:48:35Z","updated_at":"2018-07-15T14:39:33Z","pushed_at":"2018-08-02T17:49:02Z","git_url":"git://github.com/jiangxb1987/spark.git","ssh_url":"git@github.com:jiangxb1987/spark.git","clone_url":"https://github.com/jiangxb1987/spark.git","svn_url":"https://github.com/jiangxb1987/spark","homepage":null,"size":267597,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"275415777b84b82aa5409e6577e1efaff1d989e7","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"node_id":"MDEwOlJlcG9zaXRvcnkxNzE2NTY1OA==","name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2018-08-02T19:23:57Z","pushed_at":"2018-08-02T19:32:53Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":278388,"stargazers_count":18279,"watchers_count":18279,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":16560,"mirror_url":null,"archived":false,"open_issues_count":497,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":16560,"open_issues":497,"watchers":18279,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/21898"},"html":{"href":"https://github.com/apache/spark/pull/21898"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/21898"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/21898/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/21898/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/21898/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/77d082b1601d09d5e320d689b2b71a542332a01b"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"kiszk"}}