2 // This file is part of the aMule Project.
4 // Copyright (c) 2003-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
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 CEMSocket::CEMSocket(const CProxyData
*ProxyData
)
44 : CEncryptedStreamSocket(wxSOCKET_NOWAIT
, ProxyData
)
46 // If an interface has been specified,
47 // then we need to bind to it.
48 if (!thePrefs::GetAddress().IsEmpty()) {
49 amuleIPV4Address host
;
51 // No need to warn here, in case of failure to
52 // assign the hostname. That is already done
53 // in amule.cpp when starting ...
54 if (host
.Hostname(thePrefs::GetAddress())) {
59 byConnected
= ES_NOTCONNECTED
;
60 m_uTimeOut
= CONNECTION_TIMEOUT
; // default timeout for ed2k sockets
62 // Download (pseudo) rate control
64 downloadLimitEnable
= false;
65 pendingOnReceive
= false;
67 // Download partial header
68 pendingHeaderSize
= 0;
70 // Download partial packet
72 pendingPacketSize
= 0;
79 m_currentPacket_is_controlpacket
= false;
80 m_currentPackageIsFromPartFile
= false;
82 m_numberOfSentBytesCompleteFile
= 0;
83 m_numberOfSentBytesPartFile
= 0;
84 m_numberOfSentBytesControlPacket
= 0;
86 lastCalledSend
= ::GetTickCount();
87 lastSent
= ::GetTickCount()-1000;
89 m_bAccelerateUpload
= false;
91 m_actualPayloadSize
= 0;
92 m_actualPayloadSizeSent
= 0;
97 lastFinishedStandard
= 0;
103 CEMSocket::~CEMSocket()
105 // need to be locked here to know that the other methods
106 // won't be in the middle of things
108 wxMutexLocker
lock(m_sendLocker
);
109 byConnected
= ES_DISCONNECTED
;
112 // now that we know no other method will keep adding to the queue
113 // we can remove ourself from the queue
114 if (theApp
->uploadBandwidthThrottler
) {
115 theApp
->uploadBandwidthThrottler
->RemoveFromAllQueues(this);
124 void CEMSocket::Destroy() {
127 wxSocketClient::Destroy();
132 void CEMSocket::ClearQueues()
134 wxMutexLocker
lock(m_sendLocker
);
136 DeleteContents(m_control_queue
);
139 CStdPacketQueue::iterator it
= m_standard_queue
.begin();
140 for (; it
!= m_standard_queue
.end(); ++it
) {
143 m_standard_queue
.clear();
146 // Download (pseudo) rate control
148 downloadLimitEnable
= false;
149 pendingOnReceive
= false;
151 // Download partial header
152 pendingHeaderSize
= 0;
154 // Download partial packet
155 delete[] pendingPacket
;
156 pendingPacket
= NULL
;
157 pendingPacketSize
= 0;
167 void CEMSocket::OnClose(int WXUNUSED(nErrorCode
))
169 // need to be locked here to know that the other methods
170 // won't be in the middle of things
172 wxMutexLocker
lock(m_sendLocker
);
173 byConnected
= ES_DISCONNECTED
;
176 // now that we know no other method will keep adding to the queue
177 // we can remove ourself from the queue
178 theApp
->uploadBandwidthThrottler
->RemoveFromAllQueues(this);
184 void CEMSocket::OnReceive(int nErrorCode
)
187 uint32 error
= LastError();
188 if (error
!= wxSOCKET_WOULDBLOCK
) {
194 // Check current connection state
195 if (byConnected
== ES_DISCONNECTED
) {
198 byConnected
= ES_CONNECTED
; // ES_DISCONNECTED, ES_NOTCONNECTED, ES_CONNECTED
203 // CPU load improvement
204 if (downloadLimitEnable
&& downloadLimit
== 0){
205 pendingOnReceive
= true;
211 if (pendingHeaderSize
< PACKET_HEADER_SIZE
) {
212 delete[] pendingPacket
;
213 pendingPacket
= NULL
;
214 buf
= pendingHeader
+ pendingHeaderSize
;
215 readMax
= PACKET_HEADER_SIZE
- pendingHeaderSize
;
216 } else if (pendingPacket
== NULL
) {
217 pendingPacketSize
= 0;
218 readMax
= CPacket::GetPacketSizeFromHeader(pendingHeader
);
219 if (readMax
> MAX_PACKET_SIZE
) {
220 pendingHeaderSize
= 0;
224 pendingPacket
= new byte
[readMax
+ 1];
227 buf
= pendingPacket
+ pendingPacketSize
;
228 readMax
= CPacket::GetPacketSizeFromHeader(pendingHeader
) - pendingPacketSize
;
231 if (downloadLimitEnable
&& readMax
> downloadLimit
) {
232 readMax
= downloadLimit
;
237 wxMutexLocker
lock(m_sendLocker
);
238 ret
= Read(buf
, readMax
);
239 if (Error() || (ret
== 0)) {
240 if (LastError() == wxSOCKET_WOULDBLOCK
) {
241 pendingOnReceive
= true;
248 if (downloadLimitEnable
) {
250 if (ret
>= downloadLimit
) {
253 downloadLimit
-= ret
;
257 // CPU load improvement
258 // Detect if the socket's buffer is empty (or the size did match...)
259 pendingOnReceive
= (ret
== readMax
);
261 if (pendingHeaderSize
>= PACKET_HEADER_SIZE
) {
262 pendingPacketSize
+= ret
;
263 if (pendingPacketSize
>= CPacket::GetPacketSizeFromHeader(pendingHeader
)) {
264 CScopedPtr
<CPacket
> packet(new CPacket(pendingHeader
, pendingPacket
));
265 pendingPacket
= NULL
;
266 pendingPacketSize
= 0;
267 pendingHeaderSize
= 0;
269 // Bugfix We still need to check for a valid protocol
270 // Remark: the default eMule v0.26b had removed this test......
271 switch (packet
->GetProtocol()){
275 case OP_ED2KV2HEADER
:
276 case OP_ED2KV2PACKEDPROT
:
279 OnError(ERR_WRONGHEADER
);
284 PacketReceived(packet
.get());
287 pendingHeaderSize
+= ret
;
289 } while (ret
&& pendingHeaderSize
>= PACKET_HEADER_SIZE
);
293 void CEMSocket::SetDownloadLimit(uint32 limit
)
295 downloadLimit
= limit
;
296 downloadLimitEnable
= true;
298 // CPU load improvement
299 if(limit
> 0 && pendingOnReceive
== true){
305 void CEMSocket::DisableDownloadLimit()
307 downloadLimitEnable
= false;
309 // CPU load improvement
310 if (pendingOnReceive
== true){
317 * Queues up the packet to be sent. Another thread will actually send the packet.
319 * If the packet is not a control packet, and if the socket decides that its queue is
320 * full and forceAdd is false, then the socket is allowed to refuse to add the packet
321 * to its queue. It will then return false and it is up to the calling thread to try
322 * to call SendPacket for that packet again at a later time.
324 * @param packet address to the packet that should be added to the queue
326 * @param delpacket if true, the responsibility for deleting the packet after it has been sent
327 * has been transferred to this object. If false, don't delete the packet after it
330 * @param controlpacket the packet is a controlpacket
332 * @param forceAdd this packet must be added to the queue, even if it is full. If this flag is true
333 * then the method can not refuse to add the packet, and therefore not return false.
335 * @return true if the packet was added to the queue, false otherwise
337 void CEMSocket::SendPacket(CPacket
* packet
, bool delpacket
, bool controlpacket
, uint32 actualPayloadSize
)
339 //printf("* SendPacket called on socket %p\n", this);
340 wxMutexLocker
lock(m_sendLocker
);
342 if (byConnected
== ES_DISCONNECTED
) {
343 //printf("* Disconnected, drop packet\n");
349 packet
= new CPacket(*packet
);
353 //printf("* Adding a control packet\n");
354 m_control_queue
.push_back(packet
);
356 // queue up for controlpacket
357 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
359 //printf("* Adding a normal packet to the queue\n");
360 bool first
= !((sendbuffer
&& !m_currentPacket_is_controlpacket
) || !m_standard_queue
.empty());
361 StandardPacketQueueEntry queueEntry
= { actualPayloadSize
, packet
};
362 m_standard_queue
.push_back(queueEntry
);
364 // reset timeout for the first time
366 lastFinishedStandard
= ::GetTickCount();
367 m_bAccelerateUpload
= true; // Always accelerate first packet in a block
374 uint64
CEMSocket::GetSentBytesCompleteFileSinceLastCallAndReset()
376 wxMutexLocker
lock( m_sendLocker
);
378 uint64 sentBytes
= m_numberOfSentBytesCompleteFile
;
379 m_numberOfSentBytesCompleteFile
= 0;
385 uint64
CEMSocket::GetSentBytesPartFileSinceLastCallAndReset()
387 wxMutexLocker
lock( m_sendLocker
);
389 uint64 sentBytes
= m_numberOfSentBytesPartFile
;
390 m_numberOfSentBytesPartFile
= 0;
395 uint64
CEMSocket::GetSentBytesControlPacketSinceLastCallAndReset()
397 wxMutexLocker
lock( m_sendLocker
);
399 uint64 sentBytes
= m_numberOfSentBytesControlPacket
;
400 m_numberOfSentBytesControlPacket
= 0;
405 uint64
CEMSocket::GetSentPayloadSinceLastCallAndReset()
407 wxMutexLocker
lock( m_sendLocker
);
409 uint64 sentBytes
= m_actualPayloadSizeSent
;
410 m_actualPayloadSizeSent
= 0;
416 void CEMSocket::OnSend(int nErrorCode
)
423 CEncryptedStreamSocket::OnSend(0);
425 wxMutexLocker
lock( m_sendLocker
);
428 if (byConnected
!= ES_DISCONNECTED
) {
429 byConnected
= ES_CONNECTED
;
431 if (m_currentPacket_is_controlpacket
) {
432 // queue up for control packet
433 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
440 * Try to put queued up data on the socket.
442 * Control packets have higher priority, and will be sent first, if possible.
443 * Standard packets can be split up in several package containers. In that case
444 * all the parts of a split package must be sent in a row, without any control packet
447 * @param maxNumberOfBytesToSend This is the maximum number of bytes that is allowed to be put on the socket
448 * this call. The actual number of sent bytes will be returned from the method.
450 * @param onlyAllowedToSendControlPacket This call we only try to put control packets on the sockets.
451 * If there's a standard packet "in the way", and we think that this socket
452 * is no longer an upload slot, then it is ok to send the standard packet to
453 * get it out of the way. But it is not allowed to pick a new standard packet
454 * from the queue during this call. Several split packets are counted as one
455 * standard packet though, so it is ok to finish them all off if necessary.
457 * @return the actual number of bytes that were put on the socket.
459 SocketSentBytes
CEMSocket::Send(uint32 maxNumberOfBytesToSend
, uint32 minFragSize
, bool onlyAllowedToSendControlPacket
)
461 wxMutexLocker
lock(m_sendLocker
);
463 //printf("* Attempt to send a packet on socket %p\n", this);
465 if (byConnected
== ES_DISCONNECTED
) {
466 //printf("* Disconnected socket %p\n", this);
467 SocketSentBytes returnVal
= { false, 0, 0 };
469 } else if (m_bBusy
&& onlyAllowedToSendControlPacket
) {
470 //printf("* Busy socket %p\n", this);
471 SocketSentBytes returnVal
= { true, 0, 0 };
475 bool anErrorHasOccured
= false;
476 uint32 sentStandardPacketBytesThisCall
= 0;
477 uint32 sentControlPacketBytesThisCall
= 0;
479 if (byConnected
== ES_CONNECTED
&& IsEncryptionLayerReady() && (!m_bBusy
|| onlyAllowedToSendControlPacket
)) {
481 //printf("* Internal attemptto send on %p\n", this);
483 if(minFragSize
< 1) {
487 maxNumberOfBytesToSend
= GetNextFragSize(maxNumberOfBytesToSend
, minFragSize
);
489 bool bWasLongTimeSinceSend
= (::GetTickCount() - lastSent
) > 1000;
491 lastCalledSend
= ::GetTickCount();
494 while(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
< maxNumberOfBytesToSend
&& anErrorHasOccured
== false && // don't send more than allowed. Also, there should have been no error in earlier loop
495 (!m_control_queue
.empty() || !m_standard_queue
.empty() || sendbuffer
!= NULL
) && // there must exist something to send
496 (onlyAllowedToSendControlPacket
== false || // this means we are allowed to send both types of packets, so proceed
497 (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
> 0 && (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) % minFragSize
!= 0) ||
498 (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
499 (sendbuffer
!= NULL
&& m_currentPacket_is_controlpacket
== true) || // We are in the progress of sending a control packet. We are always allowed to send those
500 (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.
504 // If we are currently not in the progress of sending a packet, we will need to find the next one to send
505 if(sendbuffer
== NULL
) {
506 CPacket
* curPacket
= NULL
;
507 if(!m_control_queue
.empty()) {
508 // There's a control packet to send
509 m_currentPacket_is_controlpacket
= true;
510 curPacket
= m_control_queue
.front();
511 m_control_queue
.pop_front();
512 } else if(!m_standard_queue
.empty() /*&& onlyAllowedToSendControlPacket == false*/) {
513 // There's a standard packet to send
514 m_currentPacket_is_controlpacket
= false;
515 StandardPacketQueueEntry queueEntry
= m_standard_queue
.front();
516 m_standard_queue
.pop_front();
517 curPacket
= queueEntry
.packet
;
518 m_actualPayloadSize
= queueEntry
.actualPayloadSize
;
520 // remember this for statistics purposes.
521 m_currentPackageIsFromPartFile
= curPacket
->IsFromPF();
523 // Just to be safe. Shouldn't happen?
524 // if we reach this point, then there's something wrong with the while condition above!
526 AddDebugLogLineM(true, logGeneral
, wxT("EMSocket: Couldn't get a new packet! There's an error in the first while condition in EMSocket::Send()"));
528 SocketSentBytes returnVal
= { true, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
532 // We found a packet to send. Get the data to send from the
533 // package container and dispose of the container.
534 sendblen
= curPacket
->GetRealPacketSize();
535 sendbuffer
= curPacket
->DetachPacket();
539 CryptPrepareSendData((byte
*)sendbuffer
, sendblen
);
542 // At this point we've got a packet to send in sendbuffer. Try to send it. Loop until entire packet
543 // is sent, or until we reach maximum bytes to send for this call, or until we get an error.
544 // NOTE! If send would block (returns WOULDBLOCK), we will return from this method INSIDE this loop.
545 while (sent
< sendblen
&&
546 sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
< maxNumberOfBytesToSend
&&
548 onlyAllowedToSendControlPacket
== false || // this means we are allowed to send both types of packets, so proceed
549 m_currentPacket_is_controlpacket
||
550 (bWasLongTimeSinceSend
&& (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) < minFragSize
) ||
551 (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) % minFragSize
!= 0
553 anErrorHasOccured
== false) {
554 uint32 tosend
= sendblen
-sent
;
555 if(!onlyAllowedToSendControlPacket
|| m_currentPacket_is_controlpacket
) {
556 if (maxNumberOfBytesToSend
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> maxNumberOfBytesToSend
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
557 tosend
= maxNumberOfBytesToSend
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
558 } else if(bWasLongTimeSinceSend
&& (sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
) < minFragSize
) {
559 if (minFragSize
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> minFragSize
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
560 tosend
= minFragSize
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
562 uint32 nextFragMaxBytesToSent
= GetNextFragSize(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
, minFragSize
);
563 if (nextFragMaxBytesToSent
>= sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
&& tosend
> nextFragMaxBytesToSent
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
))
564 tosend
= nextFragMaxBytesToSent
-(sentStandardPacketBytesThisCall
+ sentControlPacketBytesThisCall
);
566 wxASSERT(tosend
!= 0 && tosend
<= sendblen
-sent
);
568 //DWORD tempStartSendTick = ::GetTickCount();
570 lastSent
= ::GetTickCount();
572 uint32 result
= CEncryptedStreamSocket::Write(sendbuffer
+sent
,tosend
);
576 uint32 error
= LastError();
577 if (error
== wxSOCKET_WOULDBLOCK
){
580 SocketSentBytes returnVal
= { true, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
582 return returnVal
; // Send() blocked, onsend will be called when ready to send again
584 // Send() gave an error
585 anErrorHasOccured
= true;
588 // we managed to send some bytes. Perform bookkeeping.
594 // Log send bytes in correct class
595 if(m_currentPacket_is_controlpacket
== false) {
596 sentStandardPacketBytesThisCall
+= result
;
598 if(m_currentPackageIsFromPartFile
== true) {
599 m_numberOfSentBytesPartFile
+= result
;
601 m_numberOfSentBytesCompleteFile
+= result
;
604 sentControlPacketBytesThisCall
+= result
;
605 m_numberOfSentBytesControlPacket
+= result
;
610 if (sent
== sendblen
){
611 // we are done sending the current packet. Delete it and set
612 // sendbuffer to NULL so a new packet can be fetched.
617 if(!m_currentPacket_is_controlpacket
) {
618 m_actualPayloadSizeSent
+= m_actualPayloadSize
;
619 m_actualPayloadSize
= 0;
621 lastFinishedStandard
= ::GetTickCount(); // reset timeout
622 m_bAccelerateUpload
= false; // Safe until told otherwise
630 if(onlyAllowedToSendControlPacket
&& (!m_control_queue
.empty() || (sendbuffer
!= NULL
&& m_currentPacket_is_controlpacket
))) {
631 // enter control packet send queue
632 // we might enter control packet queue several times for the same package,
633 // but that costs very little overhead. Less overhead than trying to make sure
634 // that we only enter the queue once.
635 //printf("* Requeueing control packet on %p\n", this);
636 theApp
->uploadBandwidthThrottler
->QueueForSendingControlPacket(this, HasSent());
639 //printf("* Finishing send debug on %p\n",this);
641 SocketSentBytes returnVal
= { !anErrorHasOccured
, sentStandardPacketBytesThisCall
, sentControlPacketBytesThisCall
};
647 uint32
CEMSocket::GetNextFragSize(uint32 current
, uint32 minFragSize
)
649 if(current
% minFragSize
== 0) {
652 return minFragSize
*(current
/minFragSize
+1);
658 * Decides the (minimum) amount the socket needs to send to prevent timeout.
662 uint32
CEMSocket::GetNeededBytes()
668 uint64 sizeleft
, sizetotal
;
671 wxMutexLocker
lock(m_sendLocker
);
673 if (byConnected
== ES_DISCONNECTED
) {
677 if (!((sendbuffer
&& !m_currentPacket_is_controlpacket
) || !m_standard_queue
.empty())) {
678 // No standard packet to send. Even if data needs to be sent to prevent timout, there's nothing to send.
682 if (((sendbuffer
&& !m_currentPacket_is_controlpacket
)) && !m_control_queue
.empty())
683 m_bAccelerateUpload
= true; // We might be trying to send a block request, accelerate packet
685 sendgap
= ::GetTickCount() - lastCalledSend
;
687 timetotal
= m_bAccelerateUpload
?45000:90000;
688 timeleft
= ::GetTickCount() - lastFinishedStandard
;
689 if (sendbuffer
&& !m_currentPacket_is_controlpacket
) {
690 sizeleft
= sendblen
-sent
;
691 sizetotal
= sendblen
;
693 sizeleft
= sizetotal
= m_standard_queue
.front().packet
->GetRealPacketSize();
697 if (timeleft
>= timetotal
)
699 timeleft
= timetotal
-timeleft
;
700 if (timeleft
*sizetotal
>= timetotal
*sizeleft
) {
701 // don't use 'GetTimeOut' here in case the timeout value is high,
702 if (sendgap
> SEC2MS(20))
703 return 1; // Don't let the socket itself time out - Might happen when switching from spread(non-focus) slot to trickle slot
706 uint64 decval
= timeleft
*sizetotal
/timetotal
;
709 if (decval
< sizeleft
)
710 return sizeleft
-decval
+1; // Round up
717 * 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.
719 * Before a socket can send a new packet, the current packet has to be finished. If the current packet is part of
720 * a split packet, then all parts of that split packet must be sent before the socket can send a control packet.
722 * This method keeps in standard queue only those packets that must be sent (rest of split packet), and removes everything
723 * after it. The method doesn't touch the control packet queue.
725 void CEMSocket::TruncateQueues()
727 wxMutexLocker
lock(m_sendLocker
);
729 // Clear the standard queue totally
730 // Please note! There may still be a standardpacket in the sendbuffer variable!
731 CStdPacketQueue::iterator it
= m_standard_queue
.begin();
732 for (; it
!= m_standard_queue
.end(); ++it
) {
736 m_standard_queue
.clear();
740 uint32
CEMSocket::GetTimeOut() const
746 void CEMSocket::SetTimeOut(uint32 uTimeOut
)
748 m_uTimeOut
= uTimeOut
;
750 // File_checked_for_headers