{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/84616347","pull_request_review_id":5382445,"id":84616347,"diff_hunk":"@@ -0,0 +1,229 @@\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.scheduler\n+\n+import scala.collection.mutable.ArrayBuffer\n+import scala.collection.mutable.PriorityQueue\n+import scala.util.Random\n+\n+import org.apache.spark.internal.{config, Logging}\n+import org.apache.spark.SparkConf\n+import org.apache.spark.util.Utils\n+\n+/** Tracks the current state of the workers with available cores and assigned task list. */\n+private[scheduler] class OfferState(val workOffer: WorkerOffer) {\n+  /** The current remaining cores that can be allocated to tasks. */\n+  var coresAvailable: Int = workOffer.cores\n+  /** The list of tasks that are assigned to this WorkerOffer. */\n+  val tasks = new ArrayBuffer[TaskDescription](coresAvailable)\n+\n+  def assignTask(task: TaskDescription, cpu: Int): Unit = {\n+    tasks += task\n+    coresAvailable -= cpu\n+    assert(coresAvailable >= 0)\n+  }\n+}\n+\n+/**\n+ * TaskAssigner is the base class for all task assigner implementations, and can be\n+ * extended to implement different task scheduling algorithms.\n+ * Together with [[org.apache.spark.scheduler.TaskScheduler TaskScheduler]], TaskAssigner\n+ * is used to assign tasks to workers with available cores. Internally, when TaskScheduler\n+ * performs task assignment given available workers, it first sorts the candidate tasksets,\n+ * and then for each taskset, it takes multiple rounds to request TaskAssigner for task\n+ * assignment with different locality restrictions until there is either no qualified\n+ * workers or no valid tasks to be assigned.\n+ *\n+ * TaskAssigner is responsible to maintain the worker availability state and task assignment\n+ * information. The contract between [[org.apache.spark.scheduler.TaskScheduler TaskScheduler]]\n+ * and TaskAssigner is as follows.\n+ *\n+ * First, TaskScheduler invokes construct() of TaskAssigner to initialize the its internal\n+ * worker states at the beginning of resource offering.\n+ *\n+ * Second, before each round of task assignment for a taskset, TaskScheduler invokes the init()\n+ * of TaskAssigner to initialize the data structure for the round.\n+ *\n+ * Third, when performing real task assignment, hasNext/next() is used by TaskScheduler\n+ * to check the worker availability and retrieve current offering from TaskAssigner.\n+ *\n+ * Fourth, TaskScheduler calls offerAccepted() to notify the TaskAssigner so that\n+ * TaskAssigner can decide whether the current offer is valid or not for the next request.\n+ *\n+ * Fifth, after task assignment is done, TaskScheduler invokes the function tasks to\n+ * retrieve all the task assignment information.\n+ */\n+\n+private[scheduler] sealed abstract class TaskAssigner {\n+  protected var offer: Seq[OfferState] = _\n+  protected var cpuPerTask = 1\n+\n+  protected def withCpuPerTask(cpuPerTask: Int): TaskAssigner = {\n+    this.cpuPerTask = cpuPerTask\n+    this\n+  }\n+\n+  /** The currently assigned offers. */\n+  final def tasks: Seq[ArrayBuffer[TaskDescription]] = offer.map(_.tasks)\n+\n+  /**\n+   * Invoked at the beginning of resource offering to construct the offer with the workoffers.\n+   * By default, offers is randomly shuffled to avoid always placing tasks on the same set of\n+   * workers.\n+   */\n+  def construct(workOffer: Seq[WorkerOffer]): Unit = {\n+    offer = Random.shuffle(workOffer.map(o => new OfferState(o)))\n+  }\n+\n+  /** Invoked at each round of Taskset assignment to initialize the internal structure. */\n+  def init(): Unit\n+\n+  /**\n+   * Tests whether there is offer available to be used inside of one round of Taskset assignment.\n+   * @return  `true` if a subsequent call to `next` will yield an element,\n+   *          `false` otherwise.\n+   */\n+  def hasNext: Boolean\n+\n+  /**\n+   * Produces next worker offer based on the task assignment strategy.\n+   * @return  the next available offer, if `hasNext` is `true`,\n+   *          undefined behavior otherwise.\n+   */\n+  def next(): OfferState\n+\n+  /**\n+   * Invoked by the TaskScheduler to indicate whether the current offer is accepted or not so that\n+   * the assigner can decide whether the current worker is valid for the next offering.\n+   *\n+   * @param isAccepted whether TaskScheduler assigns a task to current offer.\n+   */\n+  def offerAccepted(isAccepted: Boolean): Unit\n+}\n+\n+object TaskAssigner extends Logging {\n+  private val roundrobin = classOf[RoundRobinAssigner].getCanonicalName\n+  private val packed = classOf[PackedAssigner].getCanonicalName\n+  private val balanced = classOf[BalancedAssigner].getCanonicalName\n+  private val assignerMap: Map[String, String] =\n+    Map(\"roundrobin\" -> roundrobin,\n+      \"packed\" -> packed,\n+      \"balanced\" -> balanced)\n+\n+  def init(conf: SparkConf): TaskAssigner = {\n+    val assignerName = conf.get(config.SPARK_SCHEDULER_TASK_ASSIGNER.key, \"roundrobin\")\n+    val className = {\n+      val name = assignerMap.get(assignerName.toLowerCase())\n+      name.getOrElse {\n+        logWarning(s\"$assignerName cannot be constructed, fallback to default $roundrobin.\")\n+        roundrobin\n+      }\n+    }\n+    // The className is valid. No need to catch exceptions.\n+    logInfo(s\"Constructing TaskAssigner as $className\")\n+    Utils.classForName(className).getConstructor().newInstance().asInstanceOf[TaskAssigner]\n+      .withCpuPerTask(cpuPerTask = conf.getInt(\"spark.task.cpus\", 1))\n+  }\n+}\n+\n+/**\n+ * Assigns the task to workers with available cores in roundrobin manner.","path":"core/src/main/scala/org/apache/spark/scheduler/TaskAssigner.scala","position":145,"original_position":145,"commit_id":"6b29002c29fecdbe32159dd0d31f53716630de46","original_commit_id":"6b29002c29fecdbe32159dd0d31f53716630de46","user":{"login":"gatorsmile","id":11567269,"avatar_url":"https://avatars.githubusercontent.com/u/11567269?v=3","gravatar_id":"","url":"https://api.github.com/users/gatorsmile","html_url":"https://github.com/gatorsmile","followers_url":"https://api.github.com/users/gatorsmile/followers","following_url":"https://api.github.com/users/gatorsmile/following{/other_user}","gists_url":"https://api.github.com/users/gatorsmile/gists{/gist_id}","starred_url":"https://api.github.com/users/gatorsmile/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/gatorsmile/subscriptions","organizations_url":"https://api.github.com/users/gatorsmile/orgs","repos_url":"https://api.github.com/users/gatorsmile/repos","events_url":"https://api.github.com/users/gatorsmile/events{/privacy}","received_events_url":"https://api.github.com/users/gatorsmile/received_events","type":"User","site_admin":false},"body":"`in a roundrobin manner`","created_at":"2016-10-24T03:17:20Z","updated_at":"2016-10-24T03:17:20Z","html_url":"https://github.com/apache/spark/pull/15541#discussion_r84616347","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/15541","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/84616347"},"html":{"href":"https://github.com/apache/spark/pull/15541#discussion_r84616347"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/15541"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15541","id":89927393,"html_url":"https://github.com/apache/spark/pull/15541","diff_url":"https://github.com/apache/spark/pull/15541.diff","patch_url":"https://github.com/apache/spark/pull/15541.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/15541","number":15541,"state":"open","locked":false,"title":"[SPARK-17637][Scheduler]Packed scheduling for Spark tasks across executors","user":{"login":"zhzhan","id":4451616,"avatar_url":"https://avatars.githubusercontent.com/u/4451616?v=3","gravatar_id":"","url":"https://api.github.com/users/zhzhan","html_url":"https://github.com/zhzhan","followers_url":"https://api.github.com/users/zhzhan/followers","following_url":"https://api.github.com/users/zhzhan/following{/other_user}","gists_url":"https://api.github.com/users/zhzhan/gists{/gist_id}","starred_url":"https://api.github.com/users/zhzhan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhzhan/subscriptions","organizations_url":"https://api.github.com/users/zhzhan/orgs","repos_url":"https://api.github.com/users/zhzhan/repos","events_url":"https://api.github.com/users/zhzhan/events{/privacy}","received_events_url":"https://api.github.com/users/zhzhan/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nRestructure the code and implement two new task assigner.\r\nPackedAssigner: try to allocate tasks to the executors with least available cores, so that spark can release reserved executors when dynamic allocation is enabled.\r\n\r\nBalancedAssigner: try to allocate tasks to the executors with more available cores in order to balance the workload across all executors.\r\n\r\nBy default, the original round robin assigner is used.\r\n\r\nWe test a pipeline, and new PackedAssigner save around 45% regarding the reserved cpu and memory with dynamic allocation enabled.\r\n\r\n## How was this patch tested?\r\n\r\n(Please explain how this patch was tested. E.g. unit tests, integration tests, manual tests)\r\nBoth unit test in TaskSchedulerImplSuite and manual tests in production pipeline.\r\n","created_at":"2016-10-19T01:26:48Z","updated_at":"2016-10-24T03:17:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"95519ea1df0f981cb453b3fd39f1217b32592f34","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/15541/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/15541/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/15541/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/6b29002c29fecdbe32159dd0d31f53716630de46","head":{"label":"zhzhan:TaskAssigner","ref":"TaskAssigner","sha":"6b29002c29fecdbe32159dd0d31f53716630de46","user":{"login":"zhzhan","id":4451616,"avatar_url":"https://avatars.githubusercontent.com/u/4451616?v=3","gravatar_id":"","url":"https://api.github.com/users/zhzhan","html_url":"https://github.com/zhzhan","followers_url":"https://api.github.com/users/zhzhan/followers","following_url":"https://api.github.com/users/zhzhan/following{/other_user}","gists_url":"https://api.github.com/users/zhzhan/gists{/gist_id}","starred_url":"https://api.github.com/users/zhzhan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhzhan/subscriptions","organizations_url":"https://api.github.com/users/zhzhan/orgs","repos_url":"https://api.github.com/users/zhzhan/repos","events_url":"https://api.github.com/users/zhzhan/events{/privacy}","received_events_url":"https://api.github.com/users/zhzhan/received_events","type":"User","site_admin":false},"repo":{"id":22663923,"name":"spark","full_name":"zhzhan/spark","owner":{"login":"zhzhan","id":4451616,"avatar_url":"https://avatars.githubusercontent.com/u/4451616?v=3","gravatar_id":"","url":"https://api.github.com/users/zhzhan","html_url":"https://github.com/zhzhan","followers_url":"https://api.github.com/users/zhzhan/followers","following_url":"https://api.github.com/users/zhzhan/following{/other_user}","gists_url":"https://api.github.com/users/zhzhan/gists{/gist_id}","starred_url":"https://api.github.com/users/zhzhan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhzhan/subscriptions","organizations_url":"https://api.github.com/users/zhzhan/orgs","repos_url":"https://api.github.com/users/zhzhan/repos","events_url":"https://api.github.com/users/zhzhan/events{/privacy}","received_events_url":"https://api.github.com/users/zhzhan/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/zhzhan/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/zhzhan/spark","forks_url":"https://api.github.com/repos/zhzhan/spark/forks","keys_url":"https://api.github.com/repos/zhzhan/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/zhzhan/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/zhzhan/spark/teams","hooks_url":"https://api.github.com/repos/zhzhan/spark/hooks","issue_events_url":"https://api.github.com/repos/zhzhan/spark/issues/events{/number}","events_url":"https://api.github.com/repos/zhzhan/spark/events","assignees_url":"https://api.github.com/repos/zhzhan/spark/assignees{/user}","branches_url":"https://api.github.com/repos/zhzhan/spark/branches{/branch}","tags_url":"https://api.github.com/repos/zhzhan/spark/tags","blobs_url":"https://api.github.com/repos/zhzhan/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/zhzhan/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/zhzhan/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/zhzhan/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/zhzhan/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/zhzhan/spark/languages","stargazers_url":"https://api.github.com/repos/zhzhan/spark/stargazers","contributors_url":"https://api.github.com/repos/zhzhan/spark/contributors","subscribers_url":"https://api.github.com/repos/zhzhan/spark/subscribers","subscription_url":"https://api.github.com/repos/zhzhan/spark/subscription","commits_url":"https://api.github.com/repos/zhzhan/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/zhzhan/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/zhzhan/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/zhzhan/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/zhzhan/spark/contents/{+path}","compare_url":"https://api.github.com/repos/zhzhan/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/zhzhan/spark/merges","archive_url":"https://api.github.com/repos/zhzhan/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/zhzhan/spark/downloads","issues_url":"https://api.github.com/repos/zhzhan/spark/issues{/number}","pulls_url":"https://api.github.com/repos/zhzhan/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/zhzhan/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/zhzhan/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/zhzhan/spark/labels{/name}","releases_url":"https://api.github.com/repos/zhzhan/spark/releases{/id}","deployments_url":"https://api.github.com/repos/zhzhan/spark/deployments","created_at":"2014-08-06T00:18:00Z","updated_at":"2016-02-23T21:14:46Z","pushed_at":"2016-10-23T17:49:31Z","git_url":"git://github.com/zhzhan/spark.git","ssh_url":"git@github.com:zhzhan/spark.git","clone_url":"https://github.com/zhzhan/spark.git","svn_url":"https://github.com/zhzhan/spark","homepage":null,"size":236876,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":0,"forks":1,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"a81fba048fabcd413730548ab65955802508d4e4","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-10-24T02:40:35Z","pushed_at":"2016-10-24T03:16:23Z","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":601879,"stargazers_count":10466,"watchers_count":10466,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":9744,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":478,"forks":9744,"open_issues":478,"watchers":10466,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/15541"},"html":{"href":"https://github.com/apache/spark/pull/15541"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/15541"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/15541/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/15541/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/15541/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/6b29002c29fecdbe32159dd0d31f53716630de46"}}}}
{"actor":{"display_login":"gatorsmile"}}