1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/netwerk/protocol/http/SpdySession31.cpp Wed Dec 31 06:09:35 2014 +0100
1.3 @@ -0,0 +1,2866 @@
1.4 +/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
1.5 +/* vim: set sw=2 ts=8 et tw=80 : */
1.6 +/* This Source Code Form is subject to the terms of the Mozilla Public
1.7 + * License, v. 2.0. If a copy of the MPL was not distributed with this
1.8 + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
1.9 +
1.10 +// HttpLog.h should generally be included first
1.11 +#include "HttpLog.h"
1.12 +
1.13 +// Log on level :5, instead of default :4.
1.14 +#undef LOG
1.15 +#define LOG(args) LOG5(args)
1.16 +#undef LOG_ENABLED
1.17 +#define LOG_ENABLED() LOG5_ENABLED()
1.18 +
1.19 +#include "mozilla/Telemetry.h"
1.20 +#include "mozilla/Preferences.h"
1.21 +#include "nsHttp.h"
1.22 +#include "nsHttpHandler.h"
1.23 +#include "nsHttpConnection.h"
1.24 +#include "nsILoadGroup.h"
1.25 +#include "prprf.h"
1.26 +#include "prnetdb.h"
1.27 +#include "SpdyPush31.h"
1.28 +#include "SpdySession31.h"
1.29 +#include "SpdyStream31.h"
1.30 +#include "SpdyZlibReporter.h"
1.31 +
1.32 +#include <algorithm>
1.33 +
1.34 +#ifdef DEBUG
1.35 +// defined by the socket transport service while active
1.36 +extern PRThread *gSocketThread;
1.37 +#endif
1.38 +
1.39 +namespace mozilla {
1.40 +namespace net {
1.41 +
1.42 +// SpdySession31 has multiple inheritance of things that implement
1.43 +// nsISupports, so this magic is taken from nsHttpPipeline that
1.44 +// implements some of the same abstract classes.
1.45 +NS_IMPL_ADDREF(SpdySession31)
1.46 +NS_IMPL_RELEASE(SpdySession31)
1.47 +NS_INTERFACE_MAP_BEGIN(SpdySession31)
1.48 +NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsAHttpConnection)
1.49 +NS_INTERFACE_MAP_END
1.50 +
1.51 +SpdySession31::SpdySession31(nsAHttpTransaction *aHttpTransaction,
1.52 + nsISocketTransport *aSocketTransport,
1.53 + int32_t firstPriority)
1.54 + : mSocketTransport(aSocketTransport),
1.55 + mSegmentReader(nullptr),
1.56 + mSegmentWriter(nullptr),
1.57 + mNextStreamID(1),
1.58 + mConcurrentHighWater(0),
1.59 + mDownstreamState(BUFFERING_FRAME_HEADER),
1.60 + mInputFrameBufferSize(kDefaultBufferSize),
1.61 + mInputFrameBufferUsed(0),
1.62 + mInputFrameDataLast(false),
1.63 + mInputFrameDataStream(nullptr),
1.64 + mNeedsCleanup(nullptr),
1.65 + mShouldGoAway(false),
1.66 + mClosed(false),
1.67 + mCleanShutdown(false),
1.68 + mDataPending(false),
1.69 + mGoAwayID(0),
1.70 + mMaxConcurrent(kDefaultMaxConcurrent),
1.71 + mConcurrent(0),
1.72 + mServerPushedResources(0),
1.73 + mServerInitialStreamWindow(kDefaultRwin),
1.74 + mLocalSessionWindow(kDefaultRwin),
1.75 + mRemoteSessionWindow(kDefaultRwin),
1.76 + mOutputQueueSize(kDefaultQueueSize),
1.77 + mOutputQueueUsed(0),
1.78 + mOutputQueueSent(0),
1.79 + mLastReadEpoch(PR_IntervalNow()),
1.80 + mPingSentEpoch(0),
1.81 + mNextPingID(1)
1.82 +{
1.83 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.84 +
1.85 + static uint64_t sSerial;
1.86 + mSerial = ++sSerial;
1.87 +
1.88 + LOG3(("SpdySession31::SpdySession31 %p transaction 1 = %p serial=0x%X\n",
1.89 + this, aHttpTransaction, mSerial));
1.90 +
1.91 + mConnection = aHttpTransaction->Connection();
1.92 + mInputFrameBuffer = new char[mInputFrameBufferSize];
1.93 + mOutputQueueBuffer = new char[mOutputQueueSize];
1.94 + zlibInit();
1.95 +
1.96 + mPushAllowance = gHttpHandler->SpdyPushAllowance();
1.97 +
1.98 + mSendingChunkSize = gHttpHandler->SpdySendingChunkSize();
1.99 + GenerateSettings();
1.100 +
1.101 + if (!aHttpTransaction->IsNullTransaction())
1.102 + AddStream(aHttpTransaction, firstPriority);
1.103 + mLastDataReadEpoch = mLastReadEpoch;
1.104 +
1.105 + mPingThreshold = gHttpHandler->SpdyPingThreshold();
1.106 +}
1.107 +
1.108 +PLDHashOperator
1.109 + SpdySession31::ShutdownEnumerator(nsAHttpTransaction *key,
1.110 + nsAutoPtr<SpdyStream31> &stream,
1.111 + void *closure)
1.112 +{
1.113 + SpdySession31 *self = static_cast<SpdySession31 *>(closure);
1.114 +
1.115 + // On a clean server hangup the server sets the GoAwayID to be the ID of
1.116 + // the last transaction it processed. If the ID of stream in the
1.117 + // local stream is greater than that it can safely be restarted because the
1.118 + // server guarantees it was not partially processed. Streams that have not
1.119 + // registered an ID haven't actually been sent yet so they can always be
1.120 + // restarted.
1.121 + if (self->mCleanShutdown &&
1.122 + (stream->StreamID() > self->mGoAwayID || !stream->HasRegisteredID()))
1.123 + self->CloseStream(stream, NS_ERROR_NET_RESET); // can be restarted
1.124 + else
1.125 + self->CloseStream(stream, NS_ERROR_ABORT);
1.126 +
1.127 + return PL_DHASH_NEXT;
1.128 +}
1.129 +
1.130 +PLDHashOperator
1.131 +SpdySession31::GoAwayEnumerator(nsAHttpTransaction *key,
1.132 + nsAutoPtr<SpdyStream31> &stream,
1.133 + void *closure)
1.134 +{
1.135 + SpdySession31 *self = static_cast<SpdySession31 *>(closure);
1.136 +
1.137 + // these streams were not processed by the server and can be restarted.
1.138 + // Do that after the enumerator completes to avoid the risk of
1.139 + // a restart event re-entrantly modifying this hash. Be sure not to restart
1.140 + // a pushed (even numbered) stream
1.141 + if ((stream->StreamID() > self->mGoAwayID && (stream->StreamID() & 1)) ||
1.142 + !stream->HasRegisteredID()) {
1.143 + self->mGoAwayStreamsToRestart.Push(stream);
1.144 + }
1.145 +
1.146 + return PL_DHASH_NEXT;
1.147 +}
1.148 +
1.149 +SpdySession31::~SpdySession31()
1.150 +{
1.151 + LOG3(("SpdySession31::~SpdySession31 %p mDownstreamState=%X",
1.152 + this, mDownstreamState));
1.153 +
1.154 + inflateEnd(&mDownstreamZlib);
1.155 + deflateEnd(&mUpstreamZlib);
1.156 +
1.157 + mStreamTransactionHash.Enumerate(ShutdownEnumerator, this);
1.158 + Telemetry::Accumulate(Telemetry::SPDY_PARALLEL_STREAMS, mConcurrentHighWater);
1.159 + Telemetry::Accumulate(Telemetry::SPDY_REQUEST_PER_CONN, (mNextStreamID - 1) / 2);
1.160 + Telemetry::Accumulate(Telemetry::SPDY_SERVER_INITIATED_STREAMS,
1.161 + mServerPushedResources);
1.162 +}
1.163 +
1.164 +void
1.165 +SpdySession31::LogIO(SpdySession31 *self, SpdyStream31 *stream, const char *label,
1.166 + const char *data, uint32_t datalen)
1.167 +{
1.168 + if (!LOG4_ENABLED())
1.169 + return;
1.170 +
1.171 + LOG4(("SpdySession31::LogIO %p stream=%p id=0x%X [%s]",
1.172 + self, stream, stream ? stream->StreamID() : 0, label));
1.173 +
1.174 + // Max line is (16 * 3) + 10(prefix) + newline + null
1.175 + char linebuf[128];
1.176 + uint32_t index;
1.177 + char *line = linebuf;
1.178 +
1.179 + linebuf[127] = 0;
1.180 +
1.181 + for (index = 0; index < datalen; ++index) {
1.182 + if (!(index % 16)) {
1.183 + if (index) {
1.184 + *line = 0;
1.185 + LOG4(("%s", linebuf));
1.186 + }
1.187 + line = linebuf;
1.188 + PR_snprintf(line, 128, "%08X: ", index);
1.189 + line += 10;
1.190 + }
1.191 + PR_snprintf(line, 128 - (line - linebuf), "%02X ",
1.192 + ((unsigned char *)data)[index]);
1.193 + line += 3;
1.194 + }
1.195 + if (index) {
1.196 + *line = 0;
1.197 + LOG4(("%s", linebuf));
1.198 + }
1.199 +}
1.200 +
1.201 +bool
1.202 +SpdySession31::RoomForMoreConcurrent()
1.203 +{
1.204 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.205 +
1.206 + return (mConcurrent < mMaxConcurrent);
1.207 +}
1.208 +
1.209 +bool
1.210 +SpdySession31::RoomForMoreStreams()
1.211 +{
1.212 + if (mNextStreamID + mStreamTransactionHash.Count() * 2 > kMaxStreamID)
1.213 + return false;
1.214 +
1.215 + return !mShouldGoAway;
1.216 +}
1.217 +
1.218 +PRIntervalTime
1.219 +SpdySession31::IdleTime()
1.220 +{
1.221 + return PR_IntervalNow() - mLastDataReadEpoch;
1.222 +}
1.223 +
1.224 +uint32_t
1.225 +SpdySession31::ReadTimeoutTick(PRIntervalTime now)
1.226 +{
1.227 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.228 + MOZ_ASSERT(mNextPingID & 1, "Ping Counter Not Odd");
1.229 +
1.230 + LOG(("SpdySession31::ReadTimeoutTick %p delta since last read %ds\n",
1.231 + this, PR_IntervalToSeconds(now - mLastReadEpoch)));
1.232 +
1.233 + if (!mPingThreshold)
1.234 + return UINT32_MAX;
1.235 +
1.236 + if ((now - mLastReadEpoch) < mPingThreshold) {
1.237 + // recent activity means ping is not an issue
1.238 + if (mPingSentEpoch)
1.239 + mPingSentEpoch = 0;
1.240 +
1.241 + return PR_IntervalToSeconds(mPingThreshold) -
1.242 + PR_IntervalToSeconds(now - mLastReadEpoch);
1.243 + }
1.244 +
1.245 + if (mPingSentEpoch) {
1.246 + LOG(("SpdySession31::ReadTimeoutTick %p handle outstanding ping\n", this));
1.247 + if ((now - mPingSentEpoch) >= gHttpHandler->SpdyPingTimeout()) {
1.248 + LOG(("SpdySession31::ReadTimeoutTick %p Ping Timer Exhaustion\n",
1.249 + this));
1.250 + mPingSentEpoch = 0;
1.251 + Close(NS_ERROR_NET_TIMEOUT);
1.252 + return UINT32_MAX;
1.253 + }
1.254 + return 1; // run the tick aggressively while ping is outstanding
1.255 + }
1.256 +
1.257 + LOG(("SpdySession31::ReadTimeoutTick %p generating ping 0x%X\n",
1.258 + this, mNextPingID));
1.259 +
1.260 + if (mNextPingID == 0xffffffff) {
1.261 + LOG(("SpdySession31::ReadTimeoutTick %p cannot form ping - ids exhausted\n",
1.262 + this));
1.263 + return UINT32_MAX;
1.264 + }
1.265 +
1.266 + mPingSentEpoch = PR_IntervalNow();
1.267 + if (!mPingSentEpoch)
1.268 + mPingSentEpoch = 1; // avoid the 0 sentinel value
1.269 + GeneratePing(mNextPingID);
1.270 + mNextPingID += 2;
1.271 + ResumeRecv(); // read the ping reply
1.272 +
1.273 + // Check for orphaned push streams. This looks expensive, but generally the
1.274 + // list is empty.
1.275 + SpdyPushedStream31 *deleteMe;
1.276 + TimeStamp timestampNow;
1.277 + do {
1.278 + deleteMe = nullptr;
1.279 +
1.280 + for (uint32_t index = mPushedStreams.Length();
1.281 + index > 0 ; --index) {
1.282 + SpdyPushedStream31 *pushedStream = mPushedStreams[index - 1];
1.283 +
1.284 + if (timestampNow.IsNull())
1.285 + timestampNow = TimeStamp::Now(); // lazy initializer
1.286 +
1.287 + // if spdy finished, but not connected, and its been like that for too long..
1.288 + // cleanup the stream..
1.289 + if (pushedStream->IsOrphaned(timestampNow))
1.290 + {
1.291 + LOG3(("SpdySession31 Timeout Pushed Stream %p 0x%X\n",
1.292 + this, pushedStream->StreamID()));
1.293 + deleteMe = pushedStream;
1.294 + break; // don't CleanupStream() while iterating this vector
1.295 + }
1.296 + }
1.297 + if (deleteMe)
1.298 + CleanupStream(deleteMe, NS_ERROR_ABORT, RST_CANCEL);
1.299 +
1.300 + } while (deleteMe);
1.301 +
1.302 + if (mNextPingID == 0xffffffff) {
1.303 + LOG(("SpdySession31::ReadTimeoutTick %p "
1.304 + "ping ids exhausted marking goaway\n", this));
1.305 + mShouldGoAway = true;
1.306 + }
1.307 + return 1; // run the tick aggressively while ping is outstanding
1.308 +}
1.309 +
1.310 +uint32_t
1.311 +SpdySession31::RegisterStreamID(SpdyStream31 *stream, uint32_t aNewID)
1.312 +{
1.313 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.314 +
1.315 + MOZ_ASSERT(mNextStreamID < 0xfffffff0,
1.316 + "should have stopped admitting streams");
1.317 +
1.318 + MOZ_ASSERT(!(aNewID & 1),
1.319 + "0 for autoassign pull, otherwise explicit even push assignment");
1.320 + if (!aNewID) {
1.321 + // auto generate a new pull stream ID
1.322 + aNewID = mNextStreamID;
1.323 + MOZ_ASSERT(aNewID & 1, "pull ID must be odd.");
1.324 + mNextStreamID += 2;
1.325 + }
1.326 +
1.327 + LOG3(("SpdySession31::RegisterStreamID session=%p stream=%p id=0x%X "
1.328 + "concurrent=%d",this, stream, aNewID, mConcurrent));
1.329 +
1.330 + // We've used up plenty of ID's on this session. Start
1.331 + // moving to a new one before there is a crunch involving
1.332 + // server push streams or concurrent non-registered submits
1.333 + if (aNewID >= kMaxStreamID)
1.334 + mShouldGoAway = true;
1.335 +
1.336 + // integrity check
1.337 + if (mStreamIDHash.Get(aNewID)) {
1.338 + LOG3((" New ID already present\n"));
1.339 + MOZ_ASSERT(false, "New ID already present in mStreamIDHash");
1.340 + mShouldGoAway = true;
1.341 + return kDeadStreamID;
1.342 + }
1.343 +
1.344 + mStreamIDHash.Put(aNewID, stream);
1.345 + return aNewID;
1.346 +}
1.347 +
1.348 +bool
1.349 +SpdySession31::AddStream(nsAHttpTransaction *aHttpTransaction,
1.350 + int32_t aPriority)
1.351 +{
1.352 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.353 +
1.354 + // integrity check
1.355 + if (mStreamTransactionHash.Get(aHttpTransaction)) {
1.356 + LOG3((" New transaction already present\n"));
1.357 + MOZ_ASSERT(false, "AddStream duplicate transaction pointer");
1.358 + return false;
1.359 + }
1.360 +
1.361 + aHttpTransaction->SetConnection(this);
1.362 + SpdyStream31 *stream = new SpdyStream31(aHttpTransaction, this, aPriority);
1.363 +
1.364 + LOG3(("SpdySession31::AddStream session=%p stream=%p NextID=0x%X (tentative)",
1.365 + this, stream, mNextStreamID));
1.366 +
1.367 + mStreamTransactionHash.Put(aHttpTransaction, stream);
1.368 +
1.369 + if (RoomForMoreConcurrent()) {
1.370 + LOG3(("SpdySession31::AddStream %p stream %p activated immediately.",
1.371 + this, stream));
1.372 + ActivateStream(stream);
1.373 + }
1.374 + else {
1.375 + LOG3(("SpdySession31::AddStream %p stream %p queued.", this, stream));
1.376 + mQueuedStreams.Push(stream);
1.377 + }
1.378 +
1.379 + if (!(aHttpTransaction->Caps() & NS_HTTP_ALLOW_KEEPALIVE)) {
1.380 + LOG3(("SpdySession31::AddStream %p transaction %p forces keep-alive off.\n",
1.381 + this, aHttpTransaction));
1.382 + DontReuse();
1.383 + }
1.384 +
1.385 + return true;
1.386 +}
1.387 +
1.388 +void
1.389 +SpdySession31::ActivateStream(SpdyStream31 *stream)
1.390 +{
1.391 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.392 + MOZ_ASSERT(!stream->StreamID() || (stream->StreamID() & 1),
1.393 + "Do not activate pushed streams");
1.394 +
1.395 + ++mConcurrent;
1.396 + if (mConcurrent > mConcurrentHighWater)
1.397 + mConcurrentHighWater = mConcurrent;
1.398 + LOG3(("SpdySession31::AddStream %p activating stream %p Currently %d "
1.399 + "streams in session, high water mark is %d",
1.400 + this, stream, mConcurrent, mConcurrentHighWater));
1.401 +
1.402 + mReadyForWrite.Push(stream);
1.403 + SetWriteCallbacks();
1.404 +
1.405 + // Kick off the SYN transmit without waiting for the poll loop
1.406 + // This won't work for stream id=1 because there is no segment reader
1.407 + // yet.
1.408 + if (mSegmentReader) {
1.409 + uint32_t countRead;
1.410 + ReadSegments(nullptr, kDefaultBufferSize, &countRead);
1.411 + }
1.412 +}
1.413 +
1.414 +void
1.415 +SpdySession31::ProcessPending()
1.416 +{
1.417 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.418 +
1.419 + while (RoomForMoreConcurrent()) {
1.420 + SpdyStream31 *stream = static_cast<SpdyStream31 *>(mQueuedStreams.PopFront());
1.421 + if (!stream)
1.422 + return;
1.423 + LOG3(("SpdySession31::ProcessPending %p stream %p activated from queue.",
1.424 + this, stream));
1.425 + ActivateStream(stream);
1.426 + }
1.427 +}
1.428 +
1.429 +nsresult
1.430 +SpdySession31::NetworkRead(nsAHttpSegmentWriter *writer, char *buf,
1.431 + uint32_t count, uint32_t *countWritten)
1.432 +{
1.433 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.434 +
1.435 + if (!count) {
1.436 + *countWritten = 0;
1.437 + return NS_OK;
1.438 + }
1.439 +
1.440 + nsresult rv = writer->OnWriteSegment(buf, count, countWritten);
1.441 + if (NS_SUCCEEDED(rv) && *countWritten > 0)
1.442 + mLastReadEpoch = PR_IntervalNow();
1.443 + return rv;
1.444 +}
1.445 +
1.446 +void
1.447 +SpdySession31::SetWriteCallbacks()
1.448 +{
1.449 + if (mConnection && (GetWriteQueueSize() || mOutputQueueUsed))
1.450 + mConnection->ResumeSend();
1.451 +}
1.452 +
1.453 +void
1.454 +SpdySession31::RealignOutputQueue()
1.455 +{
1.456 + mOutputQueueUsed -= mOutputQueueSent;
1.457 + memmove(mOutputQueueBuffer.get(),
1.458 + mOutputQueueBuffer.get() + mOutputQueueSent,
1.459 + mOutputQueueUsed);
1.460 + mOutputQueueSent = 0;
1.461 +}
1.462 +
1.463 +void
1.464 +SpdySession31::FlushOutputQueue()
1.465 +{
1.466 + if (!mSegmentReader || !mOutputQueueUsed)
1.467 + return;
1.468 +
1.469 + nsresult rv;
1.470 + uint32_t countRead;
1.471 + uint32_t avail = mOutputQueueUsed - mOutputQueueSent;
1.472 +
1.473 + rv = mSegmentReader->
1.474 + OnReadSegment(mOutputQueueBuffer.get() + mOutputQueueSent, avail,
1.475 + &countRead);
1.476 + LOG3(("SpdySession31::FlushOutputQueue %p sz=%d rv=%x actual=%d",
1.477 + this, avail, rv, countRead));
1.478 +
1.479 + // Dont worry about errors on write, we will pick this up as a read error too
1.480 + if (NS_FAILED(rv))
1.481 + return;
1.482 +
1.483 + if (countRead == avail) {
1.484 + mOutputQueueUsed = 0;
1.485 + mOutputQueueSent = 0;
1.486 + return;
1.487 + }
1.488 +
1.489 + mOutputQueueSent += countRead;
1.490 +
1.491 + // If the output queue is close to filling up and we have sent out a good
1.492 + // chunk of data from the beginning then realign it.
1.493 +
1.494 + if ((mOutputQueueSent >= kQueueMinimumCleanup) &&
1.495 + ((mOutputQueueSize - mOutputQueueUsed) < kQueueTailRoom)) {
1.496 + RealignOutputQueue();
1.497 + }
1.498 +}
1.499 +
1.500 +void
1.501 +SpdySession31::DontReuse()
1.502 +{
1.503 + mShouldGoAway = true;
1.504 + if (!mStreamTransactionHash.Count())
1.505 + Close(NS_OK);
1.506 +}
1.507 +
1.508 +uint32_t
1.509 +SpdySession31::GetWriteQueueSize()
1.510 +{
1.511 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.512 +
1.513 + return mReadyForWrite.GetSize();
1.514 +}
1.515 +
1.516 +void
1.517 +SpdySession31::ChangeDownstreamState(enum stateType newState)
1.518 +{
1.519 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.520 +
1.521 + LOG3(("SpdyStream31::ChangeDownstreamState() %p from %X to %X",
1.522 + this, mDownstreamState, newState));
1.523 + mDownstreamState = newState;
1.524 +}
1.525 +
1.526 +void
1.527 +SpdySession31::ResetDownstreamState()
1.528 +{
1.529 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.530 +
1.531 + LOG3(("SpdyStream31::ResetDownstreamState() %p", this));
1.532 + ChangeDownstreamState(BUFFERING_FRAME_HEADER);
1.533 +
1.534 + if (mInputFrameDataLast && mInputFrameDataStream) {
1.535 + mInputFrameDataLast = false;
1.536 + if (!mInputFrameDataStream->RecvdFin()) {
1.537 + LOG3((" SetRecvdFin id=0x%x\n", mInputFrameDataStream->StreamID()));
1.538 + mInputFrameDataStream->SetRecvdFin(true);
1.539 + DecrementConcurrent(mInputFrameDataStream);
1.540 + }
1.541 + }
1.542 + mInputFrameBufferUsed = 0;
1.543 + mInputFrameDataStream = nullptr;
1.544 +}
1.545 +
1.546 +template<typename T> void
1.547 +SpdySession31::EnsureBuffer(nsAutoArrayPtr<T> &buf,
1.548 + uint32_t newSize,
1.549 + uint32_t preserve,
1.550 + uint32_t &objSize)
1.551 +{
1.552 + if (objSize >= newSize)
1.553 + return;
1.554 +
1.555 + // Leave a little slop on the new allocation - add 2KB to
1.556 + // what we need and then round the result up to a 4KB (page)
1.557 + // boundary.
1.558 +
1.559 + objSize = (newSize + 2048 + 4095) & ~4095;
1.560 +
1.561 + static_assert(sizeof(T) == 1, "sizeof(T) must be 1");
1.562 + nsAutoArrayPtr<T> tmp(new T[objSize]);
1.563 + memcpy(tmp, buf, preserve);
1.564 + buf = tmp;
1.565 +}
1.566 +
1.567 +// Instantiate supported templates explicitly.
1.568 +template void
1.569 +SpdySession31::EnsureBuffer(nsAutoArrayPtr<char> &buf,
1.570 + uint32_t newSize,
1.571 + uint32_t preserve,
1.572 + uint32_t &objSize);
1.573 +
1.574 +template void
1.575 +SpdySession31::EnsureBuffer(nsAutoArrayPtr<uint8_t> &buf,
1.576 + uint32_t newSize,
1.577 + uint32_t preserve,
1.578 + uint32_t &objSize);
1.579 +
1.580 +void
1.581 +SpdySession31::DecrementConcurrent(SpdyStream31 *aStream)
1.582 +{
1.583 + uint32_t id = aStream->StreamID();
1.584 +
1.585 + if (id && !(id & 0x1))
1.586 + return; // pushed streams aren't counted in concurrent limit
1.587 +
1.588 + MOZ_ASSERT(mConcurrent);
1.589 + --mConcurrent;
1.590 + LOG3(("DecrementConcurrent %p id=0x%X concurrent=%d\n",
1.591 + this, id, mConcurrent));
1.592 + ProcessPending();
1.593 +}
1.594 +
1.595 +void
1.596 +SpdySession31::zlibInit()
1.597 +{
1.598 + mDownstreamZlib.zalloc = SpdyZlibReporter::Alloc;
1.599 + mDownstreamZlib.zfree = SpdyZlibReporter::Free;
1.600 + mDownstreamZlib.opaque = Z_NULL;
1.601 +
1.602 + inflateInit(&mDownstreamZlib);
1.603 +
1.604 + mUpstreamZlib.zalloc = SpdyZlibReporter::Alloc;
1.605 + mUpstreamZlib.zfree = SpdyZlibReporter::Free;
1.606 + mUpstreamZlib.opaque = Z_NULL;
1.607 +
1.608 + // mixing carte blanche compression with tls subjects us to traffic
1.609 + // analysis attacks
1.610 + deflateInit(&mUpstreamZlib, Z_NO_COMPRESSION);
1.611 + deflateSetDictionary(&mUpstreamZlib,
1.612 + SpdyStream31::kDictionary,
1.613 + sizeof(SpdyStream31::kDictionary));
1.614 +}
1.615 +
1.616 +// Need to decompress some data in order to keep the compression
1.617 +// context correct, but we really don't care what the result is
1.618 +nsresult
1.619 +SpdySession31::UncompressAndDiscard(uint32_t offset,
1.620 + uint32_t blockLen)
1.621 +{
1.622 + char *blockStart = mInputFrameBuffer + offset;
1.623 + unsigned char trash[2048];
1.624 + mDownstreamZlib.avail_in = blockLen;
1.625 + mDownstreamZlib.next_in = reinterpret_cast<unsigned char *>(blockStart);
1.626 + bool triedDictionary = false;
1.627 +
1.628 + do {
1.629 + mDownstreamZlib.next_out = trash;
1.630 + mDownstreamZlib.avail_out = sizeof(trash);
1.631 + int zlib_rv = inflate(&mDownstreamZlib, Z_NO_FLUSH);
1.632 +
1.633 + if (zlib_rv == Z_NEED_DICT) {
1.634 + if (triedDictionary) {
1.635 + LOG3(("SpdySession31::UncompressAndDiscard %p Dictionary Error\n", this));
1.636 + return NS_ERROR_ILLEGAL_VALUE;
1.637 + }
1.638 +
1.639 + triedDictionary = true;
1.640 + inflateSetDictionary(&mDownstreamZlib, SpdyStream31::kDictionary,
1.641 + sizeof(SpdyStream31::kDictionary));
1.642 + }
1.643 +
1.644 + if (zlib_rv == Z_DATA_ERROR)
1.645 + return NS_ERROR_ILLEGAL_VALUE;
1.646 +
1.647 + if (zlib_rv == Z_MEM_ERROR)
1.648 + return NS_ERROR_FAILURE;
1.649 + }
1.650 + while (mDownstreamZlib.avail_in);
1.651 + return NS_OK;
1.652 +}
1.653 +
1.654 +void
1.655 +SpdySession31::GeneratePing(uint32_t aID)
1.656 +{
1.657 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.658 + LOG3(("SpdySession31::GeneratePing %p 0x%X\n", this, aID));
1.659 +
1.660 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + 12,
1.661 + mOutputQueueUsed, mOutputQueueSize);
1.662 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.663 + mOutputQueueUsed += 12;
1.664 +
1.665 + packet[0] = kFlag_Control;
1.666 + packet[1] = kVersion;
1.667 + packet[2] = 0;
1.668 + packet[3] = CONTROL_TYPE_PING;
1.669 + packet[4] = 0; /* flags */
1.670 + packet[5] = 0;
1.671 + packet[6] = 0;
1.672 + packet[7] = 4; /* length */
1.673 +
1.674 + aID = PR_htonl(aID);
1.675 + memcpy(packet + 8, &aID, 4);
1.676 +
1.677 + LogIO(this, nullptr, "Generate Ping", packet, 12);
1.678 + FlushOutputQueue();
1.679 +}
1.680 +
1.681 +void
1.682 +SpdySession31::GenerateRstStream(uint32_t aStatusCode, uint32_t aID)
1.683 +{
1.684 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.685 + LOG3(("SpdySession31::GenerateRst %p 0x%X %d\n", this, aID, aStatusCode));
1.686 +
1.687 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + 16,
1.688 + mOutputQueueUsed, mOutputQueueSize);
1.689 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.690 + mOutputQueueUsed += 16;
1.691 +
1.692 + packet[0] = kFlag_Control;
1.693 + packet[1] = kVersion;
1.694 + packet[2] = 0;
1.695 + packet[3] = CONTROL_TYPE_RST_STREAM;
1.696 + packet[4] = 0; /* flags */
1.697 + packet[5] = 0;
1.698 + packet[6] = 0;
1.699 + packet[7] = 8; /* length */
1.700 +
1.701 + aID = PR_htonl(aID);
1.702 + memcpy(packet + 8, &aID, 4);
1.703 + aStatusCode = PR_htonl(aStatusCode);
1.704 + memcpy(packet + 12, &aStatusCode, 4);
1.705 +
1.706 + LogIO(this, nullptr, "Generate Reset", packet, 16);
1.707 + FlushOutputQueue();
1.708 +}
1.709 +
1.710 +void
1.711 +SpdySession31::GenerateGoAway(uint32_t aStatusCode)
1.712 +{
1.713 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.714 + LOG3(("SpdySession31::GenerateGoAway %p code=%X\n", this, aStatusCode));
1.715 +
1.716 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + 16,
1.717 + mOutputQueueUsed, mOutputQueueSize);
1.718 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.719 + mOutputQueueUsed += 16;
1.720 +
1.721 + memset(packet, 0, 16);
1.722 + packet[0] = kFlag_Control;
1.723 + packet[1] = kVersion;
1.724 + packet[3] = CONTROL_TYPE_GOAWAY;
1.725 + packet[7] = 8; /* data length */
1.726 +
1.727 + // last-good-stream-id are bytes 8-11, when we accept server push this will
1.728 + // need to be set non zero
1.729 +
1.730 + // bytes 12-15 are the status code.
1.731 + aStatusCode = PR_htonl(aStatusCode);
1.732 + memcpy(packet + 12, &aStatusCode, 4);
1.733 +
1.734 + LogIO(this, nullptr, "Generate GoAway", packet, 16);
1.735 + FlushOutputQueue();
1.736 +}
1.737 +
1.738 +void
1.739 +SpdySession31::GenerateSettings()
1.740 +{
1.741 + uint32_t sessionWindowBump = ASpdySession::kInitialRwin - kDefaultRwin;
1.742 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.743 + LOG3(("SpdySession31::GenerateSettings %p\n", this));
1.744 +
1.745 +// sized for 3 settings and a session window update to follow
1.746 + static const uint32_t maxDataLen = 4 + 3 * 8 + 16;
1.747 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + 8 + maxDataLen,
1.748 + mOutputQueueUsed, mOutputQueueSize);
1.749 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.750 +
1.751 + memset(packet, 0, 8 + maxDataLen);
1.752 + packet[0] = kFlag_Control;
1.753 + packet[1] = kVersion;
1.754 + packet[3] = CONTROL_TYPE_SETTINGS;
1.755 +
1.756 + uint8_t numberOfEntries = 0;
1.757 +
1.758 + // entries need to be listed in order by ID
1.759 + // 1st entry is bytes 12 to 19
1.760 + // 2nd entry is bytes 20 to 27
1.761 + // 3rd entry is bytes 28 to 35
1.762 +
1.763 + if (!gHttpHandler->AllowPush()) {
1.764 + // announcing that we accept 0 incoming streams is done to
1.765 + // disable server push
1.766 + packet[15 + 8 * numberOfEntries] = SETTINGS_TYPE_MAX_CONCURRENT;
1.767 + // The value portion of the setting pair is already initialized to 0
1.768 + numberOfEntries++;
1.769 + }
1.770 +
1.771 + nsRefPtr<nsHttpConnectionInfo> ci;
1.772 + uint32_t cwnd = 0;
1.773 + GetConnectionInfo(getter_AddRefs(ci));
1.774 + if (ci)
1.775 + cwnd = gHttpHandler->ConnMgr()->GetSpdyCWNDSetting(ci);
1.776 + if (cwnd) {
1.777 + packet[12 + 8 * numberOfEntries] = PERSISTED_VALUE;
1.778 + packet[15 + 8 * numberOfEntries] = SETTINGS_TYPE_CWND;
1.779 + LOG(("SpdySession31::GenerateSettings %p sending CWND %u\n", this, cwnd));
1.780 + cwnd = PR_htonl(cwnd);
1.781 + memcpy(packet + 16 + 8 * numberOfEntries, &cwnd, 4);
1.782 + numberOfEntries++;
1.783 + }
1.784 +
1.785 + // Advertise the Push RWIN and on each client SYN_STREAM pipeline
1.786 + // a window update with it in order to use larger initial windows with pulled
1.787 + // streams.
1.788 + packet[15 + 8 * numberOfEntries] = SETTINGS_TYPE_INITIAL_WINDOW;
1.789 + uint32_t rwin = PR_htonl(mPushAllowance);
1.790 + memcpy(packet + 16 + 8 * numberOfEntries, &rwin, 4);
1.791 + numberOfEntries++;
1.792 +
1.793 + uint32_t dataLen = 4 + 8 * numberOfEntries;
1.794 + mOutputQueueUsed += 8 + dataLen;
1.795 + packet[7] = dataLen;
1.796 + packet[11] = numberOfEntries;
1.797 +
1.798 + LogIO(this, nullptr, "Generate Settings", packet, 8 + dataLen);
1.799 +
1.800 + if (kDefaultRwin >= ASpdySession::kInitialRwin)
1.801 + goto generateSettings_complete;
1.802 +
1.803 + // send a window update for the session (Stream 0) for something large
1.804 + sessionWindowBump = PR_htonl(sessionWindowBump);
1.805 + mLocalSessionWindow = ASpdySession::kInitialRwin;
1.806 +
1.807 + packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.808 + mOutputQueueUsed += 16;
1.809 +
1.810 + packet[0] = kFlag_Control;
1.811 + packet[1] = kVersion;
1.812 + packet[3] = CONTROL_TYPE_WINDOW_UPDATE;
1.813 + packet[7] = 8; // 8 data bytes after 8 byte header
1.814 +
1.815 + // 8 to 11 stay 0 bytes for id = 0
1.816 + memcpy(packet + 12, &sessionWindowBump, 4);
1.817 +
1.818 + LOG3(("Session Window increase at start of session %p %u\n",
1.819 + this, PR_ntohl(sessionWindowBump)));
1.820 + LogIO(this, nullptr, "Session Window Bump ", packet, 16);
1.821 +
1.822 +generateSettings_complete:
1.823 + FlushOutputQueue();
1.824 +}
1.825 +
1.826 +// perform a bunch of integrity checks on the stream.
1.827 +// returns true if passed, false (plus LOG and ABORT) if failed.
1.828 +bool
1.829 +SpdySession31::VerifyStream(SpdyStream31 *aStream, uint32_t aOptionalID = 0)
1.830 +{
1.831 + // This is annoying, but at least it is O(1)
1.832 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.833 +
1.834 +#ifndef DEBUG
1.835 + // Only do the real verification in debug builds
1.836 + return true;
1.837 +#endif
1.838 +
1.839 + if (!aStream)
1.840 + return true;
1.841 +
1.842 + uint32_t test = 0;
1.843 +
1.844 + do {
1.845 + if (aStream->StreamID() == kDeadStreamID)
1.846 + break;
1.847 +
1.848 + nsAHttpTransaction *trans = aStream->Transaction();
1.849 +
1.850 + test++;
1.851 + if (!trans)
1.852 + break;
1.853 +
1.854 + test++;
1.855 + if (mStreamTransactionHash.Get(trans) != aStream)
1.856 + break;
1.857 +
1.858 + if (aStream->StreamID()) {
1.859 + SpdyStream31 *idStream = mStreamIDHash.Get(aStream->StreamID());
1.860 +
1.861 + test++;
1.862 + if (idStream != aStream)
1.863 + break;
1.864 +
1.865 + if (aOptionalID) {
1.866 + test++;
1.867 + if (idStream->StreamID() != aOptionalID)
1.868 + break;
1.869 + }
1.870 + }
1.871 +
1.872 + // tests passed
1.873 + return true;
1.874 + } while (0);
1.875 +
1.876 + LOG(("SpdySession31 %p VerifyStream Failure %p stream->id=0x%X "
1.877 + "optionalID=0x%X trans=%p test=%d\n",
1.878 + this, aStream, aStream->StreamID(),
1.879 + aOptionalID, aStream->Transaction(), test));
1.880 +
1.881 + MOZ_ASSERT(false, "VerifyStream");
1.882 + return false;
1.883 +}
1.884 +
1.885 +void
1.886 +SpdySession31::CleanupStream(SpdyStream31 *aStream, nsresult aResult,
1.887 + rstReason aResetCode)
1.888 +{
1.889 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.890 + LOG3(("SpdySession31::CleanupStream %p %p 0x%X %X\n",
1.891 + this, aStream, aStream ? aStream->StreamID() : 0, aResult));
1.892 + if (!aStream) {
1.893 + return;
1.894 + }
1.895 +
1.896 + SpdyPushedStream31 *pushSource = nullptr;
1.897 +
1.898 + if (NS_SUCCEEDED(aResult) && aStream->DeferCleanupOnSuccess()) {
1.899 + LOG(("SpdySession31::CleanupStream 0x%X deferred\n", aStream->StreamID()));
1.900 + return;
1.901 + }
1.902 +
1.903 + if (!VerifyStream(aStream)) {
1.904 + LOG(("SpdySession31::CleanupStream failed to verify stream\n"));
1.905 + return;
1.906 + }
1.907 +
1.908 + pushSource = aStream->PushSource();
1.909 +
1.910 + if (!aStream->RecvdFin() && aStream->StreamID()) {
1.911 + LOG3(("Stream had not processed recv FIN, sending RST code %X\n",
1.912 + aResetCode));
1.913 + GenerateRstStream(aResetCode, aStream->StreamID());
1.914 + DecrementConcurrent(aStream);
1.915 + }
1.916 +
1.917 + CloseStream(aStream, aResult);
1.918 +
1.919 + // Remove the stream from the ID hash table and, if an even id, the pushed
1.920 + // table too.
1.921 + uint32_t id = aStream->StreamID();
1.922 + if (id > 0) {
1.923 + mStreamIDHash.Remove(id);
1.924 + if (!(id & 1))
1.925 + mPushedStreams.RemoveElement(aStream);
1.926 + }
1.927 +
1.928 + RemoveStreamFromQueues(aStream);
1.929 +
1.930 + // removing from the stream transaction hash will
1.931 + // delete the SpdyStream31 and drop the reference to
1.932 + // its transaction
1.933 + mStreamTransactionHash.Remove(aStream->Transaction());
1.934 +
1.935 + if (mShouldGoAway && !mStreamTransactionHash.Count())
1.936 + Close(NS_OK);
1.937 +
1.938 + if (pushSource) {
1.939 + pushSource->SetDeferCleanupOnSuccess(false);
1.940 + CleanupStream(pushSource, aResult, aResetCode);
1.941 + }
1.942 +}
1.943 +
1.944 +static void RemoveStreamFromQueue(SpdyStream31 *aStream, nsDeque &queue)
1.945 +{
1.946 + uint32_t size = queue.GetSize();
1.947 + for (uint32_t count = 0; count < size; ++count) {
1.948 + SpdyStream31 *stream = static_cast<SpdyStream31 *>(queue.PopFront());
1.949 + if (stream != aStream)
1.950 + queue.Push(stream);
1.951 + }
1.952 +}
1.953 +
1.954 +void
1.955 +SpdySession31::RemoveStreamFromQueues(SpdyStream31 *aStream)
1.956 +{
1.957 + RemoveStreamFromQueue(aStream, mReadyForWrite);
1.958 + RemoveStreamFromQueue(aStream, mQueuedStreams);
1.959 + RemoveStreamFromQueue(aStream, mReadyForRead);
1.960 +}
1.961 +
1.962 +void
1.963 +SpdySession31::CloseStream(SpdyStream31 *aStream, nsresult aResult)
1.964 +{
1.965 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.966 + LOG3(("SpdySession31::CloseStream %p %p 0x%x %X\n",
1.967 + this, aStream, aStream->StreamID(), aResult));
1.968 +
1.969 + // Check if partial frame reader
1.970 + if (aStream == mInputFrameDataStream) {
1.971 + LOG3(("Stream had active partial read frame on close"));
1.972 + ChangeDownstreamState(DISCARDING_DATA_FRAME);
1.973 + mInputFrameDataStream = nullptr;
1.974 + }
1.975 +
1.976 + RemoveStreamFromQueues(aStream);
1.977 +
1.978 + // Send the stream the close() indication
1.979 + aStream->Close(aResult);
1.980 +}
1.981 +
1.982 +nsresult
1.983 +SpdySession31::HandleSynStream(SpdySession31 *self)
1.984 +{
1.985 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_SYN_STREAM);
1.986 +
1.987 + if (self->mInputFrameDataSize < 18) {
1.988 + LOG3(("SpdySession31::HandleSynStream %p SYN_STREAM too short data=%d",
1.989 + self, self->mInputFrameDataSize));
1.990 + return NS_ERROR_ILLEGAL_VALUE;
1.991 + }
1.992 +
1.993 + uint32_t streamID =
1.994 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.995 + uint32_t associatedID =
1.996 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[3]);
1.997 + uint8_t flags = reinterpret_cast<uint8_t *>(self->mInputFrameBuffer.get())[4];
1.998 +
1.999 + LOG3(("SpdySession31::HandleSynStream %p recv SYN_STREAM (push) "
1.1000 + "for ID 0x%X associated with 0x%X.\n",
1.1001 + self, streamID, associatedID));
1.1002 +
1.1003 + if (streamID & 0x01) { // test for odd stream ID
1.1004 + LOG3(("SpdySession31::HandleSynStream %p recvd SYN_STREAM id must be even.",
1.1005 + self));
1.1006 + return NS_ERROR_ILLEGAL_VALUE;
1.1007 + }
1.1008 +
1.1009 + // confirm associated-to
1.1010 + nsresult rv = self->SetInputFrameDataStream(associatedID);
1.1011 + if (NS_FAILED(rv))
1.1012 + return rv;
1.1013 + SpdyStream31 *associatedStream = self->mInputFrameDataStream;
1.1014 +
1.1015 + ++(self->mServerPushedResources);
1.1016 +
1.1017 + // Anytime we start using the high bit of stream ID (either client or server)
1.1018 + // begin to migrate to a new session.
1.1019 + if (streamID >= kMaxStreamID)
1.1020 + self->mShouldGoAway = true;
1.1021 +
1.1022 + bool resetStream = true;
1.1023 + SpdyPushCache *cache = nullptr;
1.1024 +
1.1025 + if (!(flags & kFlag_Data_UNI)) {
1.1026 + // pushed streams require UNIDIRECTIONAL flag
1.1027 + LOG3(("SpdySession31::HandleSynStream %p ID %0x%X associated ID 0x%X failed.\n",
1.1028 + self, streamID, associatedID));
1.1029 + self->GenerateRstStream(RST_PROTOCOL_ERROR, streamID);
1.1030 +
1.1031 + } else if (!associatedID) {
1.1032 + // associated stream 0 will never find a match, but the spec requires a
1.1033 + // PROTOCOL_ERROR in this specific case
1.1034 + LOG3(("SpdySession31::HandleSynStream %p associated ID of 0 failed.\n", self));
1.1035 + self->GenerateRstStream(RST_PROTOCOL_ERROR, streamID);
1.1036 +
1.1037 + } else if (!gHttpHandler->AllowPush()) {
1.1038 + // MAX_CONCURRENT_STREAMS of 0 in settings should have disabled push,
1.1039 + // but some servers are buggy about that.. or the config could have
1.1040 + // been updated after the settings frame was sent. In both cases just
1.1041 + // reject the pushed stream as refused
1.1042 + LOG3(("SpdySession31::HandleSynStream Push Recevied when Disabled\n"));
1.1043 + self->GenerateRstStream(RST_REFUSED_STREAM, streamID);
1.1044 +
1.1045 + } else if (!associatedStream) {
1.1046 + LOG3(("SpdySession31::HandleSynStream %p lookup associated ID failed.\n", self));
1.1047 + self->GenerateRstStream(RST_INVALID_STREAM, streamID);
1.1048 +
1.1049 + } else {
1.1050 + nsILoadGroupConnectionInfo *loadGroupCI = associatedStream->LoadGroupConnectionInfo();
1.1051 + if (loadGroupCI) {
1.1052 + loadGroupCI->GetSpdyPushCache(&cache);
1.1053 + if (!cache) {
1.1054 + cache = new SpdyPushCache();
1.1055 + if (!cache || NS_FAILED(loadGroupCI->SetSpdyPushCache(cache))) {
1.1056 + delete cache;
1.1057 + cache = nullptr;
1.1058 + }
1.1059 + }
1.1060 + }
1.1061 + if (!cache) {
1.1062 + // this is unexpected, but we can handle it just be refusing the push
1.1063 + LOG3(("SpdySession31::HandleSynStream Push Recevied without loadgroup cache\n"));
1.1064 + self->GenerateRstStream(RST_REFUSED_STREAM, streamID);
1.1065 + }
1.1066 + else {
1.1067 + resetStream = false;
1.1068 + }
1.1069 + }
1.1070 +
1.1071 + if (resetStream) {
1.1072 + // Need to decompress the headers even though we aren't using them yet in
1.1073 + // order to keep the compression context consistent for other syn_reply frames
1.1074 + rv = self->UncompressAndDiscard(18, self->mInputFrameDataSize - 10);
1.1075 + if (NS_FAILED(rv)) {
1.1076 + LOG(("SpdySession31::HandleSynStream uncompress failed\n"));
1.1077 + return rv;
1.1078 + }
1.1079 + self->ResetDownstreamState();
1.1080 + return NS_OK;
1.1081 + }
1.1082 +
1.1083 + // Create the buffering transaction and push stream
1.1084 + nsRefPtr<SpdyPush31TransactionBuffer> transactionBuffer =
1.1085 + new SpdyPush31TransactionBuffer();
1.1086 + transactionBuffer->SetConnection(self);
1.1087 + SpdyPushedStream31 *pushedStream =
1.1088 + new SpdyPushedStream31(transactionBuffer, self,
1.1089 + associatedStream, streamID);
1.1090 +
1.1091 + // ownership of the pushed stream is by the transaction hash, just as it
1.1092 + // is for a client initiated stream. Errors that aren't fatal to the
1.1093 + // whole session must call cleanupStream() after this point in order
1.1094 + // to remove the stream from that hash.
1.1095 + self->mStreamTransactionHash.Put(transactionBuffer, pushedStream);
1.1096 + self->mPushedStreams.AppendElement(pushedStream);
1.1097 +
1.1098 + // The pushed stream is unidirectional so it is fully open immediately
1.1099 + pushedStream->SetFullyOpen();
1.1100 +
1.1101 + // Uncompress the response headers into a stream specific buffer, leaving them
1.1102 + // in spdy format for the time being.
1.1103 + rv = pushedStream->Uncompress(&self->mDownstreamZlib,
1.1104 + self->mInputFrameBuffer + 18,
1.1105 + self->mInputFrameDataSize - 10);
1.1106 + if (NS_FAILED(rv)) {
1.1107 + LOG(("SpdySession31::HandleSynStream uncompress failed\n"));
1.1108 + return rv;
1.1109 + }
1.1110 +
1.1111 + if (self->RegisterStreamID(pushedStream, streamID) == kDeadStreamID) {
1.1112 + LOG(("SpdySession31::HandleSynStream registerstreamid failed\n"));
1.1113 + return NS_ERROR_FAILURE;
1.1114 + }
1.1115 +
1.1116 + // Fake the request side of the pushed HTTP transaction. Sets up hash
1.1117 + // key and origin
1.1118 + uint32_t notUsed;
1.1119 + pushedStream->ReadSegments(nullptr, 1, ¬Used);
1.1120 +
1.1121 + nsAutoCString key;
1.1122 + if (!pushedStream->GetHashKey(key)) {
1.1123 + LOG(("SpdySession31::HandleSynStream one of :host :scheme :path missing from push\n"));
1.1124 + self->CleanupStream(pushedStream, NS_ERROR_FAILURE, RST_INVALID_STREAM);
1.1125 + self->ResetDownstreamState();
1.1126 + return NS_OK;
1.1127 + }
1.1128 +
1.1129 + if (!associatedStream->Origin().Equals(pushedStream->Origin())) {
1.1130 + LOG(("SpdySession31::HandleSynStream pushed stream mismatched origin\n"));
1.1131 + self->CleanupStream(pushedStream, NS_ERROR_FAILURE, RST_INVALID_STREAM);
1.1132 + self->ResetDownstreamState();
1.1133 + return NS_OK;
1.1134 + }
1.1135 +
1.1136 + if (!cache->RegisterPushedStreamSpdy31(key, pushedStream)) {
1.1137 + LOG(("SpdySession31::HandleSynStream registerPushedStream Failed\n"));
1.1138 + self->CleanupStream(pushedStream, NS_ERROR_FAILURE, RST_INVALID_STREAM);
1.1139 + self->ResetDownstreamState();
1.1140 + return NS_OK;
1.1141 + }
1.1142 +
1.1143 + self->ResetDownstreamState();
1.1144 + return NS_OK;
1.1145 +}
1.1146 +
1.1147 +nsresult
1.1148 +SpdySession31::SetInputFrameDataStream(uint32_t streamID)
1.1149 +{
1.1150 + mInputFrameDataStream = mStreamIDHash.Get(streamID);
1.1151 + if (VerifyStream(mInputFrameDataStream, streamID))
1.1152 + return NS_OK;
1.1153 +
1.1154 + LOG(("SpdySession31::SetInputFrameDataStream failed to verify 0x%X\n",
1.1155 + streamID));
1.1156 + mInputFrameDataStream = nullptr;
1.1157 + return NS_ERROR_UNEXPECTED;
1.1158 +}
1.1159 +
1.1160 +nsresult
1.1161 +SpdySession31::HandleSynReply(SpdySession31 *self)
1.1162 +{
1.1163 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_SYN_REPLY);
1.1164 +
1.1165 + if (self->mInputFrameDataSize < 4) {
1.1166 + LOG3(("SpdySession31::HandleSynReply %p SYN REPLY too short data=%d",
1.1167 + self, self->mInputFrameDataSize));
1.1168 + // A framing error is a session wide error that cannot be recovered
1.1169 + return NS_ERROR_ILLEGAL_VALUE;
1.1170 + }
1.1171 +
1.1172 + LOG3(("SpdySession31::HandleSynReply %p lookup via streamID in syn_reply.\n",
1.1173 + self));
1.1174 + uint32_t streamID =
1.1175 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1176 + nsresult rv = self->SetInputFrameDataStream(streamID);
1.1177 + if (NS_FAILED(rv))
1.1178 + return rv;
1.1179 +
1.1180 + if (!self->mInputFrameDataStream) {
1.1181 + // Cannot find stream. We can continue the SPDY session, but we need to
1.1182 + // uncompress the header block to maintain the correct compression context
1.1183 +
1.1184 + LOG3(("SpdySession31::HandleSynReply %p lookup streamID in syn_reply "
1.1185 + "0x%X failed. NextStreamID = 0x%X\n",
1.1186 + self, streamID, self->mNextStreamID));
1.1187 +
1.1188 + if (streamID >= self->mNextStreamID)
1.1189 + self->GenerateRstStream(RST_INVALID_STREAM, streamID);
1.1190 +
1.1191 + rv = self->UncompressAndDiscard(12, self->mInputFrameDataSize - 4);
1.1192 + if (NS_FAILED(rv)) {
1.1193 + LOG(("SpdySession31::HandleSynReply uncompress failed\n"));
1.1194 + // this is fatal to the session
1.1195 + return rv;
1.1196 + }
1.1197 +
1.1198 + self->ResetDownstreamState();
1.1199 + return NS_OK;
1.1200 + }
1.1201 +
1.1202 + // Uncompress the headers into a stream specific buffer, leaving them in
1.1203 + // spdy format for the time being. Make certain to do this
1.1204 + // step before any error handling that might abort the stream but not
1.1205 + // the session becuase the session compression context will become
1.1206 + // inconsistent if all of the compressed data is not processed.
1.1207 + rv = self->mInputFrameDataStream->Uncompress(&self->mDownstreamZlib,
1.1208 + self->mInputFrameBuffer + 12,
1.1209 + self->mInputFrameDataSize - 4);
1.1210 +
1.1211 + if (NS_FAILED(rv)) {
1.1212 + LOG(("SpdySession31::HandleSynReply uncompress failed\n"));
1.1213 + return rv;
1.1214 + }
1.1215 +
1.1216 + if (self->mInputFrameDataStream->GetFullyOpen()) {
1.1217 + // "If an endpoint receives multiple SYN_REPLY frames for the same active
1.1218 + // stream ID, it MUST issue a stream error (Section 2.4.2) with the error
1.1219 + // code STREAM_IN_USE."
1.1220 + //
1.1221 + // "STREAM_ALREADY_CLOSED. The endpoint received a data or SYN_REPLY
1.1222 + // frame for a stream which is half closed."
1.1223 + //
1.1224 + // If the stream is open then just RST_STREAM with STREAM_IN_USE
1.1225 + // If the stream is half closed then RST_STREAM with STREAM_ALREADY_CLOSED
1.1226 + // abort the session
1.1227 + //
1.1228 + LOG3(("SpdySession31::HandleSynReply %p dup SYN_REPLY for 0x%X"
1.1229 + " recvdfin=%d", self, self->mInputFrameDataStream->StreamID(),
1.1230 + self->mInputFrameDataStream->RecvdFin()));
1.1231 +
1.1232 + self->CleanupStream(self->mInputFrameDataStream, NS_ERROR_ALREADY_OPENED,
1.1233 + self->mInputFrameDataStream->RecvdFin() ?
1.1234 + RST_STREAM_ALREADY_CLOSED : RST_STREAM_IN_USE);
1.1235 + self->ResetDownstreamState();
1.1236 + return NS_OK;
1.1237 + }
1.1238 + self->mInputFrameDataStream->SetFullyOpen();
1.1239 +
1.1240 + self->mInputFrameDataLast = self->mInputFrameBuffer[4] & kFlag_Data_FIN;
1.1241 + self->mInputFrameDataStream->UpdateTransportReadEvents(self->mInputFrameDataSize);
1.1242 + self->mLastDataReadEpoch = self->mLastReadEpoch;
1.1243 +
1.1244 + if (self->mInputFrameBuffer[4] & ~kFlag_Data_FIN) {
1.1245 + LOG3(("SynReply %p had undefined flag set 0x%X\n", self, streamID));
1.1246 + self->CleanupStream(self->mInputFrameDataStream, NS_ERROR_ILLEGAL_VALUE,
1.1247 + RST_PROTOCOL_ERROR);
1.1248 + self->ResetDownstreamState();
1.1249 + return NS_OK;
1.1250 + }
1.1251 +
1.1252 + if (!self->mInputFrameDataLast) {
1.1253 + // don't process the headers yet as there could be more coming from HEADERS
1.1254 + // frames
1.1255 + self->ResetDownstreamState();
1.1256 + return NS_OK;
1.1257 + }
1.1258 +
1.1259 + rv = self->ResponseHeadersComplete();
1.1260 + if (rv == NS_ERROR_ILLEGAL_VALUE) {
1.1261 + LOG3(("SpdySession31::HandleSynReply %p PROTOCOL_ERROR detected 0x%X\n",
1.1262 + self, streamID));
1.1263 + self->CleanupStream(self->mInputFrameDataStream, rv, RST_PROTOCOL_ERROR);
1.1264 + self->ResetDownstreamState();
1.1265 + rv = NS_OK;
1.1266 + }
1.1267 + return rv;
1.1268 +}
1.1269 +
1.1270 +// ResponseHeadersComplete() returns NS_ERROR_ILLEGAL_VALUE when the stream
1.1271 +// should be reset with a PROTOCOL_ERROR, NS_OK when the SYN_REPLY was
1.1272 +// fine, and any other error is fatal to the session.
1.1273 +nsresult
1.1274 +SpdySession31::ResponseHeadersComplete()
1.1275 +{
1.1276 + LOG3(("SpdySession31::ResponseHeadersComplete %p for 0x%X fin=%d",
1.1277 + this, mInputFrameDataStream->StreamID(), mInputFrameDataLast));
1.1278 +
1.1279 + // The spdystream needs to see flattened http headers
1.1280 + // Uncompressed spdy format headers currently live in
1.1281 + // SpdyStream31::mDecompressBuffer - convert that to HTTP format in
1.1282 + // mFlatHTTPResponseHeaders via ConvertHeaders()
1.1283 +
1.1284 + mFlatHTTPResponseHeadersOut = 0;
1.1285 + nsresult rv = mInputFrameDataStream->ConvertHeaders(mFlatHTTPResponseHeaders);
1.1286 + if (NS_FAILED(rv))
1.1287 + return rv;
1.1288 +
1.1289 + ChangeDownstreamState(PROCESSING_COMPLETE_HEADERS);
1.1290 + return NS_OK;
1.1291 +}
1.1292 +
1.1293 +nsresult
1.1294 +SpdySession31::HandleRstStream(SpdySession31 *self)
1.1295 +{
1.1296 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_RST_STREAM);
1.1297 +
1.1298 + if (self->mInputFrameDataSize != 8) {
1.1299 + LOG3(("SpdySession31::HandleRstStream %p RST_STREAM wrong length data=%d",
1.1300 + self, self->mInputFrameDataSize));
1.1301 + return NS_ERROR_ILLEGAL_VALUE;
1.1302 + }
1.1303 +
1.1304 + uint8_t flags = reinterpret_cast<uint8_t *>(self->mInputFrameBuffer.get())[4];
1.1305 +
1.1306 + uint32_t streamID =
1.1307 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1308 +
1.1309 + self->mDownstreamRstReason =
1.1310 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[3]);
1.1311 +
1.1312 + LOG3(("SpdySession31::HandleRstStream %p RST_STREAM Reason Code %u ID %x "
1.1313 + "flags %x", self, self->mDownstreamRstReason, streamID, flags));
1.1314 +
1.1315 + if (flags != 0) {
1.1316 + LOG3(("SpdySession31::HandleRstStream %p RST_STREAM with flags is illegal",
1.1317 + self));
1.1318 + return NS_ERROR_ILLEGAL_VALUE;
1.1319 + }
1.1320 +
1.1321 + if (self->mDownstreamRstReason == RST_INVALID_STREAM ||
1.1322 + self->mDownstreamRstReason == RST_STREAM_IN_USE ||
1.1323 + self->mDownstreamRstReason == RST_FLOW_CONTROL_ERROR) {
1.1324 + // basically just ignore this
1.1325 + LOG3(("SpdySession31::HandleRstStream %p No Reset Processing Needed.\n"));
1.1326 + self->ResetDownstreamState();
1.1327 + return NS_OK;
1.1328 + }
1.1329 +
1.1330 + nsresult rv = self->SetInputFrameDataStream(streamID);
1.1331 +
1.1332 + if (!self->mInputFrameDataStream) {
1.1333 + if (NS_FAILED(rv))
1.1334 + LOG(("SpdySession31::HandleRstStream %p lookup streamID for RST Frame "
1.1335 + "0x%X failed reason = %d :: VerifyStream Failed\n", self, streamID,
1.1336 + self->mDownstreamRstReason));
1.1337 +
1.1338 + LOG3(("SpdySession31::HandleRstStream %p lookup streamID for RST Frame "
1.1339 + "0x%X failed reason = %d", self, streamID,
1.1340 + self->mDownstreamRstReason));
1.1341 + return NS_ERROR_ILLEGAL_VALUE;
1.1342 + }
1.1343 +
1.1344 + self->ChangeDownstreamState(PROCESSING_CONTROL_RST_STREAM);
1.1345 + return NS_OK;
1.1346 +}
1.1347 +
1.1348 +PLDHashOperator
1.1349 +SpdySession31::UpdateServerRwinEnumerator(nsAHttpTransaction *key,
1.1350 + nsAutoPtr<SpdyStream31> &stream,
1.1351 + void *closure)
1.1352 +{
1.1353 + int32_t delta = *(static_cast<int32_t *>(closure));
1.1354 + stream->UpdateRemoteWindow(delta);
1.1355 + return PL_DHASH_NEXT;
1.1356 +}
1.1357 +
1.1358 +nsresult
1.1359 +SpdySession31::HandleSettings(SpdySession31 *self)
1.1360 +{
1.1361 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_SETTINGS);
1.1362 +
1.1363 + if (self->mInputFrameDataSize < 4) {
1.1364 + LOG3(("SpdySession31::HandleSettings %p SETTINGS wrong length data=%d",
1.1365 + self, self->mInputFrameDataSize));
1.1366 + return NS_ERROR_ILLEGAL_VALUE;
1.1367 + }
1.1368 +
1.1369 + uint32_t numEntries =
1.1370 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1371 +
1.1372 + // Ensure frame is large enough for supplied number of entries
1.1373 + // Each entry is 8 bytes, frame data is reduced by 4 to account for
1.1374 + // the NumEntries value.
1.1375 + if ((self->mInputFrameDataSize - 4) < (numEntries * 8)) {
1.1376 + LOG3(("SpdySession31::HandleSettings %p SETTINGS wrong length data=%d",
1.1377 + self, self->mInputFrameDataSize));
1.1378 + return NS_ERROR_ILLEGAL_VALUE;
1.1379 + }
1.1380 +
1.1381 + LOG3(("SpdySession31::HandleSettings %p SETTINGS Control Frame with %d entries",
1.1382 + self, numEntries));
1.1383 +
1.1384 + for (uint32_t index = 0; index < numEntries; ++index) {
1.1385 + unsigned char *setting = reinterpret_cast<unsigned char *>
1.1386 + (self->mInputFrameBuffer.get()) + 12 + index * 8;
1.1387 +
1.1388 + uint32_t flags = setting[0];
1.1389 + uint32_t id = PR_ntohl(reinterpret_cast<uint32_t *>(setting)[0]) & 0xffffff;
1.1390 + uint32_t value = PR_ntohl(reinterpret_cast<uint32_t *>(setting)[1]);
1.1391 +
1.1392 + LOG3(("Settings ID %d, Flags %X, Value %d", id, flags, value));
1.1393 +
1.1394 + switch (id)
1.1395 + {
1.1396 + case SETTINGS_TYPE_UPLOAD_BW:
1.1397 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_UL_BW, value);
1.1398 + break;
1.1399 +
1.1400 + case SETTINGS_TYPE_DOWNLOAD_BW:
1.1401 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_DL_BW, value);
1.1402 + break;
1.1403 +
1.1404 + case SETTINGS_TYPE_RTT:
1.1405 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_RTT, value);
1.1406 + break;
1.1407 +
1.1408 + case SETTINGS_TYPE_MAX_CONCURRENT:
1.1409 + self->mMaxConcurrent = value;
1.1410 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_MAX_STREAMS, value);
1.1411 + break;
1.1412 +
1.1413 + case SETTINGS_TYPE_CWND:
1.1414 + if (flags & PERSIST_VALUE)
1.1415 + {
1.1416 + nsRefPtr<nsHttpConnectionInfo> ci;
1.1417 + self->GetConnectionInfo(getter_AddRefs(ci));
1.1418 + if (ci)
1.1419 + gHttpHandler->ConnMgr()->ReportSpdyCWNDSetting(ci, value);
1.1420 + }
1.1421 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_CWND, value);
1.1422 + break;
1.1423 +
1.1424 + case SETTINGS_TYPE_DOWNLOAD_RETRANS_RATE:
1.1425 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_RETRANS, value);
1.1426 + break;
1.1427 +
1.1428 + case SETTINGS_TYPE_INITIAL_WINDOW:
1.1429 + Telemetry::Accumulate(Telemetry::SPDY_SETTINGS_IW, value >> 10);
1.1430 + {
1.1431 + int32_t delta = value - self->mServerInitialStreamWindow;
1.1432 + self->mServerInitialStreamWindow = value;
1.1433 +
1.1434 + // do not use SETTINGS to adjust the session window.
1.1435 +
1.1436 + // we need to add the delta to all open streams (delta can be negative)
1.1437 + self->mStreamTransactionHash.Enumerate(UpdateServerRwinEnumerator,
1.1438 + &delta);
1.1439 + }
1.1440 + break;
1.1441 +
1.1442 + default:
1.1443 + break;
1.1444 + }
1.1445 +
1.1446 + }
1.1447 +
1.1448 + self->ResetDownstreamState();
1.1449 + return NS_OK;
1.1450 +}
1.1451 +
1.1452 +nsresult
1.1453 +SpdySession31::HandleNoop(SpdySession31 *self)
1.1454 +{
1.1455 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_NOOP);
1.1456 +
1.1457 + // Should not be receiving noop frames in spdy/3, so we'll just
1.1458 + // make a log and ignore it
1.1459 +
1.1460 + LOG3(("SpdySession31::HandleNoop %p NOP.", self));
1.1461 +
1.1462 + self->ResetDownstreamState();
1.1463 + return NS_OK;
1.1464 +}
1.1465 +
1.1466 +nsresult
1.1467 +SpdySession31::HandlePing(SpdySession31 *self)
1.1468 +{
1.1469 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_PING);
1.1470 +
1.1471 + if (self->mInputFrameDataSize != 4) {
1.1472 + LOG3(("SpdySession31::HandlePing %p PING had wrong amount of data %d",
1.1473 + self, self->mInputFrameDataSize));
1.1474 + return NS_ERROR_ILLEGAL_VALUE;
1.1475 + }
1.1476 +
1.1477 + uint32_t pingID =
1.1478 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1479 +
1.1480 + LOG3(("SpdySession31::HandlePing %p PING ID 0x%X.", self, pingID));
1.1481 +
1.1482 + if (pingID & 0x01) {
1.1483 + // presumably a reply to our timeout ping
1.1484 + self->mPingSentEpoch = 0;
1.1485 + }
1.1486 + else {
1.1487 + // Servers initiate even numbered pings, go ahead and echo it back
1.1488 + self->GeneratePing(pingID);
1.1489 + }
1.1490 +
1.1491 + self->ResetDownstreamState();
1.1492 + return NS_OK;
1.1493 +}
1.1494 +
1.1495 +nsresult
1.1496 +SpdySession31::HandleGoAway(SpdySession31 *self)
1.1497 +{
1.1498 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_GOAWAY);
1.1499 +
1.1500 + if (self->mInputFrameDataSize != 8) {
1.1501 + LOG3(("SpdySession31::HandleGoAway %p GOAWAY had wrong amount of data %d",
1.1502 + self, self->mInputFrameDataSize));
1.1503 + return NS_ERROR_ILLEGAL_VALUE;
1.1504 + }
1.1505 +
1.1506 + self->mShouldGoAway = true;
1.1507 + self->mGoAwayID =
1.1508 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1509 + self->mCleanShutdown = true;
1.1510 +
1.1511 + // Find streams greater than the last-good ID and mark them for deletion
1.1512 + // in the mGoAwayStreamsToRestart queue with the GoAwayEnumerator. The
1.1513 + // underlying transaction can be restarted.
1.1514 + self->mStreamTransactionHash.Enumerate(GoAwayEnumerator, self);
1.1515 +
1.1516 + // Process the streams marked for deletion and restart.
1.1517 + uint32_t size = self->mGoAwayStreamsToRestart.GetSize();
1.1518 + for (uint32_t count = 0; count < size; ++count) {
1.1519 + SpdyStream31 *stream =
1.1520 + static_cast<SpdyStream31 *>(self->mGoAwayStreamsToRestart.PopFront());
1.1521 +
1.1522 + self->CloseStream(stream, NS_ERROR_NET_RESET);
1.1523 + if (stream->HasRegisteredID())
1.1524 + self->mStreamIDHash.Remove(stream->StreamID());
1.1525 + self->mStreamTransactionHash.Remove(stream->Transaction());
1.1526 + }
1.1527 +
1.1528 + // Queued streams can also be deleted from this session and restarted
1.1529 + // in another one. (they were never sent on the network so they implicitly
1.1530 + // are not covered by the last-good id.
1.1531 + size = self->mQueuedStreams.GetSize();
1.1532 + for (uint32_t count = 0; count < size; ++count) {
1.1533 + SpdyStream31 *stream =
1.1534 + static_cast<SpdyStream31 *>(self->mQueuedStreams.PopFront());
1.1535 + self->CloseStream(stream, NS_ERROR_NET_RESET);
1.1536 + self->mStreamTransactionHash.Remove(stream->Transaction());
1.1537 + }
1.1538 +
1.1539 + LOG3(("SpdySession31::HandleGoAway %p GOAWAY Last-Good-ID 0x%X status 0x%X "
1.1540 + "live streams=%d\n", self, self->mGoAwayID,
1.1541 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[3]),
1.1542 + self->mStreamTransactionHash.Count()));
1.1543 +
1.1544 + self->ResetDownstreamState();
1.1545 + return NS_OK;
1.1546 +}
1.1547 +
1.1548 +nsresult
1.1549 +SpdySession31::HandleHeaders(SpdySession31 *self)
1.1550 +{
1.1551 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_HEADERS);
1.1552 +
1.1553 + if (self->mInputFrameDataSize < 4) {
1.1554 + LOG3(("SpdySession31::HandleHeaders %p HEADERS had wrong amount of data %d",
1.1555 + self, self->mInputFrameDataSize));
1.1556 + return NS_ERROR_ILLEGAL_VALUE;
1.1557 + }
1.1558 +
1.1559 + uint32_t streamID =
1.1560 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1561 + LOG3(("SpdySession31::HandleHeaders %p HEADERS for Stream 0x%X.\n",
1.1562 + self, streamID));
1.1563 + nsresult rv = self->SetInputFrameDataStream(streamID);
1.1564 + if (NS_FAILED(rv))
1.1565 + return rv;
1.1566 +
1.1567 + if (!self->mInputFrameDataStream) {
1.1568 + LOG3(("SpdySession31::HandleHeaders %p lookup streamID 0x%X failed.\n",
1.1569 + self, streamID));
1.1570 + if (streamID >= self->mNextStreamID)
1.1571 + self->GenerateRstStream(RST_INVALID_STREAM, streamID);
1.1572 +
1.1573 + rv = self->UncompressAndDiscard(12, self->mInputFrameDataSize - 4);
1.1574 + if (NS_FAILED(rv)) {
1.1575 + LOG(("SpdySession31::HandleHeaders uncompress failed\n"));
1.1576 + // this is fatal to the session
1.1577 + return rv;
1.1578 + }
1.1579 + self->ResetDownstreamState();
1.1580 + return NS_OK;
1.1581 + }
1.1582 +
1.1583 + // Uncompress the headers into local buffers in the SpdyStream, leaving
1.1584 + // them in spdy format for the time being. Make certain to do this
1.1585 + // step before any error handling that might abort the stream but not
1.1586 + // the session becuase the session compression context will become
1.1587 + // inconsistent if all of the compressed data is not processed.
1.1588 + rv = self->mInputFrameDataStream->Uncompress(&self->mDownstreamZlib,
1.1589 + self->mInputFrameBuffer + 12,
1.1590 + self->mInputFrameDataSize - 4);
1.1591 + if (NS_FAILED(rv)) {
1.1592 + LOG(("SpdySession31::HandleHeaders uncompress failed\n"));
1.1593 + return rv;
1.1594 + }
1.1595 +
1.1596 + self->mInputFrameDataLast = self->mInputFrameBuffer[4] & kFlag_Data_FIN;
1.1597 + self->mInputFrameDataStream->
1.1598 + UpdateTransportReadEvents(self->mInputFrameDataSize);
1.1599 + self->mLastDataReadEpoch = self->mLastReadEpoch;
1.1600 +
1.1601 + if (self->mInputFrameBuffer[4] & ~kFlag_Data_FIN) {
1.1602 + LOG3(("Headers %p had undefined flag set 0x%X\n", self, streamID));
1.1603 + self->CleanupStream(self->mInputFrameDataStream, NS_ERROR_ILLEGAL_VALUE,
1.1604 + RST_PROTOCOL_ERROR);
1.1605 + self->ResetDownstreamState();
1.1606 + return NS_OK;
1.1607 + }
1.1608 +
1.1609 + if (!self->mInputFrameDataLast) {
1.1610 + // don't process the headers yet as there could be more HEADERS frames
1.1611 + self->ResetDownstreamState();
1.1612 + return NS_OK;
1.1613 + }
1.1614 +
1.1615 + rv = self->ResponseHeadersComplete();
1.1616 + if (rv == NS_ERROR_ILLEGAL_VALUE) {
1.1617 + LOG3(("SpdySession31::HanndleHeaders %p PROTOCOL_ERROR detected 0x%X\n",
1.1618 + self, streamID));
1.1619 + self->CleanupStream(self->mInputFrameDataStream, rv, RST_PROTOCOL_ERROR);
1.1620 + self->ResetDownstreamState();
1.1621 + rv = NS_OK;
1.1622 + }
1.1623 + return rv;
1.1624 +}
1.1625 +
1.1626 +PLDHashOperator
1.1627 +SpdySession31::RestartBlockedOnRwinEnumerator(nsAHttpTransaction *key,
1.1628 + nsAutoPtr<SpdyStream31> &stream,
1.1629 + void *closure)
1.1630 +{
1.1631 + SpdySession31 *self = static_cast<SpdySession31 *>(closure);
1.1632 + MOZ_ASSERT(self->mRemoteSessionWindow > 0);
1.1633 +
1.1634 + if (!stream->BlockedOnRwin() || stream->RemoteWindow() <= 0)
1.1635 + return PL_DHASH_NEXT;
1.1636 +
1.1637 + self->mReadyForWrite.Push(stream);
1.1638 + self->SetWriteCallbacks();
1.1639 + return PL_DHASH_NEXT;
1.1640 +}
1.1641 +
1.1642 +nsresult
1.1643 +SpdySession31::HandleWindowUpdate(SpdySession31 *self)
1.1644 +{
1.1645 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_WINDOW_UPDATE);
1.1646 +
1.1647 + if (self->mInputFrameDataSize < 8) {
1.1648 + LOG3(("SpdySession31::HandleWindowUpdate %p Window Update wrong length %d\n",
1.1649 + self, self->mInputFrameDataSize));
1.1650 + return NS_ERROR_ILLEGAL_VALUE;
1.1651 + }
1.1652 +
1.1653 + uint32_t delta =
1.1654 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[3]);
1.1655 + delta &= 0x7fffffff;
1.1656 + uint32_t streamID =
1.1657 + PR_ntohl(reinterpret_cast<uint32_t *>(self->mInputFrameBuffer.get())[2]);
1.1658 + streamID &= 0x7fffffff;
1.1659 +
1.1660 + LOG3(("SpdySession31::HandleWindowUpdate %p len=%d for Stream 0x%X.\n",
1.1661 + self, delta, streamID));
1.1662 +
1.1663 + // ID of 0 is a session window update
1.1664 + if (streamID) {
1.1665 + nsresult rv = self->SetInputFrameDataStream(streamID);
1.1666 + if (NS_FAILED(rv))
1.1667 + return rv;
1.1668 +
1.1669 + if (!self->mInputFrameDataStream) {
1.1670 + LOG3(("SpdySession31::HandleWindowUpdate %p lookup streamID 0x%X failed.\n",
1.1671 + self, streamID));
1.1672 + if (streamID >= self->mNextStreamID)
1.1673 + self->GenerateRstStream(RST_INVALID_STREAM, streamID);
1.1674 + self->ResetDownstreamState();
1.1675 + return NS_OK;
1.1676 + }
1.1677 +
1.1678 + self->mInputFrameDataStream->UpdateRemoteWindow(delta);
1.1679 + } else {
1.1680 + int64_t oldRemoteWindow = self->mRemoteSessionWindow;
1.1681 + self->mRemoteSessionWindow += delta;
1.1682 + if ((oldRemoteWindow <= 0) && (self->mRemoteSessionWindow > 0)) {
1.1683 + LOG3(("SpdySession31::HandleWindowUpdate %p restart session window\n",
1.1684 + self));
1.1685 + self->mStreamTransactionHash.Enumerate(RestartBlockedOnRwinEnumerator, self);
1.1686 + }
1.1687 + }
1.1688 +
1.1689 + self->ResetDownstreamState();
1.1690 + return NS_OK;
1.1691 +}
1.1692 +
1.1693 +nsresult
1.1694 +SpdySession31::HandleCredential(SpdySession31 *self)
1.1695 +{
1.1696 + MOZ_ASSERT(self->mFrameControlType == CONTROL_TYPE_CREDENTIAL);
1.1697 +
1.1698 + // These aren't used yet. Just ignore the frame.
1.1699 +
1.1700 + LOG3(("SpdySession31::HandleCredential %p NOP.", self));
1.1701 +
1.1702 + self->ResetDownstreamState();
1.1703 + return NS_OK;
1.1704 +}
1.1705 +
1.1706 +//-----------------------------------------------------------------------------
1.1707 +// nsAHttpTransaction. It is expected that nsHttpConnection is the caller
1.1708 +// of these methods
1.1709 +//-----------------------------------------------------------------------------
1.1710 +
1.1711 +void
1.1712 +SpdySession31::OnTransportStatus(nsITransport* aTransport,
1.1713 + nsresult aStatus,
1.1714 + uint64_t aProgress)
1.1715 +{
1.1716 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.1717 +
1.1718 + switch (aStatus) {
1.1719 + // These should appear only once, deliver to the first
1.1720 + // transaction on the session.
1.1721 + case NS_NET_STATUS_RESOLVING_HOST:
1.1722 + case NS_NET_STATUS_RESOLVED_HOST:
1.1723 + case NS_NET_STATUS_CONNECTING_TO:
1.1724 + case NS_NET_STATUS_CONNECTED_TO:
1.1725 + {
1.1726 + SpdyStream31 *target = mStreamIDHash.Get(1);
1.1727 + nsAHttpTransaction *transaction = target ? target->Transaction() : nullptr;
1.1728 + if (transaction)
1.1729 + transaction->OnTransportStatus(aTransport, aStatus, aProgress);
1.1730 + break;
1.1731 + }
1.1732 +
1.1733 + default:
1.1734 + // The other transport events are ignored here because there is no good
1.1735 + // way to map them to the right transaction in spdy. Instead, the events
1.1736 + // are generated again from the spdy code and passed directly to the
1.1737 + // correct transaction.
1.1738 +
1.1739 + // NS_NET_STATUS_SENDING_TO:
1.1740 + // This is generated by the socket transport when (part) of
1.1741 + // a transaction is written out
1.1742 + //
1.1743 + // There is no good way to map it to the right transaction in spdy,
1.1744 + // so it is ignored here and generated separately when the SYN_STREAM
1.1745 + // is sent from SpdyStream31::TransmitFrame
1.1746 +
1.1747 + // NS_NET_STATUS_WAITING_FOR:
1.1748 + // Created by nsHttpConnection when the request has been totally sent.
1.1749 + // There is no good way to map it to the right transaction in spdy,
1.1750 + // so it is ignored here and generated separately when the same
1.1751 + // condition is complete in SpdyStream31 when there is no more
1.1752 + // request body left to be transmitted.
1.1753 +
1.1754 + // NS_NET_STATUS_RECEIVING_FROM
1.1755 + // Generated in spdysession whenever we read a data frame or a syn_reply
1.1756 + // that can be attributed to a particular stream/transaction
1.1757 +
1.1758 + break;
1.1759 + }
1.1760 +}
1.1761 +
1.1762 +// ReadSegments() is used to write data to the network. Generally, HTTP
1.1763 +// request data is pulled from the approriate transaction and
1.1764 +// converted to SPDY data. Sometimes control data like window-update are
1.1765 +// generated instead.
1.1766 +
1.1767 +nsresult
1.1768 +SpdySession31::ReadSegments(nsAHttpSegmentReader *reader,
1.1769 + uint32_t count,
1.1770 + uint32_t *countRead)
1.1771 +{
1.1772 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.1773 +
1.1774 + MOZ_ASSERT(!mSegmentReader || !reader || (mSegmentReader == reader),
1.1775 + "Inconsistent Write Function Callback");
1.1776 +
1.1777 + if (reader)
1.1778 + mSegmentReader = reader;
1.1779 +
1.1780 + nsresult rv;
1.1781 + *countRead = 0;
1.1782 +
1.1783 + LOG3(("SpdySession31::ReadSegments %p", this));
1.1784 +
1.1785 + SpdyStream31 *stream = static_cast<SpdyStream31 *>(mReadyForWrite.PopFront());
1.1786 + if (!stream) {
1.1787 + LOG3(("SpdySession31 %p could not identify a stream to write; suspending.",
1.1788 + this));
1.1789 + FlushOutputQueue();
1.1790 + SetWriteCallbacks();
1.1791 + return NS_BASE_STREAM_WOULD_BLOCK;
1.1792 + }
1.1793 +
1.1794 + LOG3(("SpdySession31 %p will write from SpdyStream31 %p 0x%X "
1.1795 + "block-input=%d block-output=%d\n", this, stream, stream->StreamID(),
1.1796 + stream->RequestBlockedOnRead(), stream->BlockedOnRwin()));
1.1797 +
1.1798 + rv = stream->ReadSegments(this, count, countRead);
1.1799 +
1.1800 + // Not every permutation of stream->ReadSegents produces data (and therefore
1.1801 + // tries to flush the output queue) - SENDING_FIN_STREAM can be an example
1.1802 + // of that. But we might still have old data buffered that would be good
1.1803 + // to flush.
1.1804 + FlushOutputQueue();
1.1805 +
1.1806 + // Allow new server reads - that might be data or control information
1.1807 + // (e.g. window updates or http replies) that are responses to these writes
1.1808 + ResumeRecv();
1.1809 +
1.1810 + if (stream->RequestBlockedOnRead()) {
1.1811 +
1.1812 + // We are blocked waiting for input - either more http headers or
1.1813 + // any request body data. When more data from the request stream
1.1814 + // becomes available the httptransaction will call conn->ResumeSend().
1.1815 +
1.1816 + LOG3(("SpdySession31::ReadSegments %p dealing with block on read", this));
1.1817 +
1.1818 + // call readsegments again if there are other streams ready
1.1819 + // to run in this session
1.1820 + if (GetWriteQueueSize())
1.1821 + rv = NS_OK;
1.1822 + else
1.1823 + rv = NS_BASE_STREAM_WOULD_BLOCK;
1.1824 + SetWriteCallbacks();
1.1825 + return rv;
1.1826 + }
1.1827 +
1.1828 + if (NS_FAILED(rv)) {
1.1829 + LOG3(("SpdySession31::ReadSegments %p returning FAIL code %X",
1.1830 + this, rv));
1.1831 + if (rv != NS_BASE_STREAM_WOULD_BLOCK)
1.1832 + CleanupStream(stream, rv, RST_CANCEL);
1.1833 + return rv;
1.1834 + }
1.1835 +
1.1836 + if (*countRead > 0) {
1.1837 + LOG3(("SpdySession31::ReadSegments %p stream=%p countread=%d",
1.1838 + this, stream, *countRead));
1.1839 + mReadyForWrite.Push(stream);
1.1840 + SetWriteCallbacks();
1.1841 + return rv;
1.1842 + }
1.1843 +
1.1844 + if (stream->BlockedOnRwin()) {
1.1845 + LOG3(("SpdySession31 %p will stream %p 0x%X suspended for flow control\n",
1.1846 + this, stream, stream->StreamID()));
1.1847 + return NS_BASE_STREAM_WOULD_BLOCK;
1.1848 + }
1.1849 +
1.1850 + LOG3(("SpdySession31::ReadSegments %p stream=%p stream send complete",
1.1851 + this, stream));
1.1852 +
1.1853 + // call readsegments again if there are other streams ready
1.1854 + // to go in this session
1.1855 + SetWriteCallbacks();
1.1856 +
1.1857 + return rv;
1.1858 +}
1.1859 +
1.1860 +// WriteSegments() is used to read data off the socket. Generally this is
1.1861 +// just the SPDY frame header and from there the appropriate SPDYStream
1.1862 +// is identified from the Stream-ID. The http transaction associated with
1.1863 +// that read then pulls in the data directly, which it will feed to
1.1864 +// OnWriteSegment(). That function will gateway it into http and feed
1.1865 +// it to the appropriate transaction.
1.1866 +
1.1867 +// we call writer->OnWriteSegment via NetworkRead() to get a spdy header..
1.1868 +// and decide if it is data or control.. if it is control, just deal with it.
1.1869 +// if it is data, identify the spdy stream
1.1870 +// call stream->WriteSegments which can call this::OnWriteSegment to get the
1.1871 +// data. It always gets full frames if they are part of the stream
1.1872 +
1.1873 +nsresult
1.1874 +SpdySession31::WriteSegments(nsAHttpSegmentWriter *writer,
1.1875 + uint32_t count,
1.1876 + uint32_t *countWritten)
1.1877 +{
1.1878 + typedef nsresult (*Control_FX) (SpdySession31 *self);
1.1879 + static const Control_FX sControlFunctions[] =
1.1880 + {
1.1881 + nullptr,
1.1882 + SpdySession31::HandleSynStream,
1.1883 + SpdySession31::HandleSynReply,
1.1884 + SpdySession31::HandleRstStream,
1.1885 + SpdySession31::HandleSettings,
1.1886 + SpdySession31::HandleNoop,
1.1887 + SpdySession31::HandlePing,
1.1888 + SpdySession31::HandleGoAway,
1.1889 + SpdySession31::HandleHeaders,
1.1890 + SpdySession31::HandleWindowUpdate,
1.1891 + SpdySession31::HandleCredential
1.1892 + };
1.1893 +
1.1894 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.1895 +
1.1896 + nsresult rv;
1.1897 + *countWritten = 0;
1.1898 +
1.1899 + if (mClosed)
1.1900 + return NS_ERROR_FAILURE;
1.1901 +
1.1902 + SetWriteCallbacks();
1.1903 +
1.1904 + // If there are http transactions attached to a push stream with filled buffers
1.1905 + // trigger that data pump here. This only reads from buffers (not the network)
1.1906 + // so mDownstreamState doesn't matter.
1.1907 + SpdyStream31 *pushConnectedStream =
1.1908 + static_cast<SpdyStream31 *>(mReadyForRead.PopFront());
1.1909 + if (pushConnectedStream) {
1.1910 + LOG3(("SpdySession31::WriteSegments %p processing pushed stream 0x%X\n",
1.1911 + this, pushConnectedStream->StreamID()));
1.1912 + mSegmentWriter = writer;
1.1913 + rv = pushConnectedStream->WriteSegments(this, count, countWritten);
1.1914 + mSegmentWriter = nullptr;
1.1915 +
1.1916 + // The pipe in nsHttpTransaction rewrites CLOSED error codes into OK
1.1917 + // so we need this check to determine the truth.
1.1918 + if (NS_SUCCEEDED(rv) && !*countWritten &&
1.1919 + pushConnectedStream->PushSource() &&
1.1920 + pushConnectedStream->PushSource()->GetPushComplete()) {
1.1921 + rv = NS_BASE_STREAM_CLOSED;
1.1922 + }
1.1923 +
1.1924 + if (rv == NS_BASE_STREAM_CLOSED) {
1.1925 + CleanupStream(pushConnectedStream, NS_OK, RST_CANCEL);
1.1926 + rv = NS_OK;
1.1927 + }
1.1928 +
1.1929 + // if we return OK to nsHttpConnection it will use mSocketInCondition
1.1930 + // to determine whether to schedule more reads, incorrectly
1.1931 + // assuming that nsHttpConnection::OnSocketWrite() was called.
1.1932 + if (NS_SUCCEEDED(rv) || rv == NS_BASE_STREAM_WOULD_BLOCK) {
1.1933 + rv = NS_BASE_STREAM_WOULD_BLOCK;
1.1934 + ResumeRecv();
1.1935 + }
1.1936 +
1.1937 + return rv;
1.1938 + }
1.1939 +
1.1940 + // We buffer all control frames and act on them in this layer.
1.1941 + // We buffer the first 8 bytes of data frames (the header) but
1.1942 + // the actual data is passed through unprocessed.
1.1943 +
1.1944 + if (mDownstreamState == BUFFERING_FRAME_HEADER) {
1.1945 + // The first 8 bytes of every frame is header information that
1.1946 + // we are going to want to strip before passing to http. That is
1.1947 + // true of both control and data packets.
1.1948 +
1.1949 + MOZ_ASSERT(mInputFrameBufferUsed < 8,
1.1950 + "Frame Buffer Used Too Large for State");
1.1951 +
1.1952 + rv = NetworkRead(writer, mInputFrameBuffer + mInputFrameBufferUsed,
1.1953 + 8 - mInputFrameBufferUsed, countWritten);
1.1954 +
1.1955 + if (NS_FAILED(rv)) {
1.1956 + LOG3(("SpdySession31 %p buffering frame header read failure %x\n",
1.1957 + this, rv));
1.1958 + // maybe just blocked reading from network
1.1959 + if (rv == NS_BASE_STREAM_WOULD_BLOCK)
1.1960 + rv = NS_OK;
1.1961 + return rv;
1.1962 + }
1.1963 +
1.1964 + LogIO(this, nullptr, "Reading Frame Header",
1.1965 + mInputFrameBuffer + mInputFrameBufferUsed, *countWritten);
1.1966 +
1.1967 + mInputFrameBufferUsed += *countWritten;
1.1968 +
1.1969 + if (mInputFrameBufferUsed < 8)
1.1970 + {
1.1971 + LOG3(("SpdySession31::WriteSegments %p "
1.1972 + "BUFFERING FRAME HEADER incomplete size=%d",
1.1973 + this, mInputFrameBufferUsed));
1.1974 + return rv;
1.1975 + }
1.1976 +
1.1977 + // For both control and data frames the second 32 bit word of the header
1.1978 + // is 8-flags, 24-length. (network byte order)
1.1979 + mInputFrameDataSize =
1.1980 + PR_ntohl(reinterpret_cast<uint32_t *>(mInputFrameBuffer.get())[1]);
1.1981 + mInputFrameDataSize &= 0x00ffffff;
1.1982 + mInputFrameDataRead = 0;
1.1983 +
1.1984 + if (mInputFrameBuffer[0] & kFlag_Control) {
1.1985 + EnsureBuffer(mInputFrameBuffer, mInputFrameDataSize + 8, 8,
1.1986 + mInputFrameBufferSize);
1.1987 + ChangeDownstreamState(BUFFERING_CONTROL_FRAME);
1.1988 +
1.1989 + // The first 32 bit word of the header is
1.1990 + // 1 ctrl - 15 version - 16 type
1.1991 + uint16_t version =
1.1992 + PR_ntohs(reinterpret_cast<uint16_t *>(mInputFrameBuffer.get())[0]);
1.1993 + version &= 0x7fff;
1.1994 +
1.1995 + mFrameControlType =
1.1996 + PR_ntohs(reinterpret_cast<uint16_t *>(mInputFrameBuffer.get())[1]);
1.1997 +
1.1998 + LOG3(("SpdySession31::WriteSegments %p - Control Frame Identified "
1.1999 + "type %d version %d data len %d",
1.2000 + this, mFrameControlType, version, mInputFrameDataSize));
1.2001 +
1.2002 + if (mFrameControlType >= CONTROL_TYPE_LAST ||
1.2003 + mFrameControlType <= CONTROL_TYPE_FIRST)
1.2004 + return NS_ERROR_ILLEGAL_VALUE;
1.2005 +
1.2006 + if (version != kVersion)
1.2007 + return NS_ERROR_ILLEGAL_VALUE;
1.2008 + }
1.2009 + else {
1.2010 + ChangeDownstreamState(PROCESSING_DATA_FRAME);
1.2011 +
1.2012 + Telemetry::Accumulate(Telemetry::SPDY_CHUNK_RECVD,
1.2013 + mInputFrameDataSize >> 10);
1.2014 + mLastDataReadEpoch = mLastReadEpoch;
1.2015 +
1.2016 + uint32_t streamID =
1.2017 + PR_ntohl(reinterpret_cast<uint32_t *>(mInputFrameBuffer.get())[0]);
1.2018 + rv = SetInputFrameDataStream(streamID);
1.2019 + if (NS_FAILED(rv)) {
1.2020 + LOG(("SpdySession31::WriteSegments %p lookup streamID 0x%X failed. "
1.2021 + "probably due to verification.\n", this, streamID));
1.2022 + return rv;
1.2023 + }
1.2024 + if (!mInputFrameDataStream) {
1.2025 + LOG3(("SpdySession31::WriteSegments %p lookup streamID 0x%X failed. "
1.2026 + "Next = 0x%X", this, streamID, mNextStreamID));
1.2027 + if (streamID >= mNextStreamID)
1.2028 + GenerateRstStream(RST_INVALID_STREAM, streamID);
1.2029 + ChangeDownstreamState(DISCARDING_DATA_FRAME);
1.2030 + }
1.2031 + else if (mInputFrameDataStream->RecvdFin()) {
1.2032 + LOG3(("SpdySession31::WriteSegments %p streamID 0x%X "
1.2033 + "Data arrived for already server closed stream.\n",
1.2034 + this, streamID));
1.2035 + GenerateRstStream(RST_STREAM_ALREADY_CLOSED, streamID);
1.2036 + ChangeDownstreamState(DISCARDING_DATA_FRAME);
1.2037 + }
1.2038 + else if (!mInputFrameDataStream->RecvdData()) {
1.2039 + LOG3(("SpdySession31 %p First Data Frame Flushes Headers stream 0x%X\n",
1.2040 + this, streamID));
1.2041 +
1.2042 + mInputFrameDataStream->SetRecvdData(true);
1.2043 + rv = ResponseHeadersComplete();
1.2044 + if (rv == NS_ERROR_ILLEGAL_VALUE) {
1.2045 + LOG3(("SpdySession31 %p PROTOCOL_ERROR detected 0x%X\n",
1.2046 + this, streamID));
1.2047 + CleanupStream(mInputFrameDataStream, rv, RST_PROTOCOL_ERROR);
1.2048 + ChangeDownstreamState(DISCARDING_DATA_FRAME);
1.2049 + }
1.2050 + else {
1.2051 + mDataPending = true;
1.2052 + }
1.2053 + }
1.2054 +
1.2055 + mInputFrameDataLast = (mInputFrameBuffer[4] & kFlag_Data_FIN);
1.2056 + LOG3(("Start Processing Data Frame. "
1.2057 + "Session=%p Stream ID 0x%X Stream Ptr %p Fin=%d Len=%d",
1.2058 + this, streamID, mInputFrameDataStream, mInputFrameDataLast,
1.2059 + mInputFrameDataSize));
1.2060 + UpdateLocalRwin(mInputFrameDataStream, mInputFrameDataSize);
1.2061 + }
1.2062 + }
1.2063 +
1.2064 + if (mDownstreamState == PROCESSING_CONTROL_RST_STREAM) {
1.2065 + if (mDownstreamRstReason == RST_REFUSED_STREAM)
1.2066 + rv = NS_ERROR_NET_RESET; //we can retry this 100% safely
1.2067 + else if (mDownstreamRstReason == RST_CANCEL ||
1.2068 + mDownstreamRstReason == RST_PROTOCOL_ERROR ||
1.2069 + mDownstreamRstReason == RST_INTERNAL_ERROR ||
1.2070 + mDownstreamRstReason == RST_UNSUPPORTED_VERSION)
1.2071 + rv = NS_ERROR_NET_INTERRUPT;
1.2072 + else if (mDownstreamRstReason == RST_FRAME_TOO_LARGE)
1.2073 + rv = NS_ERROR_FILE_TOO_BIG;
1.2074 + else
1.2075 + rv = NS_ERROR_ILLEGAL_VALUE;
1.2076 +
1.2077 + if (mDownstreamRstReason != RST_REFUSED_STREAM &&
1.2078 + mDownstreamRstReason != RST_CANCEL)
1.2079 + mShouldGoAway = true;
1.2080 +
1.2081 + // mInputFrameDataStream is reset by ChangeDownstreamState
1.2082 + SpdyStream31 *stream = mInputFrameDataStream;
1.2083 + ResetDownstreamState();
1.2084 + LOG3(("SpdySession31::WriteSegments cleanup stream on recv of rst "
1.2085 + "session=%p stream=%p 0x%X\n", this, stream,
1.2086 + stream ? stream->StreamID() : 0));
1.2087 + CleanupStream(stream, rv, RST_CANCEL);
1.2088 + return NS_OK;
1.2089 + }
1.2090 +
1.2091 + if (mDownstreamState == PROCESSING_DATA_FRAME ||
1.2092 + mDownstreamState == PROCESSING_COMPLETE_HEADERS) {
1.2093 +
1.2094 + // The cleanup stream should only be set while stream->WriteSegments is
1.2095 + // on the stack and then cleaned up in this code block afterwards.
1.2096 + MOZ_ASSERT(!mNeedsCleanup, "cleanup stream set unexpectedly");
1.2097 + mNeedsCleanup = nullptr; /* just in case */
1.2098 +
1.2099 + mSegmentWriter = writer;
1.2100 + rv = mInputFrameDataStream->WriteSegments(this, count, countWritten);
1.2101 + mSegmentWriter = nullptr;
1.2102 +
1.2103 + mLastDataReadEpoch = mLastReadEpoch;
1.2104 +
1.2105 + if (SoftStreamError(rv)) {
1.2106 + // This will happen when the transaction figures out it is EOF, generally
1.2107 + // due to a content-length match being made. Return OK from this function
1.2108 + // otherwise the whole session would be torn down.
1.2109 + SpdyStream31 *stream = mInputFrameDataStream;
1.2110 +
1.2111 + // if we were doing PROCESSING_COMPLETE_HEADERS need to pop the state
1.2112 + // back to PROCESSING_DATA_FRAME where we came from
1.2113 + mDownstreamState = PROCESSING_DATA_FRAME;
1.2114 +
1.2115 + if (mInputFrameDataRead == mInputFrameDataSize)
1.2116 + ResetDownstreamState();
1.2117 + LOG3(("SpdySession31::WriteSegments session=%p stream=%p 0x%X "
1.2118 + "needscleanup=%p. cleanup stream based on "
1.2119 + "stream->writeSegments returning code %X\n",
1.2120 + this, stream, stream ? stream->StreamID() : 0,
1.2121 + mNeedsCleanup, rv));
1.2122 + CleanupStream(stream, NS_OK, RST_CANCEL);
1.2123 + MOZ_ASSERT(!mNeedsCleanup, "double cleanup out of data frame");
1.2124 + mNeedsCleanup = nullptr; /* just in case */
1.2125 + return NS_OK;
1.2126 + }
1.2127 +
1.2128 + if (mNeedsCleanup) {
1.2129 + LOG3(("SpdySession31::WriteSegments session=%p stream=%p 0x%X "
1.2130 + "cleanup stream based on mNeedsCleanup.\n",
1.2131 + this, mNeedsCleanup, mNeedsCleanup ? mNeedsCleanup->StreamID() : 0));
1.2132 + CleanupStream(mNeedsCleanup, NS_OK, RST_CANCEL);
1.2133 + mNeedsCleanup = nullptr;
1.2134 + }
1.2135 +
1.2136 + if (NS_FAILED(rv)) {
1.2137 + LOG3(("SpdySession31 %p data frame read failure %x\n", this, rv));
1.2138 + // maybe just blocked reading from network
1.2139 + if (rv == NS_BASE_STREAM_WOULD_BLOCK)
1.2140 + rv = NS_OK;
1.2141 + }
1.2142 +
1.2143 + return rv;
1.2144 + }
1.2145 +
1.2146 + if (mDownstreamState == DISCARDING_DATA_FRAME) {
1.2147 + char trash[4096];
1.2148 + uint32_t count = std::min(4096U, mInputFrameDataSize - mInputFrameDataRead);
1.2149 +
1.2150 + if (!count) {
1.2151 + ResetDownstreamState();
1.2152 + ResumeRecv();
1.2153 + return NS_BASE_STREAM_WOULD_BLOCK;
1.2154 + }
1.2155 +
1.2156 + rv = NetworkRead(writer, trash, count, countWritten);
1.2157 +
1.2158 + if (NS_FAILED(rv)) {
1.2159 + LOG3(("SpdySession31 %p discard frame read failure %x\n", this, rv));
1.2160 + // maybe just blocked reading from network
1.2161 + if (rv == NS_BASE_STREAM_WOULD_BLOCK)
1.2162 + rv = NS_OK;
1.2163 + return rv;
1.2164 + }
1.2165 +
1.2166 + LogIO(this, nullptr, "Discarding Frame", trash, *countWritten);
1.2167 +
1.2168 + mInputFrameDataRead += *countWritten;
1.2169 +
1.2170 + if (mInputFrameDataRead == mInputFrameDataSize)
1.2171 + ResetDownstreamState();
1.2172 + return rv;
1.2173 + }
1.2174 +
1.2175 + MOZ_ASSERT(mDownstreamState == BUFFERING_CONTROL_FRAME);
1.2176 + if (mDownstreamState != BUFFERING_CONTROL_FRAME) {
1.2177 + // this cannot happen
1.2178 + return NS_ERROR_UNEXPECTED;
1.2179 + }
1.2180 +
1.2181 + MOZ_ASSERT(mInputFrameBufferUsed == 8,
1.2182 + "Frame Buffer Header Not Present");
1.2183 +
1.2184 + rv = NetworkRead(writer, mInputFrameBuffer + 8 + mInputFrameDataRead,
1.2185 + mInputFrameDataSize - mInputFrameDataRead, countWritten);
1.2186 +
1.2187 + if (NS_FAILED(rv)) {
1.2188 + LOG3(("SpdySession31 %p buffering control frame read failure %x\n",
1.2189 + this, rv));
1.2190 + // maybe just blocked reading from network
1.2191 + if (rv == NS_BASE_STREAM_WOULD_BLOCK)
1.2192 + rv = NS_OK;
1.2193 + return rv;
1.2194 + }
1.2195 +
1.2196 + LogIO(this, nullptr, "Reading Control Frame",
1.2197 + mInputFrameBuffer + 8 + mInputFrameDataRead, *countWritten);
1.2198 +
1.2199 + mInputFrameDataRead += *countWritten;
1.2200 +
1.2201 + if (mInputFrameDataRead != mInputFrameDataSize)
1.2202 + return NS_OK;
1.2203 +
1.2204 + // This check is actually redundant, the control type was previously
1.2205 + // checked to make sure it was in range, but we will check it again
1.2206 + // at time of use to make sure a regression doesn't creep in.
1.2207 + if (mFrameControlType >= CONTROL_TYPE_LAST ||
1.2208 + mFrameControlType <= CONTROL_TYPE_FIRST)
1.2209 + {
1.2210 + MOZ_ASSERT(false, "control type out of range");
1.2211 + return NS_ERROR_ILLEGAL_VALUE;
1.2212 + }
1.2213 + rv = sControlFunctions[mFrameControlType](this);
1.2214 +
1.2215 + MOZ_ASSERT(NS_FAILED(rv) ||
1.2216 + mDownstreamState != BUFFERING_CONTROL_FRAME,
1.2217 + "Control Handler returned OK but did not change state");
1.2218 +
1.2219 + if (mShouldGoAway && !mStreamTransactionHash.Count())
1.2220 + Close(NS_OK);
1.2221 + return rv;
1.2222 +}
1.2223 +
1.2224 +void
1.2225 +SpdySession31::UpdateLocalStreamWindow(SpdyStream31 *stream,
1.2226 + uint32_t bytes)
1.2227 +{
1.2228 + if (!stream) // this is ok - it means there was a data frame for a rst stream
1.2229 + return;
1.2230 +
1.2231 + stream->DecrementLocalWindow(bytes);
1.2232 +
1.2233 + // If this data packet was not for a valid or live stream then there
1.2234 + // is no reason to mess with the flow control
1.2235 + if (stream->RecvdFin())
1.2236 + return;
1.2237 +
1.2238 + // Don't necessarily ack every data packet. Only do it
1.2239 + // after a significant amount of data.
1.2240 + uint64_t unacked = stream->LocalUnAcked();
1.2241 + int64_t localWindow = stream->LocalWindow();
1.2242 +
1.2243 + LOG3(("SpdySession31::UpdateLocalStreamWindow this=%p id=0x%X newbytes=%u "
1.2244 + "unacked=%llu localWindow=%lld\n",
1.2245 + this, stream->StreamID(), bytes, unacked, localWindow));
1.2246 +
1.2247 + if (!unacked)
1.2248 + return;
1.2249 +
1.2250 + if ((unacked < kMinimumToAck) && (localWindow > kEmergencyWindowThreshold))
1.2251 + return;
1.2252 +
1.2253 + if (!stream->HasSink()) {
1.2254 + LOG3(("SpdySession31::UpdateLocalStreamWindow %p 0x%X Pushed Stream Has No Sink\n",
1.2255 + this, stream->StreamID()));
1.2256 + return;
1.2257 + }
1.2258 +
1.2259 + // Generate window updates directly out of spdysession instead of the stream
1.2260 + // in order to avoid queue delays in getting the 'ACK' out.
1.2261 + uint32_t toack = (unacked <= 0x7fffffffU) ? unacked : 0x7fffffffU;
1.2262 +
1.2263 + LOG3(("SpdySession31::UpdateLocalStreamWindow Ack this=%p id=0x%X acksize=%d\n",
1.2264 + this, stream->StreamID(), toack));
1.2265 + stream->IncrementLocalWindow(toack);
1.2266 +
1.2267 + // room for this packet needs to be ensured before calling this function
1.2268 + static const uint32_t dataLen = 8;
1.2269 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.2270 + mOutputQueueUsed += 8 + dataLen;
1.2271 + MOZ_ASSERT(mOutputQueueUsed <= mOutputQueueSize);
1.2272 +
1.2273 + memset(packet, 0, 8 + dataLen);
1.2274 + packet[0] = kFlag_Control;
1.2275 + packet[1] = kVersion;
1.2276 + packet[3] = CONTROL_TYPE_WINDOW_UPDATE;
1.2277 + packet[7] = dataLen;
1.2278 +
1.2279 + uint32_t id = PR_htonl(stream->StreamID());
1.2280 + memcpy(packet + 8, &id, 4);
1.2281 + toack = PR_htonl(toack);
1.2282 + memcpy(packet + 12, &toack, 4);
1.2283 +
1.2284 + LogIO(this, stream, "Stream Window Update", packet, 8 + dataLen);
1.2285 + // dont flush here, this write can commonly be coalesced with a
1.2286 + // session window update to immediately follow.
1.2287 +}
1.2288 +
1.2289 +void
1.2290 +SpdySession31::UpdateLocalSessionWindow(uint32_t bytes)
1.2291 +{
1.2292 + if (!bytes)
1.2293 + return;
1.2294 +
1.2295 + mLocalSessionWindow -= bytes;
1.2296 +
1.2297 + LOG3(("SpdySession31::UpdateLocalSessionWindow this=%p newbytes=%u "
1.2298 + "localWindow=%lld\n", this, bytes, mLocalSessionWindow));
1.2299 +
1.2300 + // Don't necessarily ack every data packet. Only do it
1.2301 + // after a significant amount of data.
1.2302 + if ((mLocalSessionWindow > (ASpdySession::kInitialRwin - kMinimumToAck)) &&
1.2303 + (mLocalSessionWindow > kEmergencyWindowThreshold))
1.2304 + return;
1.2305 +
1.2306 + // Only send max 31 bits of window updates at a time.
1.2307 + uint64_t toack64 = ASpdySession::kInitialRwin - mLocalSessionWindow;
1.2308 + uint32_t toack = (toack64 <= 0x7fffffffU) ? toack64 : 0x7fffffffU;
1.2309 +
1.2310 + LOG3(("SpdySession31::UpdateLocalSessionWindow Ack this=%p acksize=%u\n",
1.2311 + this, toack));
1.2312 + mLocalSessionWindow += toack;
1.2313 +
1.2314 + // room for this packet needs to be ensured before calling this function
1.2315 + static const uint32_t dataLen = 8;
1.2316 + char *packet = mOutputQueueBuffer.get() + mOutputQueueUsed;
1.2317 + mOutputQueueUsed += 8 + dataLen;
1.2318 + MOZ_ASSERT(mOutputQueueUsed <= mOutputQueueSize);
1.2319 +
1.2320 + memset(packet, 0, 8 + dataLen);
1.2321 + packet[0] = kFlag_Control;
1.2322 + packet[1] = kVersion;
1.2323 + packet[3] = CONTROL_TYPE_WINDOW_UPDATE;
1.2324 + packet[7] = dataLen;
1.2325 +
1.2326 + // packet 8-11 is ID and left at 0 for session ID
1.2327 + toack = PR_htonl(toack);
1.2328 + memcpy(packet + 12, &toack, 4);
1.2329 +
1.2330 + LogIO(this, nullptr, "Session Window Update", packet, 8 + dataLen);
1.2331 + // dont flush here, this write can commonly be coalesced with others
1.2332 +}
1.2333 +
1.2334 +void
1.2335 +SpdySession31::UpdateLocalRwin(SpdyStream31 *stream,
1.2336 + uint32_t bytes)
1.2337 +{
1.2338 + // make sure there is room for 2 window updates even though
1.2339 + // we may not generate any.
1.2340 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + (16 *2),
1.2341 + mOutputQueueUsed, mOutputQueueSize);
1.2342 +
1.2343 + UpdateLocalStreamWindow(stream, bytes);
1.2344 + UpdateLocalSessionWindow(bytes);
1.2345 + FlushOutputQueue();
1.2346 +}
1.2347 +
1.2348 +void
1.2349 +SpdySession31::Close(nsresult aReason)
1.2350 +{
1.2351 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2352 +
1.2353 + if (mClosed)
1.2354 + return;
1.2355 +
1.2356 + LOG3(("SpdySession31::Close %p %X", this, aReason));
1.2357 +
1.2358 + mClosed = true;
1.2359 +
1.2360 + mStreamTransactionHash.Enumerate(ShutdownEnumerator, this);
1.2361 + mStreamIDHash.Clear();
1.2362 + mStreamTransactionHash.Clear();
1.2363 +
1.2364 + uint32_t goAwayReason;
1.2365 + if (NS_SUCCEEDED(aReason)) {
1.2366 + goAwayReason = OK;
1.2367 + } else if (aReason == NS_ERROR_ILLEGAL_VALUE) {
1.2368 + goAwayReason = PROTOCOL_ERROR;
1.2369 + } else {
1.2370 + goAwayReason = INTERNAL_ERROR;
1.2371 + }
1.2372 + GenerateGoAway(goAwayReason);
1.2373 + mConnection = nullptr;
1.2374 + mSegmentReader = nullptr;
1.2375 + mSegmentWriter = nullptr;
1.2376 +}
1.2377 +
1.2378 +void
1.2379 +SpdySession31::CloseTransaction(nsAHttpTransaction *aTransaction,
1.2380 + nsresult aResult)
1.2381 +{
1.2382 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2383 + LOG3(("SpdySession31::CloseTransaction %p %p %x", this, aTransaction, aResult));
1.2384 +
1.2385 + // Generally this arrives as a cancel event from the connection manager.
1.2386 +
1.2387 + // need to find the stream and call CleanupStream() on it.
1.2388 + SpdyStream31 *stream = mStreamTransactionHash.Get(aTransaction);
1.2389 + if (!stream) {
1.2390 + LOG3(("SpdySession31::CloseTransaction %p %p %x - not found.",
1.2391 + this, aTransaction, aResult));
1.2392 + return;
1.2393 + }
1.2394 + LOG3(("SpdySession31::CloseTranscation probably a cancel. "
1.2395 + "this=%p, trans=%p, result=%x, streamID=0x%X stream=%p",
1.2396 + this, aTransaction, aResult, stream->StreamID(), stream));
1.2397 + CleanupStream(stream, aResult, RST_CANCEL);
1.2398 + ResumeRecv();
1.2399 +}
1.2400 +
1.2401 +//-----------------------------------------------------------------------------
1.2402 +// nsAHttpSegmentReader
1.2403 +//-----------------------------------------------------------------------------
1.2404 +
1.2405 +nsresult
1.2406 +SpdySession31::OnReadSegment(const char *buf,
1.2407 + uint32_t count,
1.2408 + uint32_t *countRead)
1.2409 +{
1.2410 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2411 +
1.2412 + nsresult rv;
1.2413 +
1.2414 + // If we can release old queued data then we can try and write the new
1.2415 + // data directly to the network without using the output queue at all
1.2416 + if (mOutputQueueUsed)
1.2417 + FlushOutputQueue();
1.2418 +
1.2419 + if (!mOutputQueueUsed && mSegmentReader) {
1.2420 + // try and write directly without output queue
1.2421 + rv = mSegmentReader->OnReadSegment(buf, count, countRead);
1.2422 +
1.2423 + if (rv == NS_BASE_STREAM_WOULD_BLOCK)
1.2424 + *countRead = 0;
1.2425 + else if (NS_FAILED(rv))
1.2426 + return rv;
1.2427 +
1.2428 + if (*countRead < count) {
1.2429 + uint32_t required = count - *countRead;
1.2430 + // assuming a commitment() happened, this ensurebuffer is a nop
1.2431 + // but just in case the queuesize is too small for the required data
1.2432 + // call ensurebuffer().
1.2433 + EnsureBuffer(mOutputQueueBuffer, required, 0, mOutputQueueSize);
1.2434 + memcpy(mOutputQueueBuffer.get(), buf + *countRead, required);
1.2435 + mOutputQueueUsed = required;
1.2436 + }
1.2437 +
1.2438 + *countRead = count;
1.2439 + return NS_OK;
1.2440 + }
1.2441 +
1.2442 + // At this point we are going to buffer the new data in the output
1.2443 + // queue if it fits. By coalescing multiple small submissions into one larger
1.2444 + // buffer we can get larger writes out to the network later on.
1.2445 +
1.2446 + // This routine should not be allowed to fill up the output queue
1.2447 + // all on its own - at least kQueueReserved bytes are always left
1.2448 + // for other routines to use - but this is an all-or-nothing function,
1.2449 + // so if it will not all fit just return WOULD_BLOCK
1.2450 +
1.2451 + if ((mOutputQueueUsed + count) > (mOutputQueueSize - kQueueReserved))
1.2452 + return NS_BASE_STREAM_WOULD_BLOCK;
1.2453 +
1.2454 + memcpy(mOutputQueueBuffer.get() + mOutputQueueUsed, buf, count);
1.2455 + mOutputQueueUsed += count;
1.2456 + *countRead = count;
1.2457 +
1.2458 + FlushOutputQueue();
1.2459 +
1.2460 + return NS_OK;
1.2461 +}
1.2462 +
1.2463 +nsresult
1.2464 +SpdySession31::CommitToSegmentSize(uint32_t count, bool forceCommitment)
1.2465 +{
1.2466 + if (mOutputQueueUsed)
1.2467 + FlushOutputQueue();
1.2468 +
1.2469 + // would there be enough room to buffer this if needed?
1.2470 + if ((mOutputQueueUsed + count) <= (mOutputQueueSize - kQueueReserved))
1.2471 + return NS_OK;
1.2472 +
1.2473 + // if we are using part of our buffers already, try again later unless
1.2474 + // forceCommitment is set.
1.2475 + if (mOutputQueueUsed && !forceCommitment)
1.2476 + return NS_BASE_STREAM_WOULD_BLOCK;
1.2477 +
1.2478 + if (mOutputQueueUsed) {
1.2479 + // normally we avoid the memmove of RealignOutputQueue, but we'll try
1.2480 + // it if forceCommitment is set before growing the buffer.
1.2481 + RealignOutputQueue();
1.2482 +
1.2483 + // is there enough room now?
1.2484 + if ((mOutputQueueUsed + count) <= (mOutputQueueSize - kQueueReserved))
1.2485 + return NS_OK;
1.2486 + }
1.2487 +
1.2488 + // resize the buffers as needed
1.2489 + EnsureBuffer(mOutputQueueBuffer, mOutputQueueUsed + count + kQueueReserved,
1.2490 + mOutputQueueUsed, mOutputQueueSize);
1.2491 +
1.2492 + MOZ_ASSERT((mOutputQueueUsed + count) <= (mOutputQueueSize - kQueueReserved),
1.2493 + "buffer not as large as expected");
1.2494 +
1.2495 + return NS_OK;
1.2496 +}
1.2497 +
1.2498 +//-----------------------------------------------------------------------------
1.2499 +// nsAHttpSegmentWriter
1.2500 +//-----------------------------------------------------------------------------
1.2501 +
1.2502 +nsresult
1.2503 +SpdySession31::OnWriteSegment(char *buf,
1.2504 + uint32_t count,
1.2505 + uint32_t *countWritten)
1.2506 +{
1.2507 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2508 + nsresult rv;
1.2509 +
1.2510 + if (!mSegmentWriter) {
1.2511 + // the only way this could happen would be if Close() were called on the
1.2512 + // stack with WriteSegments()
1.2513 + return NS_ERROR_FAILURE;
1.2514 + }
1.2515 +
1.2516 + if (mDownstreamState == PROCESSING_DATA_FRAME) {
1.2517 +
1.2518 + if (mInputFrameDataLast &&
1.2519 + mInputFrameDataRead == mInputFrameDataSize) {
1.2520 + *countWritten = 0;
1.2521 + SetNeedsCleanup();
1.2522 + return NS_BASE_STREAM_CLOSED;
1.2523 + }
1.2524 +
1.2525 + count = std::min(count, mInputFrameDataSize - mInputFrameDataRead);
1.2526 + rv = NetworkRead(mSegmentWriter, buf, count, countWritten);
1.2527 + if (NS_FAILED(rv))
1.2528 + return rv;
1.2529 +
1.2530 + LogIO(this, mInputFrameDataStream, "Reading Data Frame",
1.2531 + buf, *countWritten);
1.2532 +
1.2533 + mInputFrameDataRead += *countWritten;
1.2534 +
1.2535 + mInputFrameDataStream->UpdateTransportReadEvents(*countWritten);
1.2536 + if ((mInputFrameDataRead == mInputFrameDataSize) && !mInputFrameDataLast)
1.2537 + ResetDownstreamState();
1.2538 +
1.2539 + return rv;
1.2540 + }
1.2541 +
1.2542 + if (mDownstreamState == PROCESSING_COMPLETE_HEADERS) {
1.2543 +
1.2544 + if (mFlatHTTPResponseHeaders.Length() == mFlatHTTPResponseHeadersOut &&
1.2545 + mInputFrameDataLast) {
1.2546 + *countWritten = 0;
1.2547 + SetNeedsCleanup();
1.2548 + return NS_BASE_STREAM_CLOSED;
1.2549 + }
1.2550 +
1.2551 + count = std::min(count,
1.2552 + mFlatHTTPResponseHeaders.Length() -
1.2553 + mFlatHTTPResponseHeadersOut);
1.2554 + memcpy(buf,
1.2555 + mFlatHTTPResponseHeaders.get() + mFlatHTTPResponseHeadersOut,
1.2556 + count);
1.2557 + mFlatHTTPResponseHeadersOut += count;
1.2558 + *countWritten = count;
1.2559 +
1.2560 + if (mFlatHTTPResponseHeaders.Length() == mFlatHTTPResponseHeadersOut) {
1.2561 + if (mDataPending) {
1.2562 + // Now ready to process data frames - pop PROCESING_DATA_FRAME back onto
1.2563 + // the stack because receipt of that first data frame triggered the
1.2564 + // response header processing
1.2565 + mDataPending = false;
1.2566 + ChangeDownstreamState(PROCESSING_DATA_FRAME);
1.2567 + }
1.2568 + else if (!mInputFrameDataLast) {
1.2569 + // If more frames are expected in this stream, then reset the state so they can be
1.2570 + // handled. Otherwise (e.g. a 0 length response with the fin on the SYN_REPLY)
1.2571 + // stay in PROCESSING_COMPLETE_HEADERS state so the SetNeedsCleanup() code above can
1.2572 + // cleanup the stream.
1.2573 + ResetDownstreamState();
1.2574 + }
1.2575 + }
1.2576 +
1.2577 + return NS_OK;
1.2578 + }
1.2579 +
1.2580 + return NS_ERROR_UNEXPECTED;
1.2581 +}
1.2582 +
1.2583 +void
1.2584 +SpdySession31::SetNeedsCleanup()
1.2585 +{
1.2586 + LOG3(("SpdySession31::SetNeedsCleanup %p - recorded downstream fin of "
1.2587 + "stream %p 0x%X", this, mInputFrameDataStream,
1.2588 + mInputFrameDataStream->StreamID()));
1.2589 +
1.2590 + // This will result in Close() being called
1.2591 + MOZ_ASSERT(!mNeedsCleanup, "mNeedsCleanup unexpectedly set");
1.2592 + mNeedsCleanup = mInputFrameDataStream;
1.2593 + ResetDownstreamState();
1.2594 +}
1.2595 +
1.2596 +void
1.2597 +SpdySession31::ConnectPushedStream(SpdyStream31 *stream)
1.2598 +{
1.2599 + mReadyForRead.Push(stream);
1.2600 + ForceRecv();
1.2601 +}
1.2602 +
1.2603 +nsresult
1.2604 +SpdySession31::BufferOutput(const char *buf,
1.2605 + uint32_t count,
1.2606 + uint32_t *countRead)
1.2607 +{
1.2608 + nsAHttpSegmentReader *old = mSegmentReader;
1.2609 + mSegmentReader = nullptr;
1.2610 + nsresult rv = OnReadSegment(buf, count, countRead);
1.2611 + mSegmentReader = old;
1.2612 + return rv;
1.2613 +}
1.2614 +
1.2615 +//-----------------------------------------------------------------------------
1.2616 +// Modified methods of nsAHttpConnection
1.2617 +//-----------------------------------------------------------------------------
1.2618 +
1.2619 +void
1.2620 +SpdySession31::TransactionHasDataToWrite(nsAHttpTransaction *caller)
1.2621 +{
1.2622 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2623 + LOG3(("SpdySession31::TransactionHasDataToWrite %p trans=%p", this, caller));
1.2624 +
1.2625 + // a trapped signal from the http transaction to the connection that
1.2626 + // it is no longer blocked on read.
1.2627 +
1.2628 + SpdyStream31 *stream = mStreamTransactionHash.Get(caller);
1.2629 + if (!stream || !VerifyStream(stream)) {
1.2630 + LOG3(("SpdySession31::TransactionHasDataToWrite %p caller %p not found",
1.2631 + this, caller));
1.2632 + return;
1.2633 + }
1.2634 +
1.2635 + LOG3(("SpdySession31::TransactionHasDataToWrite %p ID is 0x%X\n",
1.2636 + this, stream->StreamID()));
1.2637 +
1.2638 + mReadyForWrite.Push(stream);
1.2639 +}
1.2640 +
1.2641 +void
1.2642 +SpdySession31::TransactionHasDataToWrite(SpdyStream31 *stream)
1.2643 +{
1.2644 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2645 + LOG3(("SpdySession31::TransactionHasDataToWrite %p stream=%p ID=%x",
1.2646 + this, stream, stream->StreamID()));
1.2647 +
1.2648 + mReadyForWrite.Push(stream);
1.2649 + SetWriteCallbacks();
1.2650 +}
1.2651 +
1.2652 +bool
1.2653 +SpdySession31::IsPersistent()
1.2654 +{
1.2655 + return true;
1.2656 +}
1.2657 +
1.2658 +nsresult
1.2659 +SpdySession31::TakeTransport(nsISocketTransport **,
1.2660 + nsIAsyncInputStream **,
1.2661 + nsIAsyncOutputStream **)
1.2662 +{
1.2663 + MOZ_ASSERT(false, "TakeTransport of SpdySession31");
1.2664 + return NS_ERROR_UNEXPECTED;
1.2665 +}
1.2666 +
1.2667 +nsHttpConnection *
1.2668 +SpdySession31::TakeHttpConnection()
1.2669 +{
1.2670 + MOZ_ASSERT(false, "TakeHttpConnection of SpdySession31");
1.2671 + return nullptr;
1.2672 +}
1.2673 +
1.2674 +uint32_t
1.2675 +SpdySession31::CancelPipeline(nsresult reason)
1.2676 +{
1.2677 + // we don't pipeline inside spdy, so this isn't an issue
1.2678 + return 0;
1.2679 +}
1.2680 +
1.2681 +nsAHttpTransaction::Classifier
1.2682 +SpdySession31::Classification()
1.2683 +{
1.2684 + if (!mConnection)
1.2685 + return nsAHttpTransaction::CLASS_GENERAL;
1.2686 + return mConnection->Classification();
1.2687 +}
1.2688 +
1.2689 +//-----------------------------------------------------------------------------
1.2690 +// unused methods of nsAHttpTransaction
1.2691 +// We can be sure of this because SpdySession31 is only constructed in
1.2692 +// nsHttpConnection and is never passed out of that object
1.2693 +//-----------------------------------------------------------------------------
1.2694 +
1.2695 +void
1.2696 +SpdySession31::SetConnection(nsAHttpConnection *)
1.2697 +{
1.2698 + // This is unexpected
1.2699 + MOZ_ASSERT(false, "SpdySession31::SetConnection()");
1.2700 +}
1.2701 +
1.2702 +void
1.2703 +SpdySession31::GetSecurityCallbacks(nsIInterfaceRequestor **)
1.2704 +{
1.2705 + // This is unexpected
1.2706 + MOZ_ASSERT(false, "SpdySession31::GetSecurityCallbacks()");
1.2707 +}
1.2708 +
1.2709 +void
1.2710 +SpdySession31::SetProxyConnectFailed()
1.2711 +{
1.2712 + MOZ_ASSERT(false, "SpdySession31::SetProxyConnectFailed()");
1.2713 +}
1.2714 +
1.2715 +bool
1.2716 +SpdySession31::IsDone()
1.2717 +{
1.2718 + return !mStreamTransactionHash.Count();
1.2719 +}
1.2720 +
1.2721 +nsresult
1.2722 +SpdySession31::Status()
1.2723 +{
1.2724 + MOZ_ASSERT(false, "SpdySession31::Status()");
1.2725 + return NS_ERROR_UNEXPECTED;
1.2726 +}
1.2727 +
1.2728 +uint32_t
1.2729 +SpdySession31::Caps()
1.2730 +{
1.2731 + MOZ_ASSERT(false, "SpdySession31::Caps()");
1.2732 + return 0;
1.2733 +}
1.2734 +
1.2735 +void
1.2736 +SpdySession31::SetDNSWasRefreshed()
1.2737 +{
1.2738 +}
1.2739 +
1.2740 +uint64_t
1.2741 +SpdySession31::Available()
1.2742 +{
1.2743 + MOZ_ASSERT(false, "SpdySession31::Available()");
1.2744 + return 0;
1.2745 +}
1.2746 +
1.2747 +nsHttpRequestHead *
1.2748 +SpdySession31::RequestHead()
1.2749 +{
1.2750 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2751 + MOZ_ASSERT(false,
1.2752 + "SpdySession31::RequestHead() "
1.2753 + "should not be called after SPDY is setup");
1.2754 + return nullptr;
1.2755 +}
1.2756 +
1.2757 +uint32_t
1.2758 +SpdySession31::Http1xTransactionCount()
1.2759 +{
1.2760 + return 0;
1.2761 +}
1.2762 +
1.2763 +// used as an enumerator by TakeSubTransactions()
1.2764 +static PLDHashOperator
1.2765 + TakeStream(nsAHttpTransaction *key,
1.2766 + nsAutoPtr<SpdyStream31> &stream,
1.2767 + void *closure)
1.2768 +{
1.2769 + nsTArray<nsRefPtr<nsAHttpTransaction> > *list =
1.2770 + static_cast<nsTArray<nsRefPtr<nsAHttpTransaction> > *>(closure);
1.2771 +
1.2772 + list->AppendElement(key);
1.2773 +
1.2774 + // removing the stream from the hash will delete the stream
1.2775 + // and drop the transaction reference the hash held
1.2776 + return PL_DHASH_REMOVE;
1.2777 +}
1.2778 +
1.2779 +nsresult
1.2780 +SpdySession31::TakeSubTransactions(
1.2781 + nsTArray<nsRefPtr<nsAHttpTransaction> > &outTransactions)
1.2782 +{
1.2783 + // Generally this cannot be done with spdy as transactions are
1.2784 + // started right away.
1.2785 +
1.2786 + LOG3(("SpdySession31::TakeSubTransactions %p\n", this));
1.2787 +
1.2788 + if (mConcurrentHighWater > 0)
1.2789 + return NS_ERROR_ALREADY_OPENED;
1.2790 +
1.2791 + LOG3((" taking %d\n", mStreamTransactionHash.Count()));
1.2792 +
1.2793 + mStreamTransactionHash.Enumerate(TakeStream, &outTransactions);
1.2794 + return NS_OK;
1.2795 +}
1.2796 +
1.2797 +nsresult
1.2798 +SpdySession31::AddTransaction(nsAHttpTransaction *)
1.2799 +{
1.2800 + // This API is meant for pipelining, SpdySession31's should be
1.2801 + // extended with AddStream()
1.2802 +
1.2803 + MOZ_ASSERT(false,
1.2804 + "SpdySession31::AddTransaction() should not be called");
1.2805 +
1.2806 + return NS_ERROR_NOT_IMPLEMENTED;
1.2807 +}
1.2808 +
1.2809 +uint32_t
1.2810 +SpdySession31::PipelineDepth()
1.2811 +{
1.2812 + return IsDone() ? 0 : 1;
1.2813 +}
1.2814 +
1.2815 +nsresult
1.2816 +SpdySession31::SetPipelinePosition(int32_t position)
1.2817 +{
1.2818 + // This API is meant for pipelining, SpdySession31's should be
1.2819 + // extended with AddStream()
1.2820 +
1.2821 + MOZ_ASSERT(false,
1.2822 + "SpdySession31::SetPipelinePosition() should not be called");
1.2823 +
1.2824 + return NS_ERROR_NOT_IMPLEMENTED;
1.2825 +}
1.2826 +
1.2827 +int32_t
1.2828 +SpdySession31::PipelinePosition()
1.2829 +{
1.2830 + return 0;
1.2831 +}
1.2832 +
1.2833 +//-----------------------------------------------------------------------------
1.2834 +// Pass through methods of nsAHttpConnection
1.2835 +//-----------------------------------------------------------------------------
1.2836 +
1.2837 +nsAHttpConnection *
1.2838 +SpdySession31::Connection()
1.2839 +{
1.2840 + MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread);
1.2841 + return mConnection;
1.2842 +}
1.2843 +
1.2844 +nsresult
1.2845 +SpdySession31::OnHeadersAvailable(nsAHttpTransaction *transaction,
1.2846 + nsHttpRequestHead *requestHead,
1.2847 + nsHttpResponseHead *responseHead,
1.2848 + bool *reset)
1.2849 +{
1.2850 + return mConnection->OnHeadersAvailable(transaction,
1.2851 + requestHead,
1.2852 + responseHead,
1.2853 + reset);
1.2854 +}
1.2855 +
1.2856 +bool
1.2857 +SpdySession31::IsReused()
1.2858 +{
1.2859 + return mConnection->IsReused();
1.2860 +}
1.2861 +
1.2862 +nsresult
1.2863 +SpdySession31::PushBack(const char *buf, uint32_t len)
1.2864 +{
1.2865 + return mConnection->PushBack(buf, len);
1.2866 +}
1.2867 +
1.2868 +} // namespace mozilla::net
1.2869 +} // namespace mozilla
