2 // This file is part of the aMule Project.
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 )
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
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.
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>
33 #include <limits> // Do_not_auto_remove (NetBSD)
34 #include "OtherFunctions.h"
35 #include "ThrottledSocket.h"
37 #include "Preferences.h"
38 #include "Statistics.h"
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();
57 /////////////////////////////////////
61 * The constructor starts the thread.
63 UploadBandwidthThrottler::UploadBandwidthThrottler()
64 : wxThread( wxTHREAD_JOINABLE
)
66 m_SentBytesSinceLastCall
= 0;
67 m_SentBytesSinceLastCallOverhead
= 0;
77 * The destructor stops the thread. If the thread has already stoppped, destructor does nothing.
79 UploadBandwidthThrottler::~UploadBandwidthThrottler()
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
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
)
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
193 void UploadBandwidthThrottler::QueueForSendingControlPacket(ThrottledControlSocket
* socket
, bool hasSent
)
195 // Get critical section
196 wxMutexLocker
lock( m_tempQueueLocker
);
200 m_TempControlQueueFirst_list
.push_back(socket
);
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
)
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
);
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
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.
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
&& !TestDestroy()) {
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;
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) {
307 doubleSendSize
= minFragSize
; // don't send two packages at a time at very low speeds to give them a smoother load
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
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 // Check after sleep in case the thread has been signaled to end
327 if (!m_doRun
|| TestDestroy()) {
331 const uint32 thisLoopTick
= GetTickCountFullRes();
332 timeSinceLastLoop
= thisLoopTick
- lastLoopTick
;
333 lastLoopTick
= thisLoopTick
;
335 if (timeSinceLastLoop
> sleepTime
+ 2000) {
336 AddDebugLogLineN(logGeneral
, CFormat(wxT("UploadBandwidthThrottler: Time since last loop too long. time: %ims wanted: %ims Max: %ims"))
337 % timeSinceLastLoop
% sleepTime
% (sleepTime
+ 2000));
339 timeSinceLastLoop
= sleepTime
+ 2000;
342 // Calculate how many bytes we can spend
344 bytesToSpend
+= (sint32
) (allowedDataRate
/ 1000.0 * timeSinceLastLoop
);
346 if (bytesToSpend
>= 1) {
347 sint32 spentBytes
= 0;
348 sint32 spentOverhead
= 0;
350 wxMutexLocker
sendLock(m_sendLocker
);
353 wxMutexLocker
queueLock(m_tempQueueLocker
);
355 // are there any sockets in m_TempControlQueue_list? Move them to normal m_ControlQueue_list;
356 m_ControlQueueFirst_list
.insert( m_ControlQueueFirst_list
.end(),
357 m_TempControlQueueFirst_list
.begin(),
358 m_TempControlQueueFirst_list
.end() );
360 m_ControlQueue_list
.insert( m_ControlQueue_list
.end(),
361 m_TempControlQueue_list
.begin(),
362 m_TempControlQueue_list
.end() );
364 m_TempControlQueue_list
.clear();
365 m_TempControlQueueFirst_list
.clear();
368 // Send any queued up control packets first
369 while (spentBytes
< bytesToSpend
&& (!m_ControlQueueFirst_list
.empty() || !m_ControlQueue_list
.empty())) {
370 ThrottledControlSocket
* socket
= NULL
;
372 if (!m_ControlQueueFirst_list
.empty()) {
373 socket
= m_ControlQueueFirst_list
.front();
374 m_ControlQueueFirst_list
.pop_front();
375 } else if (!m_ControlQueue_list
.empty()) {
376 socket
= m_ControlQueue_list
.front();
377 m_ControlQueue_list
.pop_front();
380 if (socket
!= NULL
) {
381 SocketSentBytes socketSentBytes
= socket
->SendControlData(bytesToSpend
-spentBytes
, minFragSize
);
382 spentBytes
+= socketSentBytes
.sentBytesControlPackets
+ socketSentBytes
.sentBytesStandardPackets
;
383 spentOverhead
+= socketSentBytes
.sentBytesControlPackets
;
387 // Check if any sockets haven't gotten data for a long time. Then trickle them a package.
388 uint32 slots
= m_StandardOrder_list
.size();
389 for (uint32 slotCounter
= 0; slotCounter
< slots
; slotCounter
++) {
390 ThrottledFileSocket
* socket
= m_StandardOrder_list
[ slotCounter
];
392 if (socket
!= NULL
) {
393 if (thisLoopTick
-socket
->GetLastCalledSend() > SEC2MS(1)) {
395 uint32 neededBytes
= socket
->GetNeededBytes();
397 if (neededBytes
> 0) {
398 SocketSentBytes socketSentBytes
= socket
->SendFileAndControlData(neededBytes
, minFragSize
);
399 spentBytes
+= socketSentBytes
.sentBytesControlPackets
+ socketSentBytes
.sentBytesStandardPackets
;
400 spentOverhead
+= socketSentBytes
.sentBytesControlPackets
;
404 AddDebugLogLineN(logGeneral
, CFormat( wxT("There was a NULL socket in the UploadBandwidthThrottler Standard list (trickle)! Prevented usage. Index: %i Size: %i"))
405 % slotCounter
% m_StandardOrder_list
.size());
409 // Give available bandwidth to slots, starting with the one we ended with last time.
410 // There are two passes. First pass gives packets of doubleSendSize, second pass
411 // gives as much as possible.
412 // Second pass starts with the last slot of the first pass actually.
413 for (uint32 slotCounter
= 0; (slotCounter
< slots
* 2) && spentBytes
< bytesToSpend
; slotCounter
++) {
414 if (rememberedSlotCounter
>= slots
) { // wrap around pointer
415 rememberedSlotCounter
= 0;
418 uint32 data
= (slotCounter
< slots
- 1) ? doubleSendSize
// pass 1
419 : (bytesToSpend
- spentBytes
); // pass 2
421 ThrottledFileSocket
* socket
= m_StandardOrder_list
[ rememberedSlotCounter
];
423 if (socket
!= NULL
) {
424 SocketSentBytes socketSentBytes
= socket
->SendFileAndControlData(data
, doubleSendSize
);
425 spentBytes
+= socketSentBytes
.sentBytesControlPackets
+ socketSentBytes
.sentBytesStandardPackets
;
426 spentOverhead
+= socketSentBytes
.sentBytesControlPackets
;
428 AddDebugLogLineN(logGeneral
, CFormat(wxT("There was a NULL socket in the UploadBandwidthThrottler Standard list (equal-for-all)! Prevented usage. Index: %i Size: %i"))
429 % rememberedSlotCounter
% m_StandardOrder_list
.size());
432 rememberedSlotCounter
++;
435 // Do some limiting of what we keep for the next loop.
436 bytesToSpend
-= spentBytes
;
437 sint32 minBytesToSpend
= (slots
+ 1) * minFragSize
;
439 if (bytesToSpend
< - minBytesToSpend
) {
440 bytesToSpend
= - minBytesToSpend
;
442 sint32 bandwidthSavedTolerance
= slots
* 512 + 1;
443 if (bytesToSpend
> bandwidthSavedTolerance
) {
444 bytesToSpend
= bandwidthSavedTolerance
;
448 m_SentBytesSinceLastCall
+= spentBytes
;
449 m_SentBytesSinceLastCallOverhead
+= spentOverhead
;
451 if (spentBytes
== 0) { // spentBytes includes the overhead
452 extraSleepTime
= std::min
<uint32
>(extraSleepTime
* 5, 1000); // 1s at most
454 extraSleepTime
= TIME_BETWEEN_UPLOAD_LOOPS
;
460 wxMutexLocker
queueLock(m_tempQueueLocker
);
461 m_TempControlQueue_list
.clear();
462 m_TempControlQueueFirst_list
.clear();
465 wxMutexLocker
sendLock(m_sendLocker
);
466 m_ControlQueue_list
.clear();
467 m_StandardOrder_list
.clear();
471 // File_checked_for_headers