Fix/http client bounded parallelism

addons/twitcher/lib/http/buffered_http_client.gd

@@ -2,7 +2,9 @@
 @tool
 extends Twitcher
 
-## Http client that bufferes the requests and sends them sequentialy
+## Http client that buffers the requests and sends at most [member max_parallel_requests]
+## of them at the same time. Everything above that limit waits in a queue and is
+## dispatched in the order it was requested as soon as a slot gets free.
 class_name BufferedHTTPClient
 
 
@@ -17,7 +19,7 @@ signal request_done(response: ResponseData)
 class RequestData extends RefCounted:
 	## The client that the request belongs too
 	var client: BufferedHTTPClient
-	## The request node that is executing the request
+	## The request node that is executing the request (null while the request waits in the queue)
 	var http_request: HTTPRequest
 	## Path of the request
 	var path: String
@@ -29,10 +31,14 @@ class RequestData extends RefCounted:
 	var body: String = ""
 	## Amount of retries
 	var retry: int
+	## `Time.get_ticks_msec()` when the request was put on the wire (first attempt or retry)
+	var started_at: int
 
 	## When you are done free the request
 	func queue_free() -> void:
-		http_request.queue_free()
+		if http_request != null:
+			http_request.queue_free()
+			http_request = null
 
 
 ## Contains the response data
@@ -56,19 +62,26 @@ class ResponseData extends RefCounted:
 
 ## When a request fails max_error_count then cancel that request -1 for endless amount of tries.
 @export var max_error_count : int = -1
+## How many requests may be on the wire at the same time. [code]1[/code] sends them strictly
+## one after another, which is the safe default for API calls (see PR #130: several threaded
+## requests started in the same frame can stall and time out). Raise it for clients that
+## download many independent small files like emotes and badges. [code]0[/code] or a negative
+## value removes the limit.
+@export var max_parallel_requests : int = 1
 @export var custom_header : Dictionary[String, String] = { "Accept": "*/*" }
+## Seconds until a single attempt of a request is aborted with [constant HTTPRequest.RESULT_TIMEOUT]
+@export var request_timeout : float = 30
 
+## Every request that was started and whose response wasn't consumed via `wait_for_request` yet.
 var requests : Array[RequestData] = []
-var current_request : RequestData
-var current_response_data : PackedByteArray = PackedByteArray()
+## Requests that wait for a free slot, in the order they were requested.
+var queued_requests : Array[RequestData] = []
+## Requests that are currently on the wire (including retries).
+var active_requests : Array[RequestData] = []
 var responses : Dictionary = {}
-var error_count : int
 
-
-## Only one poll at a time so block for all other tries to call it
-var polling: bool
 var processing: bool:
-	get: return not requests.is_empty() || current_request != null
+	get: return not requests.is_empty()
 
 
 ## Starts a request that will be handled as soon as the client gets free.
@@ -83,16 +96,11 @@ func request(path: String, method: int, headers: Dictionary, body: String) -> Re
 	req.body = body
 	req.headers = headers
 	req.client = self
-	req.http_request = HTTPRequest.new()
-	req.http_request.use_threads = true
-	req.http_request.timeout = 30
-	req.http_request.request_completed.connect(_on_request_completed.bind(req))
-	add_child(req.http_request)
-	var err : Error = req.http_request.request(req.path, _pack_headers(req.headers), req.method, req.body)
-	if err != OK: logError("Problems with request to %s cause of %s" % [path, error_string(err)])
 	requests.append(req)
+	queued_requests.append(req)
 	request_added.emit(req)
-	logDebug("[%s] request started " % [ path ])
+	logDebug("[%s] request queued (queued: %s, active: %s)" % [ path, queued_requests.size(), active_requests.size() ])
+	_dispatch()
 	return req
 
 
@@ -116,6 +124,39 @@ func wait_for_request(request_data: RequestData) -> ResponseData:
 	return latest_response
 
 
+## Sends queued requests as long as there are free slots
+func _dispatch() -> void:
+	while not queued_requests.is_empty() and _has_free_slot():
+		var req: RequestData = queued_requests.pop_front()
+		active_requests.append(req)
+		_send(req)
+
+
+func _has_free_slot() -> bool:
+	return max_parallel_requests <= 0 or active_requests.size() < max_parallel_requests
+
+
+## Puts a request on the wire. Used for the first attempt and for every retry.
+func _send(request_data: RequestData) -> void:
+	if request_data.http_request != null:
+		request_data.http_request.queue_free()
+	var http_request: HTTPRequest = HTTPRequest.new()
+	http_request.use_threads = true
+	http_request.timeout = request_timeout
+	http_request.request_completed.connect(_on_request_completed.bind(request_data))
+	add_child(http_request)
+	request_data.http_request = http_request
+	request_data.started_at = Time.get_ticks_msec()
+	var err : Error = http_request.request(request_data.path, _pack_headers(request_data.headers), request_data.method, request_data.body)
+	if err != OK:
+		logError("Problems with request to %s cause of %s" % [request_data.path, error_string(err)])
+		# HTTPRequest doesn't emit request_completed when request() fails, finish it
+		# ourself otherwise the request blocks its slot forever and waiters hang.
+		_on_request_completed.call_deferred(HTTPRequest.Result.RESULT_REQUEST_FAILED, 0, PackedStringArray(), PackedByteArray(), request_data)
+		return
+	logDebug("[%s] request started " % [ request_data.path ])
+
+
 func _on_request_completed(result: int, response_code: int, headers: PackedStringArray, body: PackedByteArray, request_data: RequestData) -> void:
 	var response_data : ResponseData = ResponseData.new()
 	if result != HTTPRequest.Result.RESULT_SUCCESS:
@@ -124,25 +165,27 @@ func _on_request_completed(result: int, response_code: int, headers: PackedStrin
 	if result == HTTPRequest.Result.RESULT_CONNECTION_ERROR || result == HTTPRequest.Result.RESULT_TLS_HANDSHAKE_ERROR:
 		if request_data.retry == max_error_count:
 			printerr("Maximum amount of retries for the request. Abort request: %s" % [request_data.path])
+			# Fall through and deliver the failed response, so that waiters get an
+			# answer and the slot is released for the next queued request.
+		else:
+			var wait_time = pow(2, request_data.retry)
+			wait_time = min(wait_time, 30)
+			logDebug("Error happend during connection. Wait for %s" % wait_time)
+			await get_tree().create_timer(wait_time, true, false, true).timeout
+			request_data.retry += 1
+			# The request keeps its slot while retrying
+			_send(request_data)
 			return
-		var wait_time = pow(2, request_data.retry)
-		wait_time = min(wait_time, 30)
-		logDebug("Error happend during connection. Wait for %s" % wait_time)
-		await get_tree().create_timer(wait_time, true, false, true).timeout
-		var http_request: HTTPRequest = request_data.http_request.duplicate()
-		add_child(http_request)
-		request_data.http_request = http_request
-		request_data.retry += 1
-		http_request.request(request_data.path, _pack_headers(request_data.headers), request_data.method, request_data.body)
-		http_request.request_completed.connect(_on_request_completed.bind(http_request))
 
 	response_data.result = result
 	response_data.request_data = request_data
 	response_data.response_data = body
 	response_data.response_code = response_code
 	response_data.response_header = _get_response_headers_as_dictionary(headers)
 	responses[request_data] = response_data
-	logInfo("[%s] request done with result HTTPRequest.Result[%s] " % [ request_data.path, result])
+	logInfo("[%s] request done with result HTTPRequest.Result[%s] code %s after %sms (retries: %s)" % [ request_data.path, result, response_code, Time.get_ticks_msec() - request_data.started_at, request_data.retry ])
+	active_requests.erase(request_data)
+	_dispatch()
 	request_done.emit(response_data)
 
 
@@ -170,12 +213,9 @@ func _pack_headers(headers: Dictionary) -> PackedStringArray:
 	return result
 
 
-## The amount of requests that are pending
+## The amount of requests that are pending (waiting in the queue or on the wire)
 func queued_request_size() -> int:
-	var requests_size: int = requests.size()
-	if current_request != null:
-		requests_size += 1
-	return requests_size
+	return queued_requests.size() + active_requests.size()
 
 
 func empty_response(request_data: RequestData) -> ResponseData:

addons/twitcher/media/twitch_media_loader.gd

@@ -23,6 +23,13 @@ const FALLBACK_PROFILE = preload("res://addons/twitcher/assets/no_profile.png")
 @export var fallback_texture: Texture2D = FALLBACK_TEXTURE
 @export var fallback_profile: Texture2D = FALLBACK_PROFILE
 @export var image_cdn_host: String = "https://static-cdn.jtvnw.net"
+## How many images (emotes, badges, cheermotes, profiles) are downloaded at the same time.
+## Unlike API calls, image downloads are many small independent files and get noticeably
+## slow when they are fetched one after another. See [member BufferedHTTPClient.max_parallel_requests].
+@export var max_parallel_downloads: int = 8:
+	set(val):
+		max_parallel_downloads = val
+		if _client != null: _client.max_parallel_requests = val
 ## Will preload the whole badge and emote cache also to editor time (use it when you make a Editor Plugin with Twitch Support)
 @export var load_cache_in_editor: bool
 
@@ -51,6 +58,7 @@ var _client: BufferedHTTPClient
 func _ready() -> void:
 	_client = BufferedHTTPClient.new()
 	_client.name = "TwitchMediaLoaderClient"
+	_client.max_parallel_requests = max_parallel_downloads
 	add_child(_client)
 	_load_cache()
 	if api == null: api = TwitchAPI.instance