{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/62361823","id":62361823,"diff_hunk":"@@ -428,40 +503,89 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf)\n     }\n   }\n \n+  private def removeBroadcast(bcast: Broadcast[_]): Unit = {\n+    if (null != bcast) {\n+      broadcastManager.unbroadcast(bcast.id,\n+        removeFromDriver = true, blocking = false)\n+    }\n+  }\n+\n+  private def clearCachedBroadcast(): Unit = {\n+    for (cached <- cachedSerializedBroadcast) removeBroadcast(cached._2)\n+    cachedSerializedBroadcast.clear()\n+  }\n+\n   def getSerializedMapOutputStatuses(shuffleId: Int): Array[Byte] = {\n     var statuses: Array[MapStatus] = null\n+    var retBytes: Array[Byte] = null\n     var epochGotten: Long = -1\n-    epochLock.synchronized {\n-      if (epoch > cacheEpoch) {\n-        cachedSerializedStatuses.clear()\n-        cacheEpoch = epoch\n-      }\n-      cachedSerializedStatuses.get(shuffleId) match {\n-        case Some(bytes) =>\n-          return bytes\n-        case None =>\n-          statuses = mapStatuses.getOrElse(shuffleId, Array[MapStatus]())\n-          epochGotten = epoch\n+\n+    // Check to see if we have a cached version, returns true if it does\n+    // and has side effect of setting retBytes.  If not returns false\n+    // with side effect of setting statuses\n+    def checkCachedStatuses(): Boolean = {\n+      epochLock.synchronized {\n+        if (epoch > cacheEpoch) {\n+          cachedSerializedStatuses.clear()\n+          clearCachedBroadcast()\n+          cacheEpoch = epoch\n+        }\n+        cachedSerializedStatuses.get(shuffleId) match {\n+          case Some(bytes) =>\n+            retBytes = bytes\n+            true\n+          case None =>\n+            logDebug(\"cached status not found for : \" + shuffleId)\n+            statuses = mapStatuses.getOrElse(shuffleId, Array[MapStatus]())\n+            epochGotten = epoch\n+            false\n+        }\n       }\n     }\n-    // If we got here, we failed to find the serialized locations in the cache, so we pulled\n-    // out a snapshot of the locations as \"statuses\"; let's serialize and return that\n-    val bytes = MapOutputTracker.serializeMapStatuses(statuses)\n-    logInfo(\"Size of output statuses for shuffle %d is %d bytes\".format(shuffleId, bytes.length))\n-    // Add them into the table only if the epoch hasn't changed while we were working\n-    epochLock.synchronized {\n-      if (epoch == epochGotten) {\n-        cachedSerializedStatuses(shuffleId) = bytes\n+\n+    if (checkCachedStatuses()) return retBytes\n+    var shuffleIdLock = shuffleIdLocks.get(shuffleId)\n+    if (null == shuffleIdLock) {\n+      val newLock = new Object()\n+      // in general, this condition should be false - but good to be paranoid\n+      val prevLock = shuffleIdLocks.putIfAbsent(shuffleId, newLock)","path":"core/src/main/scala/org/apache/spark/MapOutputTracker.scala","position":235,"original_position":238,"commit_id":"0382155ef73437e490abdf83af2b232adbab0eb6","original_commit_id":"4d46b9607e0ea666abd01b32acd9a244eacd1570","user":{"login":"davies","id":40902,"avatar_url":"https://avatars.githubusercontent.com/u/40902?v=3","gravatar_id":"","url":"https://api.github.com/users/davies","html_url":"https://github.com/davies","followers_url":"https://api.github.com/users/davies/followers","following_url":"https://api.github.com/users/davies/following{/other_user}","gists_url":"https://api.github.com/users/davies/gists{/gist_id}","starred_url":"https://api.github.com/users/davies/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davies/subscriptions","organizations_url":"https://api.github.com/users/davies/orgs","repos_url":"https://api.github.com/users/davies/repos","events_url":"https://api.github.com/users/davies/events{/privacy}","received_events_url":"https://api.github.com/users/davies/received_events","type":"User","site_admin":false},"body":"Thanks for the explain, that make sense.","created_at":"2016-05-06T17:29:20Z","updated_at":"2016-05-06T17:29:20Z","html_url":"https://github.com/apache/spark/pull/12113#discussion_r62361823","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/12113","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/62361823"},"html":{"href":"https://github.com/apache/spark/pull/12113#discussion_r62361823"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/12113"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/12113","id":64988721,"html_url":"https://github.com/apache/spark/pull/12113","diff_url":"https://github.com/apache/spark/pull/12113.diff","patch_url":"https://github.com/apache/spark/pull/12113.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/12113","number":12113,"state":"open","locked":false,"title":"[SPARK-1239] Improve fetching of map output statuses","user":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"body":"The main issue we are trying to solve is the memory bloat of the Driver when tasks request the map output statuses.  This means with a large number of tasks you either need a huge amount of memory on Driver or you have to repartition to smaller number.  This makes it really difficult to run over say 50000 tasks.\r\n\r\nThe main issues that cause the memory bloat are:\r\n1) no flow control on sending the map output status responses.  We serialize the map status output  and then hand off to netty to send.  netty is sending asynchronously and it can't send them fast enough to keep up with incoming requests so we end up with lots of copies of the serialized map output statuses sitting there and this causes huge bloat when you have 10's of thousands of tasks and map output status is in the 10's of MB.\r\n2) When initial reduce tasks are started up, they all request the map output statuses from the Driver. These requests are handled by multiple threads in parallel so even though we check to see if we have a cached version, initially when we don't have a cached version yet, many of initial requests can all end up serializing the exact same map output statuses.\r\n\r\nThis patch does a couple of things:\r\n- When the map output status size is over a threshold (default 512K) then it uses broadcast to send the map statuses.  This means we no longer serialize a large map output status and thus we don't have issues with memory bloat.  the messages sizes are now in the 300-400 byte range and the map status output are broadcast. If its under the threadshold it sends it as before, the message contains the DIRECT indicator now.\r\n- synchronize the incoming requests to allow one thread to cache the serialized output and broadcast the map output status  that can then be used by everyone else.  This ensures we don't create multiple broadcast variables when we don't need to.  To ensure this happens I added a second thread pool which the Dispatcher hands the requests to so that those threads can block without blocking the main dispatcher threads (which would cause things like heartbeats and such not to come through)\r\n\r\nNote that some of design and code was contributed by @mridulm\r\n\r\n## How was this patch tested?\r\n\r\nUnit tests and a lot of manually testing. \r\nRan with akka and netty rpc. Ran with both dynamic allocation on and off.\r\n\r\none of the large jobs I used to test this was a join of 15TB of data.  it had 200,000 map tasks, and  20,000 reduce tasks. Executors ranged from 200 to 2000.  This job ran successfully with 5GB of memory on the driver with these changes. Without these changes I was using 20GB and only had 500 reduce tasks.  The job has 50mb of serialized map output statuses and took roughly the same amount of time for the executors to get the map output statuses as before.\r\n\r\nRan a variety of other jobs, from large wordcounts to small ones not using broadcasts. \r\n","created_at":"2016-04-01T18:51:17Z","updated_at":"2016-05-06T17:29:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8a92af0aae9045d4c5ad9ad797c1d6e53abd4944","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/12113/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/12113/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/12113/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/0382155ef73437e490abdf83af2b232adbab0eb6","head":{"label":"tgravescs:SPARK-1239","ref":"SPARK-1239","sha":"0382155ef73437e490abdf83af2b232adbab0eb6","user":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"repo":{"id":17249522,"name":"spark","full_name":"tgravescs/spark","owner":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tgravescs/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/tgravescs/spark","forks_url":"https://api.github.com/repos/tgravescs/spark/forks","keys_url":"https://api.github.com/repos/tgravescs/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tgravescs/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tgravescs/spark/teams","hooks_url":"https://api.github.com/repos/tgravescs/spark/hooks","issue_events_url":"https://api.github.com/repos/tgravescs/spark/issues/events{/number}","events_url":"https://api.github.com/repos/tgravescs/spark/events","assignees_url":"https://api.github.com/repos/tgravescs/spark/assignees{/user}","branches_url":"https://api.github.com/repos/tgravescs/spark/branches{/branch}","tags_url":"https://api.github.com/repos/tgravescs/spark/tags","blobs_url":"https://api.github.com/repos/tgravescs/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tgravescs/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tgravescs/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/tgravescs/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tgravescs/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/tgravescs/spark/languages","stargazers_url":"https://api.github.com/repos/tgravescs/spark/stargazers","contributors_url":"https://api.github.com/repos/tgravescs/spark/contributors","subscribers_url":"https://api.github.com/repos/tgravescs/spark/subscribers","subscription_url":"https://api.github.com/repos/tgravescs/spark/subscription","commits_url":"https://api.github.com/repos/tgravescs/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/tgravescs/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/tgravescs/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/tgravescs/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/tgravescs/spark/contents/{+path}","compare_url":"https://api.github.com/repos/tgravescs/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tgravescs/spark/merges","archive_url":"https://api.github.com/repos/tgravescs/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tgravescs/spark/downloads","issues_url":"https://api.github.com/repos/tgravescs/spark/issues{/number}","pulls_url":"https://api.github.com/repos/tgravescs/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/tgravescs/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/tgravescs/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tgravescs/spark/labels{/name}","releases_url":"https://api.github.com/repos/tgravescs/spark/releases{/id}","deployments_url":"https://api.github.com/repos/tgravescs/spark/deployments","created_at":"2014-02-27T13:26:24Z","updated_at":"2015-05-08T13:44:25Z","pushed_at":"2016-05-05T19:20:33Z","git_url":"git://github.com/tgravescs/spark.git","ssh_url":"git@github.com:tgravescs/spark.git","clone_url":"https://github.com/tgravescs/spark.git","svn_url":"https://github.com/tgravescs/spark","homepage":null,"size":158232,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"77361a433adce109c2b752b11dda25b56eca0352","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","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,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","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":"2016-05-06T16:13:12Z","pushed_at":"2016-05-06T17:28: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":476853,"stargazers_count":8415,"watchers_count":8415,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":7817,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":429,"forks":7817,"open_issues":429,"watchers":8415,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/12113"},"html":{"href":"https://github.com/apache/spark/pull/12113"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/12113"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/12113/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/12113/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/12113/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/0382155ef73437e490abdf83af2b232adbab0eb6"}}}}