Github user revans2 commented on a diff in the pull request:
https://github.com/apache/incubator-storm/pull/103#discussion_r12812919
--- Diff: storm-core/src/jvm/backtype/storm/messaging/netty/Client.java ---
@@ -128,133 +181,105 @@ private long getSleepTimeMs()
}
/**
- * Enqueue a task message to be sent to server
+ * Enqueue task messages to be sent to server
*/
- public void send(int task, byte[] message) {
- //throw exception if the client is being closed
- if (being_closed.get()) {
+ synchronized public void send(Iterator<TaskMessage> msgs) {
+
+ // throw exception if the client is being closed
+ if (closing) {
throw new RuntimeException("Client is being closed, and does
not take requests any more");
}
-
- try {
- message_queue.put(new TaskMessage(task, message));
-
- //resume delivery if it is waiting for requests
- tryDeliverMessages(true);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
+
+ if (null == msgs || !msgs.hasNext()) {
+ return;
}
- }
- /**
- * Retrieve messages from queue, and delivery to server if any
- */
- synchronized void tryDeliverMessages(boolean only_if_waiting) throws
InterruptedException {
- //just skip if delivery only if waiting, and we are not waiting
currently
- if (only_if_waiting && !wait_for_requests) return;
-
- //make sure that channel was not closed
Channel channel = channelRef.get();
- if (channel == null) return;
- if (!channel.isOpen()) {
- LOG.info("Channel to {} is no longer open.",remote_addr);
- //The channel is not open yet. Reconnect?
- reconnect();
- return;
+ if (null == channel) {
+ connect();
+ channel = channelRef.get();
}
- final MessageBatch requests = tryTakeMessages();
- if (requests==null) {
- wait_for_requests = true;
- return;
- }
+ while (msgs.hasNext()) {
+ TaskMessage message = msgs.next();
+ if (null == messageBatch) {
+ messageBatch = new MessageBatch(messageBatchSize);
+ }
- //if channel is being closed and we have no outstanding messages,
let's close the channel
- if (requests.isEmpty() && being_closed.get()) {
- close_n_release();
- return;
+ messageBatch.add(message);
+ if (messageBatch.isFull()) {
+ MessageBatch toBeFlushed = messageBatch;
+ flushRequest(channel, toBeFlushed, blocking);
+ messageBatch = null;
+ }
}
- //we are busily delivering messages, and will check queue upon
response.
- //When send() is called by senders, we should not thus call
tryDeliverMessages().
- wait_for_requests = false;
-
- //write request into socket channel
- ChannelFuture future = channel.write(requests);
- future.addListener(new ChannelFutureListener() {
- public void operationComplete(ChannelFuture future)
- throws Exception {
- if (!future.isSuccess()) {
- LOG.info("failed to send "+requests.size()+" requests
to "+remote_addr, future.getCause());
- reconnect();
- } else {
- LOG.debug("{} request(s) sent", requests.size());
-
- //Now that our requests have been sent, channel could
be closed if needed
- if (being_closed.get())
- close_n_release();
- }
+ if (null != messageBatch && !messageBatch.isEmpty()) {
+ if (channel.isWritable()) {
+ flushCheckTimer.set(Long.MAX_VALUE);
+
+ // Flush as fast as we can to reduce the latency
+ MessageBatch toBeFlushed = messageBatch;
+ messageBatch = null;
+ flushRequest(channel, toBeFlushed, blocking);
+
+ } else {
+ // when channel is NOT writable, it means the internal
netty buffer is full.
+ // In this case, we can try to buffer up more incoming
messages.
+ flushCheckTimer.set(System.currentTimeMillis() +
flushCheckInterval);
}
- });
+ }
+
}
- /**
- * Take all enqueued messages from queue
- * @return batch of messages
- * @throws InterruptedException
- *
- * synchronized ... ensure that messages are delivered in the same
order
- * as they are added into queue
- */
- private MessageBatch tryTakeMessages() throws InterruptedException {
- //1st message
- Object msg = message_queue.poll();
- if (msg == null) return null;
-
- MessageBatch batch = new MessageBatch(buffer_size);
- //we will discard any message after CLOSE
- if (msg == ControlMessage.CLOSE_MESSAGE) {
- LOG.info("Connection to {} is being closed", remote_addr);
- being_closed.set(true);
- return batch;
+ public String name() {
+ if (null != remote_addr) {
+ return PREFIX + remote_addr.toString();
}
+ return "";
+ }
- batch.add((TaskMessage)msg);
- while (!batch.isFull() && ((msg = message_queue.peek())!=null)) {
- //Is it a CLOSE message?
- if (msg == ControlMessage.CLOSE_MESSAGE) {
- message_queue.take();
- LOG.info("Connection to {} is being closed", remote_addr);
- being_closed.set(true);
- break;
+ private synchronized void flush() {
+ if (!closing) {
+ if (null != messageBatch && !messageBatch.isEmpty()) {
+ MessageBatch toBeFlushed = messageBatch;
+ Channel channel = channelRef.get();
+ if (channel != null) {
+ flushCheckTimer.set(Long.MAX_VALUE);
+ flushRequest(channel, toBeFlushed, true);
+ }
+ messageBatch = null;
--- End diff --
If Channel is null do we really want to drop the messageBatch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
