Upstream tarball 10013
[amule.git] / src / UploadBandwidthThrottler.cpp
blobc06fa7b010380e4b191aa023857988b2ef2891ca
1 //
2 // This file is part of the aMule Project.
3 //
4 // Copyright (c) 2005-2008 aMule Team ( admin@amule.org / http://www.amule.org )
5 // Copyright (c) 2002-2008 Merkur ( devs@emule-project.net / http://www.emule-project.net )
6 //
7 // Any parts of this program derived from the xMule, lMule or eMule project,
8 // or contributed by third-party developers are copyrighted by their
9 // respective authors.
11 // This program is free software; you can redistribute it and/or modify
12 // it under the terms of the GNU General Public License as published by
13 // the Free Software Foundation; either version 2 of the License, or
14 // (at your option) any later version.
16 // This program is distributed in the hope that it will be useful,
17 // but WITHOUT ANY WARRANTY; without even the implied warranty of
18 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 // GNU General Public License for more details.
20 //
21 // You should have received a copy of the GNU General Public License
22 // along with this program; if not, write to the Free Software
23 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
26 #include "UploadBandwidthThrottler.h"
28 #include <protocol/ed2k/Constants.h>
29 #include <common/Macros.h>
30 #include <common/Constants.h>
32 #include <cmath>
33 #include <limits> // Do_not_auto_remove (NetBSD)
34 #include "OtherFunctions.h"
35 #include "ThrottledSocket.h"
36 #include "Logger.h"
37 #include "Preferences.h"
38 #include "Statistics.h"
40 #ifndef _MSC_VER
42 #ifdef _UI64_MAX
43 #undef _UI64_MAX
44 #endif
46 #ifdef _I64_MAX
47 #undef _I64_MAX
48 #endif
50 const uint32 _UI32_MAX = std::numeric_limits<uint32>::max();
51 const sint32 _I32_MAX = std::numeric_limits<sint32>::max();
52 const uint64 _UI64_MAX = std::numeric_limits<uint64>::max();
53 const sint64 _I64_MAX = std::numeric_limits<sint64>::max();
55 #endif
57 /////////////////////////////////////
60 /**
61 * The constructor starts the thread.
63 UploadBandwidthThrottler::UploadBandwidthThrottler()
64 : wxThread( wxTHREAD_JOINABLE )
66 m_SentBytesSinceLastCall = 0;
67 m_SentBytesSinceLastCallOverhead = 0;
69 m_doRun = true;
71 Create();
72 Run();
76 /**
77 * The destructor stops the thread. If the thread has already stoppped, destructor does nothing.
79 UploadBandwidthThrottler::~UploadBandwidthThrottler()
81 EndThread();
85 /**
86 * Find out how many bytes that has been put on the sockets since the last call to this
87 * method. Includes overhead of control packets.
89 * @return the number of bytes that has been put on the sockets since the last call
91 uint64 UploadBandwidthThrottler::GetNumberOfSentBytesSinceLastCallAndReset()
93 wxMutexLocker lock( m_sendLocker );
95 uint64 numberOfSentBytesSinceLastCall = m_SentBytesSinceLastCall;
96 m_SentBytesSinceLastCall = 0;
98 return numberOfSentBytesSinceLastCall;
102 * Find out how many bytes that has been put on the sockets since the last call to this
103 * method. Excludes overhead of control packets.
105 * @return the number of bytes that has been put on the sockets since the last call
107 uint64 UploadBandwidthThrottler::GetNumberOfSentBytesOverheadSinceLastCallAndReset()
109 wxMutexLocker lock( m_sendLocker );
111 uint64 numberOfSentBytesSinceLastCall = m_SentBytesSinceLastCallOverhead;
112 m_SentBytesSinceLastCallOverhead = 0;
114 return numberOfSentBytesSinceLastCall;
119 * Add a socket to the list of sockets that have upload slots. The main thread will
120 * continously call send on these sockets, to give them chance to work off their queues.
121 * The sockets are called in the order they exist in the list, so the top socket (index 0)
122 * will be given a chance first to use bandwidth, and then the next socket (index 1) etc.
124 * It is possible to add a socket several times to the list without removing it inbetween,
125 * but that should be avoided.
127 * @param index insert the socket at this place in the list. An index that is higher than the
128 * current number of sockets in the list will mean that the socket should be inserted
129 * last in the list.
131 * @param socket the address to the socket that should be added to the list. If the address is NULL,
132 * this method will do nothing.
134 void UploadBandwidthThrottler::AddToStandardList(uint32 index, ThrottledFileSocket* socket)
136 if ( socket ) {
137 wxMutexLocker lock( m_sendLocker );
139 RemoveFromStandardListNoLock(socket);
140 if (index > (uint32)m_StandardOrder_list.size()) {
141 index = m_StandardOrder_list.size();
144 m_StandardOrder_list.insert(m_StandardOrder_list.begin() + index, socket);
150 * Remove a socket from the list of sockets that have upload slots.
152 * If the socket has mistakenly been added several times to the list, this method
153 * will return all of the entries for the socket.
155 * @param socket the address of the socket that should be removed from the list. If this socket
156 * does not exist in the list, this method will do nothing.
158 bool UploadBandwidthThrottler::RemoveFromStandardList(ThrottledFileSocket* socket)
160 wxMutexLocker lock( m_sendLocker );
162 return RemoveFromStandardListNoLock(socket);
167 * Remove a socket from the list of sockets that have upload slots. NOT THREADSAFE!
168 * This is an internal method that doesn't take the necessary lock before it removes
169 * the socket. This method should only be called when the current thread already owns
170 * the m_sendLocker lock!
172 * @param socket address of the socket that should be removed from the list. If this socket
173 * does not exist in the list, this method will do nothing.
175 bool UploadBandwidthThrottler::RemoveFromStandardListNoLock(ThrottledFileSocket* socket)
177 return (EraseFirstValue( m_StandardOrder_list, socket ) > 0);
182 * Notifies the send thread that it should try to call controlpacket send
183 * for the supplied socket. It is allowed to call this method several times
184 * for the same socket, without having controlpacket send called for the socket
185 * first. The doublette entries are never filtered, since it is incurs less cpu
186 * overhead to simply call Send() in the socket for each double. Send() will
187 * already have done its work when the second Send() is called, and will just
188 * return with little cpu overhead.
190 * @param socket address to the socket that requests to have controlpacket send
191 * to be called on it
193 void UploadBandwidthThrottler::QueueForSendingControlPacket(ThrottledControlSocket* socket, bool hasSent)
195 // Get critical section
196 wxMutexLocker lock( m_tempQueueLocker );
198 if ( m_doRun ) {
199 if( hasSent ) {
200 m_TempControlQueueFirst_list.push_back(socket);
201 } else {
202 m_TempControlQueue_list.push_back(socket);
210 * Remove the socket from all lists and queues. This will make it safe to
211 * erase/delete the socket. It will also cause the main thread to stop calling
212 * send() for the socket.
214 * @param socket address to the socket that should be removed
216 void UploadBandwidthThrottler::DoRemoveFromAllQueues(ThrottledControlSocket* socket)
218 if ( m_doRun ) {
219 // Remove this socket from control packet queue
220 EraseValue( m_ControlQueue_list, socket );
221 EraseValue( m_ControlQueueFirst_list, socket );
223 wxMutexLocker lock( m_tempQueueLocker );
224 EraseValue( m_TempControlQueue_list, socket );
225 EraseValue( m_TempControlQueueFirst_list, socket );
230 void UploadBandwidthThrottler::RemoveFromAllQueues(ThrottledControlSocket* socket)
232 wxMutexLocker lock( m_sendLocker );
234 DoRemoveFromAllQueues( socket );
238 void UploadBandwidthThrottler::RemoveFromAllQueues(ThrottledFileSocket* socket)
240 wxMutexLocker lock( m_sendLocker );
242 if (m_doRun) {
243 DoRemoveFromAllQueues(socket);
245 // And remove it from upload slots
246 RemoveFromStandardListNoLock(socket);
252 * Make the thread exit. This method will not return until the thread has stopped
253 * looping. This guarantees that the thread will not access the CEMSockets after this
254 * call has exited.
256 void UploadBandwidthThrottler::EndThread()
258 if (m_doRun) { // do it only once
260 wxMutexLocker lock(m_sendLocker);
262 // signal the thread to stop looping and exit.
263 m_doRun = false;
266 Wait();
272 * The thread method that handles calling send for the individual sockets.
274 * Control packets will always be tried to be sent first. If there is any bandwidth leftover
275 * after that, send() for the upload slot sockets will be called in priority order until we have run
276 * out of available bandwidth for this loop. Upload slots will not be allowed to go without having sent
277 * called for more than a defined amount of time (i.e. two seconds).
279 * @return always returns 0.
281 void* UploadBandwidthThrottler::Entry()
283 const uint32 TIME_BETWEEN_UPLOAD_LOOPS = 1;
285 uint32 lastLoopTick = GetTickCountFullRes();
286 // Bytes to spend in current cycle. If we spend more this becomes negative and causes a wait next time.
287 sint32 bytesToSpend = 0;
288 uint32 allowedDataRate = 0;
289 uint32 rememberedSlotCounter = 0;
290 uint32 extraSleepTime = TIME_BETWEEN_UPLOAD_LOOPS;
292 while (m_doRun) {
293 uint32 timeSinceLastLoop = GetTickCountFullRes() - lastLoopTick;
295 // Calculate data rate
296 if (thePrefs::GetMaxUpload() == UNLIMITED) {
297 // Try to increase the upload rate from UploadSpeedSense
298 allowedDataRate = (uint32)theStats::GetUploadRate() + 5 * 1024;
299 } else {
300 allowedDataRate = thePrefs::GetMaxUpload() * 1024;
303 uint32 minFragSize = 1300;
304 uint32 doubleSendSize = minFragSize*2; // send two packages at a time so they can share an ACK
305 if (allowedDataRate < 6*1024) {
306 minFragSize = 536;
307 doubleSendSize = minFragSize; // don't send two packages at a time at very low speeds to give them a smoother load
311 uint32 sleepTime;
312 if (bytesToSpend < 1) {
313 // We have sent more than allowed in last cycle so we have to wait now
314 // until we can send at least 1 byte.
315 sleepTime = std::max((-bytesToSpend + 1) * 1000 / allowedDataRate + 2, // add 2 ms to allow for rounding inaccuracies
316 extraSleepTime);
317 } else {
318 // We could send at once, but sleep a while to not suck up all cpu
319 sleepTime = extraSleepTime;
322 if (timeSinceLastLoop < sleepTime) {
323 Sleep(sleepTime-timeSinceLastLoop);
326 const uint32 thisLoopTick = GetTickCountFullRes();
327 timeSinceLastLoop = thisLoopTick - lastLoopTick;
328 lastLoopTick = thisLoopTick;
330 if (timeSinceLastLoop > sleepTime + 2000) {
331 AddDebugLogLineM(false, logGeneral, CFormat(wxT("UploadBandwidthThrottler: Time since last loop too long. time: %ims wanted: %ims Max: %ims"))
332 % timeSinceLastLoop % sleepTime % (sleepTime + 2000));
334 timeSinceLastLoop = sleepTime + 2000;
337 // Calculate how many bytes we can spend
339 bytesToSpend += (sint32) (allowedDataRate / 1000.0 * timeSinceLastLoop);
341 if (bytesToSpend >= 1) {
342 sint32 spentBytes = 0;
343 sint32 spentOverhead = 0;
345 wxMutexLocker sendLock(m_sendLocker);
348 wxMutexLocker queueLock(m_tempQueueLocker);
350 // are there any sockets in m_TempControlQueue_list? Move them to normal m_ControlQueue_list;
351 m_ControlQueueFirst_list.insert( m_ControlQueueFirst_list.end(),
352 m_TempControlQueueFirst_list.begin(),
353 m_TempControlQueueFirst_list.end() );
355 m_ControlQueue_list.insert( m_ControlQueue_list.end(),
356 m_TempControlQueue_list.begin(),
357 m_TempControlQueue_list.end() );
359 m_TempControlQueue_list.clear();
360 m_TempControlQueueFirst_list.clear();
363 // Send any queued up control packets first
364 while (spentBytes < bytesToSpend && (!m_ControlQueueFirst_list.empty() || !m_ControlQueue_list.empty())) {
365 ThrottledControlSocket* socket = NULL;
367 if (!m_ControlQueueFirst_list.empty()) {
368 socket = m_ControlQueueFirst_list.front();
369 m_ControlQueueFirst_list.pop_front();
370 } else if (!m_ControlQueue_list.empty()) {
371 socket = m_ControlQueue_list.front();
372 m_ControlQueue_list.pop_front();
375 if (socket != NULL) {
376 SocketSentBytes socketSentBytes = socket->SendControlData(bytesToSpend-spentBytes, minFragSize);
377 spentBytes += socketSentBytes.sentBytesControlPackets + socketSentBytes.sentBytesStandardPackets;
378 spentOverhead += socketSentBytes.sentBytesControlPackets;
382 // Check if any sockets haven't gotten data for a long time. Then trickle them a package.
383 uint32 slots = m_StandardOrder_list.size();
384 for (uint32 slotCounter = 0; slotCounter < slots; slotCounter++) {
385 ThrottledFileSocket* socket = m_StandardOrder_list[ slotCounter ];
387 if (socket != NULL) {
388 if (thisLoopTick-socket->GetLastCalledSend() > SEC2MS(1)) {
389 // trickle
390 uint32 neededBytes = socket->GetNeededBytes();
392 if (neededBytes > 0) {
393 SocketSentBytes socketSentBytes = socket->SendFileAndControlData(neededBytes, minFragSize);
394 spentBytes += socketSentBytes.sentBytesControlPackets + socketSentBytes.sentBytesStandardPackets;
395 spentOverhead += socketSentBytes.sentBytesControlPackets;
398 } else {
399 AddDebugLogLineM(false, logGeneral, CFormat( wxT("There was a NULL socket in the UploadBandwidthThrottler Standard list (trickle)! Prevented usage. Index: %i Size: %i"))
400 % slotCounter % m_StandardOrder_list.size());
404 // Give available bandwidth to slots, starting with the one we ended with last time.
405 // There are two passes. First pass gives packets of doubleSendSize, second pass
406 // gives as much as possible.
407 // Second pass starts with the last slot of the first pass actually.
408 for (uint32 slotCounter = 0; (slotCounter < slots * 2) && spentBytes < bytesToSpend; slotCounter++) {
409 if (rememberedSlotCounter >= slots) { // wrap around pointer
410 rememberedSlotCounter = 0;
413 uint32 data = (slotCounter < slots - 1) ? doubleSendSize // pass 1
414 : (bytesToSpend - spentBytes); // pass 2
416 ThrottledFileSocket* socket = m_StandardOrder_list[ rememberedSlotCounter ];
418 if (socket != NULL) {
419 SocketSentBytes socketSentBytes = socket->SendFileAndControlData(data, doubleSendSize);
420 spentBytes += socketSentBytes.sentBytesControlPackets + socketSentBytes.sentBytesStandardPackets;
421 spentOverhead += socketSentBytes.sentBytesControlPackets;
422 } else {
423 AddDebugLogLineM(false, logGeneral, CFormat(wxT("There was a NULL socket in the UploadBandwidthThrottler Standard list (equal-for-all)! Prevented usage. Index: %i Size: %i"))
424 % rememberedSlotCounter % m_StandardOrder_list.size());
427 rememberedSlotCounter++;
430 // Do some limiting of what we keep for the next loop.
431 bytesToSpend -= spentBytes;
432 sint32 minBytesToSpend = (slots + 1) * minFragSize;
434 if (bytesToSpend < - minBytesToSpend) {
435 bytesToSpend = - minBytesToSpend;
436 } else {
437 sint32 bandwidthSavedTolerance = slots * 512 + 1;
438 if (bytesToSpend > bandwidthSavedTolerance) {
439 bytesToSpend = bandwidthSavedTolerance;
443 m_SentBytesSinceLastCall += spentBytes;
444 m_SentBytesSinceLastCallOverhead += spentOverhead;
446 if (spentBytes == 0) { // spentBytes includes the overhead
447 extraSleepTime = std::min<uint32>(extraSleepTime * 5, 1000); // 1s at most
448 } else {
449 extraSleepTime = TIME_BETWEEN_UPLOAD_LOOPS;
455 wxMutexLocker queueLock(m_tempQueueLocker);
456 m_TempControlQueue_list.clear();
457 m_TempControlQueueFirst_list.clear();
460 wxMutexLocker sendLock(m_sendLocker);
461 m_ControlQueue_list.clear();
462 m_StandardOrder_list.clear();
464 return 0;
466 // File_checked_for_headers