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