From fe5cb2cdfb137764d20ca56f17f607ec361e0dcf Mon Sep 17 00:00:00 2001 From: sfan5 Date: Sun, 10 Oct 2021 21:10:52 +0200 Subject: Remove broken timeout behaviour Code that relies on `resend_count` was added in 7ea4a03 and 247a1eb, but never worked. This was fixed in #11607 which caused the problem to surface. Hence undo the first commit entirely and change the logic of the second. --- src/network/connectionthreads.cpp | 27 ++++----------------------- 1 file changed, 4 insertions(+), 23 deletions(-) (limited to 'src/network/connectionthreads.cpp') diff --git a/src/network/connectionthreads.cpp b/src/network/connectionthreads.cpp index 47678dac5..a306ced9b 100644 --- a/src/network/connectionthreads.cpp +++ b/src/network/connectionthreads.cpp @@ -48,9 +48,6 @@ std::mutex log_conthread_mutex; #undef DEBUG_CONNECTION_KBPS #endif -/* maximum number of retries for reliable packets */ -#define MAX_RELIABLE_RETRY 5 - #define WINDOW_SIZE 5 static session_t readPeerId(u8 *packetdata) @@ -212,7 +209,6 @@ void ConnectionSendThread::runTimeouts(float dtime) } float resend_timeout = udpPeer->getResendTimeout(); - bool retry_count_exceeded = false; for (Channel &channel : udpPeer->channels) { // Remove timed out incomplete unreliable split packets @@ -231,24 +227,16 @@ void ConnectionSendThread::runTimeouts(float dtime) m_iteration_packets_avaialble -= timed_outs.size(); for (const auto &k : timed_outs) { - session_t peer_id = readPeerId(*k.data); u8 channelnum = readChannel(*k.data); u16 seqnum = readU16(&(k.data[BASE_HEADER_SIZE + 1])); channel.UpdateBytesLost(k.data.getSize()); - if (k.resend_count > MAX_RELIABLE_RETRY) { - retry_count_exceeded = true; - timeouted_peers.push_back(peer->id); - /* no need to check additional packets if a single one did timeout*/ - break; - } - LOG(derr_con << m_connection->getDesc() << "RE-SENDING timed-out RELIABLE to " << k.address.serializeString() << "(t/o=" << resend_timeout << "): " - << "from_peer_id=" << peer_id + << "count=" << k.resend_count << ", channel=" << ((int) channelnum & 0xff) << ", seqnum=" << seqnum << std::endl); @@ -259,17 +247,9 @@ void ConnectionSendThread::runTimeouts(float dtime) // lost or really takes more time to transmit } - if (retry_count_exceeded) { - break; /* no need to check other channels if we already did timeout */ - } - channel.UpdateTimers(dtime); } - /* skip to next peer if we did timeout */ - if (retry_count_exceeded) - continue; - /* send ping if necessary */ if (udpPeer->Ping(dtime, data)) { LOG(dout_con << m_connection->getDesc() @@ -1153,8 +1133,8 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan try { BufferedPacket p = channel->outgoing_reliables_sent.popSeqnum(seqnum); - // only calculate rtt from straight sent packets - if (p.resend_count == 0) { + // the rtt calculation will be a bit off for re-sent packets but that's okay + { // Get round trip time u64 current_time = porting::getTimeMs(); @@ -1174,6 +1154,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan dynamic_cast(peer)->reportRTT(rtt); } } + // put bytes for max bandwidth calculation channel->UpdateBytesSent(p.data.getSize(), 1); if (channel->outgoing_reliables_sent.size() == 0) -- cgit v1.2.3 From 57a59ae92d4bbfa4fdd60d7acd72c6440f63a49c Mon Sep 17 00:00:00 2001 From: SmallJoker Date: Wed, 1 Dec 2021 20:22:33 +0100 Subject: Network: Delete copy constructor and use std::move instead (#11642) This is a follow-up change which disables class copies where possible to avoid unnecessary memory movements. --- src/client/client.cpp | 2 +- src/network/connection.cpp | 445 ++++++++++++++++++++++---------------- src/network/connection.h | 416 ++++++++++++++++------------------- src/network/connectionthreads.cpp | 190 ++++++++-------- src/network/connectionthreads.h | 31 ++- src/unittest/test_connection.cpp | 10 +- src/util/pointer.h | 58 ++--- 7 files changed, 600 insertions(+), 552 deletions(-) (limited to 'src/network/connectionthreads.cpp') diff --git a/src/client/client.cpp b/src/client/client.cpp index 45cc62a33..3ee1298ff 100644 --- a/src/client/client.cpp +++ b/src/client/client.cpp @@ -877,7 +877,7 @@ void Client::ProcessData(NetworkPacket *pkt) */ if(sender_peer_id != PEER_ID_SERVER) { infostream << "Client::ProcessData(): Discarding data not " - "coming from server: peer_id=" << sender_peer_id + "coming from server: peer_id=" << sender_peer_id << " command=" << pkt->getCommand() << std::endl; return; } diff --git a/src/network/connection.cpp b/src/network/connection.cpp index 548b2e3a0..2d3cf6e88 100644 --- a/src/network/connection.cpp +++ b/src/network/connection.cpp @@ -62,18 +62,27 @@ namespace con #define PING_TIMEOUT 5.0 -BufferedPacket makePacket(Address &address, const SharedBuffer &data, +u16 BufferedPacket::getSeqnum() const +{ + if (size() < BASE_HEADER_SIZE + 3) + return 0; // should never happen + + return readU16(&data[BASE_HEADER_SIZE + 1]); +} + +BufferedPacketPtr makePacket(Address &address, const SharedBuffer &data, u32 protocol_id, session_t sender_peer_id, u8 channel) { u32 packet_size = data.getSize() + BASE_HEADER_SIZE; - BufferedPacket p(packet_size); - p.address = address; - writeU32(&p.data[0], protocol_id); - writeU16(&p.data[4], sender_peer_id); - writeU8(&p.data[6], channel); + BufferedPacketPtr p(new BufferedPacket(packet_size)); + p->address = address; + + writeU32(&p->data[0], protocol_id); + writeU16(&p->data[4], sender_peer_id); + writeU8(&p->data[6], channel); - memcpy(&p.data[BASE_HEADER_SIZE], *data, data.getSize()); + memcpy(&p->data[BASE_HEADER_SIZE], *data, data.getSize()); return p; } @@ -169,9 +178,8 @@ void ReliablePacketBuffer::print() MutexAutoLock listlock(m_list_mutex); LOG(dout_con<<"Dump of ReliablePacketBuffer:" << std::endl); unsigned int index = 0; - for (BufferedPacket &bufferedPacket : m_list) { - u16 s = readU16(&(bufferedPacket.data[BASE_HEADER_SIZE+1])); - LOG(dout_con<getSeqnum() << std::endl); index++; } } @@ -188,16 +196,13 @@ u32 ReliablePacketBuffer::size() return m_list.size(); } -RPBSearchResult ReliablePacketBuffer::findPacket(u16 seqnum) +RPBSearchResult ReliablePacketBuffer::findPacketNoLock(u16 seqnum) { - std::list::iterator i = m_list.begin(); - for(; i != m_list.end(); ++i) - { - u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1])); - if (s == seqnum) - break; + for (auto it = m_list.begin(); it != m_list.end(); ++it) { + if ((*it)->getSeqnum() == seqnum) + return it; } - return i; + return m_list.end(); } bool ReliablePacketBuffer::getFirstSeqnum(u16& result) @@ -205,54 +210,54 @@ bool ReliablePacketBuffer::getFirstSeqnum(u16& result) MutexAutoLock listlock(m_list_mutex); if (m_list.empty()) return false; - const BufferedPacket &p = m_list.front(); - result = readU16(&p.data[BASE_HEADER_SIZE + 1]); + result = m_list.front()->getSeqnum(); return true; } -BufferedPacket ReliablePacketBuffer::popFirst() +BufferedPacketPtr ReliablePacketBuffer::popFirst() { MutexAutoLock listlock(m_list_mutex); if (m_list.empty()) throw NotFoundException("Buffer is empty"); - BufferedPacket p = std::move(m_list.front()); + + BufferedPacketPtr p(m_list.front()); m_list.pop_front(); if (m_list.empty()) { m_oldest_non_answered_ack = 0; } else { - m_oldest_non_answered_ack = - readU16(&m_list.front().data[BASE_HEADER_SIZE + 1]); + m_oldest_non_answered_ack = m_list.front()->getSeqnum(); } return p; } -BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum) +BufferedPacketPtr ReliablePacketBuffer::popSeqnum(u16 seqnum) { MutexAutoLock listlock(m_list_mutex); - RPBSearchResult r = findPacket(seqnum); - if (r == notFound()) { + RPBSearchResult r = findPacketNoLock(seqnum); + if (r == m_list.end()) { LOG(dout_con<<"Sequence number: " << seqnum << " not found in reliable buffer"<getSeqnum(); } return p; } -void ReliablePacketBuffer::insert(const BufferedPacket &p, u16 next_expected) +void ReliablePacketBuffer::insert(BufferedPacketPtr &p_ptr, u16 next_expected) { MutexAutoLock listlock(m_list_mutex); - if (p.data.getSize() < BASE_HEADER_SIZE + 3) { + const BufferedPacket &p = *p_ptr; + + if (p.size() < BASE_HEADER_SIZE + 3) { errorstream << "ReliablePacketBuffer::insert(): Invalid data size for " "reliable packet" << std::endl; return; @@ -263,7 +268,7 @@ void ReliablePacketBuffer::insert(const BufferedPacket &p, u16 next_expected) << std::endl; return; } - u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE + 1]); + const u16 seqnum = p.getSeqnum(); if (!seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE)) { errorstream << "ReliablePacketBuffer::insert(): seqnum is outside of " @@ -280,44 +285,44 @@ void ReliablePacketBuffer::insert(const BufferedPacket &p, u16 next_expected) // Find the right place for the packet and insert it there // If list is empty, just add it - if (m_list.empty()) - { - m_list.push_back(p); + if (m_list.empty()) { + m_list.push_back(p_ptr); m_oldest_non_answered_ack = seqnum; // Done. return; } // Otherwise find the right place - std::list::iterator i = m_list.begin(); + auto it = m_list.begin(); // Find the first packet in the list which has a higher seqnum - u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1])); + u16 s = (*it)->getSeqnum(); /* case seqnum is smaller then next_expected seqnum */ /* this is true e.g. on wrap around */ if (seqnum < next_expected) { - while(((s < seqnum) || (s >= next_expected)) && (i != m_list.end())) { - ++i; - if (i != m_list.end()) - s = readU16(&(i->data[BASE_HEADER_SIZE+1])); + while(((s < seqnum) || (s >= next_expected)) && (it != m_list.end())) { + ++it; + if (it != m_list.end()) + s = (*it)->getSeqnum(); } } /* non wrap around case (at least for incoming and next_expected */ else { - while(((s < seqnum) && (s >= next_expected)) && (i != m_list.end())) { - ++i; - if (i != m_list.end()) - s = readU16(&(i->data[BASE_HEADER_SIZE+1])); + while(((s < seqnum) && (s >= next_expected)) && (it != m_list.end())) { + ++it; + if (it != m_list.end()) + s = (*it)->getSeqnum(); } } if (s == seqnum) { /* nothing to do this seems to be a resent packet */ /* for paranoia reason data should be compared */ + auto &i = *it; if ( - (readU16(&(i->data[BASE_HEADER_SIZE+1])) != seqnum) || - (i->data.getSize() != p.data.getSize()) || + (i->getSeqnum() != seqnum) || + (i->size() != p.size()) || (i->address != p.address) ) { @@ -325,51 +330,52 @@ void ReliablePacketBuffer::insert(const BufferedPacket &p, u16 next_expected) fprintf(stderr, "Duplicated seqnum %d non matching packet detected:\n", seqnum); - fprintf(stderr, "Old: seqnum: %05d size: %04d, address: %s\n", - readU16(&(i->data[BASE_HEADER_SIZE+1])),i->data.getSize(), + fprintf(stderr, "Old: seqnum: %05d size: %04zu, address: %s\n", + i->getSeqnum(), i->size(), i->address.serializeString().c_str()); - fprintf(stderr, "New: seqnum: %05d size: %04u, address: %s\n", - readU16(&(p.data[BASE_HEADER_SIZE+1])),p.data.getSize(), + fprintf(stderr, "New: seqnum: %05d size: %04zu, address: %s\n", + p.getSeqnum(), p.size(), p.address.serializeString().c_str()); throw IncomingDataCorruption("duplicated packet isn't same as original one"); } } /* insert or push back */ - else if (i != m_list.end()) { - m_list.insert(i, p); + else if (it != m_list.end()) { + m_list.insert(it, p_ptr); } else { - m_list.push_back(p); + m_list.push_back(p_ptr); } /* update last packet number */ - m_oldest_non_answered_ack = readU16(&m_list.front().data[BASE_HEADER_SIZE+1]); + m_oldest_non_answered_ack = m_list.front()->getSeqnum(); } void ReliablePacketBuffer::incrementTimeouts(float dtime) { MutexAutoLock listlock(m_list_mutex); - for (BufferedPacket &bufferedPacket : m_list) { - bufferedPacket.time += dtime; - bufferedPacket.totaltime += dtime; + for (auto &packet : m_list) { + packet->time += dtime; + packet->totaltime += dtime; } } -std::list +std::list> ReliablePacketBuffer::getTimedOuts(float timeout, u32 max_packets) { MutexAutoLock listlock(m_list_mutex); - std::list timed_outs; - for (BufferedPacket &bufferedPacket : m_list) { - if (bufferedPacket.time >= timeout) { - // caller will resend packet so reset time and increase counter - bufferedPacket.time = 0.0f; - bufferedPacket.resend_count++; + std::list> timed_outs; + for (auto &packet : m_list) { + if (packet->time < timeout) + continue; - timed_outs.push_back(bufferedPacket); + // caller will resend packet so reset time and increase counter + packet->time = 0.0f; + packet->resend_count++; - if (timed_outs.size() >= max_packets) - break; - } + timed_outs.emplace_back(packet); + + if (timed_outs.size() >= max_packets) + break; } return timed_outs; } @@ -428,11 +434,13 @@ IncomingSplitBuffer::~IncomingSplitBuffer() } } -SharedBuffer IncomingSplitBuffer::insert(const BufferedPacket &p, bool reliable) +SharedBuffer IncomingSplitBuffer::insert(BufferedPacketPtr &p_ptr, bool reliable) { MutexAutoLock listlock(m_map_mutex); + const BufferedPacket &p = *p_ptr; + u32 headersize = BASE_HEADER_SIZE + 7; - if (p.data.getSize() < headersize) { + if (p.size() < headersize) { errorstream << "Invalid data size for split packet" << std::endl; return SharedBuffer(); } @@ -473,7 +481,7 @@ SharedBuffer IncomingSplitBuffer::insert(const BufferedPacket &p, bool relia < chunkdata(chunkdatasize); memcpy(*chunkdata, &(p.data[headersize]), chunkdatasize); @@ -520,14 +528,67 @@ void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout) ConnectionCommand */ -void ConnectionCommand::send(session_t peer_id_, u8 channelnum_, NetworkPacket *pkt, - bool reliable_) +ConnectionCommandPtr ConnectionCommand::create(ConnectionCommandType type) +{ + return ConnectionCommandPtr(new ConnectionCommand(type)); +} + +ConnectionCommandPtr ConnectionCommand::serve(Address address) +{ + auto c = create(CONNCMD_SERVE); + c->address = address; + return c; +} + +ConnectionCommandPtr ConnectionCommand::connect(Address address) { - type = CONNCMD_SEND; - peer_id = peer_id_; - channelnum = channelnum_; - data = pkt->oldForgePacket(); - reliable = reliable_; + auto c = create(CONNCMD_CONNECT); + c->address = address; + return c; +} + +ConnectionCommandPtr ConnectionCommand::disconnect() +{ + return create(CONNCMD_DISCONNECT); +} + +ConnectionCommandPtr ConnectionCommand::disconnect_peer(session_t peer_id) +{ + auto c = create(CONNCMD_DISCONNECT_PEER); + c->peer_id = peer_id; + return c; +} + +ConnectionCommandPtr ConnectionCommand::send(session_t peer_id, u8 channelnum, + NetworkPacket *pkt, bool reliable) +{ + auto c = create(CONNCMD_SEND); + c->peer_id = peer_id; + c->channelnum = channelnum; + c->reliable = reliable; + c->data = pkt->oldForgePacket(); + return c; +} + +ConnectionCommandPtr ConnectionCommand::ack(session_t peer_id, u8 channelnum, const Buffer &data) +{ + auto c = create(CONCMD_ACK); + c->peer_id = peer_id; + c->channelnum = channelnum; + c->reliable = false; + data.copyTo(c->data); + return c; +} + +ConnectionCommandPtr ConnectionCommand::createPeer(session_t peer_id, const Buffer &data) +{ + auto c = create(CONCMD_CREATE_PEER); + c->peer_id = peer_id; + c->channelnum = 0; + c->reliable = true; + c->raw = true; + data.copyTo(c->data); + return c; } /* @@ -562,39 +623,38 @@ void Channel::setNextSplitSeqNum(u16 seqnum) u16 Channel::getOutgoingSequenceNumber(bool& successful) { MutexAutoLock internal(m_internal_mutex); + u16 retval = next_outgoing_seqnum; - u16 lowest_unacked_seqnumber; + successful = false; /* shortcut if there ain't any packet in outgoing list */ - if (outgoing_reliables_sent.empty()) - { + if (outgoing_reliables_sent.empty()) { + successful = true; next_outgoing_seqnum++; return retval; } - if (outgoing_reliables_sent.getFirstSeqnum(lowest_unacked_seqnumber)) - { + u16 lowest_unacked_seqnumber; + if (outgoing_reliables_sent.getFirstSeqnum(lowest_unacked_seqnumber)) { if (lowest_unacked_seqnumber < next_outgoing_seqnum) { // ugly cast but this one is required in order to tell compiler we // know about difference of two unsigned may be negative in general // but we already made sure it won't happen in this case if (((u16)(next_outgoing_seqnum - lowest_unacked_seqnumber)) > m_window_size) { - successful = false; return 0; } - } - else { + } else { // ugly cast but this one is required in order to tell compiler we // know about difference of two unsigned may be negative in general // but we already made sure it won't happen in this case if ((next_outgoing_seqnum + (u16)(SEQNUM_MAX - lowest_unacked_seqnumber)) > m_window_size) { - successful = false; return 0; } } } + successful = true; next_outgoing_seqnum++; return retval; } @@ -946,45 +1006,45 @@ bool UDPPeer::Ping(float dtime,SharedBuffer& data) return false; } -void UDPPeer::PutReliableSendCommand(ConnectionCommand &c, +void UDPPeer::PutReliableSendCommand(ConnectionCommandPtr &c, unsigned int max_packet_size) { if (m_pending_disconnect) return; - Channel &chan = channels[c.channelnum]; + Channel &chan = channels[c->channelnum]; if (chan.queued_commands.empty() && /* don't queue more packets then window size */ - (chan.queued_reliables.size() < chan.getWindowSize() / 2)) { + (chan.queued_reliables.size() + 1 < chan.getWindowSize() / 2)) { LOG(dout_con<getDesc() - <<" processing reliable command for peer id: " << c.peer_id - <<" data size: " << c.data.getSize() << std::endl); - if (!processReliableSendCommand(c,max_packet_size)) { - chan.queued_commands.push_back(c); - } - } - else { + <<" processing reliable command for peer id: " << c->peer_id + <<" data size: " << c->data.getSize() << std::endl); + if (processReliableSendCommand(c, max_packet_size)) + return; + } else { LOG(dout_con<getDesc() - <<" Queueing reliable command for peer id: " << c.peer_id - <<" data size: " << c.data.getSize() <= chan.getWindowSize() / 2) { + <<" Queueing reliable command for peer id: " << c->peer_id + <<" data size: " << c->data.getSize() <= chan.getWindowSize() / 2) { LOG(derr_con << m_connection->getDesc() - << "Possible packet stall to peer id: " << c.peer_id + << "Possible packet stall to peer id: " << c->peer_id << " queued_commands=" << chan.queued_commands.size() << std::endl); } } + chan.queued_commands.push_back(c); } bool UDPPeer::processReliableSendCommand( - ConnectionCommand &c, + ConnectionCommandPtr &c_ptr, unsigned int max_packet_size) { if (m_pending_disconnect) return true; + const auto &c = *c_ptr; Channel &chan = channels[c.channelnum]; u32 chunksize_max = max_packet_size @@ -1003,9 +1063,9 @@ bool UDPPeer::processReliableSendCommand( chan.setNextSplitSeqNum(split_sequence_number); } - bool have_sequence_number = true; + bool have_sequence_number = false; bool have_initial_sequence_number = false; - std::queue toadd; + std::queue toadd; volatile u16 initial_sequence_number = 0; for (SharedBuffer &original : originals) { @@ -1024,25 +1084,23 @@ bool UDPPeer::processReliableSendCommand( SharedBuffer reliable = makeReliablePacket(original, seqnum); // Add base headers and make a packet - BufferedPacket p = con::makePacket(address, reliable, + BufferedPacketPtr p = con::makePacket(address, reliable, m_connection->GetProtocolID(), m_connection->GetPeerID(), c.channelnum); - toadd.push(std::move(p)); + toadd.push(p); } if (have_sequence_number) { - volatile u16 pcount = 0; while (!toadd.empty()) { - BufferedPacket p = std::move(toadd.front()); + BufferedPacketPtr p = toadd.front(); toadd.pop(); // LOG(dout_con<getDesc() // << " queuing reliable packet for peer_id: " << c.peer_id // << " channel: " << (c.channelnum&0xFF) // << " seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1]) // << std::endl) - chan.queued_reliables.push(std::move(p)); - pcount++; + chan.queued_reliables.push(p); } sanity_check(chan.queued_reliables.size() < 0xFFFF); return true; @@ -1051,6 +1109,7 @@ bool UDPPeer::processReliableSendCommand( volatile u16 packets_available = toadd.size(); /* we didn't get a single sequence number no need to fill queue */ if (!have_initial_sequence_number) { + LOG(derr_con << m_connection->getDesc() << "Ran out of sequence numbers!" << std::endl); return false; } @@ -1096,18 +1155,18 @@ void UDPPeer::RunCommandQueues( (channel.queued_reliables.size() < maxtransfer) && (commands_processed < maxcommands)) { try { - ConnectionCommand c = channel.queued_commands.front(); + ConnectionCommandPtr c = channel.queued_commands.front(); LOG(dout_con << m_connection->getDesc() << " processing queued reliable command " << std::endl); // Packet is processed, remove it from queue - if (processReliableSendCommand(c,max_packet_size)) { + if (processReliableSendCommand(c, max_packet_size)) { channel.queued_commands.pop_front(); } else { LOG(dout_con << m_connection->getDesc() - << " Failed to queue packets for peer_id: " << c.peer_id - << ", delaying sending of " << c.data.getSize() + << " Failed to queue packets for peer_id: " << c->peer_id + << ", delaying sending of " << c->data.getSize() << " bytes" << std::endl); } } @@ -1130,13 +1189,70 @@ void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum) channels[channel].setNextSplitSeqNum(seqnum); } -SharedBuffer UDPPeer::addSplitPacket(u8 channel, const BufferedPacket &toadd, +SharedBuffer UDPPeer::addSplitPacket(u8 channel, BufferedPacketPtr &toadd, bool reliable) { assert(channel < CHANNEL_COUNT); // Pre-condition return channels[channel].incoming_splits.insert(toadd, reliable); } +/* + ConnectionEvent +*/ + +const char *ConnectionEvent::describe() const +{ + switch(type) { + case CONNEVENT_NONE: + return "CONNEVENT_NONE"; + case CONNEVENT_DATA_RECEIVED: + return "CONNEVENT_DATA_RECEIVED"; + case CONNEVENT_PEER_ADDED: + return "CONNEVENT_PEER_ADDED"; + case CONNEVENT_PEER_REMOVED: + return "CONNEVENT_PEER_REMOVED"; + case CONNEVENT_BIND_FAILED: + return "CONNEVENT_BIND_FAILED"; + } + return "Invalid ConnectionEvent"; +} + + +ConnectionEventPtr ConnectionEvent::create(ConnectionEventType type) +{ + return std::shared_ptr(new ConnectionEvent(type)); +} + +ConnectionEventPtr ConnectionEvent::dataReceived(session_t peer_id, const Buffer &data) +{ + auto e = create(CONNEVENT_DATA_RECEIVED); + e->peer_id = peer_id; + data.copyTo(e->data); + return e; +} + +ConnectionEventPtr ConnectionEvent::peerAdded(session_t peer_id, Address address) +{ + auto e = create(CONNEVENT_PEER_ADDED); + e->peer_id = peer_id; + e->address = address; + return e; +} + +ConnectionEventPtr ConnectionEvent::peerRemoved(session_t peer_id, bool is_timeout, Address address) +{ + auto e = create(CONNEVENT_PEER_REMOVED); + e->peer_id = peer_id; + e->timeout = is_timeout; + e->address = address; + return e; +} + +ConnectionEventPtr ConnectionEvent::bindFailed() +{ + return create(CONNEVENT_BIND_FAILED); +} + /* Connection */ @@ -1186,18 +1302,12 @@ Connection::~Connection() /* Internal stuff */ -void Connection::putEvent(const ConnectionEvent &e) +void Connection::putEvent(ConnectionEventPtr e) { - assert(e.type != CONNEVENT_NONE); // Pre-condition + assert(e->type != CONNEVENT_NONE); // Pre-condition m_event_queue.push_back(e); } -void Connection::putEvent(ConnectionEvent &&e) -{ - assert(e.type != CONNEVENT_NONE); // Pre-condition - m_event_queue.push_back(std::move(e)); -} - void Connection::TriggerSend() { m_sendThread->Trigger(); @@ -1260,11 +1370,9 @@ bool Connection::deletePeer(session_t peer_id, bool timeout) Address peer_address; //any peer has a primary address this never fails! peer->getAddress(MTP_PRIMARY, peer_address); - // Create event - ConnectionEvent e; - e.peerRemoved(peer_id, timeout, peer_address); - putEvent(e); + // Create event + putEvent(ConnectionEvent::peerRemoved(peer_id, timeout, peer_address)); peer->Drop(); return true; @@ -1272,18 +1380,16 @@ bool Connection::deletePeer(session_t peer_id, bool timeout) /* Interface */ -ConnectionEvent Connection::waitEvent(u32 timeout_ms) +ConnectionEventPtr Connection::waitEvent(u32 timeout_ms) { try { return m_event_queue.pop_front(timeout_ms); } catch(ItemNotFoundException &ex) { - ConnectionEvent e; - e.type = CONNEVENT_NONE; - return e; + return ConnectionEvent::create(CONNEVENT_NONE); } } -void Connection::putCommand(const ConnectionCommand &c) +void Connection::putCommand(ConnectionCommandPtr c) { if (!m_shutting_down) { m_command_queue.push_back(c); @@ -1291,26 +1397,14 @@ void Connection::putCommand(const ConnectionCommand &c) } } -void Connection::putCommand(ConnectionCommand &&c) -{ - if (!m_shutting_down) { - m_command_queue.push_back(std::move(c)); - m_sendThread->Trigger(); - } -} - void Connection::Serve(Address bind_addr) { - ConnectionCommand c; - c.serve(bind_addr); - putCommand(c); + putCommand(ConnectionCommand::serve(bind_addr)); } void Connection::Connect(Address address) { - ConnectionCommand c; - c.connect(address); - putCommand(c); + putCommand(ConnectionCommand::connect(address)); } bool Connection::Connected() @@ -1332,9 +1426,7 @@ bool Connection::Connected() void Connection::Disconnect() { - ConnectionCommand c; - c.disconnect(); - putCommand(c); + putCommand(ConnectionCommand::disconnect()); } bool Connection::Receive(NetworkPacket *pkt, u32 timeout) @@ -1345,11 +1437,15 @@ bool Connection::Receive(NetworkPacket *pkt, u32 timeout) This is not considered to be a problem (is it?) */ for(;;) { - ConnectionEvent e = waitEvent(timeout); - if (e.type != CONNEVENT_NONE) + ConnectionEventPtr e_ptr = waitEvent(timeout); + const ConnectionEvent &e = *e_ptr; + + if (e.type != CONNEVENT_NONE) { LOG(dout_con << getDesc() << ": Receive: got event: " << e.describe() << std::endl); - switch(e.type) { + } + + switch (e.type) { case CONNEVENT_NONE: return false; case CONNEVENT_DATA_RECEIVED: @@ -1397,10 +1493,7 @@ void Connection::Send(session_t peer_id, u8 channelnum, { assert(channelnum < CHANNEL_COUNT); // Pre-condition - ConnectionCommand c; - - c.send(peer_id, channelnum, pkt, reliable); - putCommand(std::move(c)); + putCommand(ConnectionCommand::send(peer_id, channelnum, pkt, reliable)); } Address Connection::GetPeerAddress(session_t peer_id) @@ -1499,41 +1592,31 @@ u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd) LOG(dout_con << getDesc() << "createPeer(): giving peer_id=" << peer_id_new << std::endl); - ConnectionCommand cmd; - Buffer reply(4); - writeU8(&reply[0], PACKET_TYPE_CONTROL); - writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID); - writeU16(&reply[2], peer_id_new); - cmd.createPeer(peer_id_new,reply); - putCommand(std::move(cmd)); + { + Buffer reply(4); + writeU8(&reply[0], PACKET_TYPE_CONTROL); + writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID); + writeU16(&reply[2], peer_id_new); + putCommand(ConnectionCommand::createPeer(peer_id_new, reply)); + } // Create peer addition event - ConnectionEvent e; - e.peerAdded(peer_id_new, sender); - putEvent(e); + putEvent(ConnectionEvent::peerAdded(peer_id_new, sender)); // We're now talking to a valid peer_id return peer_id_new; } -void Connection::PrintInfo(std::ostream &out) -{ - m_info_mutex.lock(); - out< ack(4); writeU8(&ack[0], PACKET_TYPE_CONTROL); writeU8(&ack[1], CONTROLTYPE_ACK); writeU16(&ack[2], seqnum); - c.ack(peer_id, channelnum, ack); - putCommand(std::move(c)); + putCommand(ConnectionCommand::ack(peer_id, channelnum, ack)); m_sendThread->Trigger(); } diff --git a/src/network/connection.h b/src/network/connection.h index ea74ffb1c..1afb4ae84 100644 --- a/src/network/connection.h +++ b/src/network/connection.h @@ -32,6 +32,95 @@ with this program; if not, write to the Free Software Foundation, Inc., #include #include +#define MAX_UDP_PEERS 65535 + +/* +=== NOTES === + +A packet is sent through a channel to a peer with a basic header: + Header (7 bytes): + [0] u32 protocol_id + [4] session_t sender_peer_id + [6] u8 channel +sender_peer_id: + Unique to each peer. + value 0 (PEER_ID_INEXISTENT) is reserved for making new connections + value 1 (PEER_ID_SERVER) is reserved for server + these constants are defined in constants.h +channel: + Channel numbers have no intrinsic meaning. Currently only 0, 1, 2 exist. +*/ +#define BASE_HEADER_SIZE 7 +#define CHANNEL_COUNT 3 + +/* +Packet types: + +CONTROL: This is a packet used by the protocol. +- When this is processed, nothing is handed to the user. + Header (2 byte): + [0] u8 type + [1] u8 controltype +controltype and data description: + CONTROLTYPE_ACK + [2] u16 seqnum + CONTROLTYPE_SET_PEER_ID + [2] session_t peer_id_new + CONTROLTYPE_PING + - There is no actual reply, but this can be sent in a reliable + packet to get a reply + CONTROLTYPE_DISCO +*/ +enum ControlType : u8 { + CONTROLTYPE_ACK = 0, + CONTROLTYPE_SET_PEER_ID = 1, + CONTROLTYPE_PING = 2, + CONTROLTYPE_DISCO = 3, +}; + +/* +ORIGINAL: This is a plain packet with no control and no error +checking at all. +- When this is processed, it is directly handed to the user. + Header (1 byte): + [0] u8 type +*/ +//#define TYPE_ORIGINAL 1 +#define ORIGINAL_HEADER_SIZE 1 + +/* +SPLIT: These are sequences of packets forming one bigger piece of +data. +- When processed and all the packet_nums 0...packet_count-1 are + present (this should be buffered), the resulting data shall be + directly handed to the user. +- If the data fails to come up in a reasonable time, the buffer shall + be silently discarded. +- These can be sent as-is or atop of a RELIABLE packet stream. + Header (7 bytes): + [0] u8 type + [1] u16 seqnum + [3] u16 chunk_count + [5] u16 chunk_num +*/ +//#define TYPE_SPLIT 2 + +/* +RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs, +and they shall be delivered in the same order as sent. This is done +with a buffer in the receiving and transmitting end. +- When this is processed, the contents of each packet is recursively + processed as packets. + Header (3 bytes): + [0] u8 type + [1] u16 seqnum + +*/ +//#define TYPE_RELIABLE 3 +#define RELIABLE_HEADER_SIZE 3 +#define SEQNUM_INITIAL 65500 +#define SEQNUM_MAX 65535 + class NetworkPacket; namespace con @@ -46,9 +135,13 @@ typedef enum MTProtocols { MTP_MINETEST_RELIABLE_UDP } MTProtocols; -#define MAX_UDP_PEERS 65535 - -#define SEQNUM_MAX 65535 +enum PacketType : u8 { + PACKET_TYPE_CONTROL = 0, + PACKET_TYPE_ORIGINAL = 1, + PACKET_TYPE_SPLIT = 2, + PACKET_TYPE_RELIABLE = 3, + PACKET_TYPE_MAX +}; inline bool seqnum_higher(u16 totest, u16 base) { @@ -85,24 +178,40 @@ static inline float CALC_DTIME(u64 lasttime, u64 curtime) return MYMAX(MYMIN(value,0.1),0.0); } -struct BufferedPacket -{ - BufferedPacket(u8 *a_data, u32 a_size): - data(a_data, a_size) - {} - BufferedPacket(u32 a_size): - data(a_size) - {} - Buffer data; // Data of the packet, including headers +/* + Struct for all kinds of packets. Includes following data: + BASE_HEADER + u8[] packet data (usually copied from SharedBuffer) +*/ +struct BufferedPacket { + BufferedPacket(u32 a_size) + { + m_data.resize(a_size); + data = &m_data[0]; + } + + DISABLE_CLASS_COPY(BufferedPacket) + + u16 getSeqnum() const; + + inline const size_t size() const { return m_data.size(); } + + u8 *data; // Direct memory access float time = 0.0f; // Seconds from buffering the packet or re-sending float totaltime = 0.0f; // Seconds from buffering the packet u64 absolute_send_time = -1; Address address; // Sender or destination unsigned int resend_count = 0; + +private: + std::vector m_data; // Data of the packet, including headers }; +typedef std::shared_ptr BufferedPacketPtr; + + // This adds the base headers to the data and makes a packet out of it -BufferedPacket makePacket(Address &address, const SharedBuffer &data, +BufferedPacketPtr makePacket(Address &address, const SharedBuffer &data, u32 protocol_id, session_t sender_peer_id, u8 channel); // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet @@ -136,101 +245,12 @@ private: std::map> chunks; }; -/* -=== NOTES === - -A packet is sent through a channel to a peer with a basic header: - Header (7 bytes): - [0] u32 protocol_id - [4] session_t sender_peer_id - [6] u8 channel -sender_peer_id: - Unique to each peer. - value 0 (PEER_ID_INEXISTENT) is reserved for making new connections - value 1 (PEER_ID_SERVER) is reserved for server - these constants are defined in constants.h -channel: - Channel numbers have no intrinsic meaning. Currently only 0, 1, 2 exist. -*/ -#define BASE_HEADER_SIZE 7 -#define CHANNEL_COUNT 3 -/* -Packet types: - -CONTROL: This is a packet used by the protocol. -- When this is processed, nothing is handed to the user. - Header (2 byte): - [0] u8 type - [1] u8 controltype -controltype and data description: - CONTROLTYPE_ACK - [2] u16 seqnum - CONTROLTYPE_SET_PEER_ID - [2] session_t peer_id_new - CONTROLTYPE_PING - - There is no actual reply, but this can be sent in a reliable - packet to get a reply - CONTROLTYPE_DISCO -*/ -//#define TYPE_CONTROL 0 -#define CONTROLTYPE_ACK 0 -#define CONTROLTYPE_SET_PEER_ID 1 -#define CONTROLTYPE_PING 2 -#define CONTROLTYPE_DISCO 3 - -/* -ORIGINAL: This is a plain packet with no control and no error -checking at all. -- When this is processed, it is directly handed to the user. - Header (1 byte): - [0] u8 type -*/ -//#define TYPE_ORIGINAL 1 -#define ORIGINAL_HEADER_SIZE 1 -/* -SPLIT: These are sequences of packets forming one bigger piece of -data. -- When processed and all the packet_nums 0...packet_count-1 are - present (this should be buffered), the resulting data shall be - directly handed to the user. -- If the data fails to come up in a reasonable time, the buffer shall - be silently discarded. -- These can be sent as-is or atop of a RELIABLE packet stream. - Header (7 bytes): - [0] u8 type - [1] u16 seqnum - [3] u16 chunk_count - [5] u16 chunk_num -*/ -//#define TYPE_SPLIT 2 -/* -RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs, -and they shall be delivered in the same order as sent. This is done -with a buffer in the receiving and transmitting end. -- When this is processed, the contents of each packet is recursively - processed as packets. - Header (3 bytes): - [0] u8 type - [1] u16 seqnum - -*/ -//#define TYPE_RELIABLE 3 -#define RELIABLE_HEADER_SIZE 3 -#define SEQNUM_INITIAL 65500 - -enum PacketType: u8 { - PACKET_TYPE_CONTROL = 0, - PACKET_TYPE_ORIGINAL = 1, - PACKET_TYPE_SPLIT = 2, - PACKET_TYPE_RELIABLE = 3, - PACKET_TYPE_MAX -}; /* A buffer which stores reliable packets and sorts them internally for fast access to the smallest one. */ -typedef std::list::iterator RPBSearchResult; +typedef std::list::iterator RPBSearchResult; class ReliablePacketBuffer { @@ -239,12 +259,12 @@ public: bool getFirstSeqnum(u16& result); - BufferedPacket popFirst(); - BufferedPacket popSeqnum(u16 seqnum); - void insert(const BufferedPacket &p, u16 next_expected); + BufferedPacketPtr popFirst(); + BufferedPacketPtr popSeqnum(u16 seqnum); + void insert(BufferedPacketPtr &p_ptr, u16 next_expected); void incrementTimeouts(float dtime); - std::list getTimedOuts(float timeout, u32 max_packets); + std::list> getTimedOuts(float timeout, u32 max_packets); void print(); bool empty(); @@ -252,10 +272,9 @@ public: private: - RPBSearchResult findPacket(u16 seqnum); // does not perform locking - inline RPBSearchResult notFound() { return m_list.end(); } + RPBSearchResult findPacketNoLock(u16 seqnum); - std::list m_list; + std::list m_list; u16 m_oldest_non_answered_ack; @@ -274,7 +293,7 @@ public: Returns a reference counted buffer of length != 0 when a full split packet is constructed. If not, returns one of length 0. */ - SharedBuffer insert(const BufferedPacket &p, bool reliable); + SharedBuffer insert(BufferedPacketPtr &p_ptr, bool reliable); void removeUnreliableTimedOuts(float dtime, float timeout); @@ -285,25 +304,6 @@ private: std::mutex m_map_mutex; }; -struct OutgoingPacket -{ - session_t peer_id; - u8 channelnum; - SharedBuffer data; - bool reliable; - bool ack; - - OutgoingPacket(session_t peer_id_, u8 channelnum_, const SharedBuffer &data_, - bool reliable_,bool ack_=false): - peer_id(peer_id_), - channelnum(channelnum_), - data(data_), - reliable(reliable_), - ack(ack_) - { - } -}; - enum ConnectionCommandType{ CONNCMD_NONE, CONNCMD_SERVE, @@ -316,9 +316,13 @@ enum ConnectionCommandType{ CONCMD_CREATE_PEER }; +struct ConnectionCommand; +typedef std::shared_ptr ConnectionCommandPtr; + +// This is very similar to ConnectionEvent struct ConnectionCommand { - enum ConnectionCommandType type = CONNCMD_NONE; + const ConnectionCommandType type; Address address; session_t peer_id = PEER_ID_INEXISTENT; u8 channelnum = 0; @@ -326,48 +330,21 @@ struct ConnectionCommand bool reliable = false; bool raw = false; - ConnectionCommand() = default; - - void serve(Address address_) - { - type = CONNCMD_SERVE; - address = address_; - } - void connect(Address address_) - { - type = CONNCMD_CONNECT; - address = address_; - } - void disconnect() - { - type = CONNCMD_DISCONNECT; - } - void disconnect_peer(session_t peer_id_) - { - type = CONNCMD_DISCONNECT_PEER; - peer_id = peer_id_; - } + DISABLE_CLASS_COPY(ConnectionCommand); - void send(session_t peer_id_, u8 channelnum_, NetworkPacket *pkt, bool reliable_); + static ConnectionCommandPtr serve(Address address); + static ConnectionCommandPtr connect(Address address); + static ConnectionCommandPtr disconnect(); + static ConnectionCommandPtr disconnect_peer(session_t peer_id); + static ConnectionCommandPtr send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable); + static ConnectionCommandPtr ack(session_t peer_id, u8 channelnum, const Buffer &data); + static ConnectionCommandPtr createPeer(session_t peer_id, const Buffer &data); - void ack(session_t peer_id_, u8 channelnum_, const Buffer &data_) - { - type = CONCMD_ACK; - peer_id = peer_id_; - channelnum = channelnum_; - data = data_; - reliable = false; - } +private: + ConnectionCommand(ConnectionCommandType type_) : + type(type_) {} - void createPeer(session_t peer_id_, const Buffer &data_) - { - type = CONCMD_CREATE_PEER; - peer_id = peer_id_; - data = data_; - channelnum = 0; - reliable = true; - raw = true; - } + static ConnectionCommandPtr create(ConnectionCommandType type); }; /* maximum window size to use, 0xFFFF is theoretical maximum. don't think about @@ -402,10 +379,10 @@ public: ReliablePacketBuffer outgoing_reliables_sent; //queued reliable packets - std::queue queued_reliables; + std::queue queued_reliables; //queue commands prior splitting to packets - std::deque queued_commands; + std::deque queued_commands; IncomingSplitBuffer incoming_splits; @@ -514,7 +491,7 @@ class Peer { public: friend class PeerHelper; - Peer(Address address_,u16 id_,Connection* connection) : + Peer(Address address_,session_t id_,Connection* connection) : id(id_), m_connection(connection), address(address_), @@ -528,11 +505,11 @@ class Peer { }; // Unique id of the peer - u16 id; + const session_t id; void Drop(); - virtual void PutReliableSendCommand(ConnectionCommand &c, + virtual void PutReliableSendCommand(ConnectionCommandPtr &c, unsigned int max_packet_size) {}; virtual bool getAddress(MTProtocols type, Address& toset) = 0; @@ -549,7 +526,7 @@ class Peer { virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; }; virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {}; - virtual SharedBuffer addSplitPacket(u8 channel, const BufferedPacket &toadd, + virtual SharedBuffer addSplitPacket(u8 channel, BufferedPacketPtr &toadd, bool reliable) { errorstream << "Peer::addSplitPacket called," @@ -586,7 +563,7 @@ class Peer { bool IncUseCount(); void DecUseCount(); - std::mutex m_exclusive_access_mutex; + mutable std::mutex m_exclusive_access_mutex; bool m_pending_deletion = false; @@ -634,7 +611,7 @@ public: UDPPeer(u16 a_id, Address a_address, Connection* connection); virtual ~UDPPeer() = default; - void PutReliableSendCommand(ConnectionCommand &c, + void PutReliableSendCommand(ConnectionCommandPtr &c, unsigned int max_packet_size); bool getAddress(MTProtocols type, Address& toset); @@ -642,7 +619,7 @@ public: u16 getNextSplitSequenceNumber(u8 channel); void setNextSplitSequenceNumber(u8 channel, u16 seqnum); - SharedBuffer addSplitPacket(u8 channel, const BufferedPacket &toadd, + SharedBuffer addSplitPacket(u8 channel, BufferedPacketPtr &toadd, bool reliable); protected: @@ -671,7 +648,7 @@ private: float resend_timeout = 0.5; bool processReliableSendCommand( - ConnectionCommand &c, + ConnectionCommandPtr &c_ptr, unsigned int max_packet_size); }; @@ -679,7 +656,7 @@ private: Connection */ -enum ConnectionEventType{ +enum ConnectionEventType { CONNEVENT_NONE, CONNEVENT_DATA_RECEIVED, CONNEVENT_PEER_ADDED, @@ -687,56 +664,32 @@ enum ConnectionEventType{ CONNEVENT_BIND_FAILED, }; +struct ConnectionEvent; +typedef std::shared_ptr ConnectionEventPtr; + +// This is very similar to ConnectionCommand struct ConnectionEvent { - enum ConnectionEventType type = CONNEVENT_NONE; + const ConnectionEventType type; session_t peer_id = 0; Buffer data; bool timeout = false; Address address; - ConnectionEvent() = default; + // We don't want to copy "data" + DISABLE_CLASS_COPY(ConnectionEvent); - const char *describe() const - { - switch(type) { - case CONNEVENT_NONE: - return "CONNEVENT_NONE"; - case CONNEVENT_DATA_RECEIVED: - return "CONNEVENT_DATA_RECEIVED"; - case CONNEVENT_PEER_ADDED: - return "CONNEVENT_PEER_ADDED"; - case CONNEVENT_PEER_REMOVED: - return "CONNEVENT_PEER_REMOVED"; - case CONNEVENT_BIND_FAILED: - return "CONNEVENT_BIND_FAILED"; - } - return "Invalid ConnectionEvent"; - } + static ConnectionEventPtr create(ConnectionEventType type); + static ConnectionEventPtr dataReceived(session_t peer_id, const Buffer &data); + static ConnectionEventPtr peerAdded(session_t peer_id, Address address); + static ConnectionEventPtr peerRemoved(session_t peer_id, bool is_timeout, Address address); + static ConnectionEventPtr bindFailed(); - void dataReceived(session_t peer_id_, const Buffer &data_) - { - type = CONNEVENT_DATA_RECEIVED; - peer_id = peer_id_; - data = data_; - } - void peerAdded(session_t peer_id_, Address address_) - { - type = CONNEVENT_PEER_ADDED; - peer_id = peer_id_; - address = address_; - } - void peerRemoved(session_t peer_id_, bool timeout_, Address address_) - { - type = CONNEVENT_PEER_REMOVED; - peer_id = peer_id_; - timeout = timeout_; - address = address_; - } - void bindFailed() - { - type = CONNEVENT_BIND_FAILED; - } + const char *describe() const; + +private: + ConnectionEvent(ConnectionEventType type_) : + type(type_) {} }; class PeerHandler; @@ -752,10 +705,9 @@ public: ~Connection(); /* Interface */ - ConnectionEvent waitEvent(u32 timeout_ms); - // Warning: creates an unnecessary copy, prefer putCommand(T&&) if possible - void putCommand(const ConnectionCommand &c); - void putCommand(ConnectionCommand &&c); + ConnectionEventPtr waitEvent(u32 timeout_ms); + + void putCommand(ConnectionCommandPtr c); void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; } void Serve(Address bind_addr); @@ -785,8 +737,6 @@ protected: void sendAck(session_t peer_id, u8 channelnum, u16 seqnum); - void PrintInfo(std::ostream &out); - std::vector getPeerIDs() { MutexAutoLock peerlock(m_peers_mutex); @@ -795,13 +745,11 @@ protected: UDPSocket m_udpSocket; // Command queue: user -> SendThread - MutexedQueue m_command_queue; + MutexedQueue m_command_queue; bool Receive(NetworkPacket *pkt, u32 timeout); - // Warning: creates an unnecessary copy, prefer putEvent(T&&) if possible - void putEvent(const ConnectionEvent &e); - void putEvent(ConnectionEvent &&e); + void putEvent(ConnectionEventPtr e); void TriggerSend(); @@ -811,7 +759,7 @@ protected: } private: // Event queue: ReceiveThread -> user - MutexedQueue m_event_queue; + MutexedQueue m_event_queue; session_t m_peer_id = 0; u32 m_protocol_id; @@ -823,7 +771,7 @@ private: std::unique_ptr m_sendThread; std::unique_ptr m_receiveThread; - std::mutex m_info_mutex; + mutable std::mutex m_info_mutex; // Backwards compatibility PeerHandler *m_bc_peerhandler; diff --git a/src/network/connectionthreads.cpp b/src/network/connectionthreads.cpp index a306ced9b..dca065ae1 100644 --- a/src/network/connectionthreads.cpp +++ b/src/network/connectionthreads.cpp @@ -50,11 +50,11 @@ std::mutex log_conthread_mutex; #define WINDOW_SIZE 5 -static session_t readPeerId(u8 *packetdata) +static session_t readPeerId(const u8 *packetdata) { return readU16(&packetdata[4]); } -static u8 readChannel(u8 *packetdata) +static u8 readChannel(const u8 *packetdata) { return readU8(&packetdata[6]); } @@ -114,9 +114,9 @@ void *ConnectionSendThread::run() } /* translate commands to packets */ - ConnectionCommand c = m_connection->m_command_queue.pop_frontNoEx(0); - while (c.type != CONNCMD_NONE) { - if (c.reliable) + auto c = m_connection->m_command_queue.pop_frontNoEx(0); + while (c && c->type != CONNCMD_NONE) { + if (c->reliable) processReliableCommand(c); else processNonReliableCommand(c); @@ -227,21 +227,21 @@ void ConnectionSendThread::runTimeouts(float dtime) m_iteration_packets_avaialble -= timed_outs.size(); for (const auto &k : timed_outs) { - u8 channelnum = readChannel(*k.data); - u16 seqnum = readU16(&(k.data[BASE_HEADER_SIZE + 1])); + u8 channelnum = readChannel(k->data); + u16 seqnum = k->getSeqnum(); - channel.UpdateBytesLost(k.data.getSize()); + channel.UpdateBytesLost(k->size()); LOG(derr_con << m_connection->getDesc() << "RE-SENDING timed-out RELIABLE to " - << k.address.serializeString() + << k->address.serializeString() << "(t/o=" << resend_timeout << "): " - << "count=" << k.resend_count + << "count=" << k->resend_count << ", channel=" << ((int) channelnum & 0xff) << ", seqnum=" << seqnum << std::endl); - rawSend(k); + rawSend(k.get()); // do not handle rtt here as we can't decide if this packet was // lost or really takes more time to transmit @@ -274,25 +274,24 @@ void ConnectionSendThread::runTimeouts(float dtime) } } -void ConnectionSendThread::rawSend(const BufferedPacket &packet) +void ConnectionSendThread::rawSend(const BufferedPacket *p) { try { - m_connection->m_udpSocket.Send(packet.address, *packet.data, - packet.data.getSize()); + m_connection->m_udpSocket.Send(p->address, p->data, p->size()); LOG(dout_con << m_connection->getDesc() - << " rawSend: " << packet.data.getSize() + << " rawSend: " << p->size() << " bytes sent" << std::endl); } catch (SendFailedException &e) { LOG(derr_con << m_connection->getDesc() << "Connection::rawSend(): SendFailedException: " - << packet.address.serializeString() << std::endl); + << p->address.serializeString() << std::endl); } } -void ConnectionSendThread::sendAsPacketReliable(BufferedPacket &p, Channel *channel) +void ConnectionSendThread::sendAsPacketReliable(BufferedPacketPtr &p, Channel *channel) { try { - p.absolute_send_time = porting::getTimeMs(); + p->absolute_send_time = porting::getTimeMs(); // Buffer the packet channel->outgoing_reliables_sent.insert(p, (channel->readOutgoingSequenceNumber() - MAX_RELIABLE_WINDOW_SIZE) @@ -305,7 +304,7 @@ void ConnectionSendThread::sendAsPacketReliable(BufferedPacket &p, Channel *chan } // Send the packet - rawSend(p); + rawSend(p.get()); } bool ConnectionSendThread::rawSendAsPacket(session_t peer_id, u8 channelnum, @@ -321,11 +320,10 @@ bool ConnectionSendThread::rawSendAsPacket(session_t peer_id, u8 channelnum, Channel *channel = &(dynamic_cast(&peer)->channels[channelnum]); if (reliable) { - bool have_sequence_number_for_raw_packet = true; - u16 seqnum = - channel->getOutgoingSequenceNumber(have_sequence_number_for_raw_packet); + bool have_seqnum = false; + const u16 seqnum = channel->getOutgoingSequenceNumber(have_seqnum); - if (!have_sequence_number_for_raw_packet) + if (!have_seqnum) return false; SharedBuffer reliable = makeReliablePacket(data, seqnum); @@ -333,13 +331,12 @@ bool ConnectionSendThread::rawSendAsPacket(session_t peer_id, u8 channelnum, peer->getAddress(MTP_MINETEST_RELIABLE_UDP, peer_address); // Add base headers and make a packet - BufferedPacket p = con::makePacket(peer_address, reliable, + BufferedPacketPtr p = con::makePacket(peer_address, reliable, m_connection->GetProtocolID(), m_connection->GetPeerID(), channelnum); // first check if our send window is already maxed out - if (channel->outgoing_reliables_sent.size() - < channel->getWindowSize()) { + if (channel->outgoing_reliables_sent.size() < channel->getWindowSize()) { LOG(dout_con << m_connection->getDesc() << " INFO: sending a reliable packet to peer_id " << peer_id << " channel: " << (u32)channelnum @@ -352,19 +349,19 @@ bool ConnectionSendThread::rawSendAsPacket(session_t peer_id, u8 channelnum, << " INFO: queueing reliable packet for peer_id: " << peer_id << " channel: " << (u32)channelnum << " seqnum: " << seqnum << std::endl); - channel->queued_reliables.push(std::move(p)); + channel->queued_reliables.push(p); return false; } Address peer_address; if (peer->getAddress(MTP_UDP, peer_address)) { // Add base headers and make a packet - BufferedPacket p = con::makePacket(peer_address, data, + BufferedPacketPtr p = con::makePacket(peer_address, data, m_connection->GetProtocolID(), m_connection->GetPeerID(), channelnum); // Send the packet - rawSend(p); + rawSend(p.get()); return true; } @@ -374,11 +371,11 @@ bool ConnectionSendThread::rawSendAsPacket(session_t peer_id, u8 channelnum, return false; } -void ConnectionSendThread::processReliableCommand(ConnectionCommand &c) +void ConnectionSendThread::processReliableCommand(ConnectionCommandPtr &c) { - assert(c.reliable); // Pre-condition + assert(c->reliable); // Pre-condition - switch (c.type) { + switch (c->type) { case CONNCMD_NONE: LOG(dout_con << m_connection->getDesc() << "UDP processing reliable CONNCMD_NONE" << std::endl); @@ -399,7 +396,7 @@ void ConnectionSendThread::processReliableCommand(ConnectionCommand &c) case CONCMD_CREATE_PEER: LOG(dout_con << m_connection->getDesc() << "UDP processing reliable CONCMD_CREATE_PEER" << std::endl); - if (!rawSendAsPacket(c.peer_id, c.channelnum, c.data, c.reliable)) { + if (!rawSendAsPacket(c->peer_id, c->channelnum, c->data, c->reliable)) { /* put to queue if we couldn't send it immediately */ sendReliable(c); } @@ -412,13 +409,14 @@ void ConnectionSendThread::processReliableCommand(ConnectionCommand &c) FATAL_ERROR("Got command that shouldn't be reliable as reliable command"); default: LOG(dout_con << m_connection->getDesc() - << " Invalid reliable command type: " << c.type << std::endl); + << " Invalid reliable command type: " << c->type << std::endl); } } -void ConnectionSendThread::processNonReliableCommand(ConnectionCommand &c) +void ConnectionSendThread::processNonReliableCommand(ConnectionCommandPtr &c_ptr) { + const ConnectionCommand &c = *c_ptr; assert(!c.reliable); // Pre-condition switch (c.type) { @@ -480,9 +478,7 @@ void ConnectionSendThread::serve(Address bind_address) } catch (SocketException &e) { // Create event - ConnectionEvent ce; - ce.bindFailed(); - m_connection->putEvent(ce); + m_connection->putEvent(ConnectionEvent::bindFailed()); } } @@ -495,9 +491,7 @@ void ConnectionSendThread::connect(Address address) UDPPeer *peer = m_connection->createServerPeer(address); // Create event - ConnectionEvent e; - e.peerAdded(peer->id, peer->address); - m_connection->putEvent(e); + m_connection->putEvent(ConnectionEvent::peerAdded(peer->id, peer->address)); Address bind_addr; @@ -586,9 +580,9 @@ void ConnectionSendThread::send(session_t peer_id, u8 channelnum, } } -void ConnectionSendThread::sendReliable(ConnectionCommand &c) +void ConnectionSendThread::sendReliable(ConnectionCommandPtr &c) { - PeerHelper peer = m_connection->getPeerNoEx(c.peer_id); + PeerHelper peer = m_connection->getPeerNoEx(c->peer_id); if (!peer) return; @@ -604,7 +598,7 @@ void ConnectionSendThread::sendToAll(u8 channelnum, const SharedBuffer &data } } -void ConnectionSendThread::sendToAllReliable(ConnectionCommand &c) +void ConnectionSendThread::sendToAllReliable(ConnectionCommandPtr &c) { std::vector peerids = m_connection->getPeerIDs(); @@ -663,8 +657,12 @@ void ConnectionSendThread::sendPackets(float dtime) // first send queued reliable packets for all peers (if possible) for (unsigned int i = 0; i < CHANNEL_COUNT; i++) { Channel &channel = udpPeer->channels[i]; - u16 next_to_ack = 0; + // Reduces logging verbosity + if (channel.queued_reliables.empty()) + continue; + + u16 next_to_ack = 0; channel.outgoing_reliables_sent.getFirstSeqnum(next_to_ack); u16 next_to_receive = 0; channel.incoming_reliables.getFirstSeqnum(next_to_receive); @@ -694,13 +692,13 @@ void ConnectionSendThread::sendPackets(float dtime) channel.outgoing_reliables_sent.size() < channel.getWindowSize() && peer->m_increment_packets_remaining > 0) { - BufferedPacket p = std::move(channel.queued_reliables.front()); + BufferedPacketPtr p = channel.queued_reliables.front(); channel.queued_reliables.pop(); LOG(dout_con << m_connection->getDesc() << " INFO: sending a queued reliable packet " << " channel: " << i - << ", seqnum: " << readU16(&p.data[BASE_HEADER_SIZE + 1]) + << ", seqnum: " << p->getSeqnum() << std::endl); sendAsPacketReliable(p, &channel); @@ -881,17 +879,14 @@ void ConnectionReceiveThread::receive(SharedBuffer &packetdata, try { // First, see if there any buffered packets we can process now if (packet_queued) { - bool data_left = true; session_t peer_id; SharedBuffer resultdata; - while (data_left) { + while (true) { try { - data_left = getFromBuffers(peer_id, resultdata); - if (data_left) { - ConnectionEvent e; - e.dataReceived(peer_id, resultdata); - m_connection->putEvent(std::move(e)); - } + if (!getFromBuffers(peer_id, resultdata)) + break; + + m_connection->putEvent(ConnectionEvent::dataReceived(peer_id, resultdata)); } catch (ProcessedSilentlyException &e) { /* try reading again */ @@ -908,7 +903,7 @@ void ConnectionReceiveThread::receive(SharedBuffer &packetdata, return; if ((received_size < BASE_HEADER_SIZE) || - (readU32(&packetdata[0]) != m_connection->GetProtocolID())) { + (readU32(&packetdata[0]) != m_connection->GetProtocolID())) { LOG(derr_con << m_connection->getDesc() << "Receive(): Invalid incoming packet, " << "size: " << received_size @@ -999,9 +994,7 @@ void ConnectionReceiveThread::receive(SharedBuffer &packetdata, << ", channel: " << (u32)channelnum << ", returned " << resultdata.getSize() << " bytes" << std::endl); - ConnectionEvent e; - e.dataReceived(peer_id, resultdata); - m_connection->putEvent(std::move(e)); + m_connection->putEvent(ConnectionEvent::dataReceived(peer_id, resultdata)); } catch (ProcessedSilentlyException &e) { } @@ -1026,10 +1019,11 @@ bool ConnectionReceiveThread::getFromBuffers(session_t &peer_id, SharedBuffer(&peer) == 0) + UDPPeer *p = dynamic_cast(&peer); + if (!p) continue; - for (Channel &channel : (dynamic_cast(&peer))->channels) { + for (Channel &channel : p->channels) { if (checkIncomingBuffers(&channel, peer_id, dst)) { return true; } @@ -1042,32 +1036,34 @@ bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel, session_t &peer_id, SharedBuffer &dst) { u16 firstseqnum = 0; - if (channel->incoming_reliables.getFirstSeqnum(firstseqnum)) { - if (firstseqnum == channel->readNextIncomingSeqNum()) { - BufferedPacket p = channel->incoming_reliables.popFirst(); - peer_id = readPeerId(*p.data); - u8 channelnum = readChannel(*p.data); - u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE + 1]); + if (!channel->incoming_reliables.getFirstSeqnum(firstseqnum)) + return false; - LOG(dout_con << m_connection->getDesc() - << "UNBUFFERING TYPE_RELIABLE" - << " seqnum=" << seqnum - << " peer_id=" << peer_id - << " channel=" << ((int) channelnum & 0xff) - << std::endl); + if (firstseqnum != channel->readNextIncomingSeqNum()) + return false; - channel->incNextIncomingSeqNum(); + BufferedPacketPtr p = channel->incoming_reliables.popFirst(); - u32 headers_size = BASE_HEADER_SIZE + RELIABLE_HEADER_SIZE; - // Get out the inside packet and re-process it - SharedBuffer payload(p.data.getSize() - headers_size); - memcpy(*payload, &p.data[headers_size], payload.getSize()); + peer_id = readPeerId(p->data); // Carried over to caller function + u8 channelnum = readChannel(p->data); + u16 seqnum = p->getSeqnum(); - dst = processPacket(channel, payload, peer_id, channelnum, true); - return true; - } - } - return false; + LOG(dout_con << m_connection->getDesc() + << "UNBUFFERING TYPE_RELIABLE" + << " seqnum=" << seqnum + << " peer_id=" << peer_id + << " channel=" << ((int) channelnum & 0xff) + << std::endl); + + channel->incNextIncomingSeqNum(); + + u32 headers_size = BASE_HEADER_SIZE + RELIABLE_HEADER_SIZE; + // Get out the inside packet and re-process it + SharedBuffer payload(p->size() - headers_size); + memcpy(*payload, &p->data[headers_size], payload.getSize()); + + dst = processPacket(channel, payload, peer_id, channelnum, true); + return true; } SharedBuffer ConnectionReceiveThread::processPacket(Channel *channel, @@ -1115,7 +1111,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan if (packetdata.getSize() < 2) throw InvalidIncomingDataException("packetdata.getSize() < 2"); - u8 controltype = readU8(&(packetdata[1])); + ControlType controltype = (ControlType)readU8(&(packetdata[1])); if (controltype == CONTROLTYPE_ACK) { assert(channel != NULL); @@ -1131,7 +1127,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan << seqnum << " ]" << std::endl); try { - BufferedPacket p = channel->outgoing_reliables_sent.popSeqnum(seqnum); + BufferedPacketPtr p = channel->outgoing_reliables_sent.popSeqnum(seqnum); // the rtt calculation will be a bit off for re-sent packets but that's okay { @@ -1140,14 +1136,14 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan // a overflow is quite unlikely but as it'd result in major // rtt miscalculation we handle it here - if (current_time > p.absolute_send_time) { - float rtt = (current_time - p.absolute_send_time) / 1000.0; + if (current_time > p->absolute_send_time) { + float rtt = (current_time - p->absolute_send_time) / 1000.0; // Let peer calculate stuff according to it // (avg_rtt and resend_timeout) dynamic_cast(peer)->reportRTT(rtt); - } else if (p.totaltime > 0) { - float rtt = p.totaltime; + } else if (p->totaltime > 0) { + float rtt = p->totaltime; // Let peer calculate stuff according to it // (avg_rtt and resend_timeout) @@ -1156,7 +1152,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan } // put bytes for max bandwidth calculation - channel->UpdateBytesSent(p.data.getSize(), 1); + channel->UpdateBytesSent(p->size(), 1); if (channel->outgoing_reliables_sent.size() == 0) m_connection->TriggerSend(); } catch (NotFoundException &e) { @@ -1204,7 +1200,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Control(Channel *chan throw ProcessedSilentlyException("Got a DISCO"); } else { LOG(derr_con << m_connection->getDesc() - << "INVALID TYPE_CONTROL: invalid controltype=" + << "INVALID controltype=" << ((int) controltype & 0xff) << std::endl); throw InvalidIncomingDataException("Invalid control type"); } @@ -1232,7 +1228,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Split(Channel *channe if (peer->getAddress(MTP_UDP, peer_address)) { // We have to create a packet again for buffering // This isn't actually too bad an idea. - BufferedPacket packet = makePacket(peer_address, + BufferedPacketPtr packet = con::makePacket(peer_address, packetdata, m_connection->GetProtocolID(), peer->id, @@ -1267,7 +1263,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Reliable(Channel *cha if (packetdata.getSize() < RELIABLE_HEADER_SIZE) throw InvalidIncomingDataException("packetdata.getSize() < RELIABLE_HEADER_SIZE"); - u16 seqnum = readU16(&packetdata[1]); + const u16 seqnum = readU16(&packetdata[1]); bool is_future_packet = false; bool is_old_packet = false; @@ -1311,7 +1307,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Reliable(Channel *cha // This one comes later, buffer it. // Actually we have to make a packet to buffer one. // Well, we have all the ingredients, so just do it. - BufferedPacket packet = con::makePacket( + BufferedPacketPtr packet = con::makePacket( peer_address, packetdata, m_connection->GetProtocolID(), @@ -1328,9 +1324,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Reliable(Channel *cha throw ProcessedQueued("Buffered future reliable packet"); } catch (AlreadyExistsException &e) { } catch (IncomingDataCorruption &e) { - ConnectionCommand discon; - discon.disconnect_peer(peer->id); - m_connection->putCommand(discon); + m_connection->putCommand(ConnectionCommand::disconnect_peer(peer->id)); LOG(derr_con << m_connection->getDesc() << "INVALID, TYPE_RELIABLE peer_id: " << peer->id @@ -1351,7 +1345,7 @@ SharedBuffer ConnectionReceiveThread::handlePacketType_Reliable(Channel *cha u16 queued_seqnum = 0; if (channel->incoming_reliables.getFirstSeqnum(queued_seqnum)) { if (queued_seqnum == seqnum) { - BufferedPacket queued_packet = channel->incoming_reliables.popFirst(); + BufferedPacketPtr queued_packet = channel->incoming_reliables.popFirst(); /** TODO find a way to verify the new against the old packet */ } } diff --git a/src/network/connectionthreads.h b/src/network/connectionthreads.h index 612407c3b..c2e2dae12 100644 --- a/src/network/connectionthreads.h +++ b/src/network/connectionthreads.h @@ -29,6 +29,25 @@ namespace con class Connection; +struct OutgoingPacket +{ + session_t peer_id; + u8 channelnum; + SharedBuffer data; + bool reliable; + bool ack; + + OutgoingPacket(session_t peer_id_, u8 channelnum_, const SharedBuffer &data_, + bool reliable_,bool ack_=false): + peer_id(peer_id_), + channelnum(channelnum_), + data(data_), + reliable(reliable_), + ack(ack_) + { + } +}; + class ConnectionSendThread : public Thread { @@ -51,27 +70,27 @@ public: private: void runTimeouts(float dtime); - void rawSend(const BufferedPacket &packet); + void rawSend(const BufferedPacket *p); bool rawSendAsPacket(session_t peer_id, u8 channelnum, const SharedBuffer &data, bool reliable); - void processReliableCommand(ConnectionCommand &c); - void processNonReliableCommand(ConnectionCommand &c); + void processReliableCommand(ConnectionCommandPtr &c); + void processNonReliableCommand(ConnectionCommandPtr &c); void serve(Address bind_address); void connect(Address address); void disconnect(); void disconnect_peer(session_t peer_id); void send(session_t peer_id, u8 channelnum, const SharedBuffer &data); - void sendReliable(ConnectionCommand &c); + void sendReliable(ConnectionCommandPtr &c); void sendToAll(u8 channelnum, const SharedBuffer &data); - void sendToAllReliable(ConnectionCommand &c); + void sendToAllReliable(ConnectionCommandPtr &c); void sendPackets(float dtime); void sendAsPacket(session_t peer_id, u8 channelnum, const SharedBuffer &data, bool ack = false); - void sendAsPacketReliable(BufferedPacket &p, Channel *channel); + void sendAsPacketReliable(BufferedPacketPtr &p, Channel *channel); bool packetsQueued(); diff --git a/src/unittest/test_connection.cpp b/src/unittest/test_connection.cpp index 23b7e9105..04fea90d6 100644 --- a/src/unittest/test_connection.cpp +++ b/src/unittest/test_connection.cpp @@ -124,7 +124,7 @@ void TestConnection::testHelpers() Address a(127,0,0,1, 10); const u16 seqnum = 34352; - con::BufferedPacket p1 = con::makePacket(a, data1, + con::BufferedPacketPtr p1 = con::makePacket(a, data1, proto_id, peer_id, channel); /* We should now have a packet with this data: @@ -135,10 +135,10 @@ void TestConnection::testHelpers() Data: [7] u8 data1[0] */ - UASSERT(readU32(&p1.data[0]) == proto_id); - UASSERT(readU16(&p1.data[4]) == peer_id); - UASSERT(readU8(&p1.data[6]) == channel); - UASSERT(readU8(&p1.data[7]) == data1[0]); + UASSERT(readU32(&p1->data[0]) == proto_id); + UASSERT(readU16(&p1->data[4]) == peer_id); + UASSERT(readU8(&p1->data[6]) == channel); + UASSERT(readU8(&p1->data[7]) == data1[0]); //infostream<<"initial data1[0]="<<((u32)data1[0]&0xff)< +#include // std::shared_ptr + + +template class ConstSharedPtr { +public: + ConstSharedPtr(T *ptr) : ptr(ptr) {} + ConstSharedPtr(const std::shared_ptr &ptr) : ptr(ptr) {} + + const T* get() const noexcept { return ptr.get(); } + const T& operator*() const noexcept { return *ptr.get(); } + const T* operator->() const noexcept { return ptr.get(); } + +private: + std::shared_ptr ptr; +}; template class Buffer @@ -40,17 +55,11 @@ public: else data = NULL; } - Buffer(const Buffer &buffer) - { - m_size = buffer.m_size; - if(m_size != 0) - { - data = new T[buffer.m_size]; - memcpy(data, buffer.data, buffer.m_size); - } - else - data = NULL; - } + + // Disable class copy + Buffer(const Buffer &) = delete; + Buffer &operator=(const Buffer &) = delete; + Buffer(Buffer &&buffer) { m_size = buffer.m_size; @@ -81,21 +90,6 @@ public: drop(); } - Buffer& operator=(const Buffer &buffer) - { - if(this == &buffer) - return *this; - drop(); - m_size = buffer.m_size; - if(m_size != 0) - { - data = new T[buffer.m_size]; - memcpy(data, buffer.data, buffer.m_size); - } - else - data = NULL; - return *this; - } Buffer& operator=(Buffer &&buffer) { if(this == &buffer) @@ -113,6 +107,18 @@ public: return *this; } + void copyTo(Buffer &buffer) const + { + buffer.drop(); + buffer.m_size = m_size; + if (m_size != 0) { + buffer.data = new T[m_size]; + memcpy(buffer.data, data, m_size); + } else { + buffer.data = nullptr; + } + } + T & operator[](unsigned int i) const { return data[i]; -- cgit v1.2.3 From 0704ca055059088bdd53e15be672e6b5663b8f50 Mon Sep 17 00:00:00 2001 From: paradust7 <102263465+paradust7@users.noreply.github.com> Date: Wed, 4 May 2022 11:55:01 -0700 Subject: Make logging cost free when there is no output target (#12247) The logging streams now do almost no work when there is no output target for them. For example, if LL_VERBOSE has no output targets, then `verbosestream << x` will return a StreamProxy with a null target. Any further `<<` operations applied to it will do nothing. --- builtin/settingtypes.txt | 5 +- minetest.conf.example | 5 +- src/client/game.cpp | 2 +- src/client/renderingengine.cpp | 2 +- src/log.cpp | 187 ++++++++++++----------------------- src/log.h | 202 +++++++++++++++++++++++++++++++++----- src/main.cpp | 16 ++- src/network/address.cpp | 8 +- src/network/address.h | 2 +- src/network/connection.cpp | 17 +--- src/network/connectionthreads.cpp | 10 +- src/network/socket.cpp | 4 +- src/server.cpp | 2 +- src/util/stream.h | 70 +++++++++++++ 14 files changed, 338 insertions(+), 194 deletions(-) create mode 100644 src/util/stream.h (limited to 'src/network/connectionthreads.cpp') diff --git a/builtin/settingtypes.txt b/builtin/settingtypes.txt index a983a8f6b..e3589d76f 100644 --- a/builtin/settingtypes.txt +++ b/builtin/settingtypes.txt @@ -1468,7 +1468,8 @@ language (Language) enum ,be,bg,ca,cs,da,de,el,en,eo,es,et,eu,fi,fr,gd,gl,hu,i # - action # - info # - verbose -debug_log_level (Debug log level) enum action ,none,error,warning,action,info,verbose +# - trace +debug_log_level (Debug log level) enum action ,none,error,warning,action,info,verbose,trace # If the file size of debug.txt exceeds the number of megabytes specified in # this setting when it is opened, the file is moved to debug.txt.1, @@ -1477,7 +1478,7 @@ debug_log_level (Debug log level) enum action ,none,error,warning,action,info,ve debug_log_size_max (Debug log file size threshold) int 50 # Minimal level of logging to be written to chat. -chat_log_level (Chat log level) enum error ,none,error,warning,action,info,verbose +chat_log_level (Chat log level) enum error ,none,error,warning,action,info,verbose,trace # Enable IPv6 support (for both client and server). # Required for IPv6 connections to work at all. diff --git a/minetest.conf.example b/minetest.conf.example index 3cdb15026..68757680c 100644 --- a/minetest.conf.example +++ b/minetest.conf.example @@ -1768,7 +1768,8 @@ # - action # - info # - verbose -# type: enum values: , none, error, warning, action, info, verbose +# - trace +# type: enum values: , none, error, warning, action, info, verbose, trace # debug_log_level = action # If the file size of debug.txt exceeds the number of megabytes specified in @@ -1779,7 +1780,7 @@ # debug_log_size_max = 50 # Minimal level of logging to be written to chat. -# type: enum values: , none, error, warning, action, info, verbose +# type: enum values: , none, error, warning, action, info, verbose, trace # chat_log_level = error # Enable IPv6 support (for both client and server). diff --git a/src/client/game.cpp b/src/client/game.cpp index e0e60eb2c..1e34ca286 100644 --- a/src/client/game.cpp +++ b/src/client/game.cpp @@ -1491,7 +1491,7 @@ bool Game::connectToServer(const GameStartData &start_data, client->m_simple_singleplayer_mode = simple_singleplayer_mode; infostream << "Connecting to server at "; - connect_address.print(&infostream); + connect_address.print(infostream); infostream << std::endl; client->connect(connect_address, diff --git a/src/client/renderingengine.cpp b/src/client/renderingengine.cpp index 723865db4..455d5e538 100644 --- a/src/client/renderingengine.cpp +++ b/src/client/renderingengine.cpp @@ -116,7 +116,7 @@ RenderingEngine::RenderingEngine(IEventReceiver *receiver) } SIrrlichtCreationParameters params = SIrrlichtCreationParameters(); - if (g_logger.getTraceEnabled()) + if (tracestream) params.LoggingLevel = irr::ELL_DEBUG; params.DriverType = driverType; params.WindowSize = core::dimension2d(screen_w, screen_h); diff --git a/src/log.cpp b/src/log.cpp index 3c61414e9..51652fe0a 100644 --- a/src/log.cpp +++ b/src/log.cpp @@ -35,43 +35,30 @@ with this program; if not, write to the Free Software Foundation, Inc., #include #include -const int BUFFER_LENGTH = 256; - -class StringBuffer : public std::streambuf { +class LevelTarget : public LogTarget { public: - StringBuffer() { - buffer_index = 0; - } - - int overflow(int c); - virtual void flush(const std::string &buf) = 0; - std::streamsize xsputn(const char *s, std::streamsize n); - void push_back(char c); - -private: - char buffer[BUFFER_LENGTH]; - int buffer_index; -}; - - -class LogBuffer : public StringBuffer { -public: - LogBuffer(Logger &logger, LogLevel lev) : - logger(logger), - level(lev) + LevelTarget(Logger &logger, LogLevel level, bool raw = false) : + m_logger(logger), + m_level(level), + m_raw(raw) {} - void flush(const std::string &buffer); - -private: - Logger &logger; - LogLevel level; -}; + virtual bool hasOutput() override { + return m_logger.hasOutput(m_level); + } + virtual void log(const std::string &buf) override { + if (!m_raw) { + m_logger.log(m_level, buf); + } else { + m_logger.logRaw(m_level, buf); + } + } -class RawLogBuffer : public StringBuffer { -public: - void flush(const std::string &buffer); +private: + Logger &m_logger; + LogLevel m_level; + bool m_raw; }; //// @@ -80,31 +67,33 @@ public: Logger g_logger; +#ifdef __ANDROID__ +AndroidLogOutput stdout_output; +AndroidLogOutput stderr_output; +#else StreamLogOutput stdout_output(std::cout); StreamLogOutput stderr_output(std::cerr); -std::ostream null_stream(NULL); - -RawLogBuffer raw_buf; - -LogBuffer none_buf(g_logger, LL_NONE); -LogBuffer error_buf(g_logger, LL_ERROR); -LogBuffer warning_buf(g_logger, LL_WARNING); -LogBuffer action_buf(g_logger, LL_ACTION); -LogBuffer info_buf(g_logger, LL_INFO); -LogBuffer verbose_buf(g_logger, LL_VERBOSE); - -// Connection -std::ostream *dout_con_ptr = &null_stream; -std::ostream *derr_con_ptr = &verbosestream; - -// Common streams -std::ostream rawstream(&raw_buf); -std::ostream dstream(&none_buf); -std::ostream errorstream(&error_buf); -std::ostream warningstream(&warning_buf); -std::ostream actionstream(&action_buf); -std::ostream infostream(&info_buf); -std::ostream verbosestream(&verbose_buf); +#endif + +LevelTarget none_target_raw(g_logger, LL_NONE, true); +LevelTarget none_target(g_logger, LL_NONE); +LevelTarget error_target(g_logger, LL_ERROR); +LevelTarget warning_target(g_logger, LL_WARNING); +LevelTarget action_target(g_logger, LL_ACTION); +LevelTarget info_target(g_logger, LL_INFO); +LevelTarget verbose_target(g_logger, LL_VERBOSE); +LevelTarget trace_target(g_logger, LL_TRACE); + +thread_local LogStream dstream(none_target); +thread_local LogStream rawstream(none_target_raw); +thread_local LogStream errorstream(error_target); +thread_local LogStream warningstream(warning_target); +thread_local LogStream actionstream(action_target); +thread_local LogStream infostream(info_target); +thread_local LogStream verbosestream(verbose_target); +thread_local LogStream tracestream(trace_target); +thread_local LogStream derr_con(verbose_target); +thread_local LogStream dout_con(trace_target); // Android #ifdef __ANDROID__ @@ -118,29 +107,15 @@ static unsigned int g_level_to_android[] = { //ANDROID_LOG_INFO, ANDROID_LOG_DEBUG, // LL_INFO ANDROID_LOG_VERBOSE, // LL_VERBOSE + ANDROID_LOG_VERBOSE, // LL_TRACE }; -class AndroidSystemLogOutput : public ICombinedLogOutput { - public: - AndroidSystemLogOutput() - { - g_logger.addOutput(this); - } - ~AndroidSystemLogOutput() - { - g_logger.removeOutput(this); - } - void logRaw(LogLevel lev, const std::string &line) - { - STATIC_ASSERT(ARRLEN(g_level_to_android) == LL_MAX, - mismatch_between_android_and_internal_loglevels); - __android_log_print(g_level_to_android[lev], - PROJECT_NAME_C, "%s", line.c_str()); - } -}; - -AndroidSystemLogOutput g_android_log_output; - +void AndroidLogOutput::logRaw(LogLevel lev, const std::string &line) { + STATIC_ASSERT(ARRLEN(g_level_to_android) == LL_MAX, + mismatch_between_android_and_internal_loglevels); + __android_log_print(g_level_to_android[lev], + PROJECT_NAME_C, "%s", line.c_str()); +} #endif /////////////////////////////////////////////////////////////////////////////// @@ -164,6 +139,8 @@ LogLevel Logger::stringToLevel(const std::string &name) return LL_INFO; else if (name == "verbose") return LL_VERBOSE; + else if (name == "trace") + return LL_TRACE; else return LL_MAX; } @@ -176,21 +153,26 @@ void Logger::addOutput(ILogOutput *out) void Logger::addOutput(ILogOutput *out, LogLevel lev) { m_outputs[lev].push_back(out); + m_has_outputs[lev] = true; } void Logger::addOutputMasked(ILogOutput *out, LogLevelMask mask) { for (size_t i = 0; i < LL_MAX; i++) { - if (mask & LOGLEVEL_TO_MASKLEVEL(i)) + if (mask & LOGLEVEL_TO_MASKLEVEL(i)) { m_outputs[i].push_back(out); + m_has_outputs[i] = true; + } } } void Logger::addOutputMaxLevel(ILogOutput *out, LogLevel lev) { assert(lev < LL_MAX); - for (size_t i = 0; i <= lev; i++) + for (size_t i = 0; i <= lev; i++) { m_outputs[i].push_back(out); + m_has_outputs[i] = true; + } } LogLevelMask Logger::removeOutput(ILogOutput *out) @@ -203,6 +185,7 @@ LogLevelMask Logger::removeOutput(ILogOutput *out) if (it != m_outputs[i].end()) { ret_mask |= LOGLEVEL_TO_MASKLEVEL(i); m_outputs[i].erase(it); + m_has_outputs[i] = !m_outputs[i].empty(); } } return ret_mask; @@ -236,6 +219,7 @@ const std::string Logger::getLevelLabel(LogLevel lev) "ACTION", "INFO", "VERBOSE", + "TRACE", }; assert(lev < LL_MAX && lev >= 0); STATIC_ASSERT(ARRLEN(names) == LL_MAX, @@ -297,7 +281,6 @@ void Logger::logToOutputs(LogLevel lev, const std::string &combined, m_outputs[lev][i]->log(lev, combined, time, thread_name, payload_text); } - //// //// *LogOutput methods //// @@ -349,6 +332,7 @@ void StreamLogOutput::logRaw(LogLevel lev, const std::string &line) m_stream << "\033[37m"; break; case LL_VERBOSE: + case LL_TRACE: // verbose is darker than info m_stream << "\033[2m"; break; @@ -396,6 +380,7 @@ void LogOutputBuffer::logRaw(LogLevel lev, const std::string &line) color = "\x1b(c@#BBB)"; break; case LL_VERBOSE: // dark grey + case LL_TRACE: color = "\x1b(c@#888)"; break; default: break; @@ -404,47 +389,3 @@ void LogOutputBuffer::logRaw(LogLevel lev, const std::string &line) m_buffer.push(color.append(line)); } - -//// -//// *Buffer methods -//// - -int StringBuffer::overflow(int c) -{ - push_back(c); - return c; -} - - -std::streamsize StringBuffer::xsputn(const char *s, std::streamsize n) -{ - for (int i = 0; i < n; ++i) - push_back(s[i]); - return n; -} - -void StringBuffer::push_back(char c) -{ - if (c == '\n' || c == '\r') { - if (buffer_index) - flush(std::string(buffer, buffer_index)); - buffer_index = 0; - } else { - buffer[buffer_index++] = c; - if (buffer_index >= BUFFER_LENGTH) { - flush(std::string(buffer, buffer_index)); - buffer_index = 0; - } - } -} - - -void LogBuffer::flush(const std::string &buffer) -{ - logger.log(level, buffer); -} - -void RawLogBuffer::flush(const std::string &buffer) -{ - g_logger.logRaw(LL_NONE, buffer); -} diff --git a/src/log.h b/src/log.h index 6ed6b1fb7..5c6ba7d64 100644 --- a/src/log.h +++ b/src/log.h @@ -19,6 +19,7 @@ with this program; if not, write to the Free Software Foundation, Inc., #pragma once +#include #include #include #include @@ -28,6 +29,8 @@ with this program; if not, write to the Free Software Foundation, Inc., #if !defined(_WIN32) // POSIX #include #endif +#include "util/basic_macros.h" +#include "util/stream.h" #include "irrlichttypes.h" class ILogOutput; @@ -39,6 +42,7 @@ enum LogLevel { LL_ACTION, // In-game actions LL_INFO, LL_VERBOSE, + LL_TRACE, LL_MAX, }; @@ -67,12 +71,13 @@ public: // Logs without a prefix void logRaw(LogLevel lev, const std::string &text); - void setTraceEnabled(bool enable) { m_trace_enabled = enable; } - bool getTraceEnabled() { return m_trace_enabled; } - static LogLevel stringToLevel(const std::string &name); static const std::string getLevelLabel(LogLevel lev); + bool hasOutput(LogLevel level) { + return m_has_outputs[level].load(std::memory_order_relaxed); + } + static LogColor color_mode; private: @@ -84,6 +89,7 @@ private: const std::string getThreadName(); std::vector m_outputs[LL_MAX]; + std::atomic m_has_outputs[LL_MAX]; // Should implement atomic loads and stores (even though it's only // written to when one thread has access currently). @@ -91,7 +97,6 @@ private: volatile bool m_silenced_levels[LL_MAX]; std::map m_thread_names; mutable std::mutex m_mutex; - bool m_trace_enabled; }; class ILogOutput { @@ -185,35 +190,178 @@ private: Logger &m_logger; }; +#ifdef __ANDROID__ +class AndroidLogOutput : public ICombinedLogOutput { +public: + void logRaw(LogLevel lev, const std::string &line); +}; +#endif -extern StreamLogOutput stdout_output; -extern StreamLogOutput stderr_output; -extern std::ostream null_stream; +/* + * LogTarget + * + * This is the interface that sits between the LogStreams and the global logger. + * Primarily used to route streams to log levels, but could also enable other + * custom behavior. + * + */ +class LogTarget { +public: + // Must be thread-safe. These can be called from any thread. + virtual bool hasOutput() = 0; + virtual void log(const std::string &buf) = 0; +}; -extern std::ostream *dout_con_ptr; -extern std::ostream *derr_con_ptr; -extern std::ostream *derr_server_ptr; -extern Logger g_logger; +/* + * StreamProxy + * + * An ostream-like object that can proxy to a real ostream or do nothing, + * depending on how it is configured. See LogStream below. + * + */ +class StreamProxy { +public: + StreamProxy(std::ostream *os) : m_os(os) { } + + template + StreamProxy& operator<<(T&& arg) { + if (m_os) { + *m_os << std::forward(arg); + } + return *this; + } -// Writes directly to all LL_NONE log outputs for g_logger with no prefix. -extern std::ostream rawstream; + StreamProxy& operator<<(std::ostream& (*manip)(std::ostream&)) { + if (m_os) { + *m_os << manip; + } + return *this; + } -extern std::ostream errorstream; -extern std::ostream warningstream; -extern std::ostream actionstream; -extern std::ostream infostream; -extern std::ostream verbosestream; -extern std::ostream dstream; +private: + std::ostream *m_os; +}; -#define TRACEDO(x) do { \ - if (g_logger.getTraceEnabled()) { \ - x; \ - } \ -} while (0) -#define TRACESTREAM(x) TRACEDO(verbosestream x) +/* + * LogStream + * + * The public interface for log streams (infostream, verbosestream, etc). + * + * LogStream minimizes the work done when a given stream is off. (meaning + * it has no output targets, so it goes to /dev/null) + * + * For example, consider: + * + * verbosestream << "hello world" << 123 << std::endl; + * + * The compiler evaluates this as: + * + * (((verbosestream << "hello world") << 123) << std::endl) + * ^ ^ + * + * If `verbosestream` is on, the innermost expression (marked by ^) will return + * a StreamProxy that forwards to a real ostream, that feeds into the logger. + * However, if `verbosestream` is off, it will return a StreamProxy that does + * nothing on all later operations. Specifically, CPU time won't be wasted + * writing "hello world" and 123 into a buffer, or formatting the log entry. + * + * It is also possible to directly check if the stream is on/off: + * + * if (verbosestream) { + * auto data = ComputeExpensiveDataForTheLog(); + * verbosestream << data << endl; + * } + * +*/ -#define dout_con (*dout_con_ptr) -#define derr_con (*derr_con_ptr) +class LogStream { +public: + LogStream() = delete; + DISABLE_CLASS_COPY(LogStream); + + LogStream(LogTarget &target) : + m_target(target), + m_buffer(std::bind(&LogStream::internalFlush, this, std::placeholders::_1)), + m_dummy_buffer(), + m_stream(&m_buffer), + m_dummy_stream(&m_dummy_buffer), + m_proxy(&m_stream), + m_dummy_proxy(nullptr) { } + + template + StreamProxy& operator<<(T&& arg) { + StreamProxy& sp = m_target.hasOutput() ? m_proxy : m_dummy_proxy; + sp << std::forward(arg); + return sp; + } + StreamProxy& operator<<(std::ostream& (*manip)(std::ostream&)) { + StreamProxy& sp = m_target.hasOutput() ? m_proxy : m_dummy_proxy; + sp << manip; + return sp; + } + + operator bool() { + return m_target.hasOutput(); + } + + void internalFlush(const std::string &buf) { + m_target.log(buf); + } + + operator std::ostream&() { + return m_target.hasOutput() ? m_stream : m_dummy_stream; + } + +private: + // 10 streams per thread x (256 + overhead) ~ 3K per thread + static const int BUFFER_LENGTH = 256; + LogTarget &m_target; + StringStreamBuffer m_buffer; + DummyStreamBuffer m_dummy_buffer; + std::ostream m_stream; + std::ostream m_dummy_stream; + StreamProxy m_proxy; + StreamProxy m_dummy_proxy; + +}; + +#ifdef __ANDROID__ +extern AndroidLogOutput stdout_output; +extern AndroidLogOutput stderr_output; +#else +extern StreamLogOutput stdout_output; +extern StreamLogOutput stderr_output; +#endif + +extern Logger g_logger; + +/* + * By making the streams thread_local, each thread has its own + * private buffer. Two or more threads can write to the same stream + * simultaneously (lock-free), and there won't be any interference. + * + * The finished lines are sent to a LogTarget which is a global (not thread-local) + * object, and from there relayed to g_logger. The final writes are serialized + * by the mutex in g_logger. +*/ + +extern thread_local LogStream dstream; +extern thread_local LogStream rawstream; // Writes directly to all LL_NONE log outputs with no prefix. +extern thread_local LogStream errorstream; +extern thread_local LogStream warningstream; +extern thread_local LogStream actionstream; +extern thread_local LogStream infostream; +extern thread_local LogStream verbosestream; +extern thread_local LogStream tracestream; +// TODO: Search/replace these with verbose/tracestream +extern thread_local LogStream derr_con; +extern thread_local LogStream dout_con; + +#define TRACESTREAM(x) do { \ + if (tracestream) { \ + tracestream x; \ + } \ +} while (0) diff --git a/src/main.cpp b/src/main.cpp index 5ea212d8a..ddd725134 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -453,14 +453,6 @@ static bool setup_log_params(const Settings &cmd_args) } } - // If trace is enabled, enable logging of certain things - if (cmd_args.getFlag("trace")) { - dstream << _("Enabling trace level debug output") << std::endl; - g_logger.setTraceEnabled(true); - dout_con_ptr = &verbosestream; // This is somewhat old - socket_enable_debug_output = true; // Sockets doesn't use log.h - } - // In certain cases, output info level on stderr if (cmd_args.getFlag("info") || cmd_args.getFlag("verbose") || cmd_args.getFlag("trace") || cmd_args.getFlag("speedtests")) @@ -470,6 +462,12 @@ static bool setup_log_params(const Settings &cmd_args) if (cmd_args.getFlag("verbose") || cmd_args.getFlag("trace")) g_logger.addOutput(&stderr_output, LL_VERBOSE); + if (cmd_args.getFlag("trace")) { + dstream << _("Enabling trace level debug output") << std::endl; + g_logger.addOutput(&stderr_output, LL_TRACE); + socket_enable_debug_output = true; + } + return true; } @@ -599,7 +597,7 @@ static void init_log_streams(const Settings &cmd_args) warningstream << "Deprecated use of debug_log_level with an " "integer value; please update your configuration." << std::endl; static const char *lev_name[] = - {"", "error", "action", "info", "verbose"}; + {"", "error", "action", "info", "verbose", "trace"}; int lev_i = atoi(conf_loglev.c_str()); if (lev_i < 0 || lev_i >= (int)ARRLEN(lev_name)) { warningstream << "Supplied invalid debug_log_level!" diff --git a/src/network/address.cpp b/src/network/address.cpp index 90e561802..cf2e6208d 100644 --- a/src/network/address.cpp +++ b/src/network/address.cpp @@ -230,14 +230,14 @@ void Address::setPort(u16 port) m_port = port; } -void Address::print(std::ostream *s) const +void Address::print(std::ostream& s) const { if (m_addr_family == AF_INET6) - *s << "[" << serializeString() << "]:" << m_port; + s << "[" << serializeString() << "]:" << m_port; else if (m_addr_family == AF_INET) - *s << serializeString() << ":" << m_port; + s << serializeString() << ":" << m_port; else - *s << "(undefined)"; + s << "(undefined)"; } bool Address::isLocalhost() const diff --git a/src/network/address.h b/src/network/address.h index c2f5f2eef..692bf82c5 100644 --- a/src/network/address.h +++ b/src/network/address.h @@ -59,7 +59,7 @@ public: int getFamily() const { return m_addr_family; } bool isIPv6() const { return m_addr_family == AF_INET6; } bool isZero() const; - void print(std::ostream *s) const; + void print(std::ostream &s) const; std::string serializeString() const; bool isLocalhost() const; diff --git a/src/network/connection.cpp b/src/network/connection.cpp index 2d3cf6e88..6fb676f25 100644 --- a/src/network/connection.cpp +++ b/src/network/connection.cpp @@ -41,25 +41,14 @@ namespace con /* defines used for debugging and profiling */ /******************************************************************************/ #ifdef NDEBUG - #define LOG(a) a #define PROFILE(a) #else - #if 0 - /* this mutex is used to achieve log message consistency */ - std::mutex log_message_mutex; - #define LOG(a) \ - { \ - MutexAutoLock loglock(log_message_mutex); \ - a; \ - } - #else - // Prevent deadlocks until a solution is found after 5.2.0 (TODO) - #define LOG(a) a - #endif - #define PROFILE(a) a #endif +// TODO: Clean this up. +#define LOG(a) a + #define PING_TIMEOUT 5.0 u16 BufferedPacket::getSeqnum() const diff --git a/src/network/connectionthreads.cpp b/src/network/connectionthreads.cpp index dca065ae1..90936b43d 100644 --- a/src/network/connectionthreads.cpp +++ b/src/network/connectionthreads.cpp @@ -32,22 +32,18 @@ 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 */ -std::mutex log_conthread_mutex; -#define LOG(a) \ - { \ - MutexAutoLock loglock(log_conthread_mutex); \ - a; \ - } #define PROFILE(a) a //#define DEBUG_CONNECTION_KBPS #undef DEBUG_CONNECTION_KBPS #endif +// TODO: Clean this up. +#define LOG(a) a + #define WINDOW_SIZE 5 static session_t readPeerId(const u8 *packetdata) diff --git a/src/network/socket.cpp b/src/network/socket.cpp index 97a5f19f7..df15c89ba 100644 --- a/src/network/socket.cpp +++ b/src/network/socket.cpp @@ -198,7 +198,7 @@ void UDPSocket::Send(const Address &destination, const void *data, int size) if (socket_enable_debug_output) { // Print packet destination and size dstream << (int)m_handle << " -> "; - destination.print(&dstream); + destination.print(dstream); dstream << ", size=" << size; // Print packet contents @@ -295,7 +295,7 @@ int UDPSocket::Receive(Address &sender, void *data, int size) if (socket_enable_debug_output) { // Print packet sender and size dstream << (int)m_handle << " <- "; - sender.print(&dstream); + sender.print(dstream); dstream << ", size=" << received; // Print packet contents diff --git a/src/server.cpp b/src/server.cpp index c9cd4e398..6a4349ba5 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -525,7 +525,7 @@ void Server::start() actionstream << "World at [" << m_path_world << "]" << std::endl; actionstream << "Server for gameid=\"" << m_gamespec.id << "\" listening on "; - m_bind_addr.print(&actionstream); + m_bind_addr.print(actionstream); actionstream << "." << std::endl; } diff --git a/src/util/stream.h b/src/util/stream.h new file mode 100644 index 000000000..2e61b46d2 --- /dev/null +++ b/src/util/stream.h @@ -0,0 +1,70 @@ +/* +Minetest +Copyright (C) 2022 Minetest Authors + +This program is free software; you can redistribute it and/or modify +it under the terms of the GNU Lesser General Public License as published by +the Free Software Foundation; either version 2.1 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Lesser General Public License for more details. + +You should have received a copy of the GNU Lesser General Public License along +with this program; if not, write to the Free Software Foundation, Inc., +51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +*/ + +#pragma once + +#include +#include +#include + +template > +class StringStreamBuffer : public std::streambuf { +public: + StringStreamBuffer(Emitter emitter) : m_emitter(emitter) { + buffer_index = 0; + } + + int overflow(int c) { + push_back(c); + return c; + } + + void push_back(char c) { + if (c == '\n' || c == '\r') { + if (buffer_index) + m_emitter(std::string(buffer, buffer_index)); + buffer_index = 0; + } else { + buffer[buffer_index++] = c; + if (buffer_index >= BufferLength) { + m_emitter(std::string(buffer, buffer_index)); + buffer_index = 0; + } + } + } + + std::streamsize xsputn(const char *s, std::streamsize n) { + for (int i = 0; i < n; ++i) + push_back(s[i]); + return n; + } +private: + Emitter m_emitter; + char buffer[BufferLength]; + int buffer_index; +}; + +class DummyStreamBuffer : public std::streambuf { + int overflow(int c) { + return c; + } + std::streamsize xsputn(const char *s, std::streamsize n) { + return n; + } +}; -- cgit v1.2.3