namespace con
{
+/******************************************************************************/
+/* defines used for debugging and profiling */
+/******************************************************************************/
+#ifdef NDEBUG
+#define LOG(a) a
+#define PROFILE(a)
+#undef DEBUG_CONNECTION_KBPS
+#else
/* this mutex is used to achieve log message consistency */
JMutex log_message_mutex;
#define LOG(a) \
JMutexAutoLock loglock(log_message_mutex); \
a; \
}
-
-/******************************************************************************/
-/* defines used for debugging and profiling */
-/******************************************************************************/
#define PROFILE(a) a
-//#define PROFILE(a)
-
//#define DEBUG_CONNECTION_KBPS
#undef DEBUG_CONNECTION_KBPS
+#endif
static inline float CALC_DTIME(unsigned int lasttime, unsigned int curtime) {
UDPPeer::UDPPeer(u16 a_id, Address a_address, Connection* connection) :
Peer(a_address,a_id,connection),
+ m_pending_disconnect(false),
resend_timeout(0.5),
m_legacy_peer(true)
{
}
-UDPPeer::~UDPPeer()
-{
-}
-
bool UDPPeer::getAddress(MTProtocols type,Address& toset)
{
if ((type == UDP) || (type == MINETEST_RELIABLE_UDP) || (type == PRIMARY))
return false;
}
+void UDPPeer::setNonLegacyPeer()
+{
+ m_legacy_peer = false;
+ for(unsigned int i=0; i< CHANNEL_COUNT; i++)
+ {
+ channels->setWindowSize(g_settings->getU16("max_packets_per_iteration"));
+ }
+}
+
void UDPPeer::reportRTT(float rtt)
{
if (rtt < 0.0) {
void UDPPeer::PutReliableSendCommand(ConnectionCommand &c,
unsigned int max_packet_size)
{
+ if (m_pending_disconnect)
+ return;
+
if ( channels[c.channelnum].queued_commands.empty() &&
/* don't queue more packets then window size */
(channels[c.channelnum].queued_reliables.size()
ConnectionCommand &c,
unsigned int max_packet_size)
{
+ if (m_pending_disconnect)
+ return true;
+
u32 chunksize_max = max_packet_size
- BASE_HEADER_SIZE
- RELIABLE_HEADER_SIZE;
case CONNCMD_SERVE:
case CONNCMD_CONNECT:
case CONNCMD_DISCONNECT:
- case CONNCMD_DELETE_PEER:
case CONCMD_ACK:
assert("Got command that shouldn't be reliable as reliable command" == 0);
default:
LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
sendToAll(c.channelnum, c.data);
return;
- case CONNCMD_DELETE_PEER:
- LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_DELETE_PEER"<<std::endl);
- m_connection->deletePeer(c.peer_id, false);
- return;
case CONCMD_ACK:
LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONCMD_ACK"<<std::endl);
sendAsPacket(c.peer_id,c.channelnum,c.data,true);
writeU8(&data[0], TYPE_CONTROL);
writeU8(&data[1], CONTROLTYPE_DISCO);
sendAsPacket(peer_id, 0,data,false);
+
+ PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+
+ if (!peer)
+ return;
+
+ if (dynamic_cast<UDPPeer*>(&peer) == 0)
+ {
+ return;
+ }
+
+ dynamic_cast<UDPPeer*>(&peer)->m_pending_disconnect = true;
}
void ConnectionSendThread::send(u16 peer_id, u8 channelnum,
void ConnectionSendThread::sendPackets(float dtime)
{
std::list<u16> peerIds = m_connection->getPeerIDs();
+ std::list<u16> pendingDisconnect;
+ std::map<u16,bool> pending_unreliable;
for(std::list<u16>::iterator
j = peerIds.begin();
continue;
}
+ if (dynamic_cast<UDPPeer*>(&peer)->m_pending_disconnect)
+ {
+ pendingDisconnect.push_back(*j);
+ }
+
PROFILE(std::stringstream peerIdentifier);
PROFILE(peerIdentifier << "sendPackets[" << m_connection->getDesc() << ";" << *j << ";RELIABLE]");
PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG));
}
else {
m_outgoing_queue.push_back(packet);
+ pending_unreliable[packet.peer_id] = true;
+ }
+ }
+
+ for(std::list<u16>::iterator
+ k = pendingDisconnect.begin();
+ k != pendingDisconnect.end(); ++k)
+ {
+ if (!pending_unreliable[*k])
+ {
+ m_connection->deletePeer(*k,false);
}
}
}
// Receive packets from the network and buffers and create ConnectionEvents
void ConnectionReceiveThread::receive()
{
- /* now reorder reliables */
- u32 datasize = m_max_packet_size * 2; // Double it just to be safe
- // TODO: We can not know how many layers of header there are.
- // For now, just assume there are no other than the base headers.
- u32 packet_maxsize = datasize + BASE_HEADER_SIZE;
+ // use IPv6 minimum allowed MTU as receive buffer size as this is
+ // theoretical reliable upper boundary of a udp packet for all IPv6 enabled
+ // infrastructure
+ unsigned int packet_maxsize = 1500;
SharedBuffer<u8> packetdata(packet_maxsize);
bool packet_queued = true;
LOG(dout_con<<m_connection->getDesc()
<<" ProcessPacket from peer_id: " << peer_id
- << ",channel: " << channelnum << ", returned "
+ << ",channel: " << (channelnum & 0xFF) << ", returned "
<< resultdata.getSize() << " bytes" <<std::endl);
ConnectionEvent e;
}
//put bytes for max bandwidth calculation
channel->UpdateBytesSent(p.data.getSize(),1);
+ if (channel->outgoing_reliables_sent.size() == 0)
+ {
+ m_connection->TriggerSend();
+ }
}
catch(NotFoundException &e){
LOG(derr_con<<m_connection->getDesc()
m_info_mutex(),
m_bc_peerhandler(0),
m_bc_receive_timeout(0),
- m_shutting_down(false)
+ m_shutting_down(false),
+ m_next_remote_peer_id(2)
{
m_udpSocket.setTimeoutMs(5);
m_info_mutex(),
m_bc_peerhandler(peerhandler),
m_bc_receive_timeout(0),
- m_shutting_down(false)
+ m_shutting_down(false),
+ m_next_remote_peer_id(2)
{
m_udpSocket.setTimeoutMs(5);
putCommand(c);
}
-void Connection::RunTimeouts(float dtime)
-{
- // No-op
-}
-
Address Connection::GetPeerAddress(u16 peer_id)
{
PeerHelper peer = getPeerNoEx(peer_id);
// Somebody wants to make a new connection
// Get a unique peer id (2 or higher)
- u16 peer_id_new = 2;
+ u16 peer_id_new = m_next_remote_peer_id;
u16 overflow = MAX_UDP_PEERS;
/*
Find an unused peer id
*/
- bool out_of_ids = false;
- for(;;)
{
- // Check if exists
- if(m_peers.find(peer_id_new) == m_peers.end())
- break;
- // Check for overflow
- if(peer_id_new == overflow){
- out_of_ids = true;
- break;
+ JMutexAutoLock lock(m_peers_mutex);
+ bool out_of_ids = false;
+ for(;;)
+ {
+ // Check if exists
+ if(m_peers.find(peer_id_new) == m_peers.end())
+ break;
+ // Check for overflow
+ if(peer_id_new == overflow){
+ out_of_ids = true;
+ break;
+ }
+ peer_id_new++;
+ }
+ if(out_of_ids){
+ errorstream<<getDesc()<<" ran out of peer ids"<<std::endl;
+ return PEER_ID_INEXISTENT;
}
- peer_id_new++;
- }
- if(out_of_ids){
- errorstream<<getDesc()<<" ran out of peer ids"<<std::endl;
- return PEER_ID_INEXISTENT;
- }
-
- LOG(dout_con<<getDesc()
- <<"createPeer(): giving peer_id="<<peer_id_new<<std::endl);
- // Create a peer
- Peer *peer = 0;
+ // Create a peer
+ Peer *peer = 0;
+ peer = new UDPPeer(peer_id_new, sender, this);
- peer = new UDPPeer(peer_id_new, sender, this);
+ m_peers[peer->id] = peer;
+ }
- m_peers_mutex.Lock();
- m_peers[peer->id] = peer;
- m_peers_mutex.Unlock();
+ m_next_remote_peer_id = (peer_id_new +1) % MAX_UDP_PEERS;
- // Create peer addition event
- ConnectionEvent e;
- e.peerAdded(peer_id_new, sender);
- putEvent(e);
+ LOG(dout_con<<getDesc()
+ <<"createPeer(): giving peer_id="<<peer_id_new<<std::endl);
ConnectionCommand cmd;
SharedBuffer<u8> reply(4);
cmd.createPeer(peer_id_new,reply);
this->putCommand(cmd);
+ // Create peer addition event
+ ConnectionEvent e;
+ e.peerAdded(peer_id_new, sender);
+ putEvent(e);
+
// We're now talking to a valid peer_id
return peer_id_new;
}
-void Connection::DeletePeer(u16 peer_id)
-{
- ConnectionCommand c;
- c.deletePeer(peer_id);
- putCommand(c);
-}
-
void Connection::PrintInfo(std::ostream &out)
{
m_info_mutex.Lock();
return std::string("con(")+itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")";
}
+void Connection::DisconnectPeer(u16 peer_id)
+{
+ ConnectionCommand discon;
+ discon.disconnect_peer(peer_id);
+ putCommand(discon);
+}
+
void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum) {
assert(channelnum < CHANNEL_COUNT);
projects / dragonfireclient.git / blobdiff