dom/base/BodyStream.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "BodyStream.h"
   8 #include "mozilla/CycleCollectedJSContext.h"
   9 #include "mozilla/dom/DOMException.h"
  10 #include "mozilla/dom/ScriptSettings.h"
  11 #include "mozilla/dom/WorkerCommon.h"
  12 #include "mozilla/dom/WorkerPrivate.h"
  13 #include "mozilla/dom/WorkerRunnable.h"
  14 #include "mozilla/Maybe.h"
  15 #include "mozilla/Unused.h"
  16 #include "nsProxyRelease.h"
  17 #include "nsStreamUtils.h"
  18
  19 static NS_DEFINE_CID(kStreamTransportServiceCID, NS_STREAMTRANSPORTSERVICE_CID);
  20
  21 namespace mozilla {
  22 namespace dom {
  23
  24 // BodyStreamHolder
  25 // ---------------------------------------------------------------------------
  26
  27 NS_IMPL_CYCLE_COLLECTION_CLASS(BodyStreamHolder)
  28
  29 NS_IMPL_CYCLE_COLLECTION_TRAVERSE_BEGIN(BodyStreamHolder)
  30 NS_IMPL_CYCLE_COLLECTION_TRAVERSE_END
  31
  32 NS_IMPL_CYCLE_COLLECTION_UNLINK_BEGIN(BodyStreamHolder)
  33   if (tmp->mBodyStream) {
  34     tmp->mBodyStream->ReleaseObjects();
  35     MOZ_ASSERT(!tmp->mBodyStream);
  36   }
  37 NS_IMPL_CYCLE_COLLECTION_UNLINK_END
  38
  39 NS_IMPL_CYCLE_COLLECTING_ADDREF(BodyStreamHolder)
  40 NS_IMPL_CYCLE_COLLECTING_RELEASE(BodyStreamHolder)
  41
  42 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(BodyStreamHolder)
  43   NS_INTERFACE_MAP_ENTRY(nsISupports)
  44 NS_INTERFACE_MAP_END
  45
  46 BodyStreamHolder::BodyStreamHolder() : mBodyStream(nullptr) {}
  47
  48 void BodyStreamHolder::StoreBodyStream(BodyStream* aBodyStream) {
  49   MOZ_ASSERT(aBodyStream);
  50   MOZ_ASSERT(!mBodyStream);
  51   mBodyStream = aBodyStream;
  52 }
  53
  54 void BodyStreamHolder::ForgetBodyStream() {
  55   MOZ_ASSERT_IF(mStreamCreated, mBodyStream);
  56   mBodyStream = nullptr;
  57 }
  58
  59 // BodyStream
  60 // ---------------------------------------------------------------------------
  61
  62 class BodyStream::WorkerShutdown final : public WorkerControlRunnable {
  63  public:
  64   WorkerShutdown(WorkerPrivate* aWorkerPrivate, RefPtr<BodyStream> aStream)
  65       : WorkerControlRunnable(aWorkerPrivate, WorkerThreadUnchangedBusyCount),
  66         mStream(aStream) {}
  67
  68   bool WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override {
  69     mStream->ReleaseObjects();
  70     return true;
  71   }
  72
  73   // This runnable starts from a JS Thread. We need to disable a couple of
  74   // assertions overring the following methods.
  75
  76   bool PreDispatch(WorkerPrivate* aWorkerPrivate) override { return true; }
  77
  78   void PostDispatch(WorkerPrivate* aWorkerPrivate,
  79                     bool aDispatchResult) override {}
  80
  81  private:
  82   RefPtr<BodyStream> mStream;
  83 };
  84
  85 NS_IMPL_ISUPPORTS(BodyStream, nsIInputStreamCallback, nsIObserver,
  86                   nsISupportsWeakReference)
  87
  88 /* static */
  89 void BodyStream::Create(JSContext* aCx, BodyStreamHolder* aStreamHolder,
  90                         nsIGlobalObject* aGlobal, nsIInputStream* aInputStream,
  91                         ErrorResult& aRv) {
  92   MOZ_DIAGNOSTIC_ASSERT(aCx);
  93   MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
  94   MOZ_DIAGNOSTIC_ASSERT(aInputStream);
  95
  96   RefPtr<BodyStream> stream =
  97       new BodyStream(aGlobal, aStreamHolder, aInputStream);
  98
  99   auto cleanup = MakeScopeExit([stream] { stream->Close(); });
 100
 101   if (NS_IsMainThread()) {
 102     nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
 103     if (NS_WARN_IF(!os)) {
 104       aRv.Throw(NS_ERROR_FAILURE);
 105       return;
 106     }
 107
 108     aRv = os->AddObserver(stream, DOM_WINDOW_DESTROYED_TOPIC, true);
 109     if (NS_WARN_IF(aRv.Failed())) {
 110       return;
 111     }
 112
 113   } else {
 114     WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
 115     MOZ_ASSERT(workerPrivate);
 116
 117     RefPtr<WeakWorkerRef> workerRef =
 118         WeakWorkerRef::Create(workerPrivate, [stream]() { stream->Close(); });
 119
 120     if (NS_WARN_IF(!workerRef)) {
 121       aRv.Throw(NS_ERROR_DOM_INVALID_STATE_ERR);
 122       return;
 123     }
 124
 125     // Note, this will create a ref-cycle between the holder and the stream.
 126     // The cycle is broken when the stream is closed or the worker begins
 127     // shutting down.
 128     stream->mWorkerRef = std::move(workerRef);
 129   }
 130
 131   aRv.MightThrowJSException();
 132   JS::Rooted<JSObject*> body(aCx, JS::NewReadableExternalSourceStreamObject(
 133                                       aCx, stream, aStreamHolder));
 134   if (!body) {
 135     aRv.StealExceptionFromJSContext(aCx);
 136     return;
 137   }
 138
 139   // This will be released in BodyStream::FinalizeCallback().  We are
 140   // guaranteed the jsapi will call FinalizeCallback when ReadableStream
 141   // js object is finalized.
 142   NS_ADDREF(stream.get());
 143
 144   cleanup.release();
 145
 146   aStreamHolder->StoreBodyStream(stream);
 147   aStreamHolder->SetReadableStreamBody(body);
 148
 149 #ifdef DEBUG
 150   aStreamHolder->mStreamCreated = true;
 151 #endif
 152 }
 153
 154 void BodyStream::requestData(JSContext* aCx, JS::HandleObject aStream,
 155                              size_t aDesiredSize) {
 156 #if MOZ_DIAGNOSTIC_ASSERT_ENABLED
 157   bool disturbed;
 158   if (!JS::ReadableStreamIsDisturbed(aCx, aStream, &disturbed)) {
 159     JS_ClearPendingException(aCx);
 160   } else {
 161     MOZ_DIAGNOSTIC_ASSERT(disturbed);
 162   }
 163 #endif
 164
 165   AssertIsOnOwningThread();
 166
 167   MutexAutoLock lock(mMutex);
 168
 169   MOZ_DIAGNOSTIC_ASSERT(mState == eInitializing || mState == eWaiting ||
 170                         mState == eChecking || mState == eReading);
 171
 172   if (mState == eReading) {
 173     // We are already reading data.
 174     return;
 175   }
 176
 177   if (mState == eChecking) {
 178     // If we are looking for more data, there is nothing else we should do:
 179     // let's move this checking operation in a reading.
 180     MOZ_ASSERT(mInputStream);
 181     mState = eReading;
 182     return;
 183   }
 184
 185   if (mState == eInitializing) {
 186     // The stream has been used for the first time.
 187     mStreamHolder->MarkAsRead();
 188   }
 189
 190   mState = eReading;
 191
 192   if (!mInputStream) {
 193     // This is the first use of the stream. Let's convert the
 194     // mOriginalInputStream into an nsIAsyncInputStream.
 195     MOZ_ASSERT(mOriginalInputStream);
 196
 197     nsCOMPtr<nsIAsyncInputStream> asyncStream;
 198     nsresult rv = NS_MakeAsyncNonBlockingInputStream(
 199         mOriginalInputStream.forget(), getter_AddRefs(asyncStream));
 200     if (NS_WARN_IF(NS_FAILED(rv))) {
 201       ErrorPropagation(aCx, lock, aStream, rv);
 202       return;
 203     }
 204
 205     mInputStream = asyncStream;
 206     mOriginalInputStream = nullptr;
 207   }
 208
 209   MOZ_DIAGNOSTIC_ASSERT(mInputStream);
 210   MOZ_DIAGNOSTIC_ASSERT(!mOriginalInputStream);
 211
 212   nsresult rv = mInputStream->AsyncWait(this, 0, 0, mOwningEventTarget);
 213   if (NS_WARN_IF(NS_FAILED(rv))) {
 214     ErrorPropagation(aCx, lock, aStream, rv);
 215     return;
 216   }
 217
 218   // All good.
 219 }
 220
 221 void BodyStream::writeIntoReadRequestBuffer(JSContext* aCx,
 222                                             JS::HandleObject aStream,
 223                                             void* aBuffer, size_t aLength,
 224                                             size_t* aByteWritten) {
 225   MOZ_DIAGNOSTIC_ASSERT(aBuffer);
 226   MOZ_DIAGNOSTIC_ASSERT(aByteWritten);
 227
 228   AssertIsOnOwningThread();
 229
 230   MutexAutoLock lock(mMutex);
 231
 232   MOZ_DIAGNOSTIC_ASSERT(mInputStream);
 233   MOZ_DIAGNOSTIC_ASSERT(mState == eWriting);
 234   mState = eChecking;
 235
 236   uint32_t written;
 237   nsresult rv =
 238       mInputStream->Read(static_cast<char*>(aBuffer), aLength, &written);
 239   if (NS_WARN_IF(NS_FAILED(rv))) {
 240     ErrorPropagation(aCx, lock, aStream, rv);
 241     return;
 242   }
 243
 244   *aByteWritten = written;
 245
 246   if (written == 0) {
 247     CloseAndReleaseObjects(aCx, lock, aStream);
 248     return;
 249   }
 250
 251   rv = mInputStream->AsyncWait(this, 0, 0, mOwningEventTarget);
 252   if (NS_WARN_IF(NS_FAILED(rv))) {
 253     ErrorPropagation(aCx, lock, aStream, rv);
 254     return;
 255   }
 256
 257   // All good.
 258 }
 259
 260 JS::Value BodyStream::cancel(JSContext* aCx, JS::HandleObject aStream,
 261                              JS::HandleValue aReason) {
 262   AssertIsOnOwningThread();
 263
 264   if (mState == eInitializing) {
 265     // The stream has been used for the first time.
 266     mStreamHolder->MarkAsRead();
 267   }
 268
 269   if (mInputStream) {
 270     mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
 271   }
 272
 273   // It could be that we don't have mInputStream yet, but we still have the
 274   // original stream. We need to close that too.
 275   if (mOriginalInputStream) {
 276     MOZ_ASSERT(!mInputStream);
 277     mOriginalInputStream->Close();
 278   }
 279
 280   ReleaseObjects();
 281   return JS::UndefinedValue();
 282 }
 283
 284 void BodyStream::onClosed(JSContext* aCx, JS::HandleObject aStream) {}
 285
 286 void BodyStream::onErrored(JSContext* aCx, JS::HandleObject aStream,
 287                            JS::HandleValue aReason) {
 288   AssertIsOnOwningThread();
 289
 290   if (mState == eInitializing) {
 291     // The stream has been used for the first time.
 292     mStreamHolder->MarkAsRead();
 293   }
 294
 295   if (mInputStream) {
 296     mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
 297   }
 298
 299   ReleaseObjects();
 300 }
 301
 302 void BodyStream::finalize() {
 303   // This can be called in any thread.
 304
 305   // This takes ownership of the ref created in BodyStream::Create().
 306   RefPtr<BodyStream> stream = dont_AddRef(this);
 307
 308   stream->ReleaseObjects();
 309 }
 310
 311 BodyStream::BodyStream(nsIGlobalObject* aGlobal,
 312                        BodyStreamHolder* aStreamHolder,
 313                        nsIInputStream* aInputStream)
 314     : mMutex("BodyStream::mMutex"),
 315       mState(eInitializing),
 316       mGlobal(aGlobal),
 317       mStreamHolder(aStreamHolder),
 318       mOwningEventTarget(aGlobal->EventTargetFor(TaskCategory::Other)),
 319       mOriginalInputStream(aInputStream) {
 320   MOZ_DIAGNOSTIC_ASSERT(aInputStream);
 321   MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
 322 }
 323
 324 BodyStream::~BodyStream() = default;
 325
 326 void BodyStream::ErrorPropagation(JSContext* aCx,
 327                                   const MutexAutoLock& aProofOfLock,
 328                                   JS::HandleObject aStream, nsresult aError) {
 329   AssertIsOnOwningThread();
 330
 331   // Nothing to do.
 332   if (mState == eClosed) {
 333     return;
 334   }
 335
 336   // Let's close the stream.
 337   if (aError == NS_BASE_STREAM_CLOSED) {
 338     CloseAndReleaseObjects(aCx, aProofOfLock, aStream);
 339     return;
 340   }
 341
 342   // Let's use a generic error.
 343   ErrorResult rv;
 344   // XXXbz can we come up with a better error message here to tell the
 345   // consumer what went wrong?
 346   rv.ThrowTypeError("Error in body stream");
 347
 348   JS::Rooted<JS::Value> errorValue(aCx);
 349   bool ok = ToJSValue(aCx, std::move(rv), &errorValue);
 350   MOZ_RELEASE_ASSERT(ok, "ToJSValue never fails for ErrorResult");
 351
 352   {
 353     MutexAutoUnlock unlock(mMutex);
 354     JS::ReadableStreamError(aCx, aStream, errorValue);
 355   }
 356
 357   ReleaseObjects(aProofOfLock);
 358 }
 359
 360 NS_IMETHODIMP
 361 BodyStream::OnInputStreamReady(nsIAsyncInputStream* aStream) {
 362   AssertIsOnOwningThread();
 363   MOZ_DIAGNOSTIC_ASSERT(aStream);
 364
 365   // Acquire |mMutex| in order to safely inspect |mState| and use |mGlobal|.
 366   Maybe<MutexAutoLock> lock;
 367   lock.emplace(mMutex);
 368
 369   // Already closed. We have nothing else to do here.
 370   if (mState == eClosed) {
 371     return NS_OK;
 372   }
 373
 374   // Perform a microtask checkpoint after all actions are completed.  Note that
 375   // |mMutex| *must not* be held when the checkpoint occurs -- hence, far down,
 376   // the |lock.reset()|.  (|MutexAutoUnlock| as RAII wouldn't work for this task
 377   // because its destructor would reacquire |mMutex| before these objects'
 378   // destructors run.)
 379   nsAutoMicroTask mt;
 380   AutoEntryScript aes(mGlobal, "fetch body data available");
 381
 382   MOZ_DIAGNOSTIC_ASSERT(mInputStream);
 383   MOZ_DIAGNOSTIC_ASSERT(mState == eReading || mState == eChecking);
 384
 385   JSObject* streamObj = mStreamHolder->GetReadableStreamBody();
 386   if (!streamObj) {
 387     return NS_ERROR_FAILURE;
 388   }
 389
 390   JSContext* cx = aes.cx();
 391   JS::Rooted<JSObject*> stream(cx, streamObj);
 392
 393   uint64_t size = 0;
 394   nsresult rv = mInputStream->Available(&size);
 395   if (NS_SUCCEEDED(rv) && size == 0) {
 396     // In theory this should not happen. If size is 0, the stream should be
 397     // considered closed.
 398     rv = NS_BASE_STREAM_CLOSED;
 399   }
 400
 401   // No warning for stream closed.
 402   if (rv == NS_BASE_STREAM_CLOSED || NS_WARN_IF(NS_FAILED(rv))) {
 403     ErrorPropagation(cx, *lock, stream, rv);
 404     return NS_OK;
 405   }
 406
 407   // This extra checking is completed. Let's wait for the next read request.
 408   if (mState == eChecking) {
 409     mState = eWaiting;
 410     return NS_OK;
 411   }
 412
 413   mState = eWriting;
 414
 415   // Release the mutex before the call below (which could execute JS), as well
 416   // as before the microtask checkpoint queued up above occurs.
 417   lock.reset();
 418
 419   Unused << JS::ReadableStreamUpdateDataAvailableFromSource(cx, stream, size);
 420
 421   // The previous call can execute JS (even up to running a nested event loop),
 422   // so |mState| can't be asserted to have any particular value, even if the
 423   // previous call succeeds.
 424
 425   return NS_OK;
 426 }
 427
 428 /* static */
 429 nsresult BodyStream::RetrieveInputStream(
 430     JS::ReadableStreamUnderlyingSource* aUnderlyingReadableStreamSource,
 431     nsIInputStream** aInputStream) {
 432   MOZ_ASSERT(aUnderlyingReadableStreamSource);
 433   MOZ_ASSERT(aInputStream);
 434
 435   RefPtr<BodyStream> stream =
 436       static_cast<BodyStream*>(aUnderlyingReadableStreamSource);
 437   stream->AssertIsOnOwningThread();
 438
 439   // if mOriginalInputStream is null, the reading already started. We don't want
 440   // to expose the internal inputStream.
 441   if (NS_WARN_IF(!stream->mOriginalInputStream)) {
 442     return NS_ERROR_DOM_INVALID_STATE_ERR;
 443   }
 444
 445   nsCOMPtr<nsIInputStream> inputStream = stream->mOriginalInputStream;
 446   inputStream.forget(aInputStream);
 447   return NS_OK;
 448 }
 449
 450 void BodyStream::Close() {
 451   AssertIsOnOwningThread();
 452
 453   MutexAutoLock lock(mMutex);
 454
 455   if (mState == eClosed) {
 456     return;
 457   }
 458
 459   AutoJSAPI jsapi;
 460   if (NS_WARN_IF(!jsapi.Init(mGlobal))) {
 461     ReleaseObjects(lock);
 462     return;
 463   }
 464
 465   JSObject* streamObj = mStreamHolder->GetReadableStreamBody();
 466   if (streamObj) {
 467     JSContext* cx = jsapi.cx();
 468     JS::Rooted<JSObject*> stream(cx, streamObj);
 469     CloseAndReleaseObjects(cx, lock, stream);
 470   } else {
 471     ReleaseObjects(lock);
 472   }
 473 }
 474
 475 void BodyStream::CloseAndReleaseObjects(JSContext* aCx,
 476                                         const MutexAutoLock& aProofOfLock,
 477                                         JS::HandleObject aStream) {
 478   AssertIsOnOwningThread();
 479   MOZ_DIAGNOSTIC_ASSERT(mState != eClosed);
 480
 481   ReleaseObjects(aProofOfLock);
 482
 483   MutexAutoUnlock unlock(mMutex);
 484   bool readable;
 485   if (!JS::ReadableStreamIsReadable(aCx, aStream, &readable)) {
 486     return;
 487   }
 488   if (readable) {
 489     JS::ReadableStreamClose(aCx, aStream);
 490   }
 491 }
 492
 493 void BodyStream::ReleaseObjects() {
 494   MutexAutoLock lock(mMutex);
 495   ReleaseObjects(lock);
 496 }
 497
 498 void BodyStream::ReleaseObjects(const MutexAutoLock& aProofOfLock) {
 499   // This method can be called on 2 possible threads: the owning one and a JS
 500   // thread used to release resources. If we are on the JS thread, we need to
 501   // dispatch a runnable to go back to the owning thread in order to release
 502   // resources correctly.
 503
 504   if (mState == eClosed) {
 505     // Already gone. Nothing to do.
 506     return;
 507   }
 508
 509   if (!NS_IsMainThread() && !IsCurrentThreadRunningWorker()) {
 510     // Let's dispatch a WorkerControlRunnable if the owning thread is a worker.
 511     if (mWorkerRef) {
 512       RefPtr<WorkerShutdown> r =
 513           new WorkerShutdown(mWorkerRef->GetUnsafePrivate(), this);
 514       Unused << NS_WARN_IF(!r->Dispatch());
 515       return;
 516     }
 517
 518     // A normal runnable of the owning thread is the main-thread.
 519     RefPtr<BodyStream> self = this;
 520     RefPtr<Runnable> r = NS_NewRunnableFunction(
 521         "BodyStream::ReleaseObjects", [self]() { self->ReleaseObjects(); });
 522     mOwningEventTarget->Dispatch(r.forget());
 523     return;
 524   }
 525
 526   AssertIsOnOwningThread();
 527
 528   mState = eClosed;
 529
 530   if (NS_IsMainThread()) {
 531     nsCOMPtr<nsIObserverService> obs = mozilla::services::GetObserverService();
 532     if (obs) {
 533       obs->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);
 534     }
 535   }
 536
 537   JSObject* streamObj = mStreamHolder->GetReadableStreamBody();
 538   if (streamObj) {
 539     // Let's inform the JSEngine that we are going to be released.
 540     JS::ReadableStreamReleaseCCObject(streamObj);
 541   }
 542
 543   mWorkerRef = nullptr;
 544   mGlobal = nullptr;
 545
 546   mStreamHolder->ForgetBodyStream();
 547   mStreamHolder->NullifyStream();
 548   mStreamHolder = nullptr;
 549 }
 550
 551 #ifdef DEBUG
 552 void BodyStream::AssertIsOnOwningThread() {
 553   NS_ASSERT_OWNINGTHREAD(BodyStream);
 554 }
 555 #endif
 556
 557 // nsIObserver
 558 // -----------
 559
 560 NS_IMETHODIMP
 561 BodyStream::Observe(nsISupports* aSubject, const char* aTopic,
 562                     const char16_t* aData) {
 563   AssertIsOnMainThread();
 564   AssertIsOnOwningThread();
 565
 566   MOZ_ASSERT(strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0);
 567
 568   nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
 569   if (SameCOMIdentity(aSubject, window)) {
 570     Close();
 571   }
 572
 573   return NS_OK;
 574 }
 575
 576 }  // namespace dom
 577 }  // namespace mozilla