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/logging.h"
     10 #include "base/stl_util.h"
     11 #include "net/quic/quic_blocked_writer_interface.h"
     12 #include "net/quic/quic_utils.h"
     13 #include "net/tools/quic/quic_default_packet_writer.h"
     14 #include "net/tools/quic/quic_epoll_connection_helper.h"
     15 #include "net/tools/quic/quic_socket_utils.h"
     16 
     17 namespace net {
     18 namespace tools {
     19 
     20 using std::make_pair;
     21 
     22 class DeleteSessionsAlarm : public EpollAlarm {
     23  public:
     24   explicit DeleteSessionsAlarm(QuicDispatcher* dispatcher)
     25       : dispatcher_(dispatcher) {
     26   }
     27 
     28   virtual int64 OnAlarm() OVERRIDE {
     29     EpollAlarm::OnAlarm();
     30     dispatcher_->DeleteSessions();
     31     return 0;
     32   }
     33 
     34  private:
     35   QuicDispatcher* dispatcher_;
     36 };
     37 
     38 QuicDispatcher::QuicDispatcher(const QuicConfig& config,
     39                                const QuicCryptoServerConfig& crypto_config,
     40                                const QuicVersionVector& supported_versions,
     41                                int fd,
     42                                EpollServer* epoll_server)
     43     : config_(config),
     44       crypto_config_(crypto_config),
     45       time_wait_list_manager_(
     46           new QuicTimeWaitListManager(this, epoll_server, supported_versions)),
     47       delete_sessions_alarm_(new DeleteSessionsAlarm(this)),
     48       epoll_server_(epoll_server),
     49       fd_(fd),
     50       write_blocked_(false),
     51       helper_(new QuicEpollConnectionHelper(epoll_server_)),
     52       writer_(new QuicDefaultPacketWriter(fd)),
     53       supported_versions_(supported_versions) {
     54 }
     55 
     56 QuicDispatcher::~QuicDispatcher() {
     57   STLDeleteValues(&session_map_);
     58   STLDeleteElements(&closed_session_list_);
     59 }
     60 
     61 void QuicDispatcher::set_fd(int fd) {
     62   fd_ = fd;
     63   writer_.reset(new QuicDefaultPacketWriter(fd));
     64 }
     65 
     66 WriteResult QuicDispatcher::WritePacket(const char* buffer, size_t buf_len,
     67                                         const IPAddressNumber& self_address,
     68                                         const IPEndPoint& peer_address,
     69                                         QuicBlockedWriterInterface* writer) {
     70   if (write_blocked_) {
     71     write_blocked_list_.insert(make_pair(writer, true));
     72     return WriteResult(WRITE_STATUS_BLOCKED, EAGAIN);
     73   }
     74 
     75   WriteResult result =
     76       writer_->WritePacket(buffer, buf_len, self_address, peer_address, writer);
     77   if (result.status == WRITE_STATUS_BLOCKED) {
     78     write_blocked_list_.insert(make_pair(writer, true));
     79     write_blocked_ = true;
     80   }
     81   return result;
     82 }
     83 
     84 bool QuicDispatcher::IsWriteBlockedDataBuffered() const {
     85   return writer_->IsWriteBlockedDataBuffered();
     86 }
     87 
     88 void QuicDispatcher::ProcessPacket(const IPEndPoint& server_address,
     89                                    const IPEndPoint& client_address,
     90                                    QuicGuid guid,
     91                                    bool has_version_flag,
     92                                    const QuicEncryptedPacket& packet) {
     93   QuicSession* session = NULL;
     94 
     95   SessionMap::iterator it = session_map_.find(guid);
     96   if (it == session_map_.end()) {
     97     if (time_wait_list_manager_->IsGuidInTimeWait(guid)) {
     98       time_wait_list_manager_->ProcessPacket(server_address,
     99                                              client_address,
    100                                              guid,
    101                                              packet);
    102       return;
    103     }
    104 
    105     // Ensure the packet has a version negotiation bit set before creating a new
    106     // session for it.  All initial packets for a new connection are required to
    107     // have the flag set.  Otherwise it may be a stray packet.
    108     if (has_version_flag) {
    109       session = CreateQuicSession(guid, server_address, client_address);
    110     }
    111 
    112     if (session == NULL) {
    113       DLOG(INFO) << "Failed to create session for " << guid;
    114       // Add this guid fo the time-wait state, to safely reject future packets.
    115       // We don't know the version here, so assume latest.
    116       // TODO(ianswett): Produce a no-version version negotiation packet.
    117       time_wait_list_manager_->AddGuidToTimeWait(guid,
    118                                                  supported_versions_.front(),
    119                                                  NULL);
    120       time_wait_list_manager_->ProcessPacket(server_address,
    121                                              client_address,
    122                                              guid,
    123                                              packet);
    124       return;
    125     }
    126     DLOG(INFO) << "Created new session for " << guid;
    127     session_map_.insert(make_pair(guid, session));
    128   } else {
    129     session = it->second;
    130   }
    131 
    132   session->connection()->ProcessUdpPacket(
    133       server_address, client_address, packet);
    134 }
    135 
    136 void QuicDispatcher::CleanUpSession(SessionMap::iterator it) {
    137   QuicConnection* connection = it->second->connection();
    138   QuicEncryptedPacket* connection_close_packet =
    139           connection->ReleaseConnectionClosePacket();
    140   write_blocked_list_.erase(connection);
    141   time_wait_list_manager_->AddGuidToTimeWait(it->first,
    142                                              connection->version(),
    143                                              connection_close_packet);
    144   session_map_.erase(it);
    145 }
    146 
    147 void QuicDispatcher::DeleteSessions() {
    148   STLDeleteElements(&closed_session_list_);
    149 }
    150 
    151 void QuicDispatcher::UseWriter(QuicPacketWriter* writer) {
    152   writer_.reset(writer);
    153 }
    154 
    155 bool QuicDispatcher::OnCanWrite() {
    156   // We got an EPOLLOUT: the socket should not be blocked.
    157   write_blocked_ = false;
    158 
    159   // Give each writer one attempt to write.
    160   int num_writers = write_blocked_list_.size();
    161   for (int i = 0; i < num_writers; ++i) {
    162     if (write_blocked_list_.empty()) {
    163       break;
    164     }
    165     QuicBlockedWriterInterface* writer = write_blocked_list_.begin()->first;
    166     write_blocked_list_.erase(write_blocked_list_.begin());
    167     bool can_write_more = writer->OnCanWrite();
    168     if (write_blocked_) {
    169       // We were unable to write.  Wait for the next EPOLLOUT.
    170       // In this case, the session would have been added to the blocked list
    171       // up in WritePacket.
    172       return false;
    173     }
    174     // The socket is not blocked but the writer has ceded work.  Add it to the
    175     // end of the list.
    176     if (can_write_more) {
    177       write_blocked_list_.insert(make_pair(writer, true));
    178     }
    179   }
    180 
    181   // We're not write blocked.  Return true if there's more work to do.
    182   return !write_blocked_list_.empty();
    183 }
    184 
    185 void QuicDispatcher::Shutdown() {
    186   while (!session_map_.empty()) {
    187     QuicSession* session = session_map_.begin()->second;
    188     session->connection()->SendConnectionClose(QUIC_PEER_GOING_AWAY);
    189     // Validate that the session removes itself from the session map on close.
    190     DCHECK(session_map_.empty() || session_map_.begin()->second != session);
    191   }
    192   DeleteSessions();
    193 }
    194 
    195 void QuicDispatcher::OnConnectionClosed(QuicGuid guid, QuicErrorCode error) {
    196   SessionMap::iterator it = session_map_.find(guid);
    197   if (it == session_map_.end()) {
    198     LOG(DFATAL) << "GUID " << guid << " does not exist in the session map.  "
    199                 << "Error: " << QuicUtils::ErrorToString(error);
    200     return;
    201   }
    202 
    203   DLOG_IF(INFO, error != QUIC_NO_ERROR) << "Closing connection (" << guid
    204                                         << ") due to error: "
    205                                         << QuicUtils::ErrorToString(error);
    206 
    207   if (closed_session_list_.empty()) {
    208     epoll_server_->RegisterAlarmApproximateDelta(
    209         0, delete_sessions_alarm_.get());
    210   }
    211   closed_session_list_.push_back(it->second);
    212   CleanUpSession(it);
    213 }
    214 
    215 QuicSession* QuicDispatcher::CreateQuicSession(
    216     QuicGuid guid,
    217     const IPEndPoint& server_address,
    218     const IPEndPoint& client_address) {
    219   QuicServerSession* session = new QuicServerSession(
    220       config_, new QuicConnection(guid, client_address, helper_.get(), this,
    221                                   true, supported_versions_), this);
    222   session->InitializeSession(crypto_config_);
    223   return session;
    224 }
    225 
    226 }  // namespace tools
    227 }  // namespace net
    228