[ABANDONED] Use new Websocket SendBlob IPC from renderer.

third_party/WebKit/Source/modules/websockets/DocumentWebSocketChannel.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 16 matching lines @@
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
 #include "modules/websockets/DocumentWebSocketChannel.h"
 
 #include "core/dom/DOMArrayBuffer.h"
 #include "core/dom/Document.h"
 #include "core/dom/ExecutionContext.h"
 #include "core/fetch/UniqueIdentifier.h"
-#include "core/fileapi/FileReaderLoader.h"
-#include "core/fileapi/FileReaderLoaderClient.h"
 #include "core/frame/LocalFrame.h"
 #include "core/inspector/ConsoleMessage.h"
 #include "core/inspector/InspectorInstrumentation.h"
 #include "core/loader/FrameLoader.h"
 #include "core/loader/FrameLoaderClient.h"
 #include "core/loader/MixedContentChecker.h"
 #include "modules/websockets/InspectorWebSocketEvents.h"
 #include "modules/websockets/WebSocketChannelClient.h"
 #include "modules/websockets/WebSocketFrame.h"
 #include "platform/Logging.h"
 #include "platform/network/WebSocketHandshakeRequest.h"
 #include "platform/weborigin/SecurityOrigin.h"
 #include "public/platform/Platform.h"
 #include "public/platform/WebSecurityOrigin.h"
 #include "public/platform/WebString.h"
 #include "public/platform/WebURL.h"
 #include "public/platform/WebVector.h"
 #include "public/platform/modules/websockets/WebSocketHandshakeRequestInfo.h"
 #include "public/platform/modules/websockets/WebSocketHandshakeResponseInfo.h"
 
 using blink::WebSocketHandle;
 
 namespace blink {
 
-class DocumentWebSocketChannel::BlobLoader final : public GarbageCollectedFinalized<DocumentWebSocketChannel::BlobLoader>, public FileReaderLoaderClient {
-public:
-    BlobLoader(PassRefPtr<BlobDataHandle>, DocumentWebSocketChannel*);
-    ~BlobLoader() override { }
-
-    void cancel();
-
-    // FileReaderLoaderClient functions.
-    void didStartLoading() override { }
-    void didReceiveData() override { }
-    void didFinishLoading() override;
-    void didFail(FileError::ErrorCode) override;
-
-    DEFINE_INLINE_TRACE()
-    {
-        visitor->trace(m_channel);
-    }
-
-private:
-    Member<DocumentWebSocketChannel> m_channel;
-    FileReaderLoader m_loader;
-};
-
-DocumentWebSocketChannel::BlobLoader::BlobLoader(PassRefPtr<BlobDataHandle> blobDataHandle, DocumentWebSocketChannel* channel)
-    : m_channel(channel)
-    , m_loader(FileReaderLoader::ReadAsArrayBuffer, this)
-{
-    m_loader.start(channel->executionContext(), blobDataHandle);
-}
-
-void DocumentWebSocketChannel::BlobLoader::cancel()
-{
-    m_loader.cancel();
-    // didFail will be called immediately.
-    // |this| is deleted here.
-}
-
-void DocumentWebSocketChannel::BlobLoader::didFinishLoading()
-{
-    m_channel->didFinishLoadingBlob(m_loader.arrayBufferResult());
-    // |this| is deleted here.
-}
-
-void DocumentWebSocketChannel::BlobLoader::didFail(FileError::ErrorCode errorCode)
-{
-    m_channel->didFailLoadingBlob(errorCode);
-    // |this| is deleted here.
-}
-
 DocumentWebSocketChannel::DocumentWebSocketChannel(Document* document, WebSocketChannelClient* client, const String& sourceURL, unsigned lineNumber, WebSocketHandle *handle)
     : ContextLifecycleObserver(document)
     , m_handle(adoptPtr(handle ? handle : Platform::current()->createWebSocketHandle()))
     , m_client(client)
     , m_identifier(createUniqueIdentifier())
+    , m_blobLoadingMode(false)
+    , m_blobDataPending(0)
     , m_sendingQuota(0)
     , m_receivedDataSizeForFlowControl(receivedDataSizeForFlowControlHighWaterMark * 2) // initial quota
     , m_sentSizeOfTopMessage(0)
     , m_sourceURLAtConstruction(sourceURL)
     , m_lineNumberAtConstruction(lineNumber)
 {
 }
 
 DocumentWebSocketChannel::~DocumentWebSocketChannel()
 {
-    ASSERT(!m_blobLoader);
+    ASSERT(!m_blobLoadingMode);
 }
 
 bool DocumentWebSocketChannel::connect(const KURL& url, const String& protocol)
 {
     WTF_LOG(Network, "DocumentWebSocketChannel %p connect()", this);
     if (!m_handle)
         return false;
 
     if (document()->frame()) {
         if (MixedContentChecker::shouldBlockWebSocket(document()->frame(), url))
@@ skipping 110 matching lines @@
     // handleDidClose may delete this object.
 }
 
 void DocumentWebSocketChannel::disconnect()
 {
     WTF_LOG(Network, "DocumentWebSocketChannel %p disconnect()", this);
     if (m_identifier) {
         TRACE_EVENT_INSTANT1("devtools.timeline", "WebSocketDestroy", TRACE_EVENT_SCOPE_THREAD, "data", InspectorWebSocketEvent::data(document(), m_identifier));
         InspectorInstrumentation::didCloseWebSocket(document(), m_identifier);
     }
-    abortAsyncOperations();
+    m_blobLoadingMode = false;
     m_handle.clear();
     m_client = nullptr;
     m_identifier = 0;
 }
 
 DocumentWebSocketChannel::Message::Message(const CString& text)
     : type(MessageTypeText)
     , text(text) { }
 
 DocumentWebSocketChannel::Message::Message(PassRefPtr<BlobDataHandle> blobDataHandle)
@@ skipping 32 matching lines @@
     m_sentSizeOfTopMessage += size;
     m_sendingQuota -= size;
     *consumedBufferedAmount += size;
 
     if (final) {
         m_messages.removeFirst();
         m_sentSizeOfTopMessage = 0;
     }
 }
 
+void DocumentWebSocketChannel::sendBlob(const BlobDataHandle& blob)
+{
+    ASSERT(!m_blobLoadingMode);
+    m_blobLoadingMode = true;
+    m_blobDataPending = blob.size();
+    m_handle->sendBlob(blob.uuid(), blob.size());
+    // The message is still in |m_messages| so that the renderer holds a
+    // reference to it until it is fully sent.
+}
+
 void DocumentWebSocketChannel::processSendQueue()
 {
     ASSERT(m_handle);
     uint64_t consumedBufferedAmount = 0;
-    while (!m_messages.isEmpty() && !m_blobLoader) {
+    while (!m_messages.isEmpty() && !m_blobLoadingMode) {
         Message* message = m_messages.first().get();
         if (m_sendingQuota == 0 && message->type != MessageTypeClose)
             break;
         switch (message->type) {
         case MessageTypeText:
             sendInternal(WebSocketHandle::MessageTypeText, message->text.data(), message->text.length(), &consumedBufferedAmount);
             break;
         case MessageTypeBlob:
-            ASSERT(!m_blobLoader);
-            m_blobLoader = new BlobLoader(message->blobDataHandle, this);
+            sendBlob(*message->blobDataHandle);
             break;
         case MessageTypeArrayBuffer:
             sendInternal(WebSocketHandle::MessageTypeBinary, static_cast<const char*>(message->arrayBuffer->data()), message->arrayBuffer->byteLength(), &consumedBufferedAmount);
             break;
         case MessageTypeTextAsCharVector:
             sendInternal(WebSocketHandle::MessageTypeText, message->vectorData->data(), message->vectorData->size(), &consumedBufferedAmount);
             break;
         case MessageTypeBinaryAsCharVector:
             sendInternal(WebSocketHandle::MessageTypeBinary, message->vectorData->data(), message->vectorData->size(), &consumedBufferedAmount);
             break;
@@ skipping 13 matching lines @@
 
 void DocumentWebSocketChannel::flowControlIfNecessary()
 {
     if (!m_handle || m_receivedDataSizeForFlowControl < receivedDataSizeForFlowControlHighWaterMark) {
         return;
     }
     m_handle->flowControl(m_receivedDataSizeForFlowControl);
     m_receivedDataSizeForFlowControl = 0;
 }
 
-void DocumentWebSocketChannel::abortAsyncOperations()
-{
-    if (m_blobLoader) {
-        m_blobLoader->cancel();
-        m_blobLoader.clear();
-    }
-}
-
 void DocumentWebSocketChannel::handleDidClose(bool wasClean, unsigned short code, const String& reason)
 {
     m_handle.clear();
-    abortAsyncOperations();
+    m_blobLoadingMode = false;
     if (!m_client) {
         return;
     }
     WebSocketChannelClient* client = m_client;
     m_client = nullptr;
     WebSocketChannelClient::ClosingHandshakeCompletionStatus status =
         wasClean ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete;
     client->didClose(status, code, reason);
     // client->didClose may delete this object.
 }
@@ skipping 127 matching lines @@
 
 void DocumentWebSocketChannel::didReceiveFlowControl(WebSocketHandle* handle, int64_t quota)
 {
     WTF_LOG(Network, "DocumentWebSocketChannel %p didReceiveFlowControl(%p, %ld)", this, handle, static_cast<long>(quota));
 
     ASSERT(m_handle);
     ASSERT(handle == m_handle);
     ASSERT(quota >= 0);
 
     m_sendingQuota += quota;
-    processSendQueue();
+    if (m_blobLoadingMode) {
+        uint64_t blobUsedQuota = std::min(static_cast<uint64_t>(quota), m_blobDataPending);
+        if (blobUsedQuota > 0) {
+            m_sendingQuota -= blobUsedQuota;
+            m_blobDataPending -= blobUsedQuota;
+            if (m_client)
+                m_client->didConsumeBufferedAmount(blobUsedQuota);

[tyoshino (SeeGerritForStatus), 2016/02/09 14:29:10]

IIUC, it's possible that this would process flow control replenishments for the
message sent before the blob. Is that true?

I don't think we have to be so strict about that, but wanna understand what's
strictly implemented and what's not.

[Adam Rice, 2016/02/10 01:37:55]

Correct. For example, if the user sends an 8KB Text message then immediately an
8KB Blob, bufferedAmount will go from 16KB to 8KB when the Text message is sent
to the browser process, and then from 8KB to 0KB when the flow control is
replenished.

This is obviously not quite right, but because we normally consider a message
"sent" when we send it to the browser, I don't consider it a big problem. I
don't think there's benefit to developers in spending more resources tracking it
when we still have no guarantee that the data has been transmitted by the OS
kernel.
How should I make this clearer? Add a comment? Or maybe add a test (the
behaviour is not quite right, but it is WAI)?

[tyoshino (SeeGerritForStatus), 2016/02/10 06:48:46]

Right. bufferedAmount doesn't tell the actual memory pressure to you reliably.
Adding a comment is enough. Please.

[Adam Rice, 2016/02/10 12:58:27]

Comment added. I tried to keep it short to match Blink style. PTAL.

+        }
+    } else {
+        processSendQueue();
+    }
 }
 
 void DocumentWebSocketChannel::didStartClosingHandshake(WebSocketHandle* handle)
 {
     WTF_LOG(Network, "DocumentWebSocketChannel %p didStartClosingHandshake(%p)", this, handle);
 
     ASSERT(m_handle);
     ASSERT(handle == m_handle);
 
     if (m_client)
         m_client->didStartClosingHandshake();
 }
 
-void DocumentWebSocketChannel::didFinishLoadingBlob(PassRefPtr<DOMArrayBuffer> buffer)
+void DocumentWebSocketChannel::didCompleteSendingBlob(WebSocketHandle* handle)
 {
-    m_blobLoader.clear();
+    WTF_LOG(Network, "DocumentWebSocketChannel %p didCompleteSendingBlob(%p)", this, handle);
+
+    // If the IPC was unexpected, do nothing.
+    if (!m_blobLoadingMode)
+        return;
+
     ASSERT(m_handle);
-    // The loaded blob is always placed on m_messages[0].
-    ASSERT(m_messages.size() > 0 && m_messages.first()->type == MessageTypeBlob);
-    // We replace it with the loaded blob.
-    m_messages.first() = adoptPtr(new Message(buffer));
+    ASSERT(handle == m_handle);
+    ASSERT(m_sendingQuota >= m_blobDataPending);
+
+    m_sendingQuota -= m_blobDataPending;
+    if (m_blobDataPending && m_client)
+        m_client->didConsumeBufferedAmount(m_blobDataPending);
+    m_blobDataPending = 0;
+    m_blobLoadingMode = false;
+    // Remove the reference to the Blob.
+    m_messages.removeFirst();
     processSendQueue();
 }
 
-void DocumentWebSocketChannel::didFailLoadingBlob(FileError::ErrorCode errorCode)
-{
-    m_blobLoader.clear();
-    if (errorCode == FileError::ABORT_ERR) {
-        // The error is caused by cancel().
-        return;
-    }
-    // FIXME: Generate human-friendly reason message.
-    failAsError("Failed to load Blob: error code = " + String::number(errorCode));
-    // |this| can be deleted here.
-}
-
 DEFINE_TRACE(DocumentWebSocketChannel)
 {
-    visitor->trace(m_blobLoader);
     visitor->trace(m_client);
     WebSocketChannel::trace(visitor);
     ContextLifecycleObserver::trace(visitor);
 }
 
 } // namespace blink