{"action":"created","comment":{"url":"https://api.github.com/repos/grpc/grpc-java/pulls/comments/30325836","id":30325836,"diff_hunk":"@@ -0,0 +1,148 @@\n+/*\n+ * Copyright 2015, Google Inc. All rights reserved.\n+ *\n+ * Redistribution and use in source and binary forms, with or without\n+ * modification, are permitted provided that the following conditions are\n+ * met:\n+ *\n+ *    * Redistributions of source code must retain the above copyright\n+ * notice, this list of conditions and the following disclaimer.\n+ *    * Redistributions in binary form must reproduce the above\n+ * copyright notice, this list of conditions and the following disclaimer\n+ * in the documentation and/or other materials provided with the\n+ * distribution.\n+ *\n+ *    * Neither the name of Google Inc. nor the names of its\n+ * contributors may be used to endorse or promote products derived from\n+ * this software without specific prior written permission.\n+ *\n+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n+ * \"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n+ */\n+\n+package io.grpc.transport.netty;\n+\n+import io.netty.channel.Channel;\n+import io.netty.channel.ChannelFutureListener;\n+import io.netty.channel.ChannelHandlerContext;\n+import io.netty.channel.ChannelPromise;\n+import io.netty.handler.codec.http2.Http2Exception;\n+import io.netty.util.ReferenceCountUtil;\n+\n+import java.util.ArrayDeque;\n+import java.util.concurrent.BlockingQueue;\n+import java.util.concurrent.LinkedBlockingQueue;\n+import java.util.concurrent.atomic.AtomicBoolean;\n+\n+/**\n+ * A queue of pending writes to a {@link Channel} that is flushed as a single unit.\n+ */\n+public abstract class AbstractWriteQueue {\n+\n+  private static final int DEQUE_CHUNK_SIZE = 128;\n+\n+  /**\n+   * {@link Runnable} used to schedule work onto the tail of the event loop.\n+   */\n+  private final Runnable later = new Runnable() {\n+    @Override\n+    public void run() {\n+      channel.write(AbstractWriteQueue.this);\n+    }\n+  };\n+\n+  private final Channel channel;\n+  private final BlockingQueue<Object> queue;\n+  private final AtomicBoolean scheduled = new AtomicBoolean();\n+\n+  /**\n+   * ArrayDeque to copy queue into when flushing in event loop.\n+   */\n+  private final ArrayDeque<Object> writeChunk = new ArrayDeque<Object>(DEQUE_CHUNK_SIZE);\n+\n+  public AbstractWriteQueue(Channel channel) {\n+    this.channel = channel;\n+    queue = new LinkedBlockingQueue<Object>();\n+  }\n+\n+  /**\n+   * Enqueue a flush on the channel.\n+   */\n+  void enqueueFlush() {\n+    if (scheduled.compareAndSet(false, true)) {\n+      // Add the queue to the tail of the event loop so writes will be executed immediately\n+      // inside the event loop. Note DO NOT do channel.write outside the event loop as\n+      // it will not wake up immediately without a flush.\n+      channel.eventLoop().execute(later);\n+    }\n+  }\n+\n+  /**\n+   * Enqueue a write command on the channel.\n+   */\n+  void enqueue(Object msg) {\n+    queue.add(msg);\n+    enqueueFlush();\n+  }\n+\n+  /**\n+   * Enqueue a write command with a completion listener.\n+   */\n+  void enqueue(Object msg, ChannelFutureListener completed) {\n+    queue.add(new Object[]{msg, completed});\n+    enqueueFlush();\n+  }\n+\n+  /**\n+   * Process the queue of commands and dispatch them to the stream. This method is only\n+   * called in the event loop\n+   */\n+  void flush(ChannelHandlerContext ctx, ChannelPromise promise) {\n+    try {\n+      Object event;\n+      boolean flushed = false;\n+      while (queue.drainTo(writeChunk, DEQUE_CHUNK_SIZE) > 0) {\n+        while (writeChunk.size() > 0) {\n+          event = writeChunk.poll();\n+          ChannelPromise eventPromise = ctx.newPromise();\n+          if (event instanceof Object[]) {\n+            eventPromise.addListener((ChannelFutureListener) ((Object[]) event)[1]);\n+            event = ((Object[]) event)[0];\n+          }\n+          try {\n+            handleCommand(event, eventPromise);\n+          } catch (Exception e) {\n+            promise.setFailure(e);\n+            ReferenceCountUtil.release(event);\n+          }\n+        }\n+        flushed = true;","path":"netty/src/main/java/io/grpc/transport/netty/AbstractWriteQueue.java","position":128,"original_position":128,"commit_id":"4ffcc1ef309ad82f3a2c1c91675f6ee8ca64856a","original_commit_id":"4ffcc1ef309ad82f3a2c1c91675f6ee8ca64856a","user":{"login":"nmittler","id":6493296,"avatar_url":"https://avatars.githubusercontent.com/u/6493296?v=3","gravatar_id":"","url":"https://api.github.com/users/nmittler","html_url":"https://github.com/nmittler","followers_url":"https://api.github.com/users/nmittler/followers","following_url":"https://api.github.com/users/nmittler/following{/other_user}","gists_url":"https://api.github.com/users/nmittler/gists{/gist_id}","starred_url":"https://api.github.com/users/nmittler/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nmittler/subscriptions","organizations_url":"https://api.github.com/users/nmittler/orgs","repos_url":"https://api.github.com/users/nmittler/repos","events_url":"https://api.github.com/users/nmittler/events{/privacy}","received_events_url":"https://api.github.com/users/nmittler/received_events","type":"User","site_admin":false},"body":"Why not just flush at the end?","created_at":"2015-05-14T14:39:17Z","updated_at":"2015-05-14T14:39:17Z","html_url":"https://github.com/grpc/grpc-java/pull/431#discussion_r30325836","pull_request_url":"https://api.github.com/repos/grpc/grpc-java/pulls/431","_links":{"self":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/comments/30325836"},"html":{"href":"https://github.com/grpc/grpc-java/pull/431#discussion_r30325836"},"pull_request":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/431"}}},"pull_request":{"url":"https://api.github.com/repos/grpc/grpc-java/pulls/431","id":35378856,"html_url":"https://github.com/grpc/grpc-java/pull/431","diff_url":"https://github.com/grpc/grpc-java/pull/431.diff","patch_url":"https://github.com/grpc/grpc-java/pull/431.patch","issue_url":"https://api.github.com/repos/grpc/grpc-java/issues/431","number":431,"state":"open","locked":false,"title":"Implement writes to the channel using a dedicated write queue","user":{"login":"louiscryan","id":9046588,"avatar_url":"https://avatars.githubusercontent.com/u/9046588?v=3","gravatar_id":"","url":"https://api.github.com/users/louiscryan","html_url":"https://github.com/louiscryan","followers_url":"https://api.github.com/users/louiscryan/followers","following_url":"https://api.github.com/users/louiscryan/following{/other_user}","gists_url":"https://api.github.com/users/louiscryan/gists{/gist_id}","starred_url":"https://api.github.com/users/louiscryan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/louiscryan/subscriptions","organizations_url":"https://api.github.com/users/louiscryan/orgs","repos_url":"https://api.github.com/users/louiscryan/repos","events_url":"https://api.github.com/users/louiscryan/events{/privacy}","received_events_url":"https://api.github.com/users/louiscryan/received_events","type":"User","site_admin":false},"body":"Fixes https://github.com/grpc/grpc-java/issues/305 for Netty","created_at":"2015-05-13T22:47:45Z","updated_at":"2015-05-14T14:39:17Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f705b19259878a8bce6648e0eded069ce3a12c1c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/grpc/grpc-java/pulls/431/commits","review_comments_url":"https://api.github.com/repos/grpc/grpc-java/pulls/431/comments","review_comment_url":"https://api.github.com/repos/grpc/grpc-java/pulls/comments{/number}","comments_url":"https://api.github.com/repos/grpc/grpc-java/issues/431/comments","statuses_url":"https://api.github.com/repos/grpc/grpc-java/statuses/4ffcc1ef309ad82f3a2c1c91675f6ee8ca64856a","head":{"label":"louiscryan:ChannelWriteQueue","ref":"ChannelWriteQueue","sha":"4ffcc1ef309ad82f3a2c1c91675f6ee8ca64856a","user":{"login":"louiscryan","id":9046588,"avatar_url":"https://avatars.githubusercontent.com/u/9046588?v=3","gravatar_id":"","url":"https://api.github.com/users/louiscryan","html_url":"https://github.com/louiscryan","followers_url":"https://api.github.com/users/louiscryan/followers","following_url":"https://api.github.com/users/louiscryan/following{/other_user}","gists_url":"https://api.github.com/users/louiscryan/gists{/gist_id}","starred_url":"https://api.github.com/users/louiscryan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/louiscryan/subscriptions","organizations_url":"https://api.github.com/users/louiscryan/orgs","repos_url":"https://api.github.com/users/louiscryan/repos","events_url":"https://api.github.com/users/louiscryan/events{/privacy}","received_events_url":"https://api.github.com/users/louiscryan/received_events","type":"User","site_admin":false},"repo":{"id":31382190,"name":"grpc-java","full_name":"louiscryan/grpc-java","owner":{"login":"louiscryan","id":9046588,"avatar_url":"https://avatars.githubusercontent.com/u/9046588?v=3","gravatar_id":"","url":"https://api.github.com/users/louiscryan","html_url":"https://github.com/louiscryan","followers_url":"https://api.github.com/users/louiscryan/followers","following_url":"https://api.github.com/users/louiscryan/following{/other_user}","gists_url":"https://api.github.com/users/louiscryan/gists{/gist_id}","starred_url":"https://api.github.com/users/louiscryan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/louiscryan/subscriptions","organizations_url":"https://api.github.com/users/louiscryan/orgs","repos_url":"https://api.github.com/users/louiscryan/repos","events_url":"https://api.github.com/users/louiscryan/events{/privacy}","received_events_url":"https://api.github.com/users/louiscryan/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/louiscryan/grpc-java","description":"The Java gRPC implementation. HTTP/2 based RPC","fork":true,"url":"https://api.github.com/repos/louiscryan/grpc-java","forks_url":"https://api.github.com/repos/louiscryan/grpc-java/forks","keys_url":"https://api.github.com/repos/louiscryan/grpc-java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/louiscryan/grpc-java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/louiscryan/grpc-java/teams","hooks_url":"https://api.github.com/repos/louiscryan/grpc-java/hooks","issue_events_url":"https://api.github.com/repos/louiscryan/grpc-java/issues/events{/number}","events_url":"https://api.github.com/repos/louiscryan/grpc-java/events","assignees_url":"https://api.github.com/repos/louiscryan/grpc-java/assignees{/user}","branches_url":"https://api.github.com/repos/louiscryan/grpc-java/branches{/branch}","tags_url":"https://api.github.com/repos/louiscryan/grpc-java/tags","blobs_url":"https://api.github.com/repos/louiscryan/grpc-java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/louiscryan/grpc-java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/louiscryan/grpc-java/git/refs{/sha}","trees_url":"https://api.github.com/repos/louiscryan/grpc-java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/louiscryan/grpc-java/statuses/{sha}","languages_url":"https://api.github.com/repos/louiscryan/grpc-java/languages","stargazers_url":"https://api.github.com/repos/louiscryan/grpc-java/stargazers","contributors_url":"https://api.github.com/repos/louiscryan/grpc-java/contributors","subscribers_url":"https://api.github.com/repos/louiscryan/grpc-java/subscribers","subscription_url":"https://api.github.com/repos/louiscryan/grpc-java/subscription","commits_url":"https://api.github.com/repos/louiscryan/grpc-java/commits{/sha}","git_commits_url":"https://api.github.com/repos/louiscryan/grpc-java/git/commits{/sha}","comments_url":"https://api.github.com/repos/louiscryan/grpc-java/comments{/number}","issue_comment_url":"https://api.github.com/repos/louiscryan/grpc-java/issues/comments{/number}","contents_url":"https://api.github.com/repos/louiscryan/grpc-java/contents/{+path}","compare_url":"https://api.github.com/repos/louiscryan/grpc-java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/louiscryan/grpc-java/merges","archive_url":"https://api.github.com/repos/louiscryan/grpc-java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/louiscryan/grpc-java/downloads","issues_url":"https://api.github.com/repos/louiscryan/grpc-java/issues{/number}","pulls_url":"https://api.github.com/repos/louiscryan/grpc-java/pulls{/number}","milestones_url":"https://api.github.com/repos/louiscryan/grpc-java/milestones{/number}","notifications_url":"https://api.github.com/repos/louiscryan/grpc-java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/louiscryan/grpc-java/labels{/name}","releases_url":"https://api.github.com/repos/louiscryan/grpc-java/releases{/id}","created_at":"2015-02-26T18:33:49Z","updated_at":"2015-05-01T20:01:16Z","pushed_at":"2015-05-13T22:54:32Z","git_url":"git://github.com/louiscryan/grpc-java.git","ssh_url":"git@github.com:louiscryan/grpc-java.git","clone_url":"https://github.com/louiscryan/grpc-java.git","svn_url":"https://github.com/louiscryan/grpc-java","homepage":"","size":2447,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":false,"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":"grpc:master","ref":"master","sha":"5efbb6fe73e152a74c04bfb3490a7664af2db85f","user":{"login":"grpc","id":7802525,"avatar_url":"https://avatars.githubusercontent.com/u/7802525?v=3","gravatar_id":"","url":"https://api.github.com/users/grpc","html_url":"https://github.com/grpc","followers_url":"https://api.github.com/users/grpc/followers","following_url":"https://api.github.com/users/grpc/following{/other_user}","gists_url":"https://api.github.com/users/grpc/gists{/gist_id}","starred_url":"https://api.github.com/users/grpc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/grpc/subscriptions","organizations_url":"https://api.github.com/users/grpc/orgs","repos_url":"https://api.github.com/users/grpc/repos","events_url":"https://api.github.com/users/grpc/events{/privacy}","received_events_url":"https://api.github.com/users/grpc/received_events","type":"Organization","site_admin":false},"repo":{"id":27729926,"name":"grpc-java","full_name":"grpc/grpc-java","owner":{"login":"grpc","id":7802525,"avatar_url":"https://avatars.githubusercontent.com/u/7802525?v=3","gravatar_id":"","url":"https://api.github.com/users/grpc","html_url":"https://github.com/grpc","followers_url":"https://api.github.com/users/grpc/followers","following_url":"https://api.github.com/users/grpc/following{/other_user}","gists_url":"https://api.github.com/users/grpc/gists{/gist_id}","starred_url":"https://api.github.com/users/grpc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/grpc/subscriptions","organizations_url":"https://api.github.com/users/grpc/orgs","repos_url":"https://api.github.com/users/grpc/repos","events_url":"https://api.github.com/users/grpc/events{/privacy}","received_events_url":"https://api.github.com/users/grpc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/grpc/grpc-java","description":"The Java gRPC implementation. HTTP/2 based RPC","fork":false,"url":"https://api.github.com/repos/grpc/grpc-java","forks_url":"https://api.github.com/repos/grpc/grpc-java/forks","keys_url":"https://api.github.com/repos/grpc/grpc-java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/grpc/grpc-java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/grpc/grpc-java/teams","hooks_url":"https://api.github.com/repos/grpc/grpc-java/hooks","issue_events_url":"https://api.github.com/repos/grpc/grpc-java/issues/events{/number}","events_url":"https://api.github.com/repos/grpc/grpc-java/events","assignees_url":"https://api.github.com/repos/grpc/grpc-java/assignees{/user}","branches_url":"https://api.github.com/repos/grpc/grpc-java/branches{/branch}","tags_url":"https://api.github.com/repos/grpc/grpc-java/tags","blobs_url":"https://api.github.com/repos/grpc/grpc-java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/grpc/grpc-java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/grpc/grpc-java/git/refs{/sha}","trees_url":"https://api.github.com/repos/grpc/grpc-java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/grpc/grpc-java/statuses/{sha}","languages_url":"https://api.github.com/repos/grpc/grpc-java/languages","stargazers_url":"https://api.github.com/repos/grpc/grpc-java/stargazers","contributors_url":"https://api.github.com/repos/grpc/grpc-java/contributors","subscribers_url":"https://api.github.com/repos/grpc/grpc-java/subscribers","subscription_url":"https://api.github.com/repos/grpc/grpc-java/subscription","commits_url":"https://api.github.com/repos/grpc/grpc-java/commits{/sha}","git_commits_url":"https://api.github.com/repos/grpc/grpc-java/git/commits{/sha}","comments_url":"https://api.github.com/repos/grpc/grpc-java/comments{/number}","issue_comment_url":"https://api.github.com/repos/grpc/grpc-java/issues/comments{/number}","contents_url":"https://api.github.com/repos/grpc/grpc-java/contents/{+path}","compare_url":"https://api.github.com/repos/grpc/grpc-java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/grpc/grpc-java/merges","archive_url":"https://api.github.com/repos/grpc/grpc-java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/grpc/grpc-java/downloads","issues_url":"https://api.github.com/repos/grpc/grpc-java/issues{/number}","pulls_url":"https://api.github.com/repos/grpc/grpc-java/pulls{/number}","milestones_url":"https://api.github.com/repos/grpc/grpc-java/milestones{/number}","notifications_url":"https://api.github.com/repos/grpc/grpc-java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/grpc/grpc-java/labels{/name}","releases_url":"https://api.github.com/repos/grpc/grpc-java/releases{/id}","created_at":"2014-12-08T19:00:00Z","updated_at":"2015-05-14T02:47:12Z","pushed_at":"2015-05-13T22:54:32Z","git_url":"git://github.com/grpc/grpc-java.git","ssh_url":"git@github.com:grpc/grpc-java.git","clone_url":"https://github.com/grpc/grpc-java.git","svn_url":"https://github.com/grpc/grpc-java","homepage":"","size":19115,"stargazers_count":690,"watchers_count":690,"language":"Java","has_issues":true,"has_downloads":false,"has_wiki":false,"has_pages":false,"forks_count":146,"mirror_url":null,"open_issues_count":60,"forks":146,"open_issues":60,"watchers":690,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/431"},"html":{"href":"https://github.com/grpc/grpc-java/pull/431"},"issue":{"href":"https://api.github.com/repos/grpc/grpc-java/issues/431"},"comments":{"href":"https://api.github.com/repos/grpc/grpc-java/issues/431/comments"},"review_comments":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/431/comments"},"review_comment":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/431/commits"},"statuses":{"href":"https://api.github.com/repos/grpc/grpc-java/statuses/4ffcc1ef309ad82f3a2c1c91675f6ee8ca64856a"}}}}