Changeset 8

Timestamp:

07/25/04 14:19:33 (8 years ago)

Author:

rakshasa

Message:

Throttle working, but needs fine-tuning and an API.

Fixed some incompatiblities with BSD..

Location:

trunk/libtorrent

Files:

• ChangeLog (modified) (1 diff)
• TODO (modified) (2 diffs)
• client/README (modified) (1 diff)
• client/display.cc (modified) (2 diffs)
• client/display.h (modified) (1 diff)
• client/rtorrent.cc (modified) (7 diffs)
• torrent/download.cc (modified) (2 diffs)
• torrent/download_state.cc (modified) (6 diffs)
• torrent/general.h (modified) (1 diff)
• torrent/peer_connection.cc (modified) (7 diffs)
• torrent/peer_connection.h (modified) (4 diffs)
• torrent/peer_connection_extra.cc (modified) (1 diff)
• torrent/peer_connection_sub.h (modified) (2 diffs)
• torrent/settings.cc (modified) (1 diff)
• torrent/settings.h (modified) (1 diff)
• torrent/socket_base.cc (modified) (2 diffs)
• torrent/storage.h (modified) (1 diff)
• torrent/throttle.cc (modified) (8 diffs)
• torrent/throttle.h (modified) (1 diff)
• torrent/throttle_control.cc (modified) (1 diff)
• torrent/throttle_control.h (modified) (2 diffs)
• torrent/timer.h (modified) (1 diff)
• torrent/torrent.cc (modified) (6 diffs)
• torrent/torrent.h (modified) (1 diff)
• torrent/tracker_query.h (modified) (1 diff)

trunk/libtorrent/ChangeLog

@@ +1 @@
+2004-07-25    <jaris@ifi.uio.no>
+
+        * torrent: Fixed bug that made it skip sending the stop message to the tracker when quiting.
+
+        * torrent: Changed stdint.h to inttypes.h and made it easy to disable execinfo.h in client.
+
+        * torrent: Added throttle with tree structure. Bandwidth delegation needs work.
+
 2004-07-20    <jaris@ifi.uio.no>
 

trunk/libtorrent/TODO

@@ -33 +33 @@
 Make configure do the peer id thingie automagically
 
-add unroll loops optimizing(?)
-
 Add config for how often we unchoke unknown peers vs good uploaders
 
@@ -40 +38 @@
 
 fix socketBase setAsync
+
+We're counting bytes we haven't downloaded, fixme.
+
+Recursively update the up/down rates in throttle.

trunk/libtorrent/client/README

@@ -1 +1 @@
 This is the test client for libtorrent. Dog food. Note that decreasing
-max peers does not yet disconnect. Most of the stuff is in KiB.
+max peers does not yet disconnect them.
+
+This client is just something i've been throwing togheter as i
+finished implementing stuff in the library. When i'm finished with the
+important parts of the library the API will be rewritten to something
+more usable. I will also do a complete rewrite of the client.
+
+If your system doesn't have "execinfo.h", uncomment "#define
+USE_EXECINFO" in rtorrent.cc. This will be handled by an autoconf
+script later.
 
 Main:

trunk/libtorrent/client/display.cc

@@ -1 +1 @@
 #include "display.h"
 #include <ncurses.h>
+
+int loops = 0;
 
 Display::Display() {
@@ -74 +76 @@
   }
 
-  mvprintw(maxY - 1, 0, "Port: %i Handshakes: %i",
+  mvprintw(maxY - 1, 0, "Port: %i Handshakes: %i Loops: %i",
            (int)torrent::get(torrent::LISTEN_PORT),
-           (int)torrent::get(torrent::HANDSHAKES_TOTAL));
+           (int)torrent::get(torrent::HANDSHAKES_TOTAL),
+           loops);
 
   refresh();

trunk/libtorrent/client/display.h

@@ -3 +3 @@
 
 #include <torrent/torrent.h>
+
+extern int loops;
 
 class Display {

trunk/libtorrent/client/rtorrent.cc

@@ -2 +2 @@
 #include <fstream>
 #include <signal.h>
-#include <execinfo.h>
 #include <ncurses.h>
-#include <sys/select.h>
 #include <torrent/torrent.h>
 #include <torrent/exceptions.h>
 
+#include <unistd.h>
+//#include <sys/select.h>
+
 #include "display.h"
 #include "download.h"
+
+// Uncomment this if your system doesn't have execinfo.h
+#define USE_EXECINFO
+
+#ifdef USE_EXECINFO
+#include <execinfo.h>
+#endif
 
 std::list<std::string> log_entries;
@@ -31 +39 @@
   switch (signum) {
   case SIGINT:
-    std::cout << "Shuting down" << std::endl;
+    if (shutdown) {
+      torrent::cleanup();
+      exit(0);
+    }
 
     shutdown = true;
@@ -46 +57 @@
     called = true;
 
+    std::cout << "Signal SEGFAULT recived, dumping stack:" << std::endl;
+
+#ifdef USE_EXECINFO
     // Print the stack and exit.
     stackSize = backtrace(stackPtrs, 50);
@@ -54 +68 @@
     display = NULL;
 
-    std::cout << "Signal SEGFAULT recived, dumping stack:" << std::endl;
-
     for (int i = 0; i < stackSize; ++i)
       std::cout << i << ' ' << stackStrings[i] << std::endl;
+#endif
 
     exit(-1);
@@ -71 +84 @@
   struct timeval timeout;
 
+  display = new Display();
+
   signal(SIGINT, signal_handler);
   signal(SIGSEGV, signal_handler);
@@ -79 +94 @@
   torrent::DList::const_iterator curDownload = torrent::downloads().end();
 
-  display = new Display();
   Download download(curDownload);
 
@@ -100 +114 @@
   int maxY, maxX;
 
-  while (!shutdown) {
+  while (!shutdown || !torrent::get(torrent::SHUTDOWN_DONE)) {
+    loops++;
+
     if (lastDraw + 1000000 < torrent::get(torrent::TIME_CURRENT)) {
       lastDraw = torrent::get(torrent::TIME_CURRENT);

trunk/libtorrent/torrent/download.cc

@@ -132 +132 @@
           (*itr)->lastChoked() + state().settings().chokeGracePeriod < Timer::cache() &&
           
-          (g = (*itr)->down().c_rate().rate() * 16 + (*itr)->up().c_rate().rate()) <= f) {
+          (g = (*itr)->throttle().down().rate() * 16 + (*itr)->throttle().up().rate()) <= f) {
         f = g;
         p1 = *itr;
@@ -149 +149 @@
           (*itr)->down().c_interested() &&
           
-          (g = (*itr)->down().c_rate().rate()) >= f) {
+          (g = (*itr)->throttle().down().rate()) >= f) {
         f = g;
         p2 = *itr;

trunk/libtorrent/torrent/download_state.cc

@@ -7 +7 @@
 #include "peer_connection.h"
 #include "peer_handshake.h"
+#include "throttle_control.h"
 #include <algo/algo.h>
 
@@ -25 +26 @@
 
   m_files.closeAll();
-}
-
-void DownloadState::removeConnection(PeerConnection* p) {
-  Connections::iterator itr = std::find(m_connections.begin(), m_connections.end(), p);
-
-  if (itr == m_connections.end())
-    throw internal_error("Tried to remove peer connection from download that doesn't exist");
-
-  delete *itr;
-  m_connections.erase(itr);
-
-  // TODO: Remove this when we're stable
-  if (std::find(m_connections.begin(), m_connections.end(), p) != m_connections.end())
-    throw internal_error("Duplicate PeerConnections in Download");
-
-  chokeBalance();
-  connectPeers();
 }
 
@@ -75 +59 @@
   // TODO: Optimize, do a single pass with a 's' sized list of (un)chokings.
   if (s > 0) {
-//     while (s && itr != m_connections.end()) {
-//       if (
-
-    m_connections.sort(gt(call_member(call_member(&PeerConnection::down),
-                                      &PeerConnection::Sub::c_rate),
-                          call_member(call_member(&PeerConnection::down),
-                                      &PeerConnection::Sub::c_rate)));
+    m_connections.sort(gt(call_member(call_member(&PeerConnection::throttle),
+                                      &Throttle::down),
+                          call_member(call_member(&PeerConnection::throttle),
+                                      &Throttle::down)));
 
     // unchoke peers.
@@ -97 +78 @@
     // Sort so we choke slow uploaders first.
     // TODO: Should we sort by unchoked too?
-    m_connections.sort(lt(call_member(call_member(&PeerConnection::down),
-                                      &PeerConnection::Sub::c_rate),
-                          call_member(call_member(&PeerConnection::down),
-                                      &PeerConnection::Sub::c_rate)));
+    m_connections.sort(lt(call_member(call_member(&PeerConnection::throttle),
+                                      &Throttle::down),
+                          call_member(call_member(&PeerConnection::throttle),
+                                      &Throttle::down)));
 
     for (Connections::iterator itr = m_connections.begin();
@@ -156 +137 @@
 
   c->set(fd, p, this);
+
+  c->throttle().set_parent(&ThrottleControl::global().root());
+  c->throttle().set_settings(ThrottleControl::global().settings(ThrottleControl::SETTINGS_PEER));
+  c->throttle().set_socket(c);
+
   m_connections.push_back(c);
 
@@ -162 +148 @@
   if (itr != m_availablePeers.end())
     m_availablePeers.erase(itr);
+}
+
+void DownloadState::removeConnection(PeerConnection* p) {
+  Connections::iterator itr = std::find(m_connections.begin(), m_connections.end(), p);
+
+  if (itr == m_connections.end())
+    throw internal_error("Tried to remove peer connection from download that doesn't exist");
+
+  delete *itr;
+  m_connections.erase(itr);
+
+  // TODO: Remove this when we're stable
+  if (std::find(m_connections.begin(), m_connections.end(), p) != m_connections.end())
+    throw internal_error("Duplicate PeerConnections in Download");
+
+  chokeBalance();
+  connectPeers();
 }
 

trunk/libtorrent/torrent/general.h

@@ -3 +3 @@
 
 #include <string>
-#include <stdint.h>
 #include <vector>
 

trunk/libtorrent/torrent/peer_connection.cc

@@ -269 +269 @@
     s = readChunk();
 
-    m_down.rate.add(m_down.pos - previous);
+    m_throttle.down().add(m_down.pos - previous);
     m_download->rateDown().add(m_down.pos - previous);
     
@@ -304 +304 @@
                 m_down.pos);
 
-    m_down.rate.add(m_down.pos - previous);
+    m_throttle.down().add(m_down.pos - previous);
 
     if (!s)
@@ -342 +342 @@
 void PeerConnection::write() {
   bool s;
-  int previous;
+  int previous, maxBytes;
 
   m_lastMsg = Timer::cache();
@@ -410 +410 @@
 
     previous = m_up.pos;
-    s = writeChunk();
-
-    m_up.rate.add(m_up.pos - previous);
+    maxBytes = m_throttle.left();
+    
+    if (maxBytes == 0) {
+      removeWrite();
+      return;
+    }
+
+    if (maxBytes < 0)
+      throw internal_error("PeerConnection::write() got maxBytes <= 0");
+
+    s = writeChunk(maxBytes);
+
+    m_throttle.up().add(m_up.pos - previous);
+    m_throttle.spent(m_up.pos - previous);
+
     m_download->rateUp().add(m_up.pos - previous);
 
@@ -513 +525 @@
       m_up.list.push_back(Piece(index, offset, length));
       insertWrite();
+
+      m_throttle.activate();
 
     } else if (rItr != m_up.list.end()) {
@@ -554 +568 @@
     m_sendChoked = false;
 
+    if (m_up.choked) {
+      // Clear the request queue and mmaped chunk.
+      m_up.list.clear();
+      m_up.data = Chunk();
+
+      m_throttle.idle();
+    }
+
     if ((Timer::cache() - m_lastChoked).seconds() < 10) {
       // Wait with the choke message.
@@ -564 +586 @@
       
       m_lastChoked = Timer::cache();
-      
-      if (m_up.choked) {
-        // Clear the request queue and mmaped chunk.
-        m_up.list.clear();
-        m_up.data = Chunk();
-      }
     }
   }

trunk/libtorrent/torrent/peer_connection.h

@@ -9 +9 @@
 #include "piece.h"
 #include "rate.h"
+#include "throttle.h"
 
 #include <vector>
@@ -70 +71 @@
   Sub& down() { return m_down; }
 
+  Throttle& throttle() { return m_throttle; }
+
   virtual void read();
   virtual void write();
@@ -86 +89 @@
   };
 
-  bool writeChunk();
+  bool writeChunk(int maxBytes);
   bool readChunk();
 
@@ -119 +122 @@
   Sub m_up;
   Sub m_down;
+
+  Throttle m_throttle;
 };
 

trunk/libtorrent/torrent/peer_connection_extra.cc

@@ -52 +52 @@
 }
 
-bool PeerConnection::writeChunk() {
+bool PeerConnection::writeChunk(int maxBytes) {
   Chunk::Part& part = m_up.data.get(m_up.list.front().offset() + m_up.pos);
 
+  // Length between piece start and the end of the current part of the piece.
+  unsigned int length = std::min(part.first + part.second.length() - m_up.list.front().offset(),
+                                 m_up.list.front().length());
+
   // TODO: Make this a while loop so we spit out as much of the piece as we can this work cycle.
-  if (!writeBuf(part.second.data() + m_up.list.front().offset() + m_up.pos - part.first,
-                std::min(part.first + part.second.length() - m_up.list.front().offset(),
-                         m_up.list.front().length()),
-                m_up.pos))
-    return false;
+  writeBuf(part.second.data() + m_up.list.front().offset() + m_up.pos - part.first,
+           std::min(length, m_up.pos + maxBytes),
+           m_up.pos);
 
   return m_up.pos == m_up.list.front().length();

trunk/libtorrent/torrent/peer_connection_sub.h

@@ -17 +17 @@
   bool c_interested() const { return interested; }
 
-  Rate& c_rate() { return rate; }
-
   PieceList& c_list() { return list; }
 
@@ -35 +33 @@
   Chunk data;
   PieceList list;
-
-  Rate rate;
 };

trunk/libtorrent/torrent/settings.cc

@@ -29 +29 @@
 
 int ThrottleSettings::minPeriod = 500000;
-int ThrottleSettings::wakeupPoint = 512;
+int ThrottleSettings::wakeupPoint = 2048;
 int ThrottleSettings::minChunk = 512;
 
 ThrottleSettings::ThrottleSettings() :
-  unlimited(true),
-  constantRate(0)
+  constantRate(-1)
 {}
 

trunk/libtorrent/torrent/settings.h

@@ -42 +42 @@
   ThrottleSettings();
 
-  bool unlimited;
-
   int constantRate;
 

trunk/libtorrent/torrent/socket_base.cc

@@ -7 +7 @@
 #include <arpa/inet.h>
 #include <netinet/ip.h>
-#include <errno.h>
 #include <unistd.h>
 #include <sstream>
 #include <iostream>
+#include <cerrno>
 #include <cstring>
 
@@ -67 +67 @@
 void SocketBase::setSocketAsync(int fd) {
   // Set Reuseaddr.
-  int opt = 1;
+  //int opt = 1;
 
   // TODO: this doesn't belong here
-  if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)))
-    throw local_error("Error setting socket to SO_REUSEADDR");
+  //if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)))
+  //  throw local_error("Error setting socket to SO_REUSEADDR");
 
   // Set async.

trunk/libtorrent/torrent/storage.h

@@ -5 +5 @@
 #include <vector>
 #include <string>
-#include <stdint.h>
+#include <inttypes.h>
 #include <sys/stat.h>
 

trunk/libtorrent/torrent/throttle.cc

@@ -7 +7 @@
 #include "socket_base.h"
 #include "throttle.h"
+#include <numeric>
 #include <algo/algo.h>
 
@@ -22 +23 @@
 
 Throttle::~Throttle() {
-  idle();
+  set_parent(NULL);
 
   while (!m_active.empty())
-    m_active.front()->idle();
+    m_active.front()->set_parent(NULL);
 }
 
 // Include in the calculation when parent delegates bandwidth.
+
+// TODO: active should allocate a resonable amount of bytes for immidiate download.
+// We can't do that in active, as it will be called regulary.
 void Throttle::activate() {
   if (m_parent == NULL || m_activeItr != m_parent->m_active.end())
@@ -46 +50 @@
   
   m_activeItr = m_parent->m_active.end();
+
+  m_left = 0;
+}
+
+int Throttle::left() {
+  if (m_left == UNLIMITED)
+    return std::numeric_limits<int>::max() / 2;
+
+  // TODO: Give chunks.
+  return m_left;
 }
 
 void Throttle::spent(unsigned int bytes) {
-  if (m_left < 0)
+  if (m_left == UNLIMITED)
     return;
 
@@ -62 +76 @@
 void Throttle::update(float period, int bytes) {
   // Find out how much we are allowed. Int max if unlimited.
-  if (m_settings->unlimited)
+  if (m_settings->constantRate == UNLIMITED)
     m_left = bytes;
 
   else
-    m_left = std::min<unsigned int>(bytes,
-                                    // TODO: Adjust quickly if we are behind
-                                    (unsigned int)(m_settings->constantRate * period));
+    m_left = std::min((unsigned int)bytes,
+                      // TODO: Adjust quickly if we are behind
+                      (unsigned int)(m_settings->constantRate * period));
 
   // Not needed unless i fscked up something here.
@@ -81 +95 @@
 
   // Divide the bandwidth amongst children if present.
-  for (Children::iterator itr = m_active.begin(); ++itr != m_active.end(); ++itr) {
+  for (Children::iterator itr = m_active.begin(); itr != m_active.end(); ++itr) {
     (*itr)->update(period,
                    m_left == UNLIMITED ? UNLIMITED : m_left / m_activeSize);
@@ -89 +103 @@
 void Throttle::set_parent(Throttle* parent) {
   if (m_parent) {
+    idle();
+
     m_parent->m_children.erase(m_childrenItr);
     m_parent->m_childrenSize--;
@@ -96 +112 @@
 
   if (parent) {
+    m_parent->m_childrenSize++;
+
     m_childrenItr = m_parent->m_children.insert(m_parent->m_children.end(), this);
-    m_parent->m_childrenSize++;
+    m_activeItr = m_parent->m_active.end();
   }
 }
@@ -105 +123 @@
 }
 
+void Throttle::set_socket(SocketBase* socket) {
+  m_socket = socket;
 }
+
+}

trunk/libtorrent/torrent/throttle.h

@@ -29 +29 @@
   Rate& down() { return m_down; }
   
-  int left() { return m_left; }
+  int left();
   
   void spent(unsigned int bytes);

trunk/libtorrent/torrent/throttle_control.cc

@@ -3 +3 @@
 #endif
 
+#include "exceptions.h"
 #include "throttle_control.h"
 
 namespace torrent {
 
+ThrottleControl ThrottleControl::m_global;
+
 ThrottleControl::ThrottleControl() :
-  m_settings(1) {
+  m_settings(2) {
 
   m_root.set_settings(&m_settings[SETTINGS_ROOT]);
 
-  m_settings[SETTINGS_ROOT].unlimited = false;
-  m_settings[SETTINGS_ROOT].constantRate = 5000;
+  m_settings[SETTINGS_ROOT].constantRate = Throttle::UNLIMITED;
+
+  m_settings[SETTINGS_PEER].constantRate = Throttle::UNLIMITED;
 }
 
 void ThrottleControl::service(int type) {
-  m_root.update(ThrottleSettings::minPeriod, Throttle::UNLIMITED);
+  m_root.update(ThrottleSettings::minPeriod / 1000000.0f, Throttle::UNLIMITED);
+
+  // TODO: Remove this later
+  if (inService(0))
+    throw internal_error("Duplicate ThrottleService in service");
 
   // TODO: we lose some time, adjust.

trunk/libtorrent/torrent/throttle_control.h

@@ -12 +12 @@
  public:
   typedef enum {
-    SETTINGS_ROOT
+    SETTINGS_ROOT,
+    SETTINGS_PEER
   } SettingsType;
 
@@ -25 +26 @@
   void service(int type);
 
+  static ThrottleControl& global() { return m_global; }
+
  private:
+  static ThrottleControl m_global;
+
   Throttle m_root;
 

trunk/libtorrent/torrent/timer.h

@@ -2 +2 @@
 #define LIBTORRENT_TIMER_H
 
-#include <stdint.h>
+#include <inttypes.h>
 #include <sys/time.h>
 

trunk/libtorrent/torrent/torrent.cc

@@ -27 +27 @@
 std::list<std::string> caughtExceptions;
 
-ThrottleControl throttleControl;
-
 struct add_socket {
   add_socket(fd_set* s) : fd(0), fds(s) {}
@@ -63 +61 @@
   Listen::open(beginPort, endPort);
 
-  throttleControl.insertService(Timer::current(), 0);
+  ThrottleControl::global().insertService(Timer::current(), 0);
 }
 
@@ -79 +77 @@
   Listen::close();
 
-  throttleControl.removeService();
+  ThrottleControl::global().removeService();
 
   for_each<true>(Download::downloads().begin(), Download::downloads().end(),
@@ -208 +206 @@
   case SHUTDOWN_DONE:
     return std::find_if(Download::downloads().begin(), Download::downloads().end(),
-                        call_member(&Download::isStopped))
+                        bool_not(call_member(&Download::isStopped)))
       == Download::downloads().end();
 
@@ -262 +260 @@
 
 int64_t get(DList::const_iterator d, DValue t) {
-//   if (d == Download::downloads().end())
-//     throw internal_error("torrent::get(DList::const_iterator, DValue) called on an invalid iterator");
   Timer::update();
   uint64_t a;
@@ -380 +376 @@
 
   case PEER_RATE_DOWN:
-    return (*p)->down().c_rate().rate(true);
+    return (*p)->throttle().down().rate(true);
 
   case PEER_RATE_UP:
-    return (*p)->up().c_rate().rate(true);
+    return (*p)->throttle().up().rate(true);
 
   case PEER_PORT:

trunk/libtorrent/torrent/torrent.h

@@ -4 +4 @@
 #include <list>
 #include <string>
-#include <stdint.h>
+#include <inttypes.h>
 #include <sys/types.h>
 

trunk/libtorrent/torrent/tracker_query.h

@@ -4 +4 @@
 #include <map>
 #include <iosfwd>
-#include <stdint.h>
+#include <inttypes.h>
 #include "peer.h"
 #include "socket_base.h"