{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/27847838","id":27847838,"diff_hunk":"@@ -0,0 +1,222 @@\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.api.r\n+\n+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, DataOutputStream}\n+\n+import scala.collection.mutable.HashMap\n+\n+import io.netty.channel.ChannelHandler.Sharable\n+import io.netty.channel.{ChannelHandlerContext, SimpleChannelInboundHandler}\n+\n+import org.apache.spark.Logging\n+import org.apache.spark.api.r.SerDe._\n+\n+/**\n+ * Handler for RBackend\n+ * TODO: This is marked as sharable to get a handle to RBackend. Is it safe to re-use\n+ * this across connections ?\n+ */\n+@Sharable\n+private[r] class RBackendHandler(server: RBackend)\n+  extends SimpleChannelInboundHandler[Array[Byte]] with Logging {\n+\n+  override def channelRead0(ctx: ChannelHandlerContext, msg: Array[Byte]) {\n+    val bis = new ByteArrayInputStream(msg)\n+    val dis = new DataInputStream(bis)\n+\n+    val bos = new ByteArrayOutputStream()\n+    val dos = new DataOutputStream(bos)\n+\n+    // First bit is isStatic\n+    val isStatic = readBoolean(dis)\n+    val objId = readString(dis)\n+    val methodName = readString(dis)\n+    val numArgs = readInt(dis)\n+\n+    if (objId == \"SparkRHandler\") {\n+      methodName match {\n+        case \"stopBackend\" =>\n+          writeInt(dos, 0)\n+          writeType(dos, \"void\")\n+          server.close()\n+        case \"rm\" =>\n+          try {\n+            val t = readObjectType(dis)\n+            assert(t == 'c')\n+            val objToRemove = readString(dis)\n+            JVMObjectTracker.remove(objToRemove)\n+            writeInt(dos, 0)\n+            writeObject(dos, null)\n+          } catch {\n+            case e: Exception =>\n+              logError(s\"Removing $objId failed\", e)\n+              writeInt(dos, -1)\n+          }\n+        case _ => dos.writeInt(-1)\n+      }\n+    } else {\n+      handleMethodCall(isStatic, objId, methodName, numArgs, dis, dos)\n+    }\n+\n+    val reply = bos.toByteArray\n+    ctx.write(reply)\n+  }\n+  \n+  override def channelReadComplete(ctx: ChannelHandlerContext) {\n+    ctx.flush()\n+  }\n+\n+  override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {\n+    // Close the connection when an exception is raised.\n+    cause.printStackTrace()\n+    ctx.close()\n+  }\n+\n+  def handleMethodCall(\n+      isStatic: Boolean,\n+      objId: String,\n+      methodName: String,\n+      numArgs: Int,\n+      dis: DataInputStream,\n+      dos: DataOutputStream) {\n+    var obj: Object = null\n+    try {\n+      val cls = if (isStatic) {\n+        Class.forName(objId)\n+      } else {\n+        JVMObjectTracker.get(objId) match {\n+          case None => throw new IllegalArgumentException(\"Object not found \" + objId)\n+          case Some(o) =>\n+            obj = o\n+            o.getClass\n+        }\n+      }\n+\n+      val args = readArgs(numArgs, dis)\n+\n+      val methods = cls.getMethods\n+      val selectedMethods = methods.filter(m => m.getName == methodName)\n+      if (selectedMethods.length > 0) {\n+        val methods = selectedMethods.filter { x =>\n+          matchMethod(numArgs, args, x.getParameterTypes)\n+        }\n+        if (methods.isEmpty) {\n+          logWarning(s\"cannot find matching method ${cls}.$methodName. \"\n+            + s\"Candidates are:\")\n+          selectedMethods.foreach { method =>\n+            logWarning(s\"$methodName(${method.getParameterTypes.mkString(\",\")})\")\n+          }\n+          throw new Exception(s\"No matched method found for $cls.$methodName\")\n+        }\n+        val ret = methods.head.invoke(obj, args:_*)\n+\n+        // Write status bit\n+        writeInt(dos, 0)\n+        writeObject(dos, ret.asInstanceOf[AnyRef])\n+      } else if (methodName == \"<init>\") {\n+        // methodName should be \"<init>\" for constructor\n+        val ctor = cls.getConstructors.filter { x =>\n+          matchMethod(numArgs, args, x.getParameterTypes)\n+        }.head\n+\n+        val obj = ctor.newInstance(args:_*)\n+\n+        writeInt(dos, 0)\n+        writeObject(dos, obj.asInstanceOf[AnyRef])\n+      } else {\n+        throw new IllegalArgumentException(\"invalid method \" + methodName + \" for object \" + objId)\n+      }\n+    } catch {\n+      case e: Exception =>\n+        logError(s\"$methodName on $objId failed\", e)\n+        writeInt(dos, -1)\n+    }\n+  }\n+\n+  // Read a number of arguments from the data input stream\n+  def readArgs(numArgs: Int, dis: DataInputStream): Array[java.lang.Object] =  {","path":"core/src/main/scala/org/apache/spark/api/r/RBackendHandler.scala","position":null,"original_position":153,"commit_id":"64eda24017d1d05c3dfba3a84c96b557cbd2ca76","original_commit_id":"0e788c08f3b418acc05d4d27298b65de8b6f8407","user":{"login":"shivaram","id":143893,"avatar_url":"https://avatars.githubusercontent.com/u/143893?v=3","gravatar_id":"","url":"https://api.github.com/users/shivaram","html_url":"https://github.com/shivaram","followers_url":"https://api.github.com/users/shivaram/followers","following_url":"https://api.github.com/users/shivaram/following{/other_user}","gists_url":"https://api.github.com/users/shivaram/gists{/gist_id}","starred_url":"https://api.github.com/users/shivaram/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shivaram/subscriptions","organizations_url":"https://api.github.com/users/shivaram/orgs","repos_url":"https://api.github.com/users/shivaram/repos","events_url":"https://api.github.com/users/shivaram/events{/privacy}","received_events_url":"https://api.github.com/users/shivaram/received_events","type":"User","site_admin":false},"body":"Fixed","created_at":"2015-04-07T02:32:24Z","updated_at":"2015-04-07T02:32:24Z","html_url":"https://github.com/apache/spark/pull/5096#discussion_r27847838","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/5096","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/27847838"},"html":{"href":"https://github.com/apache/spark/pull/5096#discussion_r27847838"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/5096"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5096","id":31576742,"html_url":"https://github.com/apache/spark/pull/5096","diff_url":"https://github.com/apache/spark/pull/5096.diff","patch_url":"https://github.com/apache/spark/pull/5096.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/5096","number":5096,"state":"open","locked":false,"title":"[SPARK-5654] Integrate SparkR","user":{"login":"shivaram","id":143893,"avatar_url":"https://avatars.githubusercontent.com/u/143893?v=3","gravatar_id":"","url":"https://api.github.com/users/shivaram","html_url":"https://github.com/shivaram","followers_url":"https://api.github.com/users/shivaram/followers","following_url":"https://api.github.com/users/shivaram/following{/other_user}","gists_url":"https://api.github.com/users/shivaram/gists{/gist_id}","starred_url":"https://api.github.com/users/shivaram/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shivaram/subscriptions","organizations_url":"https://api.github.com/users/shivaram/orgs","repos_url":"https://api.github.com/users/shivaram/repos","events_url":"https://api.github.com/users/shivaram/events{/privacy}","received_events_url":"https://api.github.com/users/shivaram/received_events","type":"User","site_admin":false},"body":"This pull requests integrates SparkR, an R frontend for Spark. The SparkR package contains both RDD and DataFrame APIs in R and is integrated with Spark's submission scripts to work on different cluster managers.\r\n\r\nSome integration points that would be great to get feedback on:\r\n\r\n1. Build procedure: SparkR requires R to be installed on the machine to be built. Right now we have a new Maven profile `-PsparkR` that can be used to enable SparkR builds\r\n\r\n2. YARN cluster mode: The R package that is built needs to be present on the driver and all the worker nodes during execution. The R package location is currently set using SPARK_HOME, but this might not work on YARN cluster mode.\r\n\r\nThe SparkR package represents the work of many contributors and attached below is a list of people along with areas they worked on\r\n\r\nedwardt (@edwart) - Documentation improvements\r\nFelix Cheung (@felixcheung) - Documentation improvements\r\nHossein Falaki (@falaki)  - Documentation improvements\r\nChris Freeman (@cafreeman) - DataFrame API, Programming Guide\r\nTodd Gao (@7c00) - R worker Internals\r\nRyan Hafen (@hafen) - SparkR Internals\r\nQian Huang (@hqzizania) - RDD API\r\nHao Lin (@hlin09) - RDD API, Closure cleaner\r\nEvert Lammerts (@evertlammerts) - DataFrame API\r\nDavies Liu (@davies) - DataFrame API, R worker internals, Merging with Spark \r\nYi Lu (@lythesia) - RDD API, Worker internals\r\nMatt Massie (@massie) - Jenkins build\r\nHarihar Nahak (@hnahak87) - SparkR examples\r\nOscar Olmedo (@oscaroboto) - Spark configuration\r\nAntonio Piccolboni (@piccolbo) - SparkR examples, Namespace bug fixes\r\nDan Putler (@dputler) - Dataframe API, SparkR Install Guide\r\nAshutosh Raina (@ashutoshraina) - Build improvements\r\nJosh Rosen (@joshrosen) - Travis CI build\r\nSun Rui (@sun-rui)- RDD API, JVM Backend, Shuffle improvements\r\nShivaram Venkataraman (@shivaram) - RDD API, JVM Backend, Worker Internals\r\nZongheng Yang (@concretevitamin) - RDD API, Pipelined RDDs, Examples and EC2 guide","created_at":"2015-03-19T23:35:13Z","updated_at":"2015-04-07T02:32:24Z","closed_at":null,"merged_at":null,"merge_commit_sha":"6f5d0294791ccb2d2ae6f1a1832fbd08ec104897","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/5096/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/5096/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/5096/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/64eda24017d1d05c3dfba3a84c96b557cbd2ca76","head":{"label":"amplab-extras:R","ref":"R","sha":"64eda24017d1d05c3dfba3a84c96b557cbd2ca76","user":{"login":"amplab-extras","id":6403540,"avatar_url":"https://avatars.githubusercontent.com/u/6403540?v=3","gravatar_id":"","url":"https://api.github.com/users/amplab-extras","html_url":"https://github.com/amplab-extras","followers_url":"https://api.github.com/users/amplab-extras/followers","following_url":"https://api.github.com/users/amplab-extras/following{/other_user}","gists_url":"https://api.github.com/users/amplab-extras/gists{/gist_id}","starred_url":"https://api.github.com/users/amplab-extras/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amplab-extras/subscriptions","organizations_url":"https://api.github.com/users/amplab-extras/orgs","repos_url":"https://api.github.com/users/amplab-extras/repos","events_url":"https://api.github.com/users/amplab-extras/events{/privacy}","received_events_url":"https://api.github.com/users/amplab-extras/received_events","type":"Organization","site_admin":false},"repo":{"id":32358130,"name":"spark","full_name":"amplab-extras/spark","owner":{"login":"amplab-extras","id":6403540,"avatar_url":"https://avatars.githubusercontent.com/u/6403540?v=3","gravatar_id":"","url":"https://api.github.com/users/amplab-extras","html_url":"https://github.com/amplab-extras","followers_url":"https://api.github.com/users/amplab-extras/followers","following_url":"https://api.github.com/users/amplab-extras/following{/other_user}","gists_url":"https://api.github.com/users/amplab-extras/gists{/gist_id}","starred_url":"https://api.github.com/users/amplab-extras/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amplab-extras/subscriptions","organizations_url":"https://api.github.com/users/amplab-extras/orgs","repos_url":"https://api.github.com/users/amplab-extras/repos","events_url":"https://api.github.com/users/amplab-extras/events{/privacy}","received_events_url":"https://api.github.com/users/amplab-extras/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/amplab-extras/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/amplab-extras/spark","forks_url":"https://api.github.com/repos/amplab-extras/spark/forks","keys_url":"https://api.github.com/repos/amplab-extras/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/amplab-extras/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/amplab-extras/spark/teams","hooks_url":"https://api.github.com/repos/amplab-extras/spark/hooks","issue_events_url":"https://api.github.com/repos/amplab-extras/spark/issues/events{/number}","events_url":"https://api.github.com/repos/amplab-extras/spark/events","assignees_url":"https://api.github.com/repos/amplab-extras/spark/assignees{/user}","branches_url":"https://api.github.com/repos/amplab-extras/spark/branches{/branch}","tags_url":"https://api.github.com/repos/amplab-extras/spark/tags","blobs_url":"https://api.github.com/repos/amplab-extras/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/amplab-extras/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/amplab-extras/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/amplab-extras/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/amplab-extras/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/amplab-extras/spark/languages","stargazers_url":"https://api.github.com/repos/amplab-extras/spark/stargazers","contributors_url":"https://api.github.com/repos/amplab-extras/spark/contributors","subscribers_url":"https://api.github.com/repos/amplab-extras/spark/subscribers","subscription_url":"https://api.github.com/repos/amplab-extras/spark/subscription","commits_url":"https://api.github.com/repos/amplab-extras/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/amplab-extras/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/amplab-extras/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/amplab-extras/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/amplab-extras/spark/contents/{+path}","compare_url":"https://api.github.com/repos/amplab-extras/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/amplab-extras/spark/merges","archive_url":"https://api.github.com/repos/amplab-extras/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/amplab-extras/spark/downloads","issues_url":"https://api.github.com/repos/amplab-extras/spark/issues{/number}","pulls_url":"https://api.github.com/repos/amplab-extras/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/amplab-extras/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/amplab-extras/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/amplab-extras/spark/labels{/name}","releases_url":"https://api.github.com/repos/amplab-extras/spark/releases{/id}","created_at":"2015-03-16T23:01:05Z","updated_at":"2015-04-07T02:27:50Z","pushed_at":"2015-04-07T02:27:48Z","git_url":"git://github.com/amplab-extras/spark.git","ssh_url":"git@github.com:amplab-extras/spark.git","clone_url":"https://github.com/amplab-extras/spark.git","svn_url":"https://github.com/amplab-extras/spark","homepage":null,"size":100361,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":1,"forks":2,"open_issues":1,"watchers":0,"default_branch":"R"}},"base":{"label":"apache:master","ref":"master","sha":"30363ede8635f2548e444697dbcf60a795b61a84","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}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-04-07T01:01:19Z","pushed_at":"2015-04-07T02:27:50Z","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":1475685,"stargazers_count":3513,"watchers_count":3513,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3026,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":315,"forks":3026,"open_issues":315,"watchers":3513,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/5096"},"html":{"href":"https://github.com/apache/spark/pull/5096"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/5096"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/5096/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/5096/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/5096/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/64eda24017d1d05c3dfba3a84c96b557cbd2ca76"}}}}