?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
LT SYNC
[tore.git] / libtorrent / src / session_impl.cpp
blobde8b73e836b6e8322aa2cdb68c9265732434000e
1 /*
2
3 Copyright (c) 2006, Arvid Norberg, Magnus Jonsson
4 All rights reserved.
5
6 Redistribution and use in source and binary forms, with or without
7 modification, are permitted provided that the following conditions
8 are met:
9
10 * Redistributions of source code must retain the above copyright
11 notice, this list of conditions and the following disclaimer.
12 * Redistributions in binary form must reproduce the above copyright
13 notice, this list of conditions and the following disclaimer in
14 the documentation and/or other materials provided with the distribution.
15 * Neither the name of the author nor the names of its
16 contributors may be used to endorse or promote products derived
17 from this software without specific prior written permission.
18
19 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
20 AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
21 IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
22 ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
23 LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
24 CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
25 SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
26 INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
27 CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
28 ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
29 POSSIBILITY OF SUCH DAMAGE.
30
31 */
32
33 #include "libtorrent/pch.hpp"
34
35 #include <ctime>
36 #include <iostream>
37 #include <fstream>
38 #include <iomanip>
39 #include <iterator>
40 #include <algorithm>
41 #include <set>
42 #include <cctype>
43 #include <algorithm>
44
45 #ifdef _MSC_VER
46 #pragma warning(push, 1)
47 #endif
48
49 #include <boost/lexical_cast.hpp>
50 #include <boost/filesystem/convenience.hpp>
51 #include <boost/filesystem/exception.hpp>
52 #include <boost/limits.hpp>
53 #include <boost/bind.hpp>
54
55 #ifdef _MSC_VER
56 #pragma warning(pop)
57 #endif
58
59 #include "libtorrent/peer_id.hpp"
60 #include "libtorrent/torrent_info.hpp"
61 #include "libtorrent/tracker_manager.hpp"
62 #include "libtorrent/bencode.hpp"
63 #include "libtorrent/hasher.hpp"
64 #include "libtorrent/entry.hpp"
65 #include "libtorrent/session.hpp"
66 #include "libtorrent/fingerprint.hpp"
67 #include "libtorrent/entry.hpp"
68 #include "libtorrent/alert_types.hpp"
69 #include "libtorrent/invariant_check.hpp"
70 #include "libtorrent/file.hpp"
71 #include "libtorrent/bt_peer_connection.hpp"
72 #include "libtorrent/ip_filter.hpp"
73 #include "libtorrent/socket.hpp"
74 #include "libtorrent/aux_/session_impl.hpp"
75 #include "libtorrent/kademlia/dht_tracker.hpp"
76 #include "libtorrent/enum_net.hpp"
77 #include "libtorrent/config.hpp"
78
79 #ifndef TORRENT_WINDOWS
80 #include <sys/resource.h>
81 #endif
82
83 #ifndef TORRENT_DISABLE_ENCRYPTION
84
85 #include <openssl/crypto.h>
86
87 namespace
88 {
89 // openssl requires this to clean up internal
90 // structures it allocates
91 struct openssl_cleanup
92 {
93 ~openssl_cleanup() { CRYPTO_cleanup_all_ex_data(); }
94 } openssl_global_destructor;
95 }
96
97 #endif
98 #ifdef _WIN32
99 // for ERROR_SEM_TIMEOUT
100 #include <winerror.h>
101 #endif
102
103 using boost::shared_ptr;
104 using boost::weak_ptr;
105 using boost::bind;
106 using boost::mutex;
107 using libtorrent::aux::session_impl;
108
109 namespace libtorrent {
110
111 namespace fs = boost::filesystem;
112
113 namespace detail
114 {
115
116 std::string generate_auth_string(std::string const& user
117 , std::string const& passwd)
118 {
119 if (user.empty()) return std::string();
120 return user + ":" + passwd;
121 }
122
123
124 }
125 namespace aux {
126
127 struct seed_random_generator
128 {
129 seed_random_generator()
130 {
131 std::srand(total_microseconds(time_now() - min_time()));
132 }
133 };
134
135 session_impl::session_impl(
136 std::pair<int, int> listen_port_range
137 , fingerprint const& cl_fprint
138 , char const* listen_interface
139 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
140 , fs::path const& logpath
141 #endif
142 )
143 :
144 #ifndef TORRENT_DISABLE_POOL_ALLOCATOR
145 m_send_buffers(send_buffer_size),
146 #endif
147 m_files(40)
148 , m_io_service()
149 , m_disk_thread(m_io_service)
150 , m_half_open(m_io_service)
151 , m_download_channel(m_io_service, peer_connection::download_channel)
152 #ifdef TORRENT_VERBOSE_BANDWIDTH_LIMIT
153 , m_upload_channel(m_io_service, peer_connection::upload_channel, true)
154 #else
155 , m_upload_channel(m_io_service, peer_connection::upload_channel)
156 #endif
157 , m_tracker_manager(m_settings, m_tracker_proxy)
158 , m_listen_port_retries(listen_port_range.second - listen_port_range.first)
159 , m_listen_interface(address::from_string(listen_interface), listen_port_range.first)
160 , m_abort(false)
161 , m_paused(false)
162 , m_max_uploads(8)
163 , m_allowed_upload_slots(8)
164 , m_max_connections(200)
165 , m_num_unchoked(0)
166 , m_unchoke_time_scaler(0)
167 , m_auto_manage_time_scaler(0)
168 , m_optimistic_unchoke_time_scaler(0)
169 , m_disconnect_time_scaler(90)
170 , m_auto_scrape_time_scaler(180)
171 , m_incoming_connection(false)
172 , m_last_tick(time_now())
173 #ifndef TORRENT_DISABLE_DHT
174 , m_dht_same_port(true)
175 , m_external_udp_port(0)
176 , m_dht_socket(m_io_service, bind(&session_impl::on_receive_udp, this, _1, _2, _3, _4)
177 , m_half_open)
178 #endif
179 , m_timer(m_io_service)
180 , m_next_connect_torrent(0)
181 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
182 , m_logpath(logpath)
183 #endif
184 #ifndef TORRENT_DISABLE_GEO_IP
185 , m_asnum_db(0)
186 , m_country_db(0)
187 #endif
188 , m_total_failed_bytes(0)
189 , m_total_redundant_bytes(0)
190 {
191 m_tcp_mapping[0] = -1;
192 m_tcp_mapping[1] = -1;
193 m_udp_mapping[0] = -1;
194 m_udp_mapping[1] = -1;
195 #ifdef WIN32
196 // windows XP has a limit on the number of
197 // simultaneous half-open TCP connections
198 DWORD windows_version = ::GetVersion();
199 if ((windows_version & 0xff) >= 6)
200 {
201 // on vista the limit is 5 (in home edition)
202 m_half_open.limit(4);
203 }
204 else
205 {
206 // on XP SP2 it's 10
207 m_half_open.limit(8);
208 }
209 #endif
210
211 m_bandwidth_manager[peer_connection::download_channel] = &m_download_channel;
212 m_bandwidth_manager[peer_connection::upload_channel] = &m_upload_channel;
213
214 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
215 m_logger = create_log("main_session", listen_port(), false);
216 (*m_logger) << time_now_string() << "\n";
217
218 (*m_logger) << "sizeof(torrent): " << sizeof(torrent) << "\n";
219 (*m_logger) << "sizeof(peer_connection): " << sizeof(peer_connection) << "\n";
220 (*m_logger) << "sizeof(bt_peer_connection): " << sizeof(bt_peer_connection) << "\n";
221 (*m_logger) << "sizeof(policy::peer): " << sizeof(policy::peer) << "\n";
222 (*m_logger) << "sizeof(tcp::endpoint): " << sizeof(tcp::endpoint) << "\n";
223 (*m_logger) << "sizeof(address): " << sizeof(address) << "\n";
224 (*m_logger) << "sizeof(address_v4): " << sizeof(address_v4) << "\n";
225 (*m_logger) << "sizeof(address_v6): " << sizeof(address_v6) << "\n";
226 (*m_logger) << "sizeof(void*): " << sizeof(void*) << "\n";
227 #endif
228
229 #ifdef TORRENT_STATS
230 m_stats_logger.open("session_stats.log", std::ios::trunc);
231 m_stats_logger <<
232 "1. second\n"
233 "2. upload rate\n"
234 "3. download rate\n"
235 "4. downloading torrents\n"
236 "5. seeding torrents\n"
237 "6. peers\n"
238 "7. connecting peers\n"
239 "8. disk block buffers\n"
240 "\n";
241 m_buffer_usage_logger.open("buffer_stats.log", std::ios::trunc);
242 m_second_counter = 0;
243 m_buffer_allocations = 0;
244 #endif
245
246 // ---- generate a peer id ----
247 static seed_random_generator seeder;
248
249 m_key = rand() + (rand() << 15) + (rand() << 30);
250 std::string print = cl_fprint.to_string();
251 TORRENT_ASSERT(print.length() <= 20);
252
253 // the client's fingerprint
254 std::copy(
255 print.begin()
256 , print.begin() + print.length()
257 , m_peer_id.begin());
258
259 // http-accepted characters:
260 static char const printable[] = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
261 "abcdefghijklmnopqrstuvwxyz-_.!~*'()";
262
263 // the random number
264 for (unsigned char* i = m_peer_id.begin() + print.length();
265 i != m_peer_id.end(); ++i)
266 {
267 *i = printable[rand() % (sizeof(printable)-1)];
268 }
269
270 error_code ec;
271 m_timer.expires_from_now(seconds(1), ec);
272 m_timer.async_wait(
273 bind(&session_impl::second_tick, this, _1));
274
275 m_thread.reset(new boost::thread(boost::ref(*this)));
276 }
277
278 #ifndef TORRENT_DISABLE_GEO_IP
279 namespace
280 {
281 struct free_ptr
282 {
283 void* ptr_;
284 free_ptr(void* p): ptr_(p) {}
285 ~free_ptr() { free(ptr_); }
286 };
287 }
288
289 char const* session_impl::country_for_ip(address const& a)
290 {
291 if (!a.is_v4() || m_country_db == 0) return 0;
292 return GeoIP_country_code_by_ipnum(m_country_db, a.to_v4().to_ulong());
293 }
294
295 int session_impl::as_for_ip(address const& a)
296 {
297 if (!a.is_v4() || m_asnum_db == 0) return 0;
298 char* name = GeoIP_name_by_ipnum(m_asnum_db, a.to_v4().to_ulong());
299 if (name == 0) return 0;
300 free_ptr p(name);
301 // GeoIP returns the name as AS??? where ? is the AS-number
302 return atoi(name + 2);
303 }
304
305 std::string session_impl::as_name_for_ip(address const& a)
306 {
307 if (!a.is_v4() || m_asnum_db == 0) return std::string();
308 char* name = GeoIP_name_by_ipnum(m_asnum_db, a.to_v4().to_ulong());
309 if (name == 0) return std::string();
310 free_ptr p(name);
311 char* tmp = std::strchr(name, ' ');
312 if (tmp == 0) return std::string();
313 return tmp + 1;
314 }
315
316 std::pair<const int, int>* session_impl::lookup_as(int as)
317 {
318 std::map<int, int>::iterator i = m_as_peak.lower_bound(as);
319
320 if (i == m_as_peak.end() || i->first != as)
321 {
322 // we don't have any data for this AS, insert a new entry
323 i = m_as_peak.insert(i, std::pair<int, int>(as, 0));
324 }
325 return &(*i);
326 }
327
328 bool session_impl::load_asnum_db(char const* file)
329 {
330 mutex_t::scoped_lock l(m_mutex);
331 if (m_asnum_db) GeoIP_delete(m_asnum_db);
332 m_asnum_db = GeoIP_open(file, GEOIP_STANDARD);
333 return m_asnum_db;
334 }
335
336 bool session_impl::load_country_db(char const* file)
337 {
338 mutex_t::scoped_lock l(m_mutex);
339 if (m_country_db) GeoIP_delete(m_country_db);
340 m_country_db = GeoIP_open(file, GEOIP_STANDARD);
341 return m_country_db;
342 }
343
344 #endif
345
346 void session_impl::load_state(entry const& ses_state)
347 {
348 if (ses_state.type() != entry::dictionary_t) return;
349 mutex_t::scoped_lock l(m_mutex);
350 #ifndef TORRENT_DISABLE_GEO_IP
351 entry const* as_map = ses_state.find_key("AS map");
352 if (as_map && as_map->type() == entry::dictionary_t)
353 {
354 entry::dictionary_type const& as_peak = as_map->dict();
355 for (entry::dictionary_type::const_iterator i = as_peak.begin()
356 , end(as_peak.end()); i != end; ++i)
357 {
358 int as_num = atoi(i->first.c_str());
359 if (i->second.type() != entry::int_t || i->second.integer() == 0) continue;
360 int& peak = m_as_peak[as_num];
361 if (peak < i->second.integer()) peak = i->second.integer();
362 }
363 }
364 #endif
365 }
366
367 entry session_impl::state() const
368 {
369 mutex_t::scoped_lock l(m_mutex);
370 entry ret;
371 #ifndef TORRENT_DISABLE_GEO_IP
372 entry::dictionary_type& as_map = ret["AS map"].dict();
373 char buf[10];
374 for (std::map<int, int>::const_iterator i = m_as_peak.begin()
375 , end(m_as_peak.end()); i != end; ++i)
376 {
377 if (i->second == 0) continue;
378 sprintf(buf, "%05d", i->first);
379 as_map[buf] = i->second;
380 }
381 #endif
382 return ret;
383 }
384
385 #ifndef TORRENT_DISABLE_EXTENSIONS
386 void session_impl::add_extension(
387 boost::function<boost::shared_ptr<torrent_plugin>(torrent*, void*)> ext)
388 {
389 m_extensions.push_back(ext);
390 }
391 #endif
392
393 #ifndef TORRENT_DISABLE_DHT
394 void session_impl::add_dht_node(udp::endpoint n)
395 {
396 if (m_dht) m_dht->add_node(n);
397 }
398 #endif
399
400 void session_impl::pause()
401 {
402 mutex_t::scoped_lock l(m_mutex);
403 if (m_paused) return;
404 m_paused = true;
405 for (torrent_map::iterator i = m_torrents.begin()
406 , end(m_torrents.end()); i != end; ++i)
407 {
408 torrent& t = *i->second;
409 if (!t.is_torrent_paused()) t.do_pause();
410 }
411 }
412
413 void session_impl::resume()
414 {
415 mutex_t::scoped_lock l(m_mutex);
416 if (!m_paused) return;
417 m_paused = false;
418 for (torrent_map::iterator i = m_torrents.begin()
419 , end(m_torrents.end()); i != end; ++i)
420 {
421 torrent& t = *i->second;
422 t.do_resume();
423 }
424 }
425
426 void session_impl::abort()
427 {
428 mutex_t::scoped_lock l(m_mutex);
429 if (m_abort) return;
430 #if defined(TORRENT_LOGGING)
431 (*m_logger) << time_now_string() << " *** ABORT CALLED ***\n";
432 #endif
433 // abort the main thread
434 m_abort = true;
435 m_queued_for_checking.clear();
436 if (m_lsd) m_lsd->close();
437 if (m_upnp) m_upnp->close();
438 if (m_natpmp) m_natpmp->close();
439 #ifndef TORRENT_DISABLE_DHT
440 if (m_dht) m_dht->stop();
441 m_dht_socket.close();
442 #endif
443 error_code ec;
444 m_timer.cancel(ec);
445
446 // close the listen sockets
447 for (std::list<listen_socket_t>::iterator i = m_listen_sockets.begin()
448 , end(m_listen_sockets.end()); i != end; ++i)
449 {
450 i->sock->close(ec);
451 }
452
453 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
454 (*m_logger) << time_now_string() << " aborting all torrents (" << m_torrents.size() << ")\n";
455 #endif
456 // abort all torrents
457 for (torrent_map::iterator i = m_torrents.begin()
458 , end(m_torrents.end()); i != end; ++i)
459 {
460 i->second->abort();
461 }
462
463 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
464 (*m_logger) << time_now_string() << " aborting all tracker requests\n";
465 #endif
466 m_tracker_manager.abort_all_requests();
467
468 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
469 (*m_logger) << time_now_string() << " sending event=stopped to trackers\n";
470 int counter = 0;
471 #endif
472 for (torrent_map::iterator i = m_torrents.begin();
473 i != m_torrents.end(); ++i)
474 {
475 torrent& t = *i->second;
476 t.abort();
477 }
478
479 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
480 (*m_logger) << time_now_string() << " sent " << counter << " tracker stop requests\n";
481 #endif
482
483 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
484 (*m_logger) << time_now_string() << " aborting all connections (" << m_connections.size() << ")\n";
485 #endif
486 // abort all connections
487 while (!m_connections.empty())
488 {
489 #ifndef NDEBUG
490 int conn = m_connections.size();
491 #endif
492 (*m_connections.begin())->disconnect("stopping torrent");
493 TORRENT_ASSERT(conn == int(m_connections.size()) + 1);
494 }
495
496 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
497 (*m_logger) << time_now_string() << " shutting down connection queue\n";
498 #endif
499 m_half_open.close();
500
501 m_download_channel.close();
502 m_upload_channel.close();
503 }
504
505 void session_impl::set_port_filter(port_filter const& f)
506 {
507 mutex_t::scoped_lock l(m_mutex);
508 m_port_filter = f;
509 }
510
511 void session_impl::set_ip_filter(ip_filter const& f)
512 {
513 mutex_t::scoped_lock l(m_mutex);
514
515 INVARIANT_CHECK;
516
517 m_ip_filter = f;
518
519 // Close connections whose endpoint is filtered
520 // by the new ip-filter
521 for (torrent_map::iterator i = m_torrents.begin()
522 , end(m_torrents.end()); i != end; ++i)
523 i->second->ip_filter_updated();
524 }
525
526 void session_impl::set_settings(session_settings const& s)
527 {
528 mutex_t::scoped_lock l(m_mutex);
529
530 INVARIANT_CHECK;
531
532 TORRENT_ASSERT(s.file_pool_size > 0);
533
534 // less than 5 seconds unchoke interval is insane
535 TORRENT_ASSERT(s.unchoke_interval >= 5);
536 if (m_settings.cache_size != s.cache_size)
537 m_disk_thread.set_cache_size(s.cache_size);
538 if (m_settings.cache_expiry != s.cache_expiry)
539 m_disk_thread.set_cache_size(s.cache_expiry);
540 // if queuing settings were changed, recalculate
541 // queued torrents sooner
542 if ((m_settings.active_downloads != s.active_downloads
543 || m_settings.active_seeds != s.active_seeds
544 || m_settings.active_limit != s.active_limit)
545 && m_auto_manage_time_scaler > 2)
546 m_auto_manage_time_scaler = 2;
547 m_settings = s;
548 if (m_settings.connection_speed <= 0) m_settings.connection_speed = 200;
549
550 m_files.resize(m_settings.file_pool_size);
551 if (!s.auto_upload_slots) m_allowed_upload_slots = m_max_uploads;
552 // replace all occurances of '\n' with ' '.
553 std::string::iterator i = m_settings.user_agent.begin();
554 while ((i = std::find(i, m_settings.user_agent.end(), '\n'))
555 != m_settings.user_agent.end())
556 *i = ' ';
557 }
558
559 tcp::endpoint session_impl::get_ipv6_interface() const
560 {
561 return m_ipv6_interface;
562 }
563
564 session_impl::listen_socket_t session_impl::setup_listener(tcp::endpoint ep
565 , int retries, bool v6_only)
566 {
567 error_code ec;
568 listen_socket_t s;
569 s.sock.reset(new socket_acceptor(m_io_service));
570 s.sock->open(ep.protocol(), ec);
571 s.sock->set_option(socket_acceptor::reuse_address(true), ec);
572 if (ep.protocol() == tcp::v6()) s.sock->set_option(v6only(v6_only), ec);
573 s.sock->bind(ep, ec);
574 while (ec && retries > 0)
575 {
576 ec = error_code();
577 TORRENT_ASSERT(!ec);
578 --retries;
579 ep.port(ep.port() + 1);
580 s.sock->bind(ep, ec);
581 }
582 if (ec)
583 {
584 // instead of giving up, try
585 // let the OS pick a port
586 ep.port(0);
587 ec = error_code();
588 s.sock->bind(ep, ec);
589 }
590 if (ec)
591 {
592 // not even that worked, give up
593 if (m_alerts.should_post<listen_failed_alert>())
594 m_alerts.post_alert(listen_failed_alert(ep, ec));
595 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
596 std::stringstream msg;
597 msg << "cannot bind to interface '";
598 print_endpoint(msg, ep) << "' " << ec.message();
599 (*m_logger) << msg.str() << "\n";
600 #endif
601 return listen_socket_t();
602 }
603 s.external_port = s.sock->local_endpoint(ec).port();
604 s.sock->listen(5, ec);
605 if (ec)
606 {
607 if (m_alerts.should_post<listen_failed_alert>())
608 m_alerts.post_alert(listen_failed_alert(ep, ec));
609 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
610 std::stringstream msg;
611 msg << "cannot listen on interface '";
612 print_endpoint(msg, ep) << "' " << ec.message();
613 (*m_logger) << msg.str() << "\n";
614 #endif
615 return listen_socket_t();
616 }
617
618 if (m_alerts.should_post<listen_succeeded_alert>())
619 m_alerts.post_alert(listen_succeeded_alert(ep));
620
621 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
622 (*m_logger) << "listening on: " << ep
623 << " external port: " << s.external_port << "\n";
624 #endif
625 return s;
626 }
627
628 void session_impl::open_listen_port()
629 {
630 // close the open listen sockets
631 m_listen_sockets.clear();
632 m_incoming_connection = false;
633
634 if (is_any(m_listen_interface.address()))
635 {
636 // this means we should open two listen sockets
637 // one for IPv4 and one for IPv6
638
639 listen_socket_t s = setup_listener(
640 tcp::endpoint(address_v4::any(), m_listen_interface.port())
641 , m_listen_port_retries);
642
643 if (s.sock)
644 {
645 m_listen_sockets.push_back(s);
646 async_accept(s.sock);
647 }
648
649 s = setup_listener(
650 tcp::endpoint(address_v6::any(), m_listen_interface.port())
651 , m_listen_port_retries, true);
652
653 if (s.sock)
654 {
655 m_listen_sockets.push_back(s);
656 async_accept(s.sock);
657 }
658 }
659 else
660 {
661 // we should only open a single listen socket, that
662 // binds to the given interface
663
664 listen_socket_t s = setup_listener(
665 m_listen_interface, m_listen_port_retries);
666
667 if (s.sock)
668 {
669 m_listen_sockets.push_back(s);
670 async_accept(s.sock);
671 }
672 }
673
674 m_ipv6_interface = tcp::endpoint();
675
676 for (std::list<listen_socket_t>::const_iterator i = m_listen_sockets.begin()
677 , end(m_listen_sockets.end()); i != end; ++i)
678 {
679 error_code ec;
680 tcp::endpoint ep = i->sock->local_endpoint(ec);
681 if (ec || ep.address().is_v4()) continue;
682
683 if (ep.address().to_v6() != address_v6::any())
684 {
685 // if we're listening on a specific address
686 // pick it
687 m_ipv6_interface = ep;
688 }
689 else
690 {
691 // if we're listening on any IPv6 address, enumerate them and
692 // pick the first non-local address
693 std::vector<ip_interface> const& ifs = enum_net_interfaces(m_io_service, ec);
694 for (std::vector<ip_interface>::const_iterator i = ifs.begin()
695 , end(ifs.end()); i != end; ++i)
696 {
697 if (i->interface_address.is_v4()
698 || i->interface_address.to_v6().is_link_local()
699 || i->interface_address.to_v6().is_loopback()) continue;
700 m_ipv6_interface = tcp::endpoint(i->interface_address, ep.port());
701 break;
702 }
703 break;
704 }
705 }
706
707 if (!m_listen_sockets.empty())
708 {
709 error_code ec;
710 tcp::endpoint local = m_listen_sockets.front().sock->local_endpoint(ec);
711 if (!ec)
712 {
713 if (m_natpmp.get())
714 {
715 if (m_tcp_mapping[0] != -1) m_natpmp->delete_mapping(m_tcp_mapping[0]);
716 m_tcp_mapping[0] = m_natpmp->add_mapping(natpmp::tcp
717 , local.port(), local.port());
718 }
719 if (m_upnp.get())
720 {
721 if (m_tcp_mapping[1] != -1) m_upnp->delete_mapping(m_tcp_mapping[1]);
722 m_tcp_mapping[1] = m_upnp->add_mapping(upnp::tcp
723 , local.port(), local.port());
724 }
725 }
726 }
727
728 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
729 m_logger = create_log("main_session", listen_port(), false);
730 #endif
731 }
732
733 #ifndef TORRENT_DISABLE_DHT
734
735 void session_impl::on_receive_udp(error_code const& e
736 , udp::endpoint const& ep, char const* buf, int len)
737 {
738 if (e)
739 {
740 if (e == asio::error::connection_refused
741 || e == asio::error::connection_reset
742 || e == asio::error::connection_aborted)
743 m_dht->on_unreachable(ep);
744
745 if (m_alerts.should_post<udp_error_alert>())
746 m_alerts.post_alert(udp_error_alert(ep, e));
747 return;
748 }
749
750 if (len > 20 && *buf == 'd' && m_dht)
751 {
752 // this is probably a dht message
753 m_dht->on_receive(ep, buf, len);
754 }
755 }
756
757 #endif
758
759 void session_impl::async_accept(boost::shared_ptr<socket_acceptor> const& listener)
760 {
761 shared_ptr<socket_type> c(new socket_type(m_io_service));
762 c->instantiate<stream_socket>(m_io_service);
763 listener->async_accept(c->get<stream_socket>()
764 , bind(&session_impl::on_incoming_connection, this, c
765 , boost::weak_ptr<socket_acceptor>(listener), _1));
766 }
767
768 void session_impl::on_incoming_connection(shared_ptr<socket_type> const& s
769 , weak_ptr<socket_acceptor> listen_socket, error_code const& e)
770 {
771 boost::shared_ptr<socket_acceptor> listener = listen_socket.lock();
772 if (!listener) return;
773
774 if (e == asio::error::operation_aborted) return;
775
776 mutex_t::scoped_lock l(m_mutex);
777 if (m_abort) return;
778
779 error_code ec;
780 if (e)
781 {
782 tcp::endpoint ep = listener->local_endpoint(ec);
783 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
784 std::string msg = "error accepting connection on '"
785 + boost::lexical_cast<std::string>(ep) + "' " + e.message();
786 (*m_logger) << msg << "\n";
787 #endif
788 #ifdef TORRENT_WINDOWS
789 // Windows sometimes generates this error. It seems to be
790 // non-fatal and we have to do another async_accept.
791 if (e.value() == ERROR_SEM_TIMEOUT)
792 {
793 async_accept(listener);
794 return;
795 }
796 #endif
797 #ifdef TORRENT_BSD
798 // Leopard sometimes generates an "invalid argument" error. It seems to be
799 // non-fatal and we have to do another async_accept.
800 if (e.value() == EINVAL)
801 {
802 async_accept(listener);
803 return;
804 }
805 #endif
806 if (m_alerts.should_post<listen_failed_alert>())
807 m_alerts.post_alert(listen_failed_alert(ep, e));
808 return;
809 }
810 async_accept(listener);
811
812 // we got a connection request!
813 tcp::endpoint endp = s->remote_endpoint(ec);
814
815 if (ec)
816 {
817 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
818 (*m_logger) << endp << " <== INCOMING CONNECTION FAILED, could "
819 "not retrieve remote endpoint " << ec.message() << "\n";
820 #endif
821 return;
822 }
823
824 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
825 (*m_logger) << time_now_string() << " <== INCOMING CONNECTION " << endp << "\n";
826 #endif
827
828 // local addresses do not count, since it's likely
829 // coming from our own client through local service discovery
830 // and it does not reflect whether or not a router is open
831 // for incoming connections or not.
832 if (!is_local(endp.address()))
833 m_incoming_connection = true;
834
835 if (m_ip_filter.access(endp.address()) & ip_filter::blocked)
836 {
837 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
838 (*m_logger) << "filtered blocked ip\n";
839 #endif
840 if (m_alerts.should_post<peer_blocked_alert>())
841 m_alerts.post_alert(peer_blocked_alert(endp.address()));
842 return;
843 }
844
845 // don't allow more connections than the max setting
846 if (num_connections() >= max_connections())
847 {
848 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
849 (*m_logger) << "number of connections limit exceeded (conns: "
850 << num_connections() << ", limit: " << max_connections()
851 << "), connection rejected\n";
852 #endif
853 return;
854 }
855
856 // check if we have any active torrents
857 // if we don't reject the connection
858 if (m_torrents.empty())
859 {
860 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
861 (*m_logger) << " There are no torrents, disconnect\n";
862 #endif
863 return;
864 }
865
866 bool has_active_torrent = false;
867 for (torrent_map::iterator i = m_torrents.begin()
868 , end(m_torrents.end()); i != end; ++i)
869 {
870 if (!i->second->is_paused())
871 {
872 has_active_torrent = true;
873 break;
874 }
875 }
876 if (!has_active_torrent)
877 {
878 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
879 (*m_logger) << " There are no _active_ torrents, disconnect\n";
880 #endif
881 return;
882 }
883
884 boost::intrusive_ptr<peer_connection> c(
885 new bt_peer_connection(*this, s, endp, 0));
886 #ifndef NDEBUG
887 c->m_in_constructor = false;
888 #endif
889
890 if (!c->is_disconnecting())
891 {
892 m_connections.insert(c);
893 c->start();
894 }
895 }
896 void session_impl::close_connection(peer_connection const* p
897 , char const* message)
898 {
899 mutex_t::scoped_lock l(m_mutex);
900
901 // too expensive
902 // INVARIANT_CHECK;
903
904 #ifndef NDEBUG
905 // for (aux::session_impl::torrent_map::const_iterator i = m_torrents.begin()
906 // , end(m_torrents.end()); i != end; ++i)
907 // TORRENT_ASSERT(!i->second->has_peer((peer_connection*)p));
908 #endif
909
910 #if defined(TORRENT_LOGGING)
911 (*m_logger) << time_now_string() << " CLOSING CONNECTION "
912 << p->remote() << " : " << message << "\n";
913 #endif
914
915 TORRENT_ASSERT(p->is_disconnecting());
916
917 if (!p->is_choked()) --m_num_unchoked;
918 // connection_map::iterator i = std::lower_bound(m_connections.begin(), m_connections.end()
919 // , p, bind(&boost::intrusive_ptr<peer_connection>::get, _1) < p);
920 // if (i->get() != p) i == m_connections.end();
921 connection_map::iterator i = std::find_if(m_connections.begin(), m_connections.end()
922 , bind(&boost::intrusive_ptr<peer_connection>::get, _1) == p);
923 if (i != m_connections.end()) m_connections.erase(i);
924 }
925
926 void session_impl::set_peer_id(peer_id const& id)
927 {
928 mutex_t::scoped_lock l(m_mutex);
929 m_peer_id = id;
930 }
931
932 void session_impl::set_key(int key)
933 {
934 mutex_t::scoped_lock l(m_mutex);
935 m_key = key;
936 }
937
938 int session_impl::next_port()
939 {
940 std::pair<int, int> const& out_ports = m_settings.outgoing_ports;
941 if (m_next_port < out_ports.first || m_next_port > out_ports.second)
942 m_next_port = out_ports.first;
943
944 int port = m_next_port;
945 ++m_next_port;
946 if (m_next_port > out_ports.second) m_next_port = out_ports.first;
947 #if defined TORRENT_LOGGING
948 (*m_logger) << time_now_string() << " *** BINDING OUTGOING CONNECTION [ "
949 "port: " << port << " ]\n";
950 #endif
951 return port;
952 }
953
954 void session_impl::second_tick(error_code const& e)
955 {
956 session_impl::mutex_t::scoped_lock l(m_mutex);
957
958 // too expensive
959 // INVARIANT_CHECK;
960
961 if (m_abort) return;
962
963 if (e)
964 {
965 #if defined TORRENT_LOGGING
966 (*m_logger) << "*** SECOND TIMER FAILED " << e.message() << "\n";
967 #endif
968 ::abort();
969 return;
970 }
971
972 float tick_interval = total_microseconds(time_now() - m_last_tick) / 1000000.f;
973 m_last_tick = time_now();
974
975 error_code ec;
976 m_timer.expires_from_now(seconds(1), ec);
977 m_timer.async_wait(
978 bind(&session_impl::second_tick, this, _1));
979
980 #ifdef TORRENT_STATS
981 ++m_second_counter;
982 int downloading_torrents = 0;
983 int seeding_torrents = 0;
984 for (torrent_map::iterator i = m_torrents.begin()
985 , end(m_torrents.end()); i != end; ++i)
986 {
987 if (i->second->is_seed())
988 ++seeding_torrents;
989 else
990 ++downloading_torrents;
991 }
992 int num_complete_connections = 0;
993 int num_half_open = 0;
994 for (connection_map::iterator i = m_connections.begin()
995 , end(m_connections.end()); i != end; ++i)
996 {
997 if ((*i)->is_connecting())
998 ++num_half_open;
999 else
1000 ++num_complete_connections;
1001 }
1002
1003 m_stats_logger
1004 << m_second_counter << "\t"
1005 << m_stat.upload_rate() << "\t"
1006 << m_stat.download_rate() << "\t"
1007 << downloading_torrents << "\t"
1008 << seeding_torrents << "\t"
1009 << num_complete_connections << "\t"
1010 << num_half_open << "\t"
1011 << m_disk_thread.disk_allocations() << "\t"
1012 << std::endl;
1013 #endif
1014
1015 // --------------------------------------------------------------
1016 // check for incoming connections that might have timed out
1017 // --------------------------------------------------------------
1018
1019 for (connection_map::iterator i = m_connections.begin();
1020 i != m_connections.end();)
1021 {
1022 peer_connection* p = (*i).get();
1023 ++i;
1024 // ignore connections that already have a torrent, since they
1025 // are ticket through the torrents' second_ticket
1026 if (!p->associated_torrent().expired()) continue;
1027 if (m_last_tick - p->connected_time() > seconds(m_settings.handshake_timeout))
1028 p->disconnect("timeout: incoming connection");
1029 }
1030
1031 // --------------------------------------------------------------
1032 // second_tick every torrent
1033 // --------------------------------------------------------------
1034
1035 int congested_torrents = 0;
1036 int uncongested_torrents = 0;
1037
1038 // count the number of seeding torrents vs. downloading
1039 // torrents we are running
1040 int num_seeds = 0;
1041 int num_downloads = 0;
1042
1043 // count the number of peers of downloading torrents
1044 int num_downloads_peers = 0;
1045
1046 torrent_map::iterator least_recently_scraped = m_torrents.begin();
1047 int num_paused_auto_managed = 0;
1048
1049 // check each torrent for tracker updates
1050 // TODO: do this in a timer-event in each torrent instead
1051 for (torrent_map::iterator i = m_torrents.begin();
1052 i != m_torrents.end();)
1053 {
1054 torrent& t = *i->second;
1055 TORRENT_ASSERT(!t.is_aborted());
1056 if (t.bandwidth_queue_size(peer_connection::upload_channel))
1057 ++congested_torrents;
1058 else
1059 ++uncongested_torrents;
1060
1061 if (t.is_auto_managed() && t.is_paused() && !t.has_error())
1062 {
1063 ++num_paused_auto_managed;
1064 if (!least_recently_scraped->second->is_auto_managed()
1065 || !least_recently_scraped->second->is_paused()
1066 || least_recently_scraped->second->last_scrape() > t.last_scrape())
1067 {
1068 least_recently_scraped = i;
1069 }
1070 }
1071
1072 if (t.is_finished())
1073 {
1074 ++num_seeds;
1075 }
1076 else
1077 {
1078 ++num_downloads;
1079 num_downloads_peers += t.num_peers();
1080 }
1081
1082 t.second_tick(m_stat, tick_interval);
1083 ++i;
1084 }
1085
1086 // drain the IP overhead from the bandwidth limiters
1087 m_download_channel.drain(m_stat.download_ip_overhead());
1088 m_upload_channel.drain(m_stat.upload_ip_overhead());
1089
1090 m_stat.second_tick(tick_interval);
1091
1092 // --------------------------------------------------------------
1093 // scrape paused torrents that are auto managed
1094 // --------------------------------------------------------------
1095 if (!is_paused())
1096 {
1097 --m_auto_scrape_time_scaler;
1098 if (m_auto_scrape_time_scaler <= 0)
1099 {
1100 m_auto_scrape_time_scaler = m_settings.auto_scrape_interval
1101 / (std::max)(1, num_paused_auto_managed);
1102 if (m_auto_scrape_time_scaler < m_settings.auto_scrape_min_interval)
1103 m_auto_scrape_time_scaler = m_settings.auto_scrape_min_interval;
1104
1105 if (least_recently_scraped != m_torrents.end())
1106 {
1107 least_recently_scraped->second->scrape_tracker();
1108 }
1109 }
1110 }
1111
1112 // --------------------------------------------------------------
1113 // connect new peers
1114 // --------------------------------------------------------------
1115
1116 // let torrents connect to peers if they want to
1117 // if there are any torrents and any free slots
1118
1119 // this loop will "hand out" max(connection_speed
1120 // , half_open.free_slots()) to the torrents, in a
1121 // round robin fashion, so that every torrent is
1122 // equallt likely to connect to a peer
1123
1124 int free_slots = m_half_open.free_slots();
1125 if (!m_torrents.empty()
1126 && free_slots > -m_half_open.limit()
1127 && num_connections() < m_max_connections)
1128 {
1129 // this is the maximum number of connections we will
1130 // attempt this tick
1131 int max_connections = m_settings.connection_speed;
1132 int average_peers = 0;
1133 if (num_downloads > 0)
1134 average_peers = num_downloads_peers / num_downloads;
1135
1136 torrent_map::iterator i = m_torrents.begin();
1137 if (m_next_connect_torrent < int(m_torrents.size()))
1138 std::advance(i, m_next_connect_torrent);
1139 else
1140 m_next_connect_torrent = 0;
1141 int steps_since_last_connect = 0;
1142 int num_torrents = int(m_torrents.size());
1143 for (;;)
1144 {
1145 torrent& t = *i->second;
1146 if (t.want_more_peers())
1147 {
1148 int connect_points = 100;
1149 // have a bias against torrents with more peers
1150 // than average
1151 if (!t.is_seed() && t.num_peers() > average_peers)
1152 connect_points /= 2;
1153 // if this is a seed and there is a torrent that
1154 // is downloading, lower the rate at which this
1155 // torrent gets connections.
1156 // dividing by num_seeds will have the effect
1157 // that all seed will get as many connections
1158 // together, as a single downloading torrent.
1159 if (t.is_seed() && num_downloads > 0)
1160 connect_points /= num_seeds + 1;
1161 if (connect_points <= 0) connect_points = 1;
1162 t.give_connect_points(connect_points);
1163 try
1164 {
1165 if (t.try_connect_peer())
1166 {
1167 --max_connections;
1168 --free_slots;
1169 steps_since_last_connect = 0;
1170 }
1171 }
1172 catch (std::bad_alloc&)
1173 {
1174 // we ran out of memory trying to connect to a peer
1175 // lower the global limit to the number of peers
1176 // we already have
1177 m_max_connections = num_connections();
1178 if (m_max_connections < 2) m_max_connections = 2;
1179 }
1180 }
1181 ++m_next_connect_torrent;
1182 ++steps_since_last_connect;
1183 ++i;
1184 if (i == m_torrents.end())
1185 {
1186 TORRENT_ASSERT(m_next_connect_torrent == num_torrents);
1187 i = m_torrents.begin();
1188 m_next_connect_torrent = 0;
1189 }
1190 // if we have gone two whole loops without
1191 // handing out a single connection, break
1192 if (steps_since_last_connect > num_torrents * 2) break;
1193 // if there are no more free connection slots, abort
1194 if (free_slots <= -m_half_open.limit()) break;
1195 // if we should not make any more connections
1196 // attempts this tick, abort
1197 if (max_connections == 0) break;
1198 // maintain the global limit on number of connections
1199 if (num_connections() >= m_max_connections) break;
1200 }
1201 }
1202
1203 // --------------------------------------------------------------
1204 // auto managed torrent
1205 // --------------------------------------------------------------
1206 m_auto_manage_time_scaler--;
1207 if (m_auto_manage_time_scaler <= 0)
1208 {
1209 m_auto_manage_time_scaler = settings().auto_manage_interval;
1210 recalculate_auto_managed_torrents();
1211 }
1212
1213 // --------------------------------------------------------------
1214 // unchoke set and optimistic unchoke calculations
1215 // --------------------------------------------------------------
1216 m_unchoke_time_scaler--;
1217 if (m_unchoke_time_scaler <= 0 && !m_connections.empty())
1218 {
1219 m_unchoke_time_scaler = settings().unchoke_interval;
1220 recalculate_unchoke_slots(congested_torrents
1221 , uncongested_torrents);
1222 }
1223
1224 // --------------------------------------------------------------
1225 // disconnect peers when we have too many
1226 // --------------------------------------------------------------
1227 --m_disconnect_time_scaler;
1228 if (m_disconnect_time_scaler <= 0)
1229 {
1230 m_disconnect_time_scaler = 90;
1231
1232 // every 90 seconds, disconnect the worst peers
1233 // if we have reached the connection limit
1234 if (num_connections() >= max_connections() * m_settings.peer_turnover_cutoff
1235 && !m_torrents.empty())
1236 {
1237 torrent_map::iterator i = std::max_element(m_torrents.begin(), m_torrents.end()
1238 , bind(&torrent::num_peers, bind(&torrent_map::value_type::second, _1))
1239 < bind(&torrent::num_peers, bind(&torrent_map::value_type::second, _2)));
1240
1241 TORRENT_ASSERT(i != m_torrents.end());
1242 int peers_to_disconnect = (std::min)((std::max)(int(i->second->num_peers()
1243 * m_settings.peer_turnover), 1)
1244 , i->second->get_policy().num_connect_candidates());
1245 i->second->disconnect_peers(peers_to_disconnect);
1246 }
1247 else
1248 {
1249 // if we haven't reached the global max. see if any torrent
1250 // has reached its local limit
1251 for (torrent_map::iterator i = m_torrents.begin()
1252 , end(m_torrents.end()); i != end; ++i)
1253 {
1254 boost::shared_ptr<torrent> t = i->second;
1255 if (t->num_peers() < t->max_connections() * m_settings.peer_turnover_cutoff)
1256 continue;
1257
1258 int peers_to_disconnect = (std::min)((std::max)(int(i->second->num_peers()
1259 * m_settings.peer_turnover), 1)
1260 , i->second->get_policy().num_connect_candidates());
1261 t->disconnect_peers(peers_to_disconnect);
1262 }
1263 }
1264 }
1265 }
1266
1267 namespace
1268 {
1269 bool is_active(torrent* t, session_settings const& s)
1270 {
1271 return !(s.dont_count_slow_torrents
1272 && t->statistics().upload_payload_rate() == 0.f
1273 && t->statistics().download_payload_rate() == 0.f);
1274 }
1275 }
1276
1277 void session_impl::recalculate_auto_managed_torrents()
1278 {
1279 // these vectors are filled with auto managed torrents
1280 std::vector<torrent*> downloaders;
1281 downloaders.reserve(m_torrents.size());
1282 std::vector<torrent*> seeds;
1283 seeds.reserve(m_torrents.size());
1284
1285 // these counters are set to the number of torrents
1286 // of each kind we're allowed to have active
1287 int num_downloaders = settings().active_downloads;
1288 int num_seeds = settings().active_seeds;
1289 int hard_limit = settings().active_limit;
1290
1291 if (num_downloaders == -1)
1292 num_downloaders = (std::numeric_limits<int>::max)();
1293 if (num_seeds == -1)
1294 num_seeds = (std::numeric_limits<int>::max)();
1295 if (hard_limit == -1)
1296 hard_limit = (std::numeric_limits<int>::max)();
1297
1298 for (torrent_map::iterator i = m_torrents.begin()
1299 , end(m_torrents.end()); i != end; ++i)
1300 {
1301 torrent* t = i->second.get();
1302 TORRENT_ASSERT(t);
1303 if (t->is_auto_managed() && !t->has_error())
1304 {
1305 // this torrent is auto managed, add it to
1306 // the list (depending on if it's a seed or not)
1307 if (t->is_finished())
1308 seeds.push_back(t);
1309 else
1310 downloaders.push_back(t);
1311 }
1312 else if (!t->is_paused())
1313 {
1314 --hard_limit;
1315 if (is_active(t, settings()))
1316 {
1317 // this is not an auto managed torrent,
1318 // if it's running and active, decrease the
1319 // counters.
1320 --num_downloaders;
1321 --num_seeds;
1322 }
1323 }
1324 }
1325
1326 bool handled_by_extension = false;
1327
1328 #ifndef TORRENT_DISABLE_EXTENSIONS
1329 // TODO: allow extensions to sort torrents for queuing
1330 #endif
1331
1332 if (!handled_by_extension)
1333 {
1334 std::sort(downloaders.begin(), downloaders.end()
1335 , bind(&torrent::sequence_number, _1) < bind(&torrent::sequence_number, _2));
1336
1337 std::sort(seeds.begin(), seeds.end()
1338 , bind(&torrent::seed_rank, _1, boost::ref(m_settings))
1339 > bind(&torrent::seed_rank, _2, boost::ref(m_settings)));
1340 }
1341
1342 for (std::vector<torrent*>::iterator i = downloaders.begin()
1343 , end(downloaders.end()); i != end; ++i)
1344 {
1345 torrent* t = *i;
1346 if (!t->is_paused() && !is_active(t, settings()) && hard_limit > 0)
1347 {
1348 --hard_limit;
1349 continue;
1350 }
1351
1352 if (num_downloaders > 0 && hard_limit > 0)
1353 {
1354 --hard_limit;
1355 if (t->state() != torrent_status::queued_for_checking
1356 && t->state() != torrent_status::checking_files)
1357 {
1358 --num_downloaders;
1359 if (t->is_paused()) t->resume();
1360 }
1361 }
1362 else
1363 {
1364 if (!t->is_paused()) t->pause();
1365 }
1366 }
1367
1368 for (std::vector<torrent*>::iterator i = seeds.begin()
1369 , end(seeds.end()); i != end; ++i)
1370 {
1371 torrent* t = *i;
1372 if (!t->is_paused() && !is_active(t, settings()) && hard_limit > 0)
1373 {
1374 --hard_limit;
1375 continue;
1376 }
1377
1378 if (num_seeds > 0 && hard_limit > 0)
1379 {
1380 --hard_limit;
1381 --num_seeds;
1382 if (t->is_paused()) t->resume();
1383 }
1384 else
1385 {
1386 if (!t->is_paused()) t->pause();
1387 }
1388 }
1389 }
1390
1391 void session_impl::recalculate_unchoke_slots(int congested_torrents
1392 , int uncongested_torrents)
1393 {
1394 std::vector<peer_connection*> peers;
1395 for (connection_map::iterator i = m_connections.begin()
1396 , end(m_connections.end()); i != end; ++i)
1397 {
1398 peer_connection* p = i->get();
1399 torrent* t = p->associated_torrent().lock().get();
1400 if (!p->peer_info_struct()
1401 || t == 0
1402 || !p->is_peer_interested()
1403 || p->is_disconnecting()
1404 || p->is_connecting()
1405 || (p->share_diff() < -free_upload_amount
1406 && !t->is_seed()))
1407 {
1408 if (!(*i)->is_choked() && t)
1409 {
1410 policy::peer* pi = p->peer_info_struct();
1411 if (pi && pi->optimistically_unchoked)
1412 {
1413 pi->optimistically_unchoked = false;
1414 // force a new optimistic unchoke
1415 m_optimistic_unchoke_time_scaler = 0;
1416 }
1417 t->choke_peer(*(*i));
1418 }
1419 continue;
1420 }
1421 peers.push_back(i->get());
1422 }
1423
1424 // sorts the peers that are eligible for unchoke by download rate and secondary
1425 // by total upload. The reason for this is, if all torrents are being seeded,
1426 // the download rate will be 0, and the peers we have sent the least to should
1427 // be unchoked
1428 std::sort(peers.begin(), peers.end()
1429 , bind(&peer_connection::unchoke_compare, _1, _2));
1430
1431 std::for_each(m_connections.begin(), m_connections.end()
1432 , bind(&peer_connection::reset_choke_counters, _1));
1433
1434 // auto unchoke
1435 int upload_limit = m_bandwidth_manager[peer_connection::upload_channel]->throttle();
1436 if (m_settings.auto_upload_slots && upload_limit != bandwidth_limit::inf)
1437 {
1438 // if our current upload rate is less than 90% of our
1439 // limit AND most torrents are not "congested", i.e.
1440 // they are not holding back because of a per-torrent
1441 // limit
1442 if (m_stat.upload_rate() < upload_limit * 0.9f
1443 && m_allowed_upload_slots <= m_num_unchoked + 1
1444 && congested_torrents < uncongested_torrents)
1445 {
1446 ++m_allowed_upload_slots;
1447 }
1448 else if (m_upload_channel.queue_size() > 1
1449 && m_allowed_upload_slots > m_max_uploads)
1450 {
1451 --m_allowed_upload_slots;
1452 }
1453 }
1454
1455 // reserve one upload slot for optimistic unchokes
1456 int unchoke_set_size = m_allowed_upload_slots - 1;
1457
1458 m_num_unchoked = 0;
1459 // go through all the peers and unchoke the first ones and choke
1460 // all the other ones.
1461 for (std::vector<peer_connection*>::iterator i = peers.begin()
1462 , end(peers.end()); i != end; ++i)
1463 {
1464 peer_connection* p = *i;
1465 TORRENT_ASSERT(p);
1466 torrent* t = p->associated_torrent().lock().get();
1467 TORRENT_ASSERT(t);
1468 if (unchoke_set_size > 0)
1469 {
1470 if (p->is_choked())
1471 {
1472 if (!t->unchoke_peer(*p))
1473 continue;
1474 }
1475
1476 --unchoke_set_size;
1477 ++m_num_unchoked;
1478
1479 TORRENT_ASSERT(p->peer_info_struct());
1480 if (p->peer_info_struct()->optimistically_unchoked)
1481 {
1482 // force a new optimistic unchoke
1483 m_optimistic_unchoke_time_scaler = 0;
1484 p->peer_info_struct()->optimistically_unchoked = false;
1485 }
1486 }
1487 else
1488 {
1489 TORRENT_ASSERT(p->peer_info_struct());
1490 if (!p->is_choked() && !p->peer_info_struct()->optimistically_unchoked)
1491 t->choke_peer(*p);
1492 if (!p->is_choked())
1493 ++m_num_unchoked;
1494 }
1495 }
1496
1497 m_optimistic_unchoke_time_scaler--;
1498 if (m_optimistic_unchoke_time_scaler <= 0)
1499 {
1500 m_optimistic_unchoke_time_scaler
1501 = settings().optimistic_unchoke_multiplier;
1502
1503 // find the peer that has been waiting the longest to be optimistically
1504 // unchoked
1505 connection_map::iterator current_optimistic_unchoke = m_connections.end();
1506 connection_map::iterator optimistic_unchoke_candidate = m_connections.end();
1507 ptime last_unchoke = max_time();
1508
1509 for (connection_map::iterator i = m_connections.begin()
1510 , end(m_connections.end()); i != end; ++i)
1511 {
1512 peer_connection* p = i->get();
1513 TORRENT_ASSERT(p);
1514 policy::peer* pi = p->peer_info_struct();
1515 if (!pi) continue;
1516 torrent* t = p->associated_torrent().lock().get();
1517 if (!t) continue;
1518
1519 if (pi->optimistically_unchoked)
1520 {
1521 TORRENT_ASSERT(!p->is_choked());
1522 TORRENT_ASSERT(current_optimistic_unchoke == m_connections.end());
1523 current_optimistic_unchoke = i;
1524 }
1525
1526 if (pi->last_optimistically_unchoked < last_unchoke
1527 && !p->is_connecting()
1528 && !p->is_disconnecting()
1529 && p->is_peer_interested()
1530 && t->free_upload_slots()
1531 && p->is_choked())
1532 {
1533 last_unchoke = pi->last_optimistically_unchoked;
1534 optimistic_unchoke_candidate = i;
1535 }
1536 }
1537
1538 if (optimistic_unchoke_candidate != m_connections.end()
1539 && optimistic_unchoke_candidate != current_optimistic_unchoke)
1540 {
1541 if (current_optimistic_unchoke != m_connections.end())
1542 {
1543 torrent* t = (*current_optimistic_unchoke)->associated_torrent().lock().get();
1544 TORRENT_ASSERT(t);
1545 (*current_optimistic_unchoke)->peer_info_struct()->optimistically_unchoked = false;
1546 t->choke_peer(*current_optimistic_unchoke->get());
1547 }
1548 else
1549 {
1550 ++m_num_unchoked;
1551 }
1552
1553 torrent* t = (*optimistic_unchoke_candidate)->associated_torrent().lock().get();
1554 TORRENT_ASSERT(t);
1555 bool ret = t->unchoke_peer(*optimistic_unchoke_candidate->get());
1556 TORRENT_ASSERT(ret);
1557 (*optimistic_unchoke_candidate)->peer_info_struct()->optimistically_unchoked = true;
1558 }
1559 }
1560 }
1561
1562 void session_impl::operator()()
1563 {
1564 eh_initializer();
1565
1566 {
1567 session_impl::mutex_t::scoped_lock l(m_mutex);
1568 if (m_listen_interface.port() != 0) open_listen_port();
1569 }
1570
1571 ptime timer = time_now();
1572
1573 do
1574 {
1575 #ifndef BOOST_NO_EXCEPTIONS
1576 try
1577 {
1578 #endif
1579 m_io_service.run();
1580 TORRENT_ASSERT(m_abort == true);
1581 #ifndef BOOST_NO_EXCEPTIONS
1582 }
1583 catch (std::exception& e)
1584 {
1585 #ifndef NDEBUG
1586 std::cerr << e.what() << "\n";
1587 std::string err = e.what();
1588 #endif
1589 TORRENT_ASSERT(false);
1590 }
1591 #endif
1592 }
1593 while (!m_abort);
1594
1595 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
1596 (*m_logger) << time_now_string() << " locking mutex\n";
1597 #endif
1598
1599 session_impl::mutex_t::scoped_lock l(m_mutex);
1600 /*
1601 #ifndef NDEBUG
1602 for (torrent_map::iterator i = m_torrents.begin();
1603 i != m_torrents.end(); ++i)
1604 {
1605 TORRENT_ASSERT(i->second->num_peers() == 0);
1606 }
1607 #endif
1608 */
1609 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
1610 (*m_logger) << time_now_string() << " cleaning up torrents\n";
1611 #endif
1612 m_torrents.clear();
1613
1614 TORRENT_ASSERT(m_torrents.empty());
1615 TORRENT_ASSERT(m_connections.empty());
1616 }
1617
1618
1619 // the return value from this function is valid only as long as the
1620 // session is locked!
1621 boost::weak_ptr<torrent> session_impl::find_torrent(sha1_hash const& info_hash)
1622 {
1623 std::map<sha1_hash, boost::shared_ptr<torrent> >::iterator i
1624 = m_torrents.find(info_hash);
1625 #ifndef NDEBUG
1626 for (std::map<sha1_hash, boost::shared_ptr<torrent> >::iterator j
1627 = m_torrents.begin(); j != m_torrents.end(); ++j)
1628 {
1629 torrent* p = boost::get_pointer(j->second);
1630 TORRENT_ASSERT(p);
1631 }
1632 #endif
1633 if (i != m_torrents.end()) return i->second;
1634 return boost::weak_ptr<torrent>();
1635 }
1636
1637 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
1638 boost::shared_ptr<logger> session_impl::create_log(std::string const& name
1639 , int instance, bool append)
1640 {
1641 // current options are file_logger, cout_logger and null_logger
1642 return boost::shared_ptr<logger>(new logger(m_logpath, name + ".log", instance, append));
1643 }
1644 #endif
1645
1646 std::vector<torrent_handle> session_impl::get_torrents()
1647 {
1648 mutex_t::scoped_lock l(m_mutex);
1649 std::vector<torrent_handle> ret;
1650
1651 for (session_impl::torrent_map::iterator i
1652 = m_torrents.begin(), end(m_torrents.end());
1653 i != end; ++i)
1654 {
1655 if (i->second->is_aborted()) continue;
1656 ret.push_back(torrent_handle(i->second));
1657 }
1658 return ret;
1659 }
1660
1661 torrent_handle session_impl::find_torrent_handle(sha1_hash const& info_hash)
1662 {
1663 return torrent_handle(find_torrent(info_hash));
1664 }
1665
1666 torrent_handle session_impl::add_torrent(add_torrent_params const& params)
1667 {
1668 TORRENT_ASSERT(!params.save_path.empty());
1669
1670 if (params.ti && params.ti->files().num_files() == 0)
1671 {
1672 #ifndef BOOST_NO_EXCEPTIONS
1673 throw std::runtime_error("no files in torrent");
1674 #else
1675 return torrent_handle();
1676 #endif
1677 }
1678
1679 // lock the session and the checker thread (the order is important!)
1680 mutex_t::scoped_lock l(m_mutex);
1681
1682 // INVARIANT_CHECK;
1683
1684 if (is_aborted())
1685 {
1686 #ifndef BOOST_NO_EXCEPTIONS
1687 throw std::runtime_error("session is closing");
1688 #else
1689 return torrent_handle();
1690 #endif
1691 }
1692
1693 // figure out the info hash of the torrent
1694 sha1_hash const* ih = 0;
1695 if (params.ti) ih = &params.ti->info_hash();
1696 else ih = &params.info_hash;
1697
1698 // is the torrent already active?
1699 boost::shared_ptr<torrent> torrent_ptr = find_torrent(*ih).lock();
1700 if (torrent_ptr)
1701 {
1702 if (!params.duplicate_is_error)
1703 return torrent_handle(torrent_ptr);
1704
1705 #ifndef BOOST_NO_EXCEPTIONS
1706 throw duplicate_torrent();
1707 #else
1708 return torrent_handle();
1709 #endif
1710 }
1711
1712 int queue_pos = 0;
1713 for (torrent_map::const_iterator i = m_torrents.begin()
1714 , end(m_torrents.end()); i != end; ++i)
1715 {
1716 int pos = i->second->queue_position();
1717 if (pos >= queue_pos) queue_pos = pos + 1;
1718 }
1719
1720 // create the torrent and the data associated with
1721 // the checker thread and store it before starting
1722 // the thread
1723 if (params.ti)
1724 {
1725 torrent_ptr.reset(new torrent(*this, params.ti, params.save_path
1726 , m_listen_interface, params.storage_mode, 16 * 1024
1727 , params.storage, params.paused, params.resume_data
1728 , queue_pos, params.auto_managed));
1729 }
1730 else
1731 {
1732 torrent_ptr.reset(new torrent(*this, params.tracker_url, *ih, params.name
1733 , params.save_path, m_listen_interface, params.storage_mode, 16 * 1024
1734 , params.storage, params.paused, params.resume_data
1735 , queue_pos, params.auto_managed));
1736 }
1737 torrent_ptr->start();
1738
1739 #ifndef TORRENT_DISABLE_EXTENSIONS
1740 for (extension_list_t::iterator i = m_extensions.begin()
1741 , end(m_extensions.end()); i != end; ++i)
1742 {
1743 boost::shared_ptr<torrent_plugin> tp((*i)(torrent_ptr.get(), params.userdata));
1744 if (tp) torrent_ptr->add_extension(tp);
1745 }
1746 #endif
1747
1748 #ifndef TORRENT_DISABLE_DHT
1749 if (m_dht && params.ti)
1750 {
1751 torrent_info::nodes_t const& nodes = params.ti->nodes();
1752 std::for_each(nodes.begin(), nodes.end(), bind(
1753 (void(dht::dht_tracker::*)(std::pair<std::string, int> const&))
1754 &dht::dht_tracker::add_node
1755 , boost::ref(m_dht), _1));
1756 }
1757 #endif
1758
1759 m_torrents.insert(std::make_pair(*ih, torrent_ptr));
1760
1761 // if this is an auto managed torrent, force a recalculation
1762 // of which torrents to have active
1763 if (params.auto_managed && m_auto_manage_time_scaler > 2)
1764 m_auto_manage_time_scaler = 2;
1765
1766 return torrent_handle(torrent_ptr);
1767 }
1768
1769 void session_impl::check_torrent(boost::shared_ptr<torrent> const& t)
1770 {
1771 if (m_abort) return;
1772 if (m_queued_for_checking.empty()) t->start_checking();
1773 m_queued_for_checking.push_back(t);
1774 }
1775
1776 void session_impl::done_checking(boost::shared_ptr<torrent> const& t)
1777 {
1778 if (m_queued_for_checking.empty()) return;
1779 check_queue_t::iterator next_check = m_queued_for_checking.begin();
1780 check_queue_t::iterator done = m_queued_for_checking.end();
1781 for (check_queue_t::iterator i = m_queued_for_checking.begin()
1782 , end(m_queued_for_checking.end()); i != end; ++i)
1783 {
1784 if (*i == t) done = i;
1785 if (next_check == done || (*next_check)->queue_position() > (*i)->queue_position())
1786 next_check = i;
1787 }
1788 if (next_check != done) (*next_check)->start_checking();
1789 m_queued_for_checking.erase(done);
1790 }
1791
1792 void session_impl::remove_torrent(const torrent_handle& h, int options)
1793 {
1794 boost::shared_ptr<torrent> tptr = h.m_torrent.lock();
1795 if (!tptr)
1796 #ifdef BOOST_NO_EXCEPTIONS
1797 return;
1798 #else
1799 throw invalid_handle();
1800 #endif
1801
1802 mutex_t::scoped_lock l(m_mutex);
1803
1804 INVARIANT_CHECK;
1805
1806 session_impl::torrent_map::iterator i =
1807 m_torrents.find(tptr->torrent_file().info_hash());
1808
1809 if (i != m_torrents.end())
1810 {
1811 torrent& t = *i->second;
1812 if (options & session::delete_files)
1813 t.delete_files();
1814 t.abort();
1815
1816 #ifndef NDEBUG
1817 sha1_hash i_hash = t.torrent_file().info_hash();
1818 #endif
1819 i->second->set_queue_position(-1);
1820 m_torrents.erase(i);
1821 TORRENT_ASSERT(m_torrents.find(i_hash) == m_torrents.end());
1822 return;
1823 }
1824 }
1825
1826 bool session_impl::listen_on(
1827 std::pair<int, int> const& port_range
1828 , const char* net_interface)
1829 {
1830 session_impl::mutex_t::scoped_lock l(m_mutex);
1831
1832 INVARIANT_CHECK;
1833
1834 tcp::endpoint new_interface;
1835 if (net_interface && std::strlen(net_interface) > 0)
1836 new_interface = tcp::endpoint(address::from_string(net_interface), port_range.first);
1837 else
1838 new_interface = tcp::endpoint(address_v4::any(), port_range.first);
1839
1840 m_listen_port_retries = port_range.second - port_range.first;
1841
1842 // if the interface is the same and the socket is open
1843 // don't do anything
1844 if (new_interface == m_listen_interface
1845 && !m_listen_sockets.empty()) return true;
1846
1847 m_listen_interface = new_interface;
1848
1849 open_listen_port();
1850
1851 bool new_listen_address = m_listen_interface.address() != new_interface.address();
1852
1853 #ifndef TORRENT_DISABLE_DHT
1854 if ((new_listen_address || m_dht_same_port) && m_dht)
1855 {
1856 if (m_dht_same_port)
1857 m_dht_settings.service_port = new_interface.port();
1858 // the listen interface changed, rebind the dht listen socket as well
1859 m_dht_socket.bind(m_dht_settings.service_port);
1860 if (m_natpmp.get())
1861 {
1862 if (m_udp_mapping[0] != -1) m_natpmp->delete_mapping(m_udp_mapping[0]);
1863 m_udp_mapping[0] = m_natpmp->add_mapping(natpmp::tcp
1864 , m_dht_settings.service_port
1865 , m_dht_settings.service_port);
1866 }
1867 if (m_upnp.get())
1868 {
1869 if (m_udp_mapping[1] != -1) m_upnp->delete_mapping(m_udp_mapping[1]);
1870 m_udp_mapping[1] = m_upnp->add_mapping(upnp::tcp
1871 , m_dht_settings.service_port
1872 , m_dht_settings.service_port);
1873 }
1874 }
1875 #endif
1876
1877 #if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING || defined TORRENT_ERROR_LOGGING
1878 m_logger = create_log("main_session", listen_port(), false);
1879 (*m_logger) << time_now_string() << "\n";
1880 #endif
1881
1882 return !m_listen_sockets.empty();
1883 }
1884
1885 unsigned short session_impl::listen_port() const
1886 {
1887 mutex_t::scoped_lock l(m_mutex);
1888 if (m_listen_sockets.empty()) return 0;
1889 return m_listen_sockets.front().external_port;;
1890 }
1891
1892 void session_impl::announce_lsd(sha1_hash const& ih)
1893 {
1894 mutex_t::scoped_lock l(m_mutex);
1895 // use internal listen port for local peers
1896 if (m_lsd.get())
1897 m_lsd->announce(ih, m_listen_interface.port());
1898 }
1899
1900 void session_impl::on_lsd_peer(tcp::endpoint peer, sha1_hash const& ih)
1901 {
1902 mutex_t::scoped_lock l(m_mutex);
1903
1904 INVARIANT_CHECK;
1905
1906 boost::shared_ptr<torrent> t = find_torrent(ih).lock();
1907 if (!t) return;
1908 // don't add peers from lsd to private torrents
1909 if (t->torrent_file().priv()) return;
1910
1911 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
1912 (*m_logger) << time_now_string()
1913 << ": added peer from local discovery: " << peer << "\n";
1914 #endif
1915 t->get_policy().peer_from_tracker(peer, peer_id(0), peer_info::lsd, 0);
1916 }
1917
1918 void session_impl::on_port_mapping(int mapping, int port
1919 , std::string const& errmsg, int map_transport)
1920 {
1921 #ifndef TORRENT_DISABLE_DHT
1922 if (mapping == m_udp_mapping[map_transport] && port != 0)
1923 {
1924 m_external_udp_port = port;
1925 m_dht_settings.service_port = port;
1926 if (m_alerts.should_post<portmap_alert>())
1927 m_alerts.post_alert(portmap_alert(mapping, port
1928 , map_transport));
1929 return;
1930 }
1931 #endif
1932
1933 if (mapping == m_tcp_mapping[map_transport] && port != 0)
1934 {
1935 if (!m_listen_sockets.empty())
1936 m_listen_sockets.front().external_port = port;
1937 if (m_alerts.should_post<portmap_alert>())
1938 m_alerts.post_alert(portmap_alert(mapping, port
1939 , map_transport));
1940 return;
1941 }
1942
1943 if (!errmsg.empty())
1944 {
1945 if (m_alerts.should_post<portmap_error_alert>())
1946 m_alerts.post_alert(portmap_error_alert(mapping
1947 , map_transport, errmsg));
1948 }
1949 else
1950 {
1951 if (m_alerts.should_post<portmap_alert>())
1952 m_alerts.post_alert(portmap_alert(mapping, port
1953 , map_transport));
1954 }
1955 }
1956
1957 session_status session_impl::status() const
1958 {
1959 mutex_t::scoped_lock l(m_mutex);
1960
1961 // INVARIANT_CHECK;
1962
1963 session_status s;
1964
1965 s.num_peers = (int)m_connections.size();
1966 s.num_unchoked = m_num_unchoked;
1967 s.allowed_upload_slots = m_allowed_upload_slots;
1968
1969 s.total_redundant_bytes = m_total_redundant_bytes;
1970 s.total_failed_bytes = m_total_failed_bytes;
1971
1972 s.up_bandwidth_queue = m_upload_channel.queue_size();
1973 s.down_bandwidth_queue = m_download_channel.queue_size();
1974
1975 s.has_incoming_connections = m_incoming_connection;
1976
1977 s.download_rate = m_stat.download_rate();
1978 s.upload_rate = m_stat.upload_rate();
1979
1980 s.payload_download_rate = m_stat.download_payload_rate();
1981 s.payload_upload_rate = m_stat.upload_payload_rate();
1982
1983 s.total_download = m_stat.total_protocol_download()
1984 + m_stat.total_payload_download();
1985
1986 s.total_upload = m_stat.total_protocol_upload()
1987 + m_stat.total_payload_upload();
1988
1989 s.total_payload_download = m_stat.total_payload_download();
1990 s.total_payload_upload = m_stat.total_payload_upload();
1991
1992 #ifndef TORRENT_DISABLE_DHT
1993 if (m_dht)
1994 {
1995 m_dht->dht_status(s);
1996 }
1997 else
1998 {
1999 s.dht_nodes = 0;
2000 s.dht_node_cache = 0;
2001 s.dht_torrents = 0;
2002 s.dht_global_nodes = 0;
2003 }
2004 #endif
2005
2006 return s;
2007 }
2008
2009 #ifndef TORRENT_DISABLE_DHT
2010
2011 void session_impl::start_dht(entry const& startup_state)
2012 {
2013 mutex_t::scoped_lock l(m_mutex);
2014
2015 INVARIANT_CHECK;
2016
2017 if (m_dht)
2018 {
2019 m_dht->stop();
2020 m_dht = 0;
2021 }
2022 if (m_dht_settings.service_port == 0
2023 || m_dht_same_port)
2024 {
2025 m_dht_same_port = true;
2026 // if you hit this assert you are trying to start the
2027 // DHT with the same port as the tcp listen port
2028 // (which is default) _before_ you have opened the
2029 // tcp listen port (so there is no configured port to use)
2030 // basically, make sure you call listen_on() before
2031 // start_dht(). See documentation for listen_on() for
2032 // more information.
2033 TORRENT_ASSERT(m_listen_interface.port() > 0);
2034 m_dht_settings.service_port = m_listen_interface.port();
2035 }
2036 m_external_udp_port = m_dht_settings.service_port;
2037 if (m_natpmp.get() && m_udp_mapping[0] == -1)
2038 {
2039 m_udp_mapping[0] = m_natpmp->add_mapping(natpmp::udp
2040 , m_dht_settings.service_port
2041 , m_dht_settings.service_port);
2042 }
2043 if (m_upnp.get() && m_udp_mapping[1] == -1)
2044 {
2045 m_udp_mapping[1] = m_upnp->add_mapping(upnp::udp
2046 , m_dht_settings.service_port
2047 , m_dht_settings.service_port);
2048 }
2049 m_dht = new dht::dht_tracker(m_dht_socket, m_dht_settings, startup_state);
2050 if (!m_dht_socket.is_open() || m_dht_socket.local_port() != m_dht_settings.service_port)
2051 {
2052 m_dht_socket.bind(m_dht_settings.service_port);
2053 }
2054 }
2055
2056 void session_impl::stop_dht()
2057 {
2058 mutex_t::scoped_lock l(m_mutex);
2059 if (!m_dht) return;
2060 m_dht->stop();
2061 m_dht = 0;
2062 }
2063
2064 void session_impl::set_dht_settings(dht_settings const& settings)
2065 {
2066 mutex_t::scoped_lock l(m_mutex);
2067 // only change the dht listen port in case the settings
2068 // contains a vaiid port, and if it is different from
2069 // the current setting
2070 if (settings.service_port != 0)
2071 m_dht_same_port = false;
2072 else
2073 m_dht_same_port = true;
2074 if (!m_dht_same_port
2075 && settings.service_port != m_dht_settings.service_port
2076 && m_dht)
2077 {
2078 m_dht_socket.bind(settings.service_port);
2079
2080 if (m_natpmp.get())
2081 {
2082 if (m_udp_mapping[0] != -1) m_upnp->delete_mapping(m_udp_mapping[0]);
2083 m_udp_mapping[0] = m_natpmp->add_mapping(natpmp::udp
2084 , m_dht_settings.service_port
2085 , m_dht_settings.service_port);
2086 }
2087 if (m_upnp.get())
2088 {
2089 if (m_udp_mapping[1] != -1) m_upnp->delete_mapping(m_udp_mapping[1]);
2090 m_udp_mapping[1] = m_upnp->add_mapping(upnp::udp
2091 , m_dht_settings.service_port
2092 , m_dht_settings.service_port);
2093 }
2094 m_external_udp_port = settings.service_port;
2095 }
2096 m_dht_settings = settings;
2097 if (m_dht_same_port)
2098 m_dht_settings.service_port = m_listen_interface.port();
2099 }
2100
2101 entry session_impl::dht_state() const
2102 {
2103 mutex_t::scoped_lock l(m_mutex);
2104 if (!m_dht) return entry();
2105 return m_dht->state();
2106 }
2107
2108 void session_impl::add_dht_node(std::pair<std::string, int> const& node)
2109 {
2110 TORRENT_ASSERT(m_dht);
2111 mutex_t::scoped_lock l(m_mutex);
2112 m_dht->add_node(node);
2113 }
2114
2115 void session_impl::add_dht_router(std::pair<std::string, int> const& node)
2116 {
2117 TORRENT_ASSERT(m_dht);
2118 mutex_t::scoped_lock l(m_mutex);
2119 m_dht->add_router_node(node);
2120 }
2121
2122 #endif
2123
2124 #ifndef TORRENT_DISABLE_ENCRYPTION
2125 void session_impl::set_pe_settings(pe_settings const& settings)
2126 {
2127 mutex_t::scoped_lock l(m_mutex);
2128 m_pe_settings = settings;
2129 }
2130 #endif
2131
2132 bool session_impl::is_listening() const
2133 {
2134 mutex_t::scoped_lock l(m_mutex);
2135 return !m_listen_sockets.empty();
2136 }
2137
2138 session_impl::~session_impl()
2139 {
2140 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
2141 (*m_logger) << time_now_string() << "\n\n *** shutting down session *** \n\n";
2142 #endif
2143 abort();
2144
2145 #ifndef TORRENT_DISABLE_GEO_IP
2146 if (m_asnum_db) GeoIP_delete(m_asnum_db);
2147 if (m_country_db) GeoIP_delete(m_country_db);
2148 #endif
2149 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
2150 (*m_logger) << time_now_string() << " waiting for main thread\n";
2151 #endif
2152 m_thread->join();
2153
2154 TORRENT_ASSERT(m_torrents.empty());
2155
2156 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
2157 (*m_logger) << time_now_string() << " waiting for disk io thread\n";
2158 #endif
2159 m_disk_thread.join();
2160
2161 TORRENT_ASSERT(m_torrents.empty());
2162 TORRENT_ASSERT(m_connections.empty());
2163 #if defined(TORRENT_VERBOSE_LOGGING) || defined(TORRENT_LOGGING)
2164 (*m_logger) << time_now_string() << " shutdown complete!\n";
2165 #endif
2166 }
2167
2168 void session_impl::set_max_uploads(int limit)
2169 {
2170 TORRENT_ASSERT(limit > 0 || limit == -1);
2171 mutex_t::scoped_lock l(m_mutex);
2172
2173 INVARIANT_CHECK;
2174
2175 if (limit <= 0) limit = (std::numeric_limits<int>::max)();
2176 if (m_max_uploads == limit) return;
2177 m_max_uploads = limit;
2178 m_allowed_upload_slots = limit;
2179 }
2180
2181 void session_impl::set_max_connections(int limit)
2182 {
2183 TORRENT_ASSERT(limit > 0 || limit == -1);
2184 mutex_t::scoped_lock l(m_mutex);
2185
2186 INVARIANT_CHECK;
2187
2188 if (limit <= 0)
2189 {
2190 limit = (std::numeric_limits<int>::max)();
2191 #ifndef TORRENT_WINDOWS
2192 rlimit l;
2193 if (getrlimit(RLIMIT_NOFILE, &l) == 0
2194 && l.rlim_cur != RLIM_INFINITY)
2195 {
2196 limit = l.rlim_cur - m_settings.file_pool_size;
2197 if (limit < 5) limit = 5;
2198 }
2199 #endif
2200 }
2201 m_max_connections = limit;
2202 }
2203
2204 void session_impl::set_max_half_open_connections(int limit)
2205 {
2206 TORRENT_ASSERT(limit > 0 || limit == -1);
2207 mutex_t::scoped_lock l(m_mutex);
2208
2209 INVARIANT_CHECK;
2210
2211 if (limit <= 0) limit = (std::numeric_limits<int>::max)();
2212 m_half_open.limit(limit);
2213 }
2214
2215 void session_impl::set_download_rate_limit(int bytes_per_second)
2216 {
2217 TORRENT_ASSERT(bytes_per_second > 0 || bytes_per_second == -1);
2218 mutex_t::scoped_lock l(m_mutex);
2219
2220 INVARIANT_CHECK;
2221
2222 if (bytes_per_second <= 0) bytes_per_second = bandwidth_limit::inf;
2223 m_bandwidth_manager[peer_connection::download_channel]->throttle(bytes_per_second);
2224 }
2225
2226 void session_impl::set_upload_rate_limit(int bytes_per_second)
2227 {
2228 TORRENT_ASSERT(bytes_per_second > 0 || bytes_per_second == -1);
2229 mutex_t::scoped_lock l(m_mutex);
2230
2231 INVARIANT_CHECK;
2232
2233 if (bytes_per_second <= 0) bytes_per_second = bandwidth_limit::inf;
2234 m_bandwidth_manager[peer_connection::upload_channel]->throttle(bytes_per_second);
2235 }
2236
2237 std::auto_ptr<alert> session_impl::pop_alert()
2238 {
2239 mutex_t::scoped_lock l(m_mutex);
2240
2241 // too expensive
2242 // INVARIANT_CHECK;
2243
2244 if (m_alerts.pending())
2245 return m_alerts.get();
2246 return std::auto_ptr<alert>(0);
2247 }
2248
2249 alert const* session_impl::wait_for_alert(time_duration max_wait)
2250 {
2251 return m_alerts.wait_for_alert(max_wait);
2252 }
2253
2254 void session_impl::set_alert_mask(int m)
2255 {
2256 mutex_t::scoped_lock l(m_mutex);
2257 m_alerts.set_alert_mask(m);
2258 }
2259
2260 int session_impl::upload_rate_limit() const
2261 {
2262 mutex_t::scoped_lock l(m_mutex);
2263
2264 INVARIANT_CHECK;
2265
2266 int ret = m_bandwidth_manager[peer_connection::upload_channel]->throttle();
2267 return ret == (std::numeric_limits<int>::max)() ? -1 : ret;
2268 }
2269
2270 int session_impl::download_rate_limit() const
2271 {
2272 mutex_t::scoped_lock l(m_mutex);
2273 int ret = m_bandwidth_manager[peer_connection::download_channel]->throttle();
2274 return ret == (std::numeric_limits<int>::max)() ? -1 : ret;
2275 }
2276
2277 void session_impl::start_lsd()
2278 {
2279 mutex_t::scoped_lock l(m_mutex);
2280
2281 INVARIANT_CHECK;
2282
2283 if (m_lsd) return;
2284
2285 m_lsd = new lsd(m_io_service
2286 , m_listen_interface.address()
2287 , bind(&session_impl::on_lsd_peer, this, _1, _2));
2288 }
2289
2290 natpmp* session_impl::start_natpmp()
2291 {
2292 mutex_t::scoped_lock l(m_mutex);
2293
2294 INVARIANT_CHECK;
2295
2296 if (m_natpmp) return m_natpmp.get();
2297
2298 m_natpmp = new natpmp(m_io_service
2299 , m_listen_interface.address()
2300 , bind(&session_impl::on_port_mapping
2301 , this, _1, _2, _3, 0));
2302
2303 if (m_listen_interface.port() > 0)
2304 {
2305 m_tcp_mapping[0] = m_natpmp->add_mapping(natpmp::tcp
2306 , m_listen_interface.port(), m_listen_interface.port());
2307 }
2308 #ifndef TORRENT_DISABLE_DHT
2309 if (m_dht)
2310 m_udp_mapping[0] = m_natpmp->add_mapping(natpmp::udp
2311 , m_dht_settings.service_port
2312 , m_dht_settings.service_port);
2313 #endif
2314 return m_natpmp.get();
2315 }
2316
2317 upnp* session_impl::start_upnp()
2318 {
2319 mutex_t::scoped_lock l(m_mutex);
2320
2321 INVARIANT_CHECK;
2322
2323 if (m_upnp) return m_upnp.get();
2324
2325 m_upnp = new upnp(m_io_service, m_half_open
2326 , m_listen_interface.address()
2327 , m_settings.user_agent
2328 , bind(&session_impl::on_port_mapping
2329 , this, _1, _2, _3, 1)
2330 , m_settings.upnp_ignore_nonrouters);
2331
2332 m_upnp->discover_device();
2333 if (m_listen_interface.port() > 0)
2334 {
2335 m_tcp_mapping[1] = m_upnp->add_mapping(upnp::tcp
2336 , m_listen_interface.port(), m_listen_interface.port());
2337 }
2338 #ifndef TORRENT_DISABLE_DHT
2339 if (m_dht)
2340 m_udp_mapping[1] = m_upnp->add_mapping(upnp::udp
2341 , m_dht_settings.service_port
2342 , m_dht_settings.service_port);
2343 #endif
2344 return m_upnp.get();
2345 }
2346
2347 void session_impl::stop_lsd()
2348 {
2349 mutex_t::scoped_lock l(m_mutex);
2350 if (m_lsd.get())
2351 m_lsd->close();
2352 m_lsd = 0;
2353 }
2354
2355 void session_impl::stop_natpmp()
2356 {
2357 mutex_t::scoped_lock l(m_mutex);
2358 if (m_natpmp.get())
2359 m_natpmp->close();
2360 m_natpmp = 0;
2361 }
2362
2363 void session_impl::stop_upnp()
2364 {
2365 mutex_t::scoped_lock l(m_mutex);
2366 if (m_upnp.get())
2367 {
2368 m_upnp->close();
2369 m_udp_mapping[1] = -1;
2370 m_tcp_mapping[1] = -1;
2371 }
2372 m_upnp = 0;
2373 }
2374
2375 void session_impl::set_external_address(address const& ip)
2376 {
2377 TORRENT_ASSERT(ip != address());
2378
2379 if (is_local(ip)) return;
2380 if (is_loopback(ip)) return;
2381 if (m_external_address == ip) return;
2382
2383 m_external_address = ip;
2384 if (m_alerts.should_post<external_ip_alert>())
2385 m_alerts.post_alert(external_ip_alert(ip));
2386 }
2387
2388 void session_impl::free_disk_buffer(char* buf)
2389 {
2390 m_disk_thread.free_buffer(buf);
2391 }
2392
2393 char* session_impl::allocate_disk_buffer()
2394 {
2395 return m_disk_thread.allocate_buffer();
2396 }
2397
2398 std::pair<char*, int> session_impl::allocate_buffer(int size)
2399 {
2400 TORRENT_ASSERT(size > 0);
2401 int num_buffers = (size + send_buffer_size - 1) / send_buffer_size;
2402 TORRENT_ASSERT(num_buffers > 0);
2403
2404 boost::mutex::scoped_lock l(m_send_buffer_mutex);
2405 #ifdef TORRENT_STATS
2406 TORRENT_ASSERT(m_buffer_allocations >= 0);
2407 m_buffer_allocations += num_buffers;
2408 m_buffer_usage_logger << log_time() << " protocol_buffer: "
2409 << (m_buffer_allocations * send_buffer_size) << std::endl;
2410 #endif
2411 #ifdef TORRENT_DISABLE_POOL_ALLOCATOR
2412 int num_bytes = num_buffers * send_buffer_size;
2413 return std::make_pair((char*)malloc(num_bytes), num_bytes);
2414 #else
2415 return std::make_pair((char*)m_send_buffers.ordered_malloc(num_buffers)
2416 , num_buffers * send_buffer_size);
2417 #endif
2418 }
2419
2420 void session_impl::free_buffer(char* buf, int size)
2421 {
2422 TORRENT_ASSERT(size > 0);
2423 TORRENT_ASSERT(size % send_buffer_size == 0);
2424 int num_buffers = size / send_buffer_size;
2425 TORRENT_ASSERT(num_buffers > 0);
2426
2427 boost::mutex::scoped_lock l(m_send_buffer_mutex);
2428 #ifdef TORRENT_STATS
2429 m_buffer_allocations -= num_buffers;
2430 TORRENT_ASSERT(m_buffer_allocations >= 0);
2431 m_buffer_usage_logger << log_time() << " protocol_buffer: "
2432 << (m_buffer_allocations * send_buffer_size) << std::endl;
2433 #endif
2434 #ifdef TORRENT_DISABLE_POOL_ALLOCATOR
2435 free(buf);
2436 #else
2437 m_send_buffers.ordered_free(buf, num_buffers);
2438 #endif
2439 }
2440
2441 #ifndef NDEBUG
2442 void session_impl::check_invariant() const
2443 {
2444 std::set<int> unique;
2445 int total_downloaders = 0;
2446 for (torrent_map::const_iterator i = m_torrents.begin()
2447 , end(m_torrents.end()); i != end; ++i)
2448 {
2449 int pos = i->second->queue_position();
2450 if (pos < 0)
2451 {
2452 TORRENT_ASSERT(pos == -1);
2453 continue;
2454 }
2455 ++total_downloaders;
2456 unique.insert(i->second->queue_position());
2457 }
2458 TORRENT_ASSERT(unique.size() == total_downloaders);
2459
2460 TORRENT_ASSERT(m_max_connections > 0);
2461 TORRENT_ASSERT(m_max_uploads > 0);
2462 TORRENT_ASSERT(m_allowed_upload_slots >= m_max_uploads);
2463 int unchokes = 0;
2464 int num_optimistic = 0;
2465 for (connection_map::const_iterator i = m_connections.begin();
2466 i != m_connections.end(); ++i)
2467 {
2468 TORRENT_ASSERT(*i);
2469 boost::shared_ptr<torrent> t = (*i)->associated_torrent().lock();
2470
2471 peer_connection* p = i->get();
2472 TORRENT_ASSERT(!p->is_disconnecting());
2473 if (!p->is_choked()) ++unchokes;
2474 if (p->peer_info_struct()
2475 && p->peer_info_struct()->optimistically_unchoked)
2476 {
2477 ++num_optimistic;
2478 TORRENT_ASSERT(!p->is_choked());
2479 }
2480 if (t && p->peer_info_struct())
2481 {
2482 TORRENT_ASSERT(t->get_policy().has_connection(p));
2483 }
2484 }
2485 TORRENT_ASSERT(num_optimistic == 0 || num_optimistic == 1);
2486 if (m_num_unchoked != unchokes)
2487 {
2488 TORRENT_ASSERT(false);
2489 }
2490 for (std::map<sha1_hash, boost::shared_ptr<torrent> >::const_iterator j
2491 = m_torrents.begin(); j != m_torrents.end(); ++j)
2492 {
2493 TORRENT_ASSERT(boost::get_pointer(j->second));
2494 }
2495 }
2496 #endif
2497
2498 }}
2499
A bittorrent daemon and client