[WebSocket] bufferedAmount should not decrease inside a task.

Source/modules/websockets/NewWebSocketChannelImpl.cpp

 /*
  * Copyright (C) 2013 Google Inc. All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *     * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *     * Redistributions in binary form must reproduce the above
@@ skipping 97 matching lines @@
     // |this| is deleted here.
 }
 
 NewWebSocketChannelImpl::NewWebSocketChannelImpl(ExecutionContext* context, WebSocketChannelClient* client, const String& sourceURL, unsigned lineNumber)
     : ContextLifecycleObserver(context)
     , m_handle(adoptPtr(blink::Platform::current()->createWebSocketHandle()))
     , m_client(client)
     , m_identifier(0)
     , m_sendingQuota(0)
     , m_receivedDataSizeForFlowControl(receivedDataSizeForFlowControlHighWaterMark * 2) // initial quota
-    , m_bufferedAmount(0)
     , m_sentSizeOfTopMessage(0)
     , m_sourceURLAtConstruction(sourceURL)
     , m_lineNumberAtConstruction(lineNumber)
 {
     if (context->isDocument() && toDocument(context)->page())
         m_identifier = createUniqueIdentifier();
 }
 
 NewWebSocketChannelImpl::~NewWebSocketChannelImpl()
 {
@@ skipping 105 matching lines @@
     if (m_identifier) {
         // FIXME: Change the inspector API to show the entire message instead
         // of individual frames.
         InspectorInstrumentation::didSendWebSocketFrame(document(), m_identifier, WebSocketFrame::OpCodeBinary, true, data->data(), data->size());
     }
     m_messages.append(adoptPtr(new Message(data)));
     sendInternal();
     return SendSuccess;
 }
 
-unsigned long NewWebSocketChannelImpl::bufferedAmount() const
-{
-    WTF_LOG(Network, "NewWebSocketChannelImpl %p bufferedAmount()", this);
-    return m_bufferedAmount;
-}
-
 void NewWebSocketChannelImpl::close(int code, const String& reason)
 {
     WTF_LOG(Network, "NewWebSocketChannelImpl %p close(%d, %s)", this, code, reason.utf8().data());
     ASSERT(m_handle);
     unsigned short codeToSend = static_cast<unsigned short>(code == CloseEventCodeNotSpecified ? CloseEventCodeNoStatusRcvd : code);
     m_handle->close(codeToSend, reason);
 }
 
 void NewWebSocketChannelImpl::fail(const String& reason, MessageLevel level, const String& sourceURL, unsigned lineNumber)
 {
@@ skipping 50 matching lines @@
     : type(MessageTypeArrayBuffer)
     , arrayBuffer(arrayBuffer) { }
 
 NewWebSocketChannelImpl::Message::Message(PassOwnPtr<Vector<char> > vectorData)
     : type(MessageTypeVector)
     , vectorData(vectorData) { }
 
 void NewWebSocketChannelImpl::sendInternal()
 {
     ASSERT(m_handle);
-    unsigned long bufferedAmount = m_bufferedAmount;
+    int64_t quota = m_sendingQuota;

[tyoshino (SeeGerritForStatus), 2014/06/16 02:29:29]

quota and bufferedAmount are different concept. as impact to performance should
be small, let's have another variable to accumulate amount of data sent, or call
didConsumeBufferedAmount() with "size" at the end of each case block.

[yhirano, 2014/06/16 04:44:27]

Done.

     while (!m_messages.isEmpty() && m_sendingQuota > 0 && !m_blobLoader) {
         bool final = false;
         Message* message = m_messages.first().get();
         switch (message->type) {
         case MessageTypeText: {
             WebSocketHandle::MessageType type =
                 m_sentSizeOfTopMessage ? WebSocketHandle::MessageTypeContinuation : WebSocketHandle::MessageTypeText;
             size_t size = std::min(static_cast<size_t>(m_sendingQuota), message->text.length() - m_sentSizeOfTopMessage);
             final = (m_sentSizeOfTopMessage + size == message->text.length());
             m_handle->send(final, type, message->text.data() + m_sentSizeOfTopMessage, size);
@@ skipping 24 matching lines @@
             m_sentSizeOfTopMessage += size;
             m_sendingQuota -= size;
             break;
         }
         }
         if (final) {
             m_messages.removeFirst();
             m_sentSizeOfTopMessage = 0;
         }
     }
-    if (m_client && m_bufferedAmount != bufferedAmount) {
-        m_client->didUpdateBufferedAmount(m_bufferedAmount);
+    if (m_client && quota != m_sendingQuota) {
+        unsigned long diff = quota - m_sendingQuota;
+        m_client->didConsumeBufferedAmount(diff);
     }
 }
 
 void NewWebSocketChannelImpl::flowControlIfNecessary()
 {
     if (!m_handle || m_receivedDataSizeForFlowControl < receivedDataSizeForFlowControlHighWaterMark) {
         return;
     }
     m_handle->flowControl(m_receivedDataSizeForFlowControl);
     m_receivedDataSizeForFlowControl = 0;
@@ skipping 11 matching lines @@
 {
     m_handle.clear();
     abortAsyncOperations();
     if (!m_client) {
         return;
     }
     WebSocketChannelClient* client = m_client;
     m_client = 0;
     WebSocketChannelClient::ClosingHandshakeCompletionStatus status =
         wasClean ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete;
-    client->didClose(m_bufferedAmount, status, code, reason);
+    client->didClose(status, code, reason);
     // client->didClose may delete this object.
 }
 
 Document* NewWebSocketChannelImpl::document()
 {
     ASSERT(m_identifier);
     ExecutionContext* context = executionContext();
     ASSERT(context->isDocument());
     return toDocument(context);
 }
@@ skipping 153 matching lines @@
     // |this| can be deleted here.
 }
 
 void NewWebSocketChannelImpl::trace(Visitor* visitor)
 {
     visitor->trace(m_blobLoader);
     WebSocketChannel::trace(visitor);
 }
 
 } // namespace WebCore