2 // This file is part of the aMule Project.
4 // Copyright (c) 2003-2011 aMule Team ( admin@amule.org / http://www.amule.org )
5 // Copyright (c) 2002-2011 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
27 #include "EMSocket.h" // Interface declarations.
29 #include <protocol/Protocols.h>
30 #include <protocol/ed2k/Constants.h>
32 #include "Packet.h" // Needed for CPacket
34 #include "GetTickCount.h"
35 #include "UploadBandwidthThrottler.h"
37 #include "Preferences.h"
38 #include "ScopedPtr.h"
41 const uint32 MAX_PACKET_SIZE
= 2000000;
43 // cppcheck-suppress uninitMemberVar CEMSocket::pendingHeader
44 CEMSocket::CEMSocket(const CProxyData
*ProxyData
)
45 : CEncryptedStreamSocket(wxSOCKET_NOWAIT
, ProxyData
)
47 // If an interface has been specified,
48 // then we need to bind to it.
49 if (!thePrefs::GetAddress().IsEmpty()) {
50 amuleIPV4Address host
;
52 // No need to warn here, in case of failure to
53 // assign the hostname. That is already done
54 // in amule.cpp when starting ...
55 if (host
.Hostname(thePrefs::GetAddress())) {
60 byConnected
= ES_NOTCONNECTED
;
61 m_uTimeOut
= CONNECTION_TIMEOUT
; // default timeout for ed2k sockets
63 // Download (pseudo) rate control
65 downloadLimitEnable
= false;
66 pendingOnReceive
= false;
68 // Download partial header
69 pendingHeaderSize
= 0;
71 // Download partial packet
73 pendingPacketSize
= 0;
80 m_currentPacket_is_controlpacket
= false;
81 m_currentPackageIsFromPartFile
= false;
83 m_numberOfSentBytesCompleteFile
= 0;
84 m_numberOfSentBytesPartFile
= 0;
85 m_numberOfSentBytesControlPacket
= 0;
87 lastCalledSend
= ::GetTickCount();
88 lastSent
= ::GetTickCount()-1000;
90 m_bAccelerateUpload
= false;
92 m_actualPayloadSize
= 0;
93 m_actualPayloadSizeSent
= 0;
98 lastFinishedStandard
= 0;
101 CEMSocket::~CEMSocket()
103 // need to be locked here to know that the other methods
104 // won't be in the middle of things
106 wxMutexLocker
lock(m_sendLocker
);
107 byConnected
= ES_DISCONNECTED
;
110 // now that we know no other method will keep adding to the queue
111 // we can remove ourself from the queue
112 if (theApp
->uploadBandwidthThrottler
) {
113 theApp
->uploadBandwidthThrottler
->RemoveFromAllQueues(this);
118 SetNotify(0); // this is already done in Destroy()
123 void CEMSocket::ClearQueues()
125 wxMutexLocker
lock(m_sendLocker
);
127 DeleteContents(m_control_queue
);
130 CStdPacketQueue::iterator it
= m_standard_queue
.begin();
131 for (; it
!= m_standard_queue
.end(); ++it
) {
134 m_standard_queue
.clear();
137 // Download (pseudo) rate control
139 downloadLimitEnable
= false;
140 pendingOnReceive
= false;
142 // Download partial header
143 pendingHeaderSize
= 0;
145 // Download partial packet
146 delete[] pendingPacket
;
147 pendingPacket
= NULL
;
148 pendingPacketSize
= 0;
158 void CEMSocket::OnClose(int WXUNUSED(nErrorCode
))
160 // need to be locked here to know that the other methods
161 // won't be in the middle of things
163 wxMutexLocker
lock(m_sendLocker
);
164 byConnected
= ES_DISCONNECTED
;
167 // now that we know no other method will keep adding to the queue
168 // we can remove ourself from the queue
169 theApp
->uploadBandwidthThrottler
->RemoveFromAllQueues(this);
175 void CEMSocket::OnReceive(int nErrorCode
)
184 // Check current connection state
185 if (byConnected
== ES_DISCONNECTED
) {
188 byConnected
= ES_CONNECTED
; // ES_DISCONNECTED, ES_NOTCONNECTED, ES_CONNECTED
193 // CPU load improvement
194 if (downloadLimitEnable
&& downloadLimit
== 0){
195 pendingOnReceive
= true;
201 if (pendingHeaderSize
< PACKET_HEADER_SIZE
) {
202 delete[] pendingPacket
;
203 pendingPacket
= NULL
;
204 buf
= pendingHeader
+ pendingHeaderSize
;
205 readMax
= PACKET_HEADER_SIZE
- pendingHeaderSize
;
206 } else if (pendingPacket
== NULL
) {
207 pendingPacketSize
= 0;
208 readMax
= CPacket::GetPacketSizeFromHeader(pendingHeader
);
209 if (readMax
> MAX_PACKET_SIZE
) {
210 pendingHeaderSize
= 0;
214 pendingPacket
= new byte
[readMax
+ 1];
217 buf
= pendingPacket
+ pendingPacketSize
;
218 readMax
= CPacket::GetPacketSizeFromHeader(pendingHeader
) - pendingPacketSize
;
221 if (downloadLimitEnable
&& readMax
> downloadLimit
) {
222 readMax
= downloadLimit
;
227 wxMutexLocker
lock(m_sendLocker
);
228 ret
= Read(buf
, readMax
);
230 pendingOnReceive
= true;
233 if (LastError() || ret
== 0) {
239 if (downloadLimitEnable
) {
241 if (ret
>= downloadLimit
) {
244 downloadLimit
-= ret
;
248 // CPU load improvement
249 // Detect if the socket's buffer is empty (or the size did match...)
250 pendingOnReceive
= (ret
== readMax
);
252 if (pendingHeaderSize
>= PACKET_HEADER_SIZE
) {
253 pendingPacketSize
+= ret
;
254 if (pendingPacketSize
>= CPacket::GetPacketSizeFromHeader(pendingHeader
)) {
255 CScopedPtr
<CPacket
> packet(new CPacket(pendingHeader
, pendingPacket
));
256 pendingPacket
= NULL
;
257 pendingPacketSize
= 0;
258 pendingHeaderSize
= 0;
260 // Bugfix We still need to check for a valid protocol
261 // Remark: the default eMule v0.26b had removed this test......
262 switch (packet
->GetProtocol()){
266 case OP_ED2KV2HEADER
:
267 case OP_ED2KV2PACKEDPROT
:
270 OnError(ERR_WRONGHEADER
);
275 PacketReceived(packet
.get());
278 pendingHeaderSize
+= ret
;
280 } while (ret
&& pendingHeaderSize
>= PACKET_HEADER_SIZE
);
284 void CEMSocket::SetDownloadLimit(uint32 limit
)
286 downloadLimit
= limit
;
287 downloadLimitEnable
= true;
289 // CPU load improvement
290 if(limit
> 0 && pendingOnReceive
== true){
296 void CEMSocket::DisableDownloadLimit()
298 downloadLimitEnable
= false;
300 // CPU load improvement
301 if (pendingOnReceive
== true){
308 * Queues up the packet to be sent. Another thread will actually send the packet.
310 * If the packet is not a control packet, and if the socket decides that its queue is
311 * full and forceAdd is false, then the socket is allowed to refuse to add the packet
312 * to its queue. It will then return false and it is up to the calling thread to try
313 * to call SendPacket for that packet again at a later time.
315 * @param packet address to the packet that should be added to the queue
317 * @param delpacket if true, the responsibility for deleting the packet after it has been sent
318 * has been transferred to this object. If false, don't delete the packet after it
321 * @param controlpacket the packet is a controlpacket
323 * @param forceAdd this packet must be added to the queue, even if it is full. If this flag is true
324 * then the method can not refuse to add the packet, and therefore not return false.
326 * @return true if the packet was added to the queue, false otherwise
328 void CEMSocket::SendPacket(CPacket
* packet
, bool delpacket
, bool controlpacket
, uint32 actualPayloadSize
)
330 //printf("* SendPacket called on socket %p\n", this);
331 wxMutexLocker
lock(m_sendLocker
);
333 if (byConnected
== ES_DISCONNECTED
) {
334 //printf("* Disconnected, drop packet\n");
340 packet
= new CPacket(*packet
);
344 //printf("* Adding a control packet\n");
345 m_control_queue
.push_back(packet
);
347 // queue up for controlpacket
348 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
350 //printf("* Adding a normal packet to the queue\n");
351 bool first
= !((sendbuffer
&& !m_currentPacket_is_controlpacket
) || !m_standard_queue
.empty());
352 StandardPacketQueueEntry queueEntry
= { actualPayloadSize
, packet
};
353 m_standard_queue
.push_back(queueEntry
);
355 // reset timeout for the first time
357 lastFinishedStandard
= ::GetTickCount();
358 m_bAccelerateUpload
= true; // Always accelerate first packet in a block
365 uint64
CEMSocket::GetSentBytesCompleteFileSinceLastCallAndReset()
367 wxMutexLocker
lock( m_sendLocker
);
369 uint64 sentBytes
= m_numberOfSentBytesCompleteFile
;
370 m_numberOfSentBytesCompleteFile
= 0;
376 uint64
CEMSocket::GetSentBytesPartFileSinceLastCallAndReset()
378 wxMutexLocker
lock( m_sendLocker
);
380 uint64 sentBytes
= m_numberOfSentBytesPartFile
;
381 m_numberOfSentBytesPartFile
= 0;
386 uint64
CEMSocket::GetSentBytesControlPacketSinceLastCallAndReset()
388 wxMutexLocker
lock( m_sendLocker
);
390 uint64 sentBytes
= m_numberOfSentBytesControlPacket
;
391 m_numberOfSentBytesControlPacket
= 0;
396 uint64
CEMSocket::GetSentPayloadSinceLastCallAndReset()
398 wxMutexLocker
lock( m_sendLocker
);
400 uint64 sentBytes
= m_actualPayloadSizeSent
;
401 m_actualPayloadSizeSent
= 0;
407 void CEMSocket::OnSend(int nErrorCode
)
414 CEncryptedStreamSocket::OnSend(0);
416 wxMutexLocker
lock( m_sendLocker
);
419 if (byConnected
!= ES_DISCONNECTED
) {
420 byConnected
= ES_CONNECTED
;
422 if (m_currentPacket_is_controlpacket
) {
423 // queue up for control packet
424 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
431 * Try to put queued up data on the socket.
433 * Control packets have higher priority, and will be sent first, if possible.
434 * Standard packets can be split up in several package containers. In that case
435 * all the parts of a split package must be sent in a row, without any control packet
438 * @param maxNumberOfBytesToSend This is the maximum number of bytes that is allowed to be put on the socket
439 * this call. The actual number of sent bytes will be returned from the method.
441 * @param onlyAllowedToSendControlPacket This call we only try to put control packets on the sockets.
442 * If there's a standard packet "in the way", and we think that this socket
443 * is no longer an upload slot, then it is ok to send the standard packet to
444 * get it out of the way. But it is not allowed to pick a new standard packet
445 * from the queue during this call. Several split packets are counted as one
446 * standard packet though, so it is ok to finish them all off if necessary.
448 * @return the actual number of bytes that were put on the socket.
450 SocketSentBytes
CEMSocket::Send(uint32 maxNumberOfBytesToSend
, uint32 minFragSize
, bool onlyAllowedToSendControlPacket
)
452 wxMutexLocker
lock(m_sendLocker
);
454 //printf("* Attempt to send a packet on socket %p\n", this);
456 if (byConnected
== ES_DISCONNECTED
) {
457 //printf("* Disconnected socket %p\n", this);
458 SocketSentBytes returnVal
= { false, 0, 0 };
460 } else if (m_bBusy
&& onlyAllowedToSendControlPacket
) {
461 //printf("* Busy socket %p\n", this);
462 SocketSentBytes returnVal
= { true, 0, 0 };
466 bool anErrorHasOccured
= false;
467 uint32 sentStandardPacketBytesThisCall
= 0;
468 uint32 sentControlPacketBytesThisCall
= 0;
470 if (byConnected
== ES_CONNECTED
&& IsEncryptionLayerReady() && (!m_bBusy
|| onlyAllowedToSendControlPacket
)) {
472 //printf("* Internal attemptto send on %p\n", this);
474 if(minFragSize
< 1) {
478 maxNumberOfBytesToSend
= GetNextFragSize(maxNumberOfBytesToSend
, minFragSize
);
480 bool bWasLongTimeSinceSend
= (::GetTickCount() - lastSent
) > 1000;
482 lastCalledSend
= ::GetTickCount();
485 while(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
< maxNumberOfBytesToSend
&& anErrorHasOccured
== false && // don't send more than allowed. Also, there should have been no error in earlier loop
486 (!m_control_queue
.empty() || !m_standard_queue
.empty() || sendbuffer
!= NULL
) && // there must exist something to send
487 (onlyAllowedToSendControlPacket
== false || // this means we are allowed to send both types of packets, so proceed
488 (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
> 0 && (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) % minFragSize
!= 0) ||
489 (sendbuffer
== NULL
&& !m_control_queue
.empty()) || // There's a control packet in queue, and we are not currently sending anything, so we will handle the control packet next
490 (sendbuffer
!= NULL
&& m_currentPacket_is_controlpacket
== true) || // We are in the progress of sending a control packet. We are always allowed to send those
491 (sendbuffer
!= NULL
&& m_currentPacket_is_controlpacket
== false && bWasLongTimeSinceSend
&& !m_control_queue
.empty() && m_standard_queue
.empty() && (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) < minFragSize
) // We have waited to long to clean the current packet (which may be a standard packet that is in the way). Proceed no matter what the value of onlyAllowedToSendControlPacket.
495 // If we are currently not in the progress of sending a packet, we will need to find the next one to send
496 if(sendbuffer
== NULL
) {
497 CPacket
* curPacket
= NULL
;
498 if(!m_control_queue
.empty()) {
499 // There's a control packet to send
500 m_currentPacket_is_controlpacket
= true;
501 curPacket
= m_control_queue
.front();
502 m_control_queue
.pop_front();
503 } else if(!m_standard_queue
.empty() /*&& onlyAllowedToSendControlPacket == false*/) {
504 // There's a standard packet to send
505 m_currentPacket_is_controlpacket
= false;
506 StandardPacketQueueEntry queueEntry
= m_standard_queue
.front();
507 m_standard_queue
.pop_front();
508 curPacket
= queueEntry
.packet
;
509 m_actualPayloadSize
= queueEntry
.actualPayloadSize
;
511 // remember this for statistics purposes.
512 m_currentPackageIsFromPartFile
= curPacket
->IsFromPF();
514 // Just to be safe. Shouldn't happen?
515 // if we reach this point, then there's something wrong with the while condition above!
517 AddDebugLogLineC(logGeneral
, wxT("EMSocket: Couldn't get a new packet! There's an error in the first while condition in EMSocket::Send()"));
519 SocketSentBytes returnVal
= { true, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
523 // We found a packet to send. Get the data to send from the
524 // package container and dispose of the container.
525 sendblen
= curPacket
->GetRealPacketSize();
526 sendbuffer
= curPacket
->DetachPacket();
530 CryptPrepareSendData((byte
*)sendbuffer
, sendblen
);
533 // At this point we've got a packet to send in sendbuffer. Try to send it. Loop until entire packet
534 // is sent, or until we reach maximum bytes to send for this call, or until we get an error.
535 // NOTE! If send would block (returns WOULDBLOCK), we will return from this method INSIDE this loop.
536 while (sent
< sendblen
&&
537 sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
< maxNumberOfBytesToSend
&&
539 onlyAllowedToSendControlPacket
== false || // this means we are allowed to send both types of packets, so proceed
540 m_currentPacket_is_controlpacket
||
541 (bWasLongTimeSinceSend
&& (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) < minFragSize
) ||
542 (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) % minFragSize
!= 0
544 anErrorHasOccured
== false) {
545 uint32 tosend
= sendblen
-sent
;
546 if(!onlyAllowedToSendControlPacket
|| m_currentPacket_is_controlpacket
) {
547 if (maxNumberOfBytesToSend
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> maxNumberOfBytesToSend
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
548 tosend
= maxNumberOfBytesToSend
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
549 } else if(bWasLongTimeSinceSend
&& (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) < minFragSize
) {
550 if (minFragSize
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> minFragSize
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
551 tosend
= minFragSize
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
553 uint32 nextFragMaxBytesToSent
= GetNextFragSize(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
, minFragSize
);
554 if (nextFragMaxBytesToSent
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> nextFragMaxBytesToSent
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
555 tosend
= nextFragMaxBytesToSent
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
557 wxASSERT(tosend
!= 0 && tosend
<= sendblen
-sent
);
559 //DWORD tempStartSendTick = ::GetTickCount();
561 lastSent
= ::GetTickCount();
563 uint32 result
= CEncryptedStreamSocket::Write(sendbuffer
+sent
,tosend
);
567 SocketSentBytes returnVal
= { true, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
568 return returnVal
; // Send() blocked, onsend will be called when ready to send again
569 } else if (LastError()) {
570 // Send() gave an error
571 anErrorHasOccured
= true;
573 // we managed to send some bytes. Perform bookkeeping.
579 // Log send bytes in correct class
580 if(m_currentPacket_is_controlpacket
== false) {
581 sentStandardPacketBytesThisCall
+= result
;
583 if(m_currentPackageIsFromPartFile
== true) {
584 m_numberOfSentBytesPartFile
+= result
;
586 m_numberOfSentBytesCompleteFile
+= result
;
589 sentControlPacketBytesThisCall
+= result
;
590 m_numberOfSentBytesControlPacket
+= result
;
595 if (sent
== sendblen
){
596 // we are done sending the current packet. Delete it and set
597 // sendbuffer to NULL so a new packet can be fetched.
602 if(!m_currentPacket_is_controlpacket
) {
603 m_actualPayloadSizeSent
+= m_actualPayloadSize
;
604 m_actualPayloadSize
= 0;
606 lastFinishedStandard
= ::GetTickCount(); // reset timeout
607 m_bAccelerateUpload
= false; // Safe until told otherwise
615 if(onlyAllowedToSendControlPacket
&& (!m_control_queue
.empty() || (sendbuffer
!= NULL
&& m_currentPacket_is_controlpacket
))) {
616 // enter control packet send queue
617 // we might enter control packet queue several times for the same package,
618 // but that costs very little overhead. Less overhead than trying to make sure
619 // that we only enter the queue once.
620 //printf("* Requeueing control packet on %p\n", this);
621 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
624 //printf("* Finishing send debug on %p\n",this);
626 SocketSentBytes returnVal
= { !anErrorHasOccured
, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
632 uint32
CEMSocket::GetNextFragSize(uint32 current
, uint32 minFragSize
)
634 if(current
% minFragSize
== 0) {
637 return minFragSize
*(current
/minFragSize
+1);
643 * Decides the (minimum) amount the socket needs to send to prevent timeout.
647 uint32
CEMSocket::GetNeededBytes()
653 uint64 sizeleft
, sizetotal
;
656 wxMutexLocker
lock(m_sendLocker
);
658 if (byConnected
== ES_DISCONNECTED
) {
662 if (!((sendbuffer
&& !m_currentPacket_is_controlpacket
) || !m_standard_queue
.empty())) {
663 // No standard packet to send. Even if data needs to be sent to prevent timout, there's nothing to send.
667 if (((sendbuffer
&& !m_currentPacket_is_controlpacket
)) && !m_control_queue
.empty())
668 m_bAccelerateUpload
= true; // We might be trying to send a block request, accelerate packet
670 sendgap
= ::GetTickCount() - lastCalledSend
;
672 timetotal
= m_bAccelerateUpload
?45000:90000;
673 timeleft
= ::GetTickCount() - lastFinishedStandard
;
674 if (sendbuffer
&& !m_currentPacket_is_controlpacket
) {
675 sizeleft
= sendblen
-sent
;
676 sizetotal
= sendblen
;
678 sizeleft
= sizetotal
= m_standard_queue
.front().packet
->GetRealPacketSize();
682 if (timeleft
>= timetotal
)
684 timeleft
= timetotal
-timeleft
;
685 if (timeleft
*sizetotal
>= timetotal
*sizeleft
) {
686 // don't use 'GetTimeOut' here in case the timeout value is high,
687 if (sendgap
> SEC2MS(20))
688 return 1; // Don't let the socket itself time out - Might happen when switching from spread(non-focus) slot to trickle slot
691 uint64 decval
= timeleft
*sizetotal
/timetotal
;
694 if (decval
< sizeleft
)
695 return sizeleft
-decval
+1; // Round up
702 * Removes all packets from the standard queue that don't have to be sent for the socket to be able to send a control packet.
704 * Before a socket can send a new packet, the current packet has to be finished. If the current packet is part of
705 * a split packet, then all parts of that split packet must be sent before the socket can send a control packet.
707 * This method keeps in standard queue only those packets that must be sent (rest of split packet), and removes everything
708 * after it. The method doesn't touch the control packet queue.
710 void CEMSocket::TruncateQueues()
712 wxMutexLocker
lock(m_sendLocker
);
714 // Clear the standard queue totally
715 // Please note! There may still be a standardpacket in the sendbuffer variable!
716 CStdPacketQueue::iterator it
= m_standard_queue
.begin();
717 for (; it
!= m_standard_queue
.end(); ++it
) {
721 m_standard_queue
.clear();
725 uint32
CEMSocket::GetTimeOut() const
731 void CEMSocket::SetTimeOut(uint32 uTimeOut
)
733 m_uTimeOut
= uTimeOut
;
735 // File_checked_for_headers