GitHub PullRequestReviewCommentEvent at 2014-05-20 04:12:22
Repository: incubator-storm
Actor: clockfly
Payload: {"comment":{"url":"https://api.github.com/repos/apache/incubator-storm/pulls/comments/12827326","id":12827326,"diff_hunk":"@@ -31,35 +31,69 @@\n import org.slf4j.LoggerFactory;\n \n import java.net.InetSocketAddress;\n+import java.util.ArrayList;\n+import java.util.Arrays;\n+import java.util.HashMap;\n+import java.util.Iterator;\n+import java.util.List;\n import java.util.Map;\n import java.util.concurrent.Executors;\n import java.util.concurrent.LinkedBlockingQueue;\n+import java.util.concurrent.ThreadFactory;\n \n class Server implements IConnection {\n     private static final Logger LOG = LoggerFactory.getLogger(Server.class);\n     @SuppressWarnings(\"rawtypes\")\n     Map storm_conf;\n     int port;\n-    private LinkedBlockingQueue<TaskMessage> message_queue;\n+    \n+    // Create multiple queues for incoming messages. The size equals the number of receiver threads.\n+    // For message which is sent to same task, it will be stored in the same queue to preserve the message order.\n+    private LinkedBlockingQueue<ArrayList<TaskMessage>>[] message_queue;\n+    \n     volatile ChannelGroup allChannels = new DefaultChannelGroup(\"storm-server\");\n     final ChannelFactory factory;\n     final ServerBootstrap bootstrap;\n-\n+    \n+    private int queueCount;\n+    HashMap<Integer, Integer> taskToQueueId = null;\n+    int roundRobinQueueId;\n+\t\n+    boolean closing = false;\n+    List<TaskMessage> closeMessage = Arrays.asList(new TaskMessage(-1, null));\n+    \n+    \n     @SuppressWarnings(\"rawtypes\")\n     Server(Map storm_conf, int port) {\n         this.storm_conf = storm_conf;\n         this.port = port;\n-        message_queue = new LinkedBlockingQueue<TaskMessage>();\n-\n+        \n+        queueCount = Utils.getInt(storm_conf.get(\"worker.receiver.thread.count\"), 1);\n+        roundRobinQueueId = 0;\n+        taskToQueueId = new HashMap<Integer, Integer>();\n+    \n+        message_queue = new LinkedBlockingQueue[queueCount];\n+\t\t    for (int i = 0; i < queueCount; i++) {","path":"storm-core/src/jvm/backtype/storm/messaging/netty/Server.java","position":null,"original_position":50,"commit_id":"44e47e82d03743cbec3fe6aff17c76711520c061","original_commit_id":"c5e7a0d0f137cc5f0ccf6116541c16d4a7727598","user":{"login":"clockfly","id":2595532,"avatar_url":"https://avatars.githubusercontent.com/u/2595532?","gravatar_id":"06227f12c6fbe90b07a476034c7f96b9","url":"https://api.github.com/users/clockfly","html_url":"https://github.com/clockfly","followers_url":"https://api.github.com/users/clockfly/followers","following_url":"https://api.github.com/users/clockfly/following{/other_user}","gists_url":"https://api.github.com/users/clockfly/gists{/gist_id}","starred_url":"https://api.github.com/users/clockfly/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/clockfly/subscriptions","organizations_url":"https://api.github.com/users/clockfly/orgs","repos_url":"https://api.github.com/users/clockfly/repos","events_url":"https://api.github.com/users/clockfly/events{/privacy}","received_events_url":"https://api.github.com/users/clockfly/received_events","type":"User","site_admin":false},"body":"fixed","created_at":"2014-05-20T04:12:22Z","updated_at":"2014-05-20T04:12:22Z","html_url":"https://github.com/apache/incubator-storm/pull/103#discussion_r12827326","pull_request_url":"https://api.github.com/repos/apache/incubator-storm/pulls/103","_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-storm/pulls/comments/12827326"},"html":{"href":"https://github.com/apache/incubator-storm/pull/103#discussion_r12827326"},"pull_request":{"href":"https://api.github.com/repos/apache/incubator-storm/pulls/103"}}}}