src/connection_queue.cpp

   1
   2 /*
   3
   4 Copyright (c) 2007, Arvid Norberg
   5 All rights reserved.
   6
   7 Redistribution and use in source and binary forms, with or without
   8 modification, are permitted provided that the following conditions
   9 are met:
  10
  11     * Redistributions of source code must retain the above copyright
  12       notice, this list of conditions and the following disclaimer.
  13     * Redistributions in binary form must reproduce the above copyright
  14       notice, this list of conditions and the following disclaimer in
  15       the documentation and/or other materials provided with the distribution.
  16     * Neither the name of the author nor the names of its
  17       contributors may be used to endorse or promote products derived
  18       from this software without specific prior written permission.
  19
  20 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21 AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22 IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  23 ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  24 LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  25 CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  26 SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  27 INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  28 CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  29 ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  30 POSSIBILITY OF SUCH DAMAGE.
  31
  32 */
  33
  34 #include <boost/bind.hpp>
  35 #include "libtorrent/invariant_check.hpp"
  36 #include "libtorrent/connection_queue.hpp"
  37 #include "libtorrent/socket.hpp"
  38
  39 namespace libtorrent
  40 {
  41
  42         connection_queue::connection_queue(io_service& ios): m_next_ticket(0)
  43                 , m_num_connecting(0)
  44                 , m_half_open_limit(0)
  45                 , m_timer(ios)
  46 #ifndef NDEBUG
  47                 , m_in_timeout_function(false)
  48 #endif
  49         {
  50 #ifdef TORRENT_CONNECTION_LOGGING
  51                 m_log.open("connection_queue.log");
  52 #endif
  53         }
  54
  55         int connection_queue::free_slots() const
  56         {
  57                 mutex_t::scoped_lock l(m_mutex);
  58                 return m_half_open_limit == 0 ? (std::numeric_limits<int>::max)()
  59                         : m_half_open_limit - m_queue.size();
  60         }
  61
  62         void connection_queue::enqueue(boost::function<void(int)> const& on_connect
  63                 , boost::function<void()> const& on_timeout
  64                 , time_duration timeout, int priority)
  65         {
  66                 mutex_t::scoped_lock l(m_mutex);
  67
  68                 INVARIANT_CHECK;
  69
  70                 TORRENT_ASSERT(priority >= 0);
  71                 TORRENT_ASSERT(priority < 2);
  72
  73                 entry* e = 0;
  74
  75                 switch (priority)
  76                 {
  77                         case 0:
  78                                 m_queue.push_back(entry());
  79                                 e = &m_queue.back();
  80                                 break;
  81                         case 1:
  82                                 m_queue.push_front(entry());
  83                                 e = &m_queue.front();
  84                                 break;
  85                 }
  86
  87                 e->priority = priority;
  88                 e->on_connect = on_connect;
  89                 e->on_timeout = on_timeout;
  90                 e->ticket = m_next_ticket;
  91                 e->timeout = timeout;
  92                 ++m_next_ticket;
  93                 try_connect();
  94         }
  95
  96         void connection_queue::done(int ticket)
  97         {
  98                 mutex_t::scoped_lock l(m_mutex);
  99
 100                 INVARIANT_CHECK;
 101
 102                 std::list<entry>::iterator i = std::find_if(m_queue.begin()
 103                         , m_queue.end(), boost::bind(&entry::ticket, _1) == ticket);
 104                 if (i == m_queue.end())
 105                 {
 106                         // this might not be here in case on_timeout calls remove
 107                         return;
 108                 }
 109                 if (i->connecting) --m_num_connecting;
 110                 m_queue.erase(i);
 111                 try_connect();
 112         }
 113
 114         void connection_queue::close()
 115         {
 116                 error_code ec;
 117                 m_timer.cancel(ec);
 118         }
 119
 120         void connection_queue::limit(int limit)
 121         {
 122                 TORRENT_ASSERT(limit >= 0);
 123                 m_half_open_limit = limit;
 124         }
 125
 126         int connection_queue::limit() const
 127         { return m_half_open_limit; }
 128
 129 #ifndef NDEBUG
 130
 131         void connection_queue::check_invariant() const
 132         {
 133                 int num_connecting = 0;
 134                 for (std::list<entry>::const_iterator i = m_queue.begin();
 135                         i != m_queue.end(); ++i)
 136                 {
 137                         if (i->connecting) ++num_connecting;
 138                 }
 139                 TORRENT_ASSERT(num_connecting == m_num_connecting);
 140         }
 141
 142 #endif
 143
 144         void connection_queue::try_connect()
 145         {
 146                 INVARIANT_CHECK;
 147
 148 #ifdef TORRENT_CONNECTION_LOGGING
 149                 m_log << log_time() << " " << free_slots() << std::endl;
 150 #endif
 151
 152                 if (m_num_connecting >= m_half_open_limit
 153                         && m_half_open_limit > 0) return;
 154
 155                 if (m_queue.empty())
 156                 {
 157                         error_code ec;
 158                         m_timer.cancel(ec);
 159                         return;
 160                 }
 161
 162                 std::list<entry>::iterator i = std::find_if(m_queue.begin()
 163                         , m_queue.end(), boost::bind(&entry::connecting, _1) == false);
 164                 while (i != m_queue.end())
 165                 {
 166                         TORRENT_ASSERT(i->connecting == false);
 167                         ptime expire = time_now() + i->timeout;
 168                         if (m_num_connecting == 0)
 169                         {
 170                                 error_code ec;
 171                                 m_timer.expires_at(expire, ec);
 172                                 m_timer.async_wait(boost::bind(&connection_queue::on_timeout, this, _1));
 173                         }
 174                         i->connecting = true;
 175                         ++m_num_connecting;
 176                         i->expires = expire;
 177
 178                         INVARIANT_CHECK;
 179
 180                         entry& ent = *i;
 181                         ++i;
 182 #ifndef BOOST_NO_EXCEPTIONS
 183                         try {
 184 #endif
 185                                 ent.on_connect(ent.ticket);
 186 #ifndef BOOST_NO_EXCEPTIONS
 187                         } catch (std::exception&) {}
 188 #endif
 189
 190 #ifdef TORRENT_CONNECTION_LOGGING
 191                         m_log << log_time() << " " << free_slots() << std::endl;
 192 #endif
 193
 194                         if (m_num_connecting >= m_half_open_limit
 195                                 && m_half_open_limit > 0) break;
 196                         i = std::find_if(i, m_queue.end(), boost::bind(&entry::connecting, _1) == false);
 197                 }
 198         }
 199
 200 #ifndef NDEBUG
 201         struct function_guard
 202         {
 203                 function_guard(bool& v): val(v) { TORRENT_ASSERT(!val); val = true; }
 204                 ~function_guard() { val = false; }
 205
 206                 bool& val;
 207         };
 208 #endif
 209
 210         void connection_queue::on_timeout(error_code const& e)
 211         {
 212                 mutex_t::scoped_lock l(m_mutex);
 213
 214                 INVARIANT_CHECK;
 215 #ifndef NDEBUG
 216                 function_guard guard_(m_in_timeout_function);
 217 #endif
 218
 219                 TORRENT_ASSERT(!e || e == asio::error::operation_aborted);
 220                 if (e) return;
 221
 222                 ptime next_expire = max_time();
 223                 ptime now = time_now();
 224                 std::list<entry> timed_out;
 225                 for (std::list<entry>::iterator i = m_queue.begin();
 226                         !m_queue.empty() && i != m_queue.end();)
 227                 {
 228                         if (i->connecting && i->expires < now)
 229                         {
 230                                 std::list<entry>::iterator j = i;
 231                                 ++i;
 232                                 timed_out.splice(timed_out.end(), m_queue, j, i);
 233                                 --m_num_connecting;
 234                                 continue;
 235                         }
 236                         if (i->expires < next_expire)
 237                                 next_expire = i->expires;
 238                         ++i;
 239                 }
 240
 241                 // we don't want to call the timeout callback while we're locked
 242                 // since that is a recepie for dead-locks
 243                 l.unlock();
 244
 245                 for (std::list<entry>::iterator i = timed_out.begin()
 246                         , end(timed_out.end()); i != end; ++i)
 247                 {
 248                         try { i->on_timeout(); } catch (std::exception&) {}
 249                 }
 250
 251                 l.lock();
 252
 253                 if (next_expire < max_time())
 254                 {
 255                         error_code ec;
 256                         m_timer.expires_at(next_expire, ec);
 257                         m_timer.async_wait(boost::bind(&connection_queue::on_timeout, this, _1));
 258                 }
 259                 try_connect();
 260         }
 261
 262 }
 263