Fixed voice disconnecting

2019-10-22 18:39:52 +02:00
parent 4781e75a11
commit d5ffbff3be
3 changed files with 145 additions and 79 deletions
@@ -42,9 +42,12 @@ VoiceClientConnection::~VoiceClientConnection() {
 		lock_guard write_queue_lock(this->write_queue_lock);
 		this->write_queue.clear();
 	}
-	{
-		lock_guard write_queue_lock(this->write_prepare_queue_lock);
-		this->write_prepare_queue.clear();
+
+	for(auto& category : this->write_preprocess_queues) {
+		lock_guard work_lock{category.work_lock};
+		lock_guard queue_lock{category.queue_lock};
+
+		category.queue.clear();
 	}
 	this->client = nullptr;
 	memtrack::freed<VoiceClientConnection>(this);
@@ -451,7 +454,8 @@ unique_ptr<protocol::ClientPacket> VoiceClientConnection::next_reassembled_packe


 void VoiceClientConnection::sendPacket(const shared_ptr<protocol::ServerPacket>& original_packet, bool copy, bool prepare_directly) {
-	if(this->client->state == ConnectionState::DISCONNECTED) return;
+	if(this->client->state == ConnectionState::DISCONNECTED)
+		return;

 	shared_ptr<protocol::ServerPacket> packet;
 	if(copy) {
@@ -459,16 +463,23 @@ void VoiceClientConnection::sendPacket(const shared_ptr<protocol::ServerPacket>&
 		if(original_packet->getListener())
 			packet->setListener(std::move(original_packet->getListener()));
 		packet->memory_state.flags = original_packet->memory_state.flags;
-	} else
+	} else {
 		packet = original_packet;
+	}

+	auto type = WritePreprocessCategory::from_type(packet->type().type());
+	auto& queue = this->write_preprocess_queues[type];
 	if(prepare_directly) {
 		vector<pipes::buffer> buffers;
 		this->prepare_process_count++;
-		if(!this->prepare_packet_for_write(buffers, packet)) {
-			logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
-			this->prepare_process_count--;
-			return;
+
+		{
+			unique_lock work_lock{queue.work_lock};
+			if(!this->prepare_packet_for_write(buffers, packet, work_lock)) {
+				logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
+				this->prepare_process_count--;
+				return;
+			}
 		}

 		/* enqueue buffers for write */
@@ -478,13 +489,16 @@ void VoiceClientConnection::sendPacket(const shared_ptr<protocol::ServerPacket>&
 		}
 		this->prepare_process_count--; /* we're now done preparing */
 	} else {
-		lock_guard prepare_queue_lock(this->write_prepare_queue_lock);
-		this->write_prepare_queue.push_back(packet);
+		lock_guard queue_lock{queue.queue_lock};
+		queue.queue.push_back(packet);
+		queue.has_work = true;
 	}
 	this->triggerWrite();
 }

-bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &result, const shared_ptr<ServerPacket> &packet) {
+bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &result, const shared_ptr<ServerPacket> &packet, std::unique_lock<std::mutex>& work_lock) {
+	assert(work_lock.owns_lock());
+
 	string error = "success";

 	if(packet->type().compressable() && !packet->memory_state.fragment_entry) {
@@ -498,7 +512,7 @@ bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &resu
 	std::vector<shared_ptr<ServerPacket>> fragments;
 	fragments.reserve((size_t) (packet->data().length() / packet->type().max_length()) + 1);

-	if(packet->data().length() > packet->type().max_length()){
+	if(packet->data().length() > packet->type().max_length()) {
 		if(!packet->type().fragmentable()) {
 			logError(this->client->getServerId(), "{} We've tried to send a too long, not fragmentable, packet. Dropping packet of type {} with length {}", CLIENT_STR_LOG_PREFIX_(this->client), packet->type().name(), packet->data().length());
 			return false;
@@ -534,21 +548,21 @@ bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &resu

 		if(packet->getListener())
 			fragments.back()->setListener(std::move(packet->getListener())); //Move the listener to the last :)
-	} else
+	} else {
 		fragments.push_back(packet);
+	}
+
 	result.reserve(fragments.size());

 	/* apply packet ids */
-	{
-		lock_guard id_lock(this->packet_id_manager_lock);
-		for(const auto& fragment : fragments) {
-			if(!fragment->memory_state.id_branded)
-				fragment->applyPacketId(this->packet_id_manager);
-		}
+	for(const auto& fragment : fragments) {
+		if(!fragment->memory_state.id_branded)
+			fragment->applyPacketId(this->packet_id_manager);
 	}
+	work_lock.unlock(); /* the rest could be unordered */


-	auto statistics = this->client && this->client->getServer() ? this->client->connectionStatistics : nullptr;
+	auto statistics = this->client ? this->client->connectionStatistics : nullptr;
 	for(const auto& fragment : fragments) {
 		if(!this->crypt_handler.progressPacketOut(fragment.get(), error, false)){
 			logError(this->client->getServerId(), "{} Failed to encrypt packet. Error: {}", CLIENT_STR_LOG_PREFIX_(this->client), error);
@@ -565,50 +579,50 @@ bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &resu
 	return true;
 }

-#ifdef VC_USE_READ_QUEUE
-bool VoiceClientConnection::handleNextDatagram() {
-
-	bool flag_empty;
-	pipes::buffer buffer;
-	{
-		lock_guard<recursive_mutex> queue_lock(this->queueLock);
-		if(this->readQueue.empty()) return false;
-		buffer = std::move(this->readQueue.front());
-		this->readQueue.pop_front();
-		flag_empty = this->readQueue.empty();
-	};
-
-	try {
-		this->handleDatagramReceived(buffer);
-	} catch (std::exception& e) {
-		logCritical(this->client->getServerId(), "Handling of raw packet thrown an uncaught exception! (Message: " + string(e.what()) + ")");
-	}
-	return flag_empty;
-}
-#endif
-
-bool VoiceClientConnection::prepare_write_packets() {
-	/* get the next packet to prepare */
-	unique_lock prepare_queue_lock(this->write_prepare_queue_lock);
-	if(this->write_prepare_queue.empty())
-		return false;
+bool VoiceClientConnection::preprocess_write_packets() {
+	std::shared_ptr<ServerPacket> packet{nullptr};
+	vector<pipes::buffer> buffers{};
+	bool flag_more{false};

 	prepare_process_count++; /* we're not preparing a packet */
-	auto packet = move(this->write_prepare_queue.front());
-	this->write_prepare_queue.pop_front();
-	auto flag_more = !this->write_prepare_queue.empty();
-	prepare_queue_lock.unlock();
+	for(auto& category : this->write_preprocess_queues) {
+		if(!category.has_work) continue;
+		else if(packet) {
+			flag_more = true;
+			break;
+		}

-	/* prepare the next packet */
-	vector<pipes::buffer> buffers;
-	if(!this->prepare_packet_for_write(buffers, packet)) {
-		logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
-		this->prepare_process_count--;
-		return flag_more;
+		unique_lock work_lock{category.work_lock, try_to_lock};
+		if(!work_lock) continue; /* This particular category will already be processed */
+
+		{
+			lock_guard buffer_lock{category.queue_lock};
+			if(category.queue.empty()) {
+				category.has_work = false;
+				continue;
+			}
+
+			packet = std::move(category.queue.front());
+			category.queue.pop_front();
+			category.has_work = !category.queue.empty();
+		}
+
+		if(!this->prepare_packet_for_write(buffers, packet, work_lock)) {
+			logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
+			if(flag_more)
+				break;
+			else
+				continue; /* find out if we have more */
+		}
+
+		if(flag_more)
+			break;
+		else
+			continue; /* find out if we have more */
 	}

 	/* enqueue buffers for write */
-	{
+	if(!buffers.empty()) {
 		lock_guard write_queue_lock(this->write_queue_lock);
 		this->write_queue.insert(this->write_queue.end(), buffers.begin(), buffers.end());
 	}
@@ -646,12 +660,18 @@ int VoiceClientConnection::pop_write_buffer(pipes::buffer& target) {

 bool VoiceClientConnection::wait_empty_write_and_prepare_queue(chrono::time_point<chrono::system_clock> until) {
 	while(true) {
-		{
-			lock_guard prepare_lock{this->write_prepare_queue_lock};
-			if(!this->write_prepare_queue.empty())
-				goto _wait;
-			if(this->prepare_process_count != 0)
-				goto _wait;
+		for(auto& queue : this->write_preprocess_queues) {
+			{
+				lock_guard lock{queue.queue_lock};
+				if(!queue.queue.empty())
+					goto _wait;
+			}
+
+			{
+				unique_lock lock{queue.work_lock, try_to_lock};
+				if(!lock.owns_lock())
+					goto _wait;
+			}
 		}
 		{
 			lock_guard buffer_lock{this->write_queue_lock};
@@ -672,12 +692,16 @@ bool VoiceClientConnection::wait_empty_write_and_prepare_queue(chrono::time_poin
 }

 void VoiceClientConnection::reset() {
+	for(auto& queue : this->write_preprocess_queues) {
+		{
+			lock_guard lock{queue.queue_lock};
+			queue.queue.clear();
+		}
+	}
+
 	this->acknowledge_handler.reset();
 	this->crypt_handler.reset();
-	{
-		lock_guard manager_lock(this->packet_id_manager_lock);
-		this->packet_id_manager.reset();
-	}
+	this->packet_id_manager.reset();

 	{
 		lock_guard buffer_lock(this->packet_buffer_lock);