Home | History | Annotate | Download | only in quic 
      1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
      2 // Use of this source code is governed by a BSD-style license that can be
      3 // found in the LICENSE file.
      4 
      5 #include "net/tools/quic/quic_dispatcher.h"
      6 
      7 #include <errno.h>
      8 
      9 #include "base/debug/stack_trace.h"
     10 #include "base/logging.h"
     11 #include "base/stl_util.h"
     12 #include "net/quic/quic_blocked_writer_interface.h"
     13 #include "net/quic/quic_flags.h"
     14 #include "net/quic/quic_utils.h"
     15 #include "net/tools/epoll_server/epoll_server.h"
     16 #include "net/tools/quic/quic_default_packet_writer.h"
     17 #include "net/tools/quic/quic_epoll_connection_helper.h"
     18 #include "net/tools/quic/quic_per_connection_packet_writer.h"
     19 #include "net/tools/quic/quic_socket_utils.h"
     20 #include "net/tools/quic/quic_time_wait_list_manager.h"
     21 
     22 namespace net {
     23 
     24 namespace tools {
     25 
     26 using base::StringPiece;
     27 using std::make_pair;
     28 
     29 class DeleteSessionsAlarm : public EpollAlarm {
     30  public:
     31   explicit DeleteSessionsAlarm(QuicDispatcher* dispatcher)
     32       : dispatcher_(dispatcher) {
     33   }
     34 
     35   virtual int64 OnAlarm() OVERRIDE {
     36     EpollAlarm::OnAlarm();
     37     dispatcher_->DeleteSessions();
     38     return 0;
     39   }
     40 
     41  private:
     42   QuicDispatcher* dispatcher_;
     43 };
     44 
     45 class QuicDispatcher::QuicFramerVisitor : public QuicFramerVisitorInterface {
     46  public:
     47   explicit QuicFramerVisitor(QuicDispatcher* dispatcher)
     48       : dispatcher_(dispatcher),
     49         connection_id_(0) {}
     50 
     51   // QuicFramerVisitorInterface implementation
     52   virtual void OnPacket() OVERRIDE {}
     53   virtual bool OnUnauthenticatedPublicHeader(
     54       const QuicPacketPublicHeader& header) OVERRIDE {
     55     connection_id_ = header.connection_id;
     56     return dispatcher_->OnUnauthenticatedPublicHeader(header);
     57   }
     58   virtual bool OnUnauthenticatedHeader(
     59       const QuicPacketHeader& header) OVERRIDE {
     60     dispatcher_->OnUnauthenticatedHeader(header);
     61     return false;
     62   }
     63   virtual void OnError(QuicFramer* framer) OVERRIDE {
     64     DVLOG(1) << QuicUtils::ErrorToString(framer->error());
     65   }
     66 
     67   virtual bool OnProtocolVersionMismatch(
     68       QuicVersion /*received_version*/) OVERRIDE {
     69     if (dispatcher_->time_wait_list_manager()->IsConnectionIdInTimeWait(
     70             connection_id_)) {
     71       // Keep processing after protocol mismatch - this will be dealt with by
     72       // the TimeWaitListManager.
     73       return true;
     74     } else {
     75       DLOG(DFATAL) << "Version mismatch, connection ID (" << connection_id_
     76                    << ") not in time wait list.";
     77       return false;
     78     }
     79   }
     80 
     81   // The following methods should never get called because we always return
     82   // false from OnUnauthenticatedHeader().  As a result, we never process the
     83   // payload of the packet.
     84   virtual void OnPublicResetPacket(
     85       const QuicPublicResetPacket& /*packet*/) OVERRIDE {
     86     DCHECK(false);
     87   }
     88   virtual void OnVersionNegotiationPacket(
     89       const QuicVersionNegotiationPacket& /*packet*/) OVERRIDE {
     90     DCHECK(false);
     91   }
     92   virtual void OnDecryptedPacket(EncryptionLevel level) OVERRIDE {
     93     DCHECK(false);
     94   }
     95   virtual bool OnPacketHeader(const QuicPacketHeader& /*header*/) OVERRIDE {
     96     DCHECK(false);
     97     return false;
     98   }
     99   virtual void OnRevivedPacket() OVERRIDE {
    100     DCHECK(false);
    101   }
    102   virtual void OnFecProtectedPayload(StringPiece /*payload*/) OVERRIDE {
    103     DCHECK(false);
    104   }
    105   virtual bool OnStreamFrame(const QuicStreamFrame& /*frame*/) OVERRIDE {
    106     DCHECK(false);
    107     return false;
    108   }
    109   virtual bool OnAckFrame(const QuicAckFrame& /*frame*/) OVERRIDE {
    110     DCHECK(false);
    111     return false;
    112   }
    113   virtual bool OnCongestionFeedbackFrame(
    114       const QuicCongestionFeedbackFrame& /*frame*/) OVERRIDE {
    115     DCHECK(false);
    116     return false;
    117   }
    118   virtual bool OnStopWaitingFrame(
    119       const QuicStopWaitingFrame& /*frame*/) OVERRIDE {
    120     DCHECK(false);
    121     return false;
    122   }
    123   virtual bool OnPingFrame(const QuicPingFrame& /*frame*/) OVERRIDE {
    124     DCHECK(false);
    125     return false;
    126   }
    127   virtual bool OnRstStreamFrame(const QuicRstStreamFrame& /*frame*/) OVERRIDE {
    128     DCHECK(false);
    129     return false;
    130   }
    131   virtual bool OnConnectionCloseFrame(
    132       const QuicConnectionCloseFrame & /*frame*/) OVERRIDE {
    133     DCHECK(false);
    134     return false;
    135   }
    136   virtual bool OnGoAwayFrame(const QuicGoAwayFrame& /*frame*/) OVERRIDE {
    137     DCHECK(false);
    138     return false;
    139   }
    140   virtual bool OnWindowUpdateFrame(const QuicWindowUpdateFrame& /*frame*/)
    141       OVERRIDE {
    142     DCHECK(false);
    143     return false;
    144   }
    145   virtual bool OnBlockedFrame(const QuicBlockedFrame& frame) OVERRIDE {
    146     DCHECK(false);
    147     return false;
    148   }
    149   virtual void OnFecData(const QuicFecData& /*fec*/) OVERRIDE {
    150     DCHECK(false);
    151   }
    152   virtual void OnPacketComplete() OVERRIDE {
    153     DCHECK(false);
    154   }
    155 
    156  private:
    157   QuicDispatcher* dispatcher_;
    158 
    159   // Latched in OnUnauthenticatedPublicHeader for use later.
    160   QuicConnectionId connection_id_;
    161 };
    162 
    163 QuicPacketWriter* QuicDispatcher::DefaultPacketWriterFactory::Create(
    164     QuicPacketWriter* writer,
    165     QuicConnection* connection) {
    166   return new QuicPerConnectionPacketWriter(writer, connection);
    167 }
    168 
    169 QuicDispatcher::PacketWriterFactoryAdapter::PacketWriterFactoryAdapter(
    170     QuicDispatcher* dispatcher)
    171     : dispatcher_(dispatcher) {}
    172 
    173 QuicDispatcher::PacketWriterFactoryAdapter::~PacketWriterFactoryAdapter() {}
    174 
    175 QuicPacketWriter* QuicDispatcher::PacketWriterFactoryAdapter::Create(
    176     QuicConnection* connection) const {
    177   return dispatcher_->packet_writer_factory_->Create(
    178       dispatcher_->writer_.get(),
    179       connection);
    180 }
    181 
    182 QuicDispatcher::QuicDispatcher(const QuicConfig& config,
    183                                const QuicCryptoServerConfig& crypto_config,
    184                                const QuicVersionVector& supported_versions,
    185                                PacketWriterFactory* packet_writer_factory,
    186                                EpollServer* epoll_server)
    187     : config_(config),
    188       crypto_config_(crypto_config),
    189       delete_sessions_alarm_(new DeleteSessionsAlarm(this)),
    190       epoll_server_(epoll_server),
    191       helper_(new QuicEpollConnectionHelper(epoll_server_)),
    192       packet_writer_factory_(packet_writer_factory),
    193       connection_writer_factory_(this),
    194       supported_versions_(supported_versions),
    195       current_packet_(NULL),
    196       framer_(supported_versions, /*unused*/ QuicTime::Zero(), true),
    197       framer_visitor_(new QuicFramerVisitor(this)) {
    198   framer_.set_visitor(framer_visitor_.get());
    199 }
    200 
    201 QuicDispatcher::~QuicDispatcher() {
    202   STLDeleteValues(&session_map_);
    203   STLDeleteElements(&closed_session_list_);
    204 }
    205 
    206 void QuicDispatcher::Initialize(int fd) {
    207   DCHECK(writer_ == NULL);
    208   writer_.reset(CreateWriter(fd));
    209   time_wait_list_manager_.reset(CreateQuicTimeWaitListManager());
    210 }
    211 
    212 void QuicDispatcher::ProcessPacket(const IPEndPoint& server_address,
    213                                    const IPEndPoint& client_address,
    214                                    const QuicEncryptedPacket& packet) {
    215   current_server_address_ = server_address;
    216   current_client_address_ = client_address;
    217   current_packet_ = &packet;
    218   // ProcessPacket will cause the packet to be dispatched in
    219   // OnUnauthenticatedPublicHeader, or sent to the time wait list manager
    220   // in OnAuthenticatedHeader.
    221   framer_.ProcessPacket(packet);
    222   // TODO(rjshade): Return a status describing if/why a packet was dropped,
    223   //                and log somehow.  Maybe expose as a varz.
    224 }
    225 
    226 bool QuicDispatcher::OnUnauthenticatedPublicHeader(
    227     const QuicPacketPublicHeader& header) {
    228   QuicSession* session = NULL;
    229 
    230   QuicConnectionId connection_id = header.connection_id;
    231   SessionMap::iterator it = session_map_.find(connection_id);
    232   if (it == session_map_.end()) {
    233     if (header.reset_flag) {
    234       return false;
    235     }
    236     if (time_wait_list_manager_->IsConnectionIdInTimeWait(connection_id)) {
    237       return HandlePacketForTimeWait(header);
    238     }
    239 
    240     // Ensure the packet has a version negotiation bit set before creating a new
    241     // session for it.  All initial packets for a new connection are required to
    242     // have the flag set.  Otherwise it may be a stray packet.
    243     if (header.version_flag) {
    244       session = CreateQuicSession(connection_id, current_server_address_,
    245                                   current_client_address_);
    246     }
    247 
    248     if (session == NULL) {
    249       DVLOG(1) << "Failed to create session for " << connection_id;
    250       // Add this connection_id fo the time-wait state, to safely reject future
    251       // packets.
    252 
    253       if (header.version_flag &&
    254           !framer_.IsSupportedVersion(header.versions.front())) {
    255         // TODO(ianswett): Produce a no-version version negotiation packet.
    256         return false;
    257       }
    258 
    259       // Use the version in the packet if possible, otherwise assume the latest.
    260       QuicVersion version = header.version_flag ? header.versions.front() :
    261           supported_versions_.front();
    262       time_wait_list_manager_->AddConnectionIdToTimeWait(
    263           connection_id, version, NULL);
    264       DCHECK(time_wait_list_manager_->IsConnectionIdInTimeWait(connection_id));
    265       return HandlePacketForTimeWait(header);
    266     }
    267     DVLOG(1) << "Created new session for " << connection_id;
    268     session_map_.insert(make_pair(connection_id, session));
    269   } else {
    270     session = it->second;
    271   }
    272 
    273   session->connection()->ProcessUdpPacket(
    274       current_server_address_, current_client_address_, *current_packet_);
    275 
    276   // Do not parse the packet further.  The session will process it completely.
    277   return false;
    278 }
    279 
    280 void QuicDispatcher::OnUnauthenticatedHeader(const QuicPacketHeader& header) {
    281   DCHECK(time_wait_list_manager_->IsConnectionIdInTimeWait(
    282       header.public_header.connection_id));
    283   time_wait_list_manager_->ProcessPacket(current_server_address_,
    284                                          current_client_address_,
    285                                          header.public_header.connection_id,
    286                                          header.packet_sequence_number,
    287                                          *current_packet_);
    288 }
    289 
    290 void QuicDispatcher::CleanUpSession(SessionMap::iterator it) {
    291   QuicConnection* connection = it->second->connection();
    292   QuicEncryptedPacket* connection_close_packet =
    293       connection->ReleaseConnectionClosePacket();
    294   write_blocked_list_.erase(connection);
    295   time_wait_list_manager_->AddConnectionIdToTimeWait(it->first,
    296                                                      connection->version(),
    297                                                      connection_close_packet);
    298   session_map_.erase(it);
    299 }
    300 
    301 void QuicDispatcher::DeleteSessions() {
    302   STLDeleteElements(&closed_session_list_);
    303 }
    304 
    305 void QuicDispatcher::OnCanWrite() {
    306   // We got an EPOLLOUT: the socket should not be blocked.
    307   writer_->SetWritable();
    308 
    309   // Give all the blocked writers one chance to write, until we're blocked again
    310   // or there's no work left.
    311   while (!write_blocked_list_.empty() && !writer_->IsWriteBlocked()) {
    312     QuicBlockedWriterInterface* blocked_writer =
    313         write_blocked_list_.begin()->first;
    314     write_blocked_list_.erase(write_blocked_list_.begin());
    315     blocked_writer->OnCanWrite();
    316   }
    317 }
    318 
    319 bool QuicDispatcher::HasPendingWrites() const {
    320   return !write_blocked_list_.empty();
    321 }
    322 
    323 void QuicDispatcher::Shutdown() {
    324   while (!session_map_.empty()) {
    325     QuicSession* session = session_map_.begin()->second;
    326     session->connection()->SendConnectionClose(QUIC_PEER_GOING_AWAY);
    327     // Validate that the session removes itself from the session map on close.
    328     DCHECK(session_map_.empty() || session_map_.begin()->second != session);
    329   }
    330   DeleteSessions();
    331 }
    332 
    333 void QuicDispatcher::OnConnectionClosed(QuicConnectionId connection_id,
    334                                         QuicErrorCode error) {
    335   SessionMap::iterator it = session_map_.find(connection_id);
    336   if (it == session_map_.end()) {
    337     LOG(DFATAL) << "ConnectionId " << connection_id
    338                 << " does not exist in the session map.  "
    339                 << "Error: " << QuicUtils::ErrorToString(error);
    340     LOG(DFATAL) << base::debug::StackTrace().ToString();
    341     return;
    342   }
    343 
    344   DLOG_IF(INFO, error != QUIC_NO_ERROR) << "Closing connection ("
    345                                         << connection_id
    346                                         << ") due to error: "
    347                                         << QuicUtils::ErrorToString(error);
    348 
    349   if (closed_session_list_.empty()) {
    350     epoll_server_->RegisterAlarmApproximateDelta(
    351         0, delete_sessions_alarm_.get());
    352   }
    353   closed_session_list_.push_back(it->second);
    354   CleanUpSession(it);
    355 }
    356 
    357 void QuicDispatcher::OnWriteBlocked(
    358     QuicBlockedWriterInterface* blocked_writer) {
    359   if (!writer_->IsWriteBlocked()) {
    360     LOG(DFATAL) <<
    361         "QuicDispatcher::OnWriteBlocked called when the writer is not blocked.";
    362     // Return without adding the connection to the blocked list, to avoid
    363     // infinite loops in OnCanWrite.
    364     return;
    365   }
    366   write_blocked_list_.insert(make_pair(blocked_writer, true));
    367 }
    368 
    369 QuicPacketWriter* QuicDispatcher::CreateWriter(int fd) {
    370   return new QuicDefaultPacketWriter(fd);
    371 }
    372 
    373 QuicSession* QuicDispatcher::CreateQuicSession(
    374     QuicConnectionId connection_id,
    375     const IPEndPoint& server_address,
    376     const IPEndPoint& client_address) {
    377   QuicServerSession* session = new QuicServerSession(
    378       config_,
    379       CreateQuicConnection(connection_id, server_address, client_address),
    380       this);
    381   session->InitializeSession(crypto_config_);
    382   return session;
    383 }
    384 
    385 QuicConnection* QuicDispatcher::CreateQuicConnection(
    386     QuicConnectionId connection_id,
    387     const IPEndPoint& server_address,
    388     const IPEndPoint& client_address) {
    389   return new QuicConnection(connection_id,
    390                             client_address,
    391                             helper_.get(),
    392                             connection_writer_factory_,
    393                             /* owns_writer= */ true,
    394                             /* is_server= */ true,
    395                             supported_versions_);
    396 }
    397 
    398 QuicTimeWaitListManager* QuicDispatcher::CreateQuicTimeWaitListManager() {
    399   return new QuicTimeWaitListManager(
    400       writer_.get(), this, epoll_server(), supported_versions());
    401 }
    402 
    403 bool QuicDispatcher::HandlePacketForTimeWait(
    404     const QuicPacketPublicHeader& header) {
    405   if (header.reset_flag) {
    406     // Public reset packets do not have sequence numbers, so ignore the packet.
    407     return false;
    408   }
    409 
    410   // Switch the framer to the correct version, so that the sequence number can
    411   // be parsed correctly.
    412   framer_.set_version(time_wait_list_manager_->GetQuicVersionFromConnectionId(
    413       header.connection_id));
    414 
    415   // Continue parsing the packet to extract the sequence number.  Then
    416   // send it to the time wait manager in OnUnathenticatedHeader.
    417   return true;
    418 }
    419 
    420 }  // namespace tools
    421 }  // namespace net
    422