mozilla
/
gecko-dev
зеркало из https://github.com/mozilla/gecko-dev.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность
gecko-dev/dom/base/BodyStream.cpp

592 строки
18 KiB
C++
Исходник Ответственный История

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "BodyStream.h"
#include "js/GCAPI.h"
#include "mozilla/CycleCollectedJSContext.h"
#include "mozilla/dom/AutoEntryScript.h"
#include "mozilla/dom/DOMException.h"
#include "mozilla/dom/ReadableStream.h"
#include "mozilla/dom/ReadableByteStreamController.h"
#include "mozilla/dom/ScriptSettings.h"
#include "mozilla/dom/WorkerCommon.h"
#include "mozilla/dom/WorkerPrivate.h"
#include "mozilla/dom/WorkerRunnable.h"
#include "mozilla/Maybe.h"
#include "mozilla/ScopeExit.h"
#include "mozilla/Services.h"
#include "mozilla/Unused.h"
#include "nsIObserverService.h"
#include "nsProxyRelease.h"
#include "nsStreamUtils.h"
#include <cstdint>
#include <utility>
namespace mozilla::dom {
// BodyStreamHolder
// ---------------------------------------------------------------------------
NS_IMPL_CYCLE_COLLECTION_CLASS(BodyStreamHolder)
NS_IMPL_CYCLE_COLLECTION_TRAVERSE_BEGIN(BodyStreamHolder)
NS_IMPL_CYCLE_COLLECTION_TRAVERSE(mReadableStreamBody)
NS_IMPL_CYCLE_COLLECTION_TRAVERSE_END
NS_IMPL_CYCLE_COLLECTION_UNLINK_BEGIN(BodyStreamHolder)
if (tmp->mBodyStream) {
tmp->mBodyStream->ReleaseObjects();
MOZ_ASSERT(!tmp->mBodyStream);
}
NS_IMPL_CYCLE_COLLECTION_UNLINK(mReadableStreamBody);
NS_IMPL_CYCLE_COLLECTION_UNLINK_END
NS_IMPL_CYCLE_COLLECTING_ADDREF(BodyStreamHolder)
NS_IMPL_CYCLE_COLLECTING_RELEASE(BodyStreamHolder)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(BodyStreamHolder)
NS_INTERFACE_MAP_ENTRY(nsISupports)
NS_INTERFACE_MAP_END
BodyStreamHolder::BodyStreamHolder() : mBodyStream(nullptr) {}
void BodyStreamHolder::StoreBodyStream(BodyStream* aBodyStream) {
MOZ_ASSERT(aBodyStream);
MOZ_ASSERT(!mBodyStream);
mBodyStream = aBodyStream;
}
// BodyStream
//
// This class has thread ownership assertions everywhere because historically it
// could be touched by random threads. See also:
// https://searchfox.org/mozilla-esr60/rev/02b4ae79b24aae2346b1338e2bf095a571192061/dom/fetch/FetchStream.cpp#302,306,312
//
// ---------------------------------------------------------------------------
NS_IMPL_ISUPPORTS(BodyStream, nsIInputStreamCallback, nsIObserver,
nsISupportsWeakReference)
class BodyStreamUnderlyingSourceAlgorithms final
: public UnderlyingSourceAlgorithmsWrapper {
public:
NS_DECL_ISUPPORTS_INHERITED
NS_DECL_CYCLE_COLLECTION_CLASS_INHERITED(BodyStreamUnderlyingSourceAlgorithms,
UnderlyingSourceAlgorithmsBase)
BodyStreamUnderlyingSourceAlgorithms(nsIGlobalObject& aGlobal,
BodyStreamHolder& aUnderlyingSource)
: mGlobal(&aGlobal), mUnderlyingSource(&aUnderlyingSource) {}
already_AddRefed<Promise> PullCallbackImpl(
JSContext* aCx, ReadableStreamController& aController,
ErrorResult& aRv) override {
RefPtr<BodyStream> bodyStream = mUnderlyingSource->GetBodyStream();
return bodyStream->PullCallback(aCx, aController, aRv);
}
void ReleaseObjects() override {
RefPtr<BodyStreamHolder> holder = mUnderlyingSource.forget();
// BodyStream may not be available if this cleanup happened first from
// BodyStream side.
if (RefPtr<BodyStream> bodyStream = holder->GetBodyStream()) {
bodyStream->CloseInputAndReleaseObjects();
}
}
BodyStreamHolder* GetBodyStreamHolder() override { return mUnderlyingSource; }
protected:
~BodyStreamUnderlyingSourceAlgorithms() override = default;
private:
nsCOMPtr<nsIGlobalObject> mGlobal;
RefPtr<BodyStreamHolder> mUnderlyingSource;
};
NS_IMPL_CYCLE_COLLECTION_INHERITED(BodyStreamUnderlyingSourceAlgorithms,
UnderlyingSourceAlgorithmsBase, mGlobal,
mUnderlyingSource)
NS_IMPL_ADDREF_INHERITED(BodyStreamUnderlyingSourceAlgorithms,
UnderlyingSourceAlgorithmsBase)
NS_IMPL_RELEASE_INHERITED(BodyStreamUnderlyingSourceAlgorithms,
UnderlyingSourceAlgorithmsBase)
NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(BodyStreamUnderlyingSourceAlgorithms)
NS_INTERFACE_MAP_END_INHERITING(UnderlyingSourceAlgorithmsBase)
/* static */
void BodyStream::Create(JSContext* aCx, BodyStreamHolder* aStreamHolder,
nsIGlobalObject* aGlobal, nsIInputStream* aInputStream,
ErrorResult& aRv) {
MOZ_DIAGNOSTIC_ASSERT(aCx);
MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
RefPtr<BodyStream> stream =
new BodyStream(aGlobal, aStreamHolder, aInputStream);
auto cleanup = MakeScopeExit([stream] { stream->Close(); });
if (NS_IsMainThread()) {
nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
if (NS_WARN_IF(!os)) {
aRv.Throw(NS_ERROR_FAILURE);
return;
}
aRv = os->AddObserver(stream, DOM_WINDOW_DESTROYED_TOPIC, true);
if (NS_WARN_IF(aRv.Failed())) {
return;
}
} else {
WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
MOZ_ASSERT(workerPrivate);
workerPrivate->AssertIsOnWorkerThread();
RefPtr<StrongWorkerRef> workerRef = StrongWorkerRef::Create(
workerPrivate, "BodyStream", [stream]() { stream->Close(); });
if (NS_WARN_IF(!workerRef)) {
aRv.Throw(NS_ERROR_DOM_INVALID_STATE_ERR);
return;
}
// Note, this will create a ref-cycle between the holder and the stream.
// The cycle is broken when the stream is closed or the worker begins
// shutting down.
stream->mWorkerRef = std::move(workerRef);
}
auto algorithms = MakeRefPtr<BodyStreamUnderlyingSourceAlgorithms>(
*aGlobal, *aStreamHolder);
RefPtr<ReadableStream> body = ReadableStream::CreateByteNative(
aCx, aGlobal, *algorithms, Nothing(), aRv);
if (aRv.Failed()) {
return;
}
cleanup.release();
aStreamHolder->StoreBodyStream(stream);
aStreamHolder->SetReadableStreamBody(body);
#ifdef DEBUG
aStreamHolder->mStreamCreated = true;
#endif
}
// UnderlyingSource.pull, implemented for BodyStream.
already_AddRefed<Promise> BodyStream::PullCallback(
JSContext* aCx, ReadableStreamController& aController, ErrorResult& aRv) {
MOZ_ASSERT(aController.IsByte());
ReadableStream* stream = aController.Stream();
MOZ_ASSERT(stream);
MOZ_DIAGNOSTIC_ASSERT(stream->Disturbed());
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
MOZ_DIAGNOSTIC_ASSERT(!IsClosed());
MOZ_ASSERT(!mPullPromise);
mPullPromise = Promise::CreateInfallible(aController.GetParentObject());
// Reading the stream, let's mark it as such
mStreamHolder->MarkAsRead();
if (!mInputStream) {
// This is the first use of the stream. Let's convert the
// mOriginalInputStream into an nsIAsyncInputStream.
MOZ_ASSERT(mOriginalInputStream);
nsCOMPtr<nsIAsyncInputStream> asyncStream;
nsresult rv = NS_MakeAsyncNonBlockingInputStream(
mOriginalInputStream.forget(), getter_AddRefs(asyncStream));
if (NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(aCx, stream, rv);
return nullptr;
}
mInputStream = asyncStream;
mOriginalInputStream = nullptr;
}
MOZ_DIAGNOSTIC_ASSERT(mInputStream);
MOZ_DIAGNOSTIC_ASSERT(!mOriginalInputStream);
// Cancel previous wait if we were observing closure, because waiting twice
// may cause an error for some streams
mInputStream->AsyncWait(nullptr, 0, 0, nullptr);
// And then wait for data
nsresult rv = mInputStream->AsyncWait(this, 0, 0, mOwningEventTarget);
if (NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(aCx, stream, rv);
return nullptr;
}
mAsyncWaitWorkerRef = mWorkerRef;
// All good.
return do_AddRef(mPullPromise);
}
void BodyStream::WriteIntoReadRequestBuffer(JSContext* aCx,
ReadableStream* aStream,
JS::Handle<JSObject*> aBuffer,
uint32_t aLength,
uint32_t* aByteWritten) {
MOZ_DIAGNOSTIC_ASSERT(aBuffer);
MOZ_DIAGNOSTIC_ASSERT(aByteWritten);
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
MOZ_DIAGNOSTIC_ASSERT(mInputStream);
MOZ_DIAGNOSTIC_ASSERT(!IsClosed());
MOZ_DIAGNOSTIC_ASSERT(mPullPromise->State() ==
Promise::PromiseState::Pending);
uint32_t written;
nsresult rv;
void* buffer;
{
// Bug 1754513: Hazard suppression.
//
// Because mInputStream->Read is detected as possibly GCing by the
// current state of our static hazard analysis, we need to do the
// suppression here. This can be removed with future improvements
// to the static analysis.
JS::AutoSuppressGCAnalysis suppress;
JS::AutoCheckCannotGC noGC;
bool isSharedMemory;
buffer = JS_GetArrayBufferViewData(aBuffer, &isSharedMemory, noGC);
MOZ_ASSERT(!isSharedMemory);
rv = mInputStream->Read(static_cast<char*>(buffer), aLength, &written);
if (NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(aCx, aStream, rv);
return;
}
}
*aByteWritten = written;
if (written == 0) {
CloseAndReleaseObjects(aCx, aStream);
return;
}
rv = mInputStream->AsyncWait(this, nsIAsyncOutputStream::WAIT_CLOSURE_ONLY, 0,
mOwningEventTarget);
if (NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(aCx, aStream, rv);
return;
}
mAsyncWaitWorkerRef = mWorkerRef;
// All good.
}
void BodyStream::CloseInputAndReleaseObjects() {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
if (mStreamHolder) {
// Being closed means someone touched the stream, let's mark it as read.
mStreamHolder->MarkAsRead();
}
if (mInputStream) {
mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
// It could be that we don't have mInputStream yet, but we still have the
// original stream. We need to close that too.
if (mOriginalInputStream) {
MOZ_ASSERT(!mInputStream);
mOriginalInputStream->Close();
}
ReleaseObjects();
}
BodyStream::BodyStream(nsIGlobalObject* aGlobal,
BodyStreamHolder* aStreamHolder,
nsIInputStream* aInputStream)
: mGlobal(aGlobal),
mStreamHolder(aStreamHolder),
mOwningEventTarget(aGlobal->EventTargetFor(TaskCategory::Other)),
mOriginalInputStream(aInputStream) {
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
}
BodyStream::~BodyStream() {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
}
void BodyStream::ErrorPropagation(JSContext* aCx, ReadableStream* aStream,
nsresult aError) {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
// Nothing to do.
if (IsClosed()) {
return;
}
// Let's close the stream.
if (aError == NS_BASE_STREAM_CLOSED) {
CloseAndReleaseObjects(aCx, aStream);
return;
}
// Let's use a generic error.
ErrorResult rv;
// XXXbz can we come up with a better error message here to tell the
// consumer what went wrong?
rv.ThrowTypeError("Error in body stream");
JS::Rooted<JS::Value> errorValue(aCx);
bool ok = ToJSValue(aCx, std::move(rv), &errorValue);
MOZ_RELEASE_ASSERT(ok, "ToJSValue never fails for ErrorResult");
{
// This will be ignored if it's already errored.
IgnoredErrorResult rv;
aStream->ErrorNative(aCx, errorValue, rv);
NS_WARNING_ASSERTION(!rv.Failed(), "Failed to error BodyStream");
}
if (mStreamHolder) {
// Being errored means someone touched the stream, let's mark it as read.
mStreamHolder->MarkAsRead();
}
if (mInputStream) {
mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
ReleaseObjects();
}
// https://fetch.spec.whatwg.org/#concept-bodyinit-extract
// Step 12.1: Whenever one or more bytes are available and stream is not
// errored, enqueue a Uint8Array wrapping an ArrayBuffer containing the
// available bytes into stream.
void BodyStream::EnqueueChunkWithSizeIntoStream(JSContext* aCx,
ReadableStream* aStream,
uint64_t aAvailableData,
ErrorResult& aRv) {
// To avoid OOMing up on huge amounts of available data on a 32 bit system,
// as well as potentially overflowing nsIInputStream's Read method's
// parameter, let's limit our maximum chunk size to 256MB.
uint32_t ableToRead =
std::min(static_cast<uint64_t>(256 * 1024 * 1024), aAvailableData);
// Create Chunk
aRv.MightThrowJSException();
JS::Rooted<JSObject*> chunk(aCx, JS_NewUint8Array(aCx, ableToRead));
if (!chunk) {
aRv.StealExceptionFromJSContext(aCx);
return;
}
{
uint32_t bytesWritten = 0;
WriteIntoReadRequestBuffer(aCx, aStream, chunk, ableToRead, &bytesWritten);
// If bytesWritten is zero, then the stream has been closed; return
// rather than enqueueing a chunk filled with zeros.
if (bytesWritten == 0) {
return;
}
// If we don't read every byte we've allocated in the Uint8Array
// we risk enqueuing a chunk that is padded with trailing zeros,
// corrupting future processing of the chunks:
MOZ_DIAGNOSTIC_ASSERT((ableToRead - bytesWritten) == 0);
}
MOZ_ASSERT(aStream->Controller()->IsByte());
JS::Rooted<JS::Value> chunkValue(aCx);
chunkValue.setObject(*chunk);
aStream->EnqueueNative(aCx, chunkValue, aRv);
}
NS_IMETHODIMP
BodyStream::OnInputStreamReady(nsIAsyncInputStream* aStream) {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
MOZ_DIAGNOSTIC_ASSERT(aStream);
mAsyncWaitWorkerRef = nullptr;
// Already closed. We have nothing else to do here.
if (IsClosed()) {
return NS_OK;
}
// Perform a microtask checkpoint after all actions are completed. Note that
// |mMutex| *must not* be held when the checkpoint occurs -- hence, far down,
// the |lock.reset()|. (|MutexAutoUnlock| as RAII wouldn't work for this task
// because its destructor would reacquire |mMutex| before these objects'
// destructors run.)
nsAutoMicroTask mt;
AutoEntryScript aes(mGlobal, "fetch body data available");
MOZ_DIAGNOSTIC_ASSERT(mInputStream);
JSContext* cx = aes.cx();
ReadableStream* stream = mStreamHolder->GetReadableStreamBody();
if (!stream) {
return NS_ERROR_FAILURE;
}
uint64_t size = 0;
nsresult rv = mInputStream->Available(&size);
MOZ_ASSERT_IF(NS_SUCCEEDED(rv), size > 0);
// No warning for stream closed.
if (rv == NS_BASE_STREAM_CLOSED || NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(cx, stream, rv);
return NS_OK;
}
// Not having a promise means we are pinged by stream closure, but here we
// still have more data to read. Let's wait for the next read request in that
// case.
if (!mPullPromise) {
return NS_OK;
}
MOZ_DIAGNOSTIC_ASSERT(mPullPromise->State() ==
Promise::PromiseState::Pending);
ErrorResult errorResult;
EnqueueChunkWithSizeIntoStream(cx, stream, size, errorResult);
errorResult.WouldReportJSException();
if (errorResult.Failed()) {
ErrorPropagation(cx, stream, errorResult.StealNSResult());
return NS_OK;
}
// The previous call can execute JS (even up to running a nested event
// loop, including calling this OnInputStreamReady again before it ends, see
// the above nsAutoMicroTask), so |mPullPromise| can't be asserted to have any
// particular value, even if the previous call succeeds.
MOZ_ASSERT_IF(!mPullPromise, IsClosed());
if (mPullPromise) {
mPullPromise->MaybeResolveWithUndefined();
mPullPromise = nullptr;
}
return NS_OK;
}
/* static */
nsresult BodyStream::RetrieveInputStream(BodyStreamHolder* aStream,
nsIInputStream** aInputStream) {
MOZ_ASSERT(aStream);
MOZ_ASSERT(aInputStream);
BodyStream* stream = aStream->GetBodyStream();
if (NS_WARN_IF(!stream)) {
return NS_ERROR_DOM_INVALID_STATE_ERR;
}
MOZ_DIAGNOSTIC_ASSERT(stream->mOwningEventTarget->IsOnCurrentThread());
// if mOriginalInputStream is null, the reading already started. We don't want
// to expose the internal inputStream.
if (NS_WARN_IF(!stream->mOriginalInputStream)) {
return NS_ERROR_DOM_INVALID_STATE_ERR;
}
nsCOMPtr<nsIInputStream> inputStream = stream->mOriginalInputStream;
inputStream.forget(aInputStream);
return NS_OK;
}
void BodyStream::Close() {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
if (IsClosed()) {
return;
}
AutoJSAPI jsapi;
if (NS_WARN_IF(!jsapi.Init(mGlobal))) {
ReleaseObjects();
return;
}
ReadableStream* stream = mStreamHolder->GetReadableStreamBody();
if (stream) {
JSContext* cx = jsapi.cx();
CloseAndReleaseObjects(cx, stream);
} else {
ReleaseObjects();
}
}
void BodyStream::CloseAndReleaseObjects(JSContext* aCx,
ReadableStream* aStream) {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
MOZ_DIAGNOSTIC_ASSERT(!IsClosed());
ReleaseObjects();
if (aStream->State() == ReadableStream::ReaderState::Readable) {
IgnoredErrorResult rv;
aStream->CloseNative(aCx, rv);
NS_WARNING_ASSERTION(!rv.Failed(), "Failed to Close Stream");
}
}
void BodyStream::ReleaseObjects() {
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
if (IsClosed()) {
// Already gone. Nothing to do.
return;
}
if (NS_IsMainThread()) {
nsCOMPtr<nsIObserverService> obs = mozilla::services::GetObserverService();
if (obs) {
obs->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);
}
}
mWorkerRef = nullptr;
mGlobal = nullptr;
// It's okay to leave a potentially unsettled promise as-is as this is only
// used to prevent reentrant to PullCallback. CloseNative() or ErrorNative()
// will settle the read requests for us.
mPullPromise = nullptr;
RefPtr<BodyStream> self = mStreamHolder->TakeBodyStream();
mStreamHolder->NullifyStream();
mStreamHolder = nullptr;
}
// nsIObserver
// -----------
NS_IMETHODIMP
BodyStream::Observe(nsISupports* aSubject, const char* aTopic,
const char16_t* aData) {
AssertIsOnMainThread();
MOZ_DIAGNOSTIC_ASSERT(mOwningEventTarget->IsOnCurrentThread());
MOZ_ASSERT(strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0);
nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
if (SameCOMIdentity(aSubject, window)) {
Close();
}
return NS_OK;
}
} // namespace mozilla::dom
Ссылка в новой задаче Показать git blame Копировать постоянную ссылку