obs-outputs: Various fixes to new network code

Fixes another 100% CPU spin bug, improves low latency mode, fixes some
crashes on stopping stream due to race conditions.

plugins/obs-outputs/rtmp-stream.c

@@ -92,9 +92,6 @@ static void rtmp_stream_destroy(void *data)
 		}
 	}
 
-	if (stream->socket_thread_active)
-		pthread_join(stream->socket_thread, NULL);
-
 	free_packets(stream);
 	dstr_free(&stream->path);
 	dstr_free(&stream->key);
@@ -390,6 +387,12 @@ static void *send_thread(void *data)
 		info("User stopped the stream");
 	}
 
+	if (stream->new_socket_loop) {
+		os_event_signal(stream->send_thread_signaled_exit);
+		os_event_signal(stream->buffer_has_data_event);
+		pthread_join(stream->socket_thread, NULL);
+	}
+
 	RTMP_Close(&stream->rtmp);
 
 	if (!stopping(stream)) {
@@ -546,20 +549,25 @@ static int init_send(struct rtmp_stream *stream)
 		}
 
 		if (os_event_init(&stream->buffer_space_available_event,
-					OS_EVENT_TYPE_MANUAL) != 0) {
+					OS_EVENT_TYPE_AUTO) != 0) {
 			warn("Failed to initialize write buffer event");
 			return OBS_OUTPUT_ERROR;
 		}
 		if (os_event_init(&stream->buffer_has_data_event,
-					OS_EVENT_TYPE_MANUAL) != 0) {
+					OS_EVENT_TYPE_AUTO) != 0) {
 			warn("Failed to initialize data buffer event");
 			return OBS_OUTPUT_ERROR;
 		}
 		if (os_event_init(&stream->socket_available_event,
-					OS_EVENT_TYPE_MANUAL) != 0) {
+					OS_EVENT_TYPE_AUTO) != 0) {
 			warn("Failed to initialize socket buffer event");
 			return OBS_OUTPUT_ERROR;
 		}
+		if (os_event_init(&stream->send_thread_signaled_exit,
+					OS_EVENT_TYPE_MANUAL) != 0) {
+			warn("Failed to initialize socket exit event");
+			return OBS_OUTPUT_ERROR;
+		}
 
 		info("New socket loop enabled by user");
 		if (stream->low_latency_mode)
@@ -568,8 +576,37 @@ static int init_send(struct rtmp_stream *stream)
 		if (stream->write_buf)
 			bfree(stream->write_buf);
 
-		stream->write_buf_size = STREAM_WRITE_BUFFER_SIZE;
-		stream->write_buf = bmalloc(STREAM_WRITE_BUFFER_SIZE);
+		int total_bitrate = 0;
+		obs_output_t  *context  = stream->output;
+
+		obs_encoder_t *vencoder = obs_output_get_video_encoder(context);
+		if (vencoder) {
+			obs_data_t *params = obs_encoder_get_settings(vencoder);
+			if (params) {
+				int bitrate = obs_data_get_int(params, "bitrate");
+				total_bitrate += bitrate;
+				obs_data_release(params);
+			}
+		}
+
+		obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, 0);
+		if (aencoder) {
+			obs_data_t *params = obs_encoder_get_settings(aencoder);
+			if (params) {
+				int bitrate = obs_data_get_int(params, "bitrate");
+				total_bitrate += bitrate;
+				obs_data_release(params);
+			}
+		}
+
+		// to bytes/sec
+		int ideal_buffer_size = total_bitrate * 128;
+
+		if (ideal_buffer_size < 131072)
+			ideal_buffer_size = 131072;
+
+		stream->write_buf_size = ideal_buffer_size;
+		stream->write_buf = bmalloc(ideal_buffer_size);
 
 #ifdef _WIN32
 		ret = pthread_create(&stream->socket_thread, NULL,

plugins/obs-outputs/rtmp-stream.h

@@ -31,8 +31,6 @@
 #define OPT_NEWSOCKETLOOP_ENABLED "new_socket_loop_enabled"
 #define OPT_LOWLATENCY_ENABLED "low_latency_mode_enabled"
 
-#define STREAM_WRITE_BUFFER_SIZE 524288
-
 //#define TEST_FRAMEDROPS
 
 #ifdef TEST_FRAMEDROPS
@@ -104,6 +102,7 @@ struct rtmp_stream {
 	os_event_t       *buffer_space_available_event;
 	os_event_t       *buffer_has_data_event;
 	os_event_t       *socket_available_event;
+	os_event_t       *send_thread_signaled_exit;
 };
 
 #ifdef _WIN32

plugins/obs-outputs/rtmp-windows.c

@@ -7,6 +7,7 @@ static void fatal_sock_shutdown(struct rtmp_stream *stream)
 	closesocket(stream->rtmp.m_sb.sb_socket);
 	stream->rtmp.m_sb.sb_socket = -1;
 	stream->write_buf_len = 0;
+	os_event_signal(stream->buffer_space_available_event);
 }
 
 static bool socket_event(struct rtmp_stream *stream, bool *can_write,
@@ -15,9 +16,9 @@ static bool socket_event(struct rtmp_stream *stream, bool *can_write,
 	WSANETWORKEVENTS net_events;
 	bool success;
 
-	success = !!WSAEnumNetworkEvents(stream->rtmp.m_sb.sb_socket, NULL,
+	success = !WSAEnumNetworkEvents(stream->rtmp.m_sb.sb_socket, NULL,
 			&net_events);
-	if (success) {
+	if (!success) {
 		blog(LOG_ERROR, "socket_thread_windows: Aborting due to "
 				"WSAEnumNetworkEvents failure, %d",
 				WSAGetLastError());
@@ -231,6 +232,8 @@ static enum data_ret write_data(struct rtmp_stream *stream, bool *can_write,
 	return exit_loop ? RET_BREAK : RET_CONTINUE;
 }
 
+#define LATENCY_FACTOR 20
+
 static inline void socket_thread_windows_internal(struct rtmp_stream *stream)
 {
 	bool can_write = false;
@@ -251,8 +254,8 @@ static inline void socket_thread_windows_internal(struct rtmp_stream *stream)
 	send_backlog_event = CreateEvent(NULL, true, false, NULL);
 
 	if (stream->low_latency_mode) {
-		delay_time = 1400.0f / (stream->write_buf_size / 1000.0f);
-		latency_packet_size = 1460;
+		delay_time = 1000 / LATENCY_FACTOR;
+		latency_packet_size = stream->write_buf_size / (LATENCY_FACTOR - 2);
 	} else {
 		latency_packet_size = stream->write_buf_size;
 		delay_time = 0;
@@ -276,11 +279,12 @@ static inline void socket_thread_windows_internal(struct rtmp_stream *stream)
 	objs[2] = send_backlog_event;
 
 	for (;;) {
-		if (os_event_try(stream->stop_event) != EAGAIN) {
+		if (os_event_try(stream->send_thread_signaled_exit) != EAGAIN) {
 			pthread_mutex_lock(&stream->write_buf_mutex);
 			if (stream->write_buf_len == 0) {
 				//blog(LOG_DEBUG, "Exiting on empty buffer");
 				pthread_mutex_unlock(&stream->write_buf_mutex);
+				os_event_reset(stream->send_thread_signaled_exit);
 				break;
 			}