netwerk/base/nsAsyncStreamCopier.cpp

   1 /* This Source Code Form is subject to the terms of the Mozilla Public
   2  * License, v. 2.0. If a copy of the MPL was not distributed with this
   3  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   4
   5 #include "nsAsyncStreamCopier.h"
   6 #include "nsComponentManagerUtils.h"
   7 #include "nsIOService.h"
   8 #include "nsIEventTarget.h"
   9 #include "nsStreamUtils.h"
  10 #include "nsThreadUtils.h"
  11 #include "nsNetUtil.h"
  12 #include "nsNetCID.h"
  13 #include "nsIBufferedStreams.h"
  14 #include "nsIRequestObserver.h"
  15 #include "mozilla/Logging.h"
  16
  17 using namespace mozilla;
  18 using namespace mozilla::net;
  19
  20 #undef LOG
  21 //
  22 // MOZ_LOG=nsStreamCopier:5
  23 //
  24 static LazyLogModule gStreamCopierLog("nsStreamCopier");
  25 #define LOG(args) MOZ_LOG(gStreamCopierLog, mozilla::LogLevel::Debug, args)
  26
  27 /**
  28  * An event used to perform initialization off the main thread.
  29  */
  30 class AsyncApplyBufferingPolicyEvent final : public Runnable {
  31  public:
  32   /**
  33    * @param aCopier
  34    *        The nsAsyncStreamCopier requesting the information.
  35    */
  36   explicit AsyncApplyBufferingPolicyEvent(nsAsyncStreamCopier* aCopier)
  37       : mozilla::Runnable("AsyncApplyBufferingPolicyEvent"),
  38         mCopier(aCopier),
  39         mTarget(GetCurrentSerialEventTarget()) {}
  40
  41   NS_IMETHOD Run() override {
  42     nsresult rv = mCopier->ApplyBufferingPolicy();
  43     if (NS_FAILED(rv)) {
  44       mCopier->Cancel(rv);
  45       return NS_OK;
  46     }
  47
  48     rv = mTarget->Dispatch(
  49         NewRunnableMethod("nsAsyncStreamCopier::AsyncCopyInternal", mCopier,
  50                           &nsAsyncStreamCopier::AsyncCopyInternal),
  51         NS_DISPATCH_NORMAL);
  52     MOZ_ASSERT(NS_SUCCEEDED(rv));
  53
  54     if (NS_FAILED(rv)) {
  55       mCopier->Cancel(rv);
  56     }
  57     return NS_OK;
  58   }
  59
  60  private:
  61   RefPtr<nsAsyncStreamCopier> mCopier;
  62   nsCOMPtr<nsIEventTarget> mTarget;
  63 };
  64
  65 //-----------------------------------------------------------------------------
  66
  67 nsAsyncStreamCopier::nsAsyncStreamCopier()
  68     : mChunkSize(nsIOService::gDefaultSegmentSize) {
  69   LOG(("Creating nsAsyncStreamCopier @%p\n", this));
  70 }
  71
  72 nsAsyncStreamCopier::~nsAsyncStreamCopier() {
  73   LOG(("Destroying nsAsyncStreamCopier @%p\n", this));
  74 }
  75
  76 bool nsAsyncStreamCopier::IsComplete(nsresult* status) {
  77   MutexAutoLock lock(mLock);
  78   if (status) *status = mStatus;
  79   return !mIsPending;
  80 }
  81
  82 nsIRequest* nsAsyncStreamCopier::AsRequest() {
  83   return static_cast<nsIRequest*>(static_cast<nsIAsyncStreamCopier*>(this));
  84 }
  85
  86 void nsAsyncStreamCopier::Complete(nsresult status) {
  87   LOG(("nsAsyncStreamCopier::Complete [this=%p status=%" PRIx32 "]\n", this,
  88        static_cast<uint32_t>(status)));
  89
  90   nsCOMPtr<nsIRequestObserver> observer;
  91   nsCOMPtr<nsISupports> ctx;
  92   {
  93     MutexAutoLock lock(mLock);
  94     mCopierCtx = nullptr;
  95
  96     if (mIsPending) {
  97       mIsPending = false;
  98       mStatus = status;
  99
 100       // setup OnStopRequest callback and release references...
 101       observer = mObserver;
 102       mObserver = nullptr;
 103     }
 104   }
 105
 106   if (observer) {
 107     LOG(("  calling OnStopRequest [status=%" PRIx32 "]\n",
 108          static_cast<uint32_t>(status)));
 109     observer->OnStopRequest(AsRequest(), status);
 110   }
 111 }
 112
 113 void nsAsyncStreamCopier::OnAsyncCopyComplete(void* closure, nsresult status) {
 114   // AddRef'd in AsyncCopy. Will be released at the end of the method.
 115   RefPtr<nsAsyncStreamCopier> self = dont_AddRef((nsAsyncStreamCopier*)closure);
 116   self->Complete(status);
 117 }
 118
 119 //-----------------------------------------------------------------------------
 120 // nsISupports
 121
 122 // We cannot use simply NS_IMPL_ISUPPORTSx as both
 123 // nsIAsyncStreamCopier and nsIAsyncStreamCopier2 implement nsIRequest
 124
 125 NS_IMPL_ADDREF(nsAsyncStreamCopier)
 126 NS_IMPL_RELEASE(nsAsyncStreamCopier)
 127 NS_INTERFACE_TABLE_HEAD(nsAsyncStreamCopier)
 128   NS_INTERFACE_TABLE_BEGIN
 129     NS_INTERFACE_TABLE_ENTRY(nsAsyncStreamCopier, nsIAsyncStreamCopier)
 130     NS_INTERFACE_TABLE_ENTRY(nsAsyncStreamCopier, nsIAsyncStreamCopier2)
 131     NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsAsyncStreamCopier, nsIRequest,
 132                                        nsIAsyncStreamCopier)
 133     NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsAsyncStreamCopier, nsISupports,
 134                                        nsIAsyncStreamCopier)
 135   NS_INTERFACE_TABLE_END
 136 NS_INTERFACE_TABLE_TAIL
 137
 138 //-----------------------------------------------------------------------------
 139 // nsIRequest
 140
 141 NS_IMETHODIMP
 142 nsAsyncStreamCopier::GetName(nsACString& name) {
 143   name.Truncate();
 144   return NS_OK;
 145 }
 146
 147 NS_IMETHODIMP
 148 nsAsyncStreamCopier::IsPending(bool* result) {
 149   *result = !IsComplete();
 150   return NS_OK;
 151 }
 152
 153 NS_IMETHODIMP
 154 nsAsyncStreamCopier::GetStatus(nsresult* status) {
 155   IsComplete(status);
 156   return NS_OK;
 157 }
 158
 159 NS_IMETHODIMP nsAsyncStreamCopier::SetCanceledReason(
 160     const nsACString& aReason) {
 161   return nsIAsyncStreamCopier::SetCanceledReasonImpl(aReason);
 162 }
 163
 164 NS_IMETHODIMP nsAsyncStreamCopier::GetCanceledReason(nsACString& aReason) {
 165   return nsIAsyncStreamCopier::GetCanceledReasonImpl(aReason);
 166 }
 167
 168 NS_IMETHODIMP nsAsyncStreamCopier::CancelWithReason(nsresult aStatus,
 169                                                     const nsACString& aReason) {
 170   return nsIAsyncStreamCopier::CancelWithReasonImpl(aStatus, aReason);
 171 }
 172
 173 NS_IMETHODIMP
 174 nsAsyncStreamCopier::Cancel(nsresult status) {
 175   nsCOMPtr<nsISupports> copierCtx;
 176   {
 177     MutexAutoLock lock(mLock);
 178     if (!mIsPending) {
 179       return NS_OK;
 180     }
 181     copierCtx.swap(mCopierCtx);
 182   }
 183
 184   if (NS_SUCCEEDED(status)) {
 185     NS_WARNING("cancel with non-failure status code");
 186     status = NS_BASE_STREAM_CLOSED;
 187   }
 188
 189   if (copierCtx) NS_CancelAsyncCopy(copierCtx, status);
 190
 191   return NS_OK;
 192 }
 193
 194 NS_IMETHODIMP
 195 nsAsyncStreamCopier::Suspend() {
 196   MOZ_ASSERT_UNREACHABLE("nsAsyncStreamCopier::Suspend");
 197   return NS_ERROR_NOT_IMPLEMENTED;
 198 }
 199
 200 NS_IMETHODIMP
 201 nsAsyncStreamCopier::Resume() {
 202   MOZ_ASSERT_UNREACHABLE("nsAsyncStreamCopier::Resume");
 203   return NS_ERROR_NOT_IMPLEMENTED;
 204 }
 205
 206 NS_IMETHODIMP
 207 nsAsyncStreamCopier::GetLoadFlags(nsLoadFlags* aLoadFlags) {
 208   *aLoadFlags = LOAD_NORMAL;
 209   return NS_OK;
 210 }
 211
 212 NS_IMETHODIMP
 213 nsAsyncStreamCopier::SetLoadFlags(nsLoadFlags aLoadFlags) { return NS_OK; }
 214
 215 NS_IMETHODIMP
 216 nsAsyncStreamCopier::GetTRRMode(nsIRequest::TRRMode* aTRRMode) {
 217   return nsIAsyncStreamCopier::GetTRRModeImpl(aTRRMode);
 218 }
 219
 220 NS_IMETHODIMP
 221 nsAsyncStreamCopier::SetTRRMode(nsIRequest::TRRMode aTRRMode) {
 222   return nsIAsyncStreamCopier::SetTRRModeImpl(aTRRMode);
 223 }
 224
 225 NS_IMETHODIMP
 226 nsAsyncStreamCopier::GetLoadGroup(nsILoadGroup** aLoadGroup) {
 227   *aLoadGroup = nullptr;
 228   return NS_OK;
 229 }
 230
 231 NS_IMETHODIMP
 232 nsAsyncStreamCopier::SetLoadGroup(nsILoadGroup* aLoadGroup) { return NS_OK; }
 233
 234 // Can't be accessed by multiple threads yet
 235 nsresult nsAsyncStreamCopier::InitInternal(
 236     nsIInputStream* source, nsIOutputStream* sink, nsIEventTarget* target,
 237     uint32_t chunkSize, bool closeSource,
 238     bool closeSink) MOZ_NO_THREAD_SAFETY_ANALYSIS {
 239   NS_ASSERTION(!mSource && !mSink, "Init() called more than once");
 240   if (chunkSize == 0) {
 241     chunkSize = nsIOService::gDefaultSegmentSize;
 242   }
 243   mChunkSize = chunkSize;
 244
 245   mSource = source;
 246   mSink = sink;
 247   mCloseSource = closeSource;
 248   mCloseSink = closeSink;
 249
 250   if (target) {
 251     mTarget = target;
 252   } else {
 253     nsresult rv;
 254     mTarget = do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID, &rv);
 255     if (NS_FAILED(rv)) {
 256       return rv;
 257     }
 258   }
 259
 260   return NS_OK;
 261 }
 262
 263 //-----------------------------------------------------------------------------
 264 // nsIAsyncStreamCopier
 265
 266 NS_IMETHODIMP
 267 nsAsyncStreamCopier::Init(nsIInputStream* source, nsIOutputStream* sink,
 268                           nsIEventTarget* target, bool sourceBuffered,
 269                           bool sinkBuffered, uint32_t chunkSize,
 270                           bool closeSource, bool closeSink) {
 271   NS_ASSERTION(sourceBuffered || sinkBuffered,
 272                "at least one stream must be buffered");
 273   mMode = sourceBuffered ? NS_ASYNCCOPY_VIA_READSEGMENTS
 274                          : NS_ASYNCCOPY_VIA_WRITESEGMENTS;
 275
 276   return InitInternal(source, sink, target, chunkSize, closeSource, closeSink);
 277 }
 278
 279 //-----------------------------------------------------------------------------
 280 // nsIAsyncStreamCopier2
 281
 282 NS_IMETHODIMP
 283 nsAsyncStreamCopier::Init(nsIInputStream* source, nsIOutputStream* sink,
 284                           nsIEventTarget* target, uint32_t chunkSize,
 285                           bool closeSource, bool closeSink) {
 286   mShouldSniffBuffering = true;
 287
 288   return InitInternal(source, sink, target, chunkSize, closeSource, closeSink);
 289 }
 290
 291 /**
 292  * Detect whether the input or the output stream is buffered,
 293  * bufferize one of them if neither is buffered.
 294  */
 295 nsresult nsAsyncStreamCopier::ApplyBufferingPolicy() {
 296   // This function causes I/O, it must not be executed on the main
 297   // thread.
 298   MOZ_ASSERT(!NS_IsMainThread());
 299
 300   if (NS_OutputStreamIsBuffered(mSink)) {
 301     // Sink is buffered, no need to perform additional buffering
 302     mMode = NS_ASYNCCOPY_VIA_WRITESEGMENTS;
 303     return NS_OK;
 304   }
 305   if (NS_InputStreamIsBuffered(mSource)) {
 306     // Source is buffered, no need to perform additional buffering
 307     mMode = NS_ASYNCCOPY_VIA_READSEGMENTS;
 308     return NS_OK;
 309   }
 310
 311   // No buffering, let's buffer the sink
 312   nsresult rv;
 313   nsCOMPtr<nsIBufferedOutputStream> sink =
 314       do_CreateInstance(NS_BUFFEREDOUTPUTSTREAM_CONTRACTID, &rv);
 315   if (NS_FAILED(rv)) {
 316     return rv;
 317   }
 318
 319   rv = sink->Init(mSink, mChunkSize);
 320   if (NS_FAILED(rv)) {
 321     return rv;
 322   }
 323
 324   mMode = NS_ASYNCCOPY_VIA_WRITESEGMENTS;
 325   mSink = sink;
 326   return NS_OK;
 327 }
 328
 329 //-----------------------------------------------------------------------------
 330 // Both nsIAsyncStreamCopier and nsIAsyncStreamCopier2
 331
 332 NS_IMETHODIMP
 333 nsAsyncStreamCopier::AsyncCopy(nsIRequestObserver* observer, nsISupports* ctx) {
 334   LOG(("nsAsyncStreamCopier::AsyncCopy [this=%p observer=%p]\n", this,
 335        observer));
 336
 337   NS_ASSERTION(mSource && mSink, "not initialized");
 338   nsresult rv;
 339
 340   if (observer) {
 341     // build proxy for observer events
 342     rv = NS_NewRequestObserverProxy(getter_AddRefs(mObserver), observer, ctx);
 343     if (NS_FAILED(rv)) return rv;
 344   }
 345
 346   // from this point forward, AsyncCopy is going to return NS_OK.  any errors
 347   // will be reported via OnStopRequest.
 348   {
 349     MutexAutoLock lock(mLock);
 350     mIsPending = true;
 351   }
 352
 353   if (mObserver) {
 354     rv = mObserver->OnStartRequest(AsRequest());
 355     if (NS_FAILED(rv)) Cancel(rv);
 356   }
 357
 358   if (!mShouldSniffBuffering) {
 359     // No buffer sniffing required, let's proceed
 360     AsyncCopyInternal();
 361     return NS_OK;
 362   }
 363
 364   if (NS_IsMainThread()) {
 365     // Don't perform buffer sniffing on the main thread
 366     nsCOMPtr<nsIRunnable> event = new AsyncApplyBufferingPolicyEvent(this);
 367     rv = mTarget->Dispatch(event, NS_DISPATCH_NORMAL);
 368     if (NS_FAILED(rv)) {
 369       Cancel(rv);
 370     }
 371     return NS_OK;
 372   }
 373
 374   // We're not going to block the main thread, so let's sniff here
 375   rv = ApplyBufferingPolicy();
 376   if (NS_FAILED(rv)) {
 377     Cancel(rv);
 378   }
 379   AsyncCopyInternal();
 380   return NS_OK;
 381 }
 382
 383 // Launch async copy.
 384 // All errors are reported through the observer.
 385 void nsAsyncStreamCopier::AsyncCopyInternal() {
 386   MOZ_ASSERT(mMode == NS_ASYNCCOPY_VIA_READSEGMENTS ||
 387              mMode == NS_ASYNCCOPY_VIA_WRITESEGMENTS);
 388
 389   nsresult rv;
 390   // We want to receive progress notifications; release happens in
 391   // OnAsyncCopyComplete.
 392   RefPtr<nsAsyncStreamCopier> self = this;
 393   {
 394     MutexAutoLock lock(mLock);
 395     rv = NS_AsyncCopy(mSource, mSink, mTarget, mMode, mChunkSize,
 396                       OnAsyncCopyComplete, this, mCloseSource, mCloseSink,
 397                       getter_AddRefs(mCopierCtx));
 398   }
 399   if (NS_FAILED(rv)) {
 400     Cancel(rv);
 401     return;  // release self
 402   }
 403
 404   Unused << self.forget();  // Will be released in OnAsyncCopyComplete
 405 }