Fix: race-conditions in GUI updates when downloading HTTP files

(cherry picked from commit 56c6df4702015fda7cc7a05b67bfe90b3ede1ad0)

See: https://github.com/OpenTTD/OpenTTD/issues/11636
See: https://github.com/OpenTTD/OpenTTD/pull/11639
wip-string
Patric Stout 6 months ago committed by Jonathan G Rennison
parent 673a0dc5de
commit 6e7c92e3af

@ -26,7 +26,7 @@ struct HTTPCallback {
/**
* We're receiving data.
* @param data the received data, nullptr when all data has been received.
* @param data the received data, nullptr when all data has been received. The implementation is responsible for freeing the data.
* @param length the amount of received data, 0 when all data has been received.
* @note When nullptr is sent the HTTP socket handler is closed/freed.
*/

@ -18,6 +18,7 @@
#include "../network_internal.h"
#include "http.h"
#include "http_shared.h"
#include <atomic>
#include <condition_variable>
@ -45,6 +46,11 @@ static auto _certificate_directories = {
};
#endif /* UNIX */
static std::vector<HTTPThreadSafeCallback *> _http_callbacks;
static std::vector<HTTPThreadSafeCallback *> _new_http_callbacks;
static std::mutex _http_callback_mutex;
static std::mutex _new_http_callback_mutex;
/** Single HTTP request. */
class NetworkHTTPRequest {
public:
@ -60,11 +66,19 @@ public:
callback(callback),
data(data)
{
std::lock_guard<std::mutex> lock(_new_http_callback_mutex);
_new_http_callbacks.push_back(&this->callback);
}
~NetworkHTTPRequest()
{
std::lock_guard<std::mutex> lock(_http_callback_mutex);
_http_callbacks.erase(std::remove(_http_callbacks.begin(), _http_callbacks.end(), &this->callback), _http_callbacks.end());
}
const std::string uri; ///< URI to connect to.
HTTPCallback *callback; ///< Callback to send data back on.
const std::string data; ///< Data to send, if any.
const std::string uri; ///< URI to connect to.
HTTPThreadSafeCallback callback; ///< Callback to send data back on.
const std::string data; ///< Data to send, if any.
};
static std::thread _http_thread;
@ -93,6 +107,20 @@ static std::string _http_ca_path = "";
/* static */ void NetworkHTTPSocketHandler::HTTPReceive()
{
std::lock_guard<std::mutex> lock(_http_callback_mutex);
{
std::lock_guard<std::mutex> lock_new(_new_http_callback_mutex);
if (!_new_http_callbacks.empty()) {
/* We delay adding new callbacks, as HandleQueue() below might add a new callback. */
_http_callbacks.insert(_http_callbacks.end(), _new_http_callbacks.begin(), _new_http_callbacks.end());
_new_http_callbacks.clear();
}
}
for (auto &callback : _http_callbacks) {
callback->HandleQueue();
}
}
void HttpThread()
@ -164,11 +192,18 @@ void HttpThread()
/* Setup our (C-style) callback function which we pipe back into the callback. */
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, +[](char *ptr, size_t size, size_t nmemb, void *userdata) -> size_t {
Debug(net, 4, "HTTP callback: {} bytes", size * nmemb);
HTTPCallback *callback = static_cast<HTTPCallback *>(userdata);
callback->OnReceiveData(ptr, size * nmemb);
HTTPThreadSafeCallback *callback = static_cast<HTTPThreadSafeCallback *>(userdata);
/* Copy the buffer out of CURL. OnReceiveData() will free it when done. */
char *buffer = size * nmemb == 0 ? nullptr : MallocT<char>(size * nmemb);
if (buffer != nullptr) {
memcpy(buffer, ptr, size * nmemb);
}
callback->OnReceiveData(buffer, size * nmemb);
return size * nmemb;
});
curl_easy_setopt(curl, CURLOPT_WRITEDATA, request->callback);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &request->callback);
/* Create a callback from which we can cancel. Sadly, there is no other
* thread-safe way to do this. If the connection went idle, it can take
@ -176,10 +211,10 @@ void HttpThread()
* do about this. */
curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);
curl_easy_setopt(curl, CURLOPT_XFERINFOFUNCTION, +[](void *userdata, curl_off_t /*dltotal*/, curl_off_t /*dlnow*/, curl_off_t /*ultotal*/, curl_off_t /*ulnow*/) -> int {
const HTTPCallback *callback = static_cast<HTTPCallback *>(userdata);
return (callback->IsCancelled() || _http_thread_exit) ? 1 : 0;
const HTTPThreadSafeCallback *callback = static_cast<HTTPThreadSafeCallback *>(userdata);
return (callback->cancelled || _http_thread_exit) ? 1 : 0;
});
curl_easy_setopt(curl, CURLOPT_XFERINFODATA, request->callback);
curl_easy_setopt(curl, CURLOPT_XFERINFODATA, &request->callback);
/* Perform the request. */
CURLcode res = curl_easy_perform(curl);
@ -188,15 +223,18 @@ void HttpThread()
if (res == CURLE_OK) {
Debug(net, 1, "HTTP request succeeded");
request->callback->OnReceiveData(nullptr, 0);
request->callback.OnReceiveData(nullptr, 0);
} else {
long status_code = 0;
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status_code);
/* No need to be verbose about rate limiting. */
Debug(net, (request->callback->IsCancelled() || _http_thread_exit || status_code == HTTP_429_TOO_MANY_REQUESTS) ? 1 : 0, "HTTP request failed: status_code: {}, error: {}", status_code, curl_easy_strerror(res));
request->callback->OnFailure();
Debug(net, (request->callback.cancelled || _http_thread_exit || status_code == HTTP_429_TOO_MANY_REQUESTS) ? 1 : 0, "HTTP request failed: status_code: {}, error: {}", status_code, curl_easy_strerror(res));
request->callback.OnFailure();
}
/* Wait till the callback tells us all data is dequeued. */
request->callback.WaitTillEmpty();
}
curl_easy_cleanup(curl);

@ -0,0 +1,124 @@
/*
* This file is part of OpenTTD.
* OpenTTD is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, version 2.
* OpenTTD 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 General Public License for more details. You should have received a copy of the GNU General Public License along with OpenTTD. If not, see <http://www.gnu.org/licenses/>.
*/
/**
* @file http.h Shared functions for implementations of HTTP requests.
*/
#ifndef NETWORK_CORE_HTTP_SHARED_H
#define NETWORK_CORE_HTTP_SHARED_H
#include "http.h"
#include <condition_variable>
#include <mutex>
#include <vector>
/** Converts a HTTPCallback to a Thread-Safe variant. */
class HTTPThreadSafeCallback {
private:
/** Entries on the queue for later handling. */
class Callback {
public:
Callback(const char *data, size_t length) : data(data), length(length), failure(false) {}
Callback() : data(nullptr), length(0), failure(true) {}
const char *data;
size_t length;
bool failure;
};
public:
/**
* Similar to HTTPCallback::OnFailure, but thread-safe.
*/
void OnFailure()
{
std::lock_guard<std::mutex> lock(this->mutex);
this->queue.emplace_back();
}
/**
* Similar to HTTPCallback::OnReceiveData, but thread-safe.
*/
void OnReceiveData(const char *data, size_t length)
{
std::lock_guard<std::mutex> lock(this->mutex);
this->queue.emplace_back(data, length);
}
/**
* Process everything on the queue.
*
* Should be called from the Game Thread.
*/
void HandleQueue()
{
this->cancelled = callback->IsCancelled();
std::lock_guard<std::mutex> lock(this->mutex);
for (auto &item : this->queue) {
if (item.failure) {
this->callback->OnFailure();
} else {
this->callback->OnReceiveData(item.data, item.length);
}
}
this->queue.clear();
this->queue_cv.notify_all();
}
/**
* Wait till the queue is dequeued.
*/
void WaitTillEmpty()
{
std::unique_lock<std::mutex> lock(this->mutex);
while (!queue.empty()) {
this->queue_cv.wait(lock);
}
}
/**
* Check if the queue is empty.
*/
bool IsQueueEmpty()
{
std::lock_guard<std::mutex> lock(this->mutex);
return this->queue.empty();
}
HTTPThreadSafeCallback(HTTPCallback *callback) : callback(callback) {}
~HTTPThreadSafeCallback()
{
std::lock_guard<std::mutex> lock(this->mutex);
/* Free all data that was not handled. */
for (auto &item : this->queue) {
if (!item.failure) {
free(item.data);
}
}
queue.clear();
queue_cv.notify_all();
}
std::atomic<bool> cancelled = false;
private:
HTTPCallback *callback; ///< The callback to send data back on.
std::mutex mutex; ///< Mutex to protect the queue.
std::vector<Callback> queue; ///< Queue of data to send back.
std::condition_variable queue_cv; ///< Condition variable to wait for the queue to be empty.
};
#endif /* NETWORK_CORE_HTTP_SHARED_H */

@ -15,6 +15,7 @@
#include "../network_internal.h"
#include "http.h"
#include "http_shared.h"
#include <mutex>
#include <winhttp.h>
@ -26,9 +27,9 @@ static HINTERNET _winhttp_session = nullptr;
/** Single HTTP request. */
class NetworkHTTPRequest {
private:
const std::wstring uri; ///< URI to connect to.
HTTPCallback *callback; ///< Callback to send data back on.
const std::string data; ///< Data to send, if any.
const std::wstring uri; ///< URI to connect to.
HTTPThreadSafeCallback callback; ///< Callback to send data back on.
const std::string data; ///< Data to send, if any.
HINTERNET connection = nullptr; ///< Current connection object.
HINTERNET request = nullptr; ///< Current request object.
@ -49,6 +50,11 @@ static std::vector<NetworkHTTPRequest *> _http_requests;
static std::vector<NetworkHTTPRequest *> _new_http_requests;
static std::mutex _new_http_requests_mutex;
static std::vector<HTTPThreadSafeCallback *> _http_callbacks;
static std::vector<HTTPThreadSafeCallback *> _new_http_callbacks;
static std::mutex _http_callback_mutex;
static std::mutex _new_http_callback_mutex;
/**
* Create a new HTTP request.
*
@ -61,6 +67,8 @@ NetworkHTTPRequest::NetworkHTTPRequest(const std::wstring &uri, HTTPCallback *ca
callback(callback),
data(data)
{
std::lock_guard<std::mutex> lock(_new_http_callback_mutex);
_new_http_callbacks.push_back(&this->callback);
}
static std::string GetLastErrorAsString()
@ -113,7 +121,7 @@ void NetworkHTTPRequest::WinHttpCallback(DWORD code, void *info, DWORD length)
if (this->depth++ > 5) {
Debug(net, 0, "HTTP request failed: too many redirects");
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
return;
}
break;
@ -136,7 +144,7 @@ void NetworkHTTPRequest::WinHttpCallback(DWORD code, void *info, DWORD length)
/* No need to be verbose about rate limiting. */
Debug(net, status_code == HTTP_429_TOO_MANY_REQUESTS ? 1 : 0, "HTTP request failed: status-code {}", status_code);
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
return;
}
@ -150,7 +158,7 @@ void NetworkHTTPRequest::WinHttpCallback(DWORD code, void *info, DWORD length)
DWORD size = *(DWORD *)info;
/* Next step: read the data in a temporary allocated buffer.
* The buffer will be free'd in the next step. */
* The buffer will be free'd by OnReceiveData() in the next step. */
char *buffer = size == 0 ? nullptr : MallocT<char>(size);
WinHttpReadData(this->request, buffer, size, 0);
} break;
@ -158,9 +166,7 @@ void NetworkHTTPRequest::WinHttpCallback(DWORD code, void *info, DWORD length)
case WINHTTP_CALLBACK_STATUS_READ_COMPLETE:
Debug(net, 4, "HTTP callback: {} bytes", length);
this->callback->OnReceiveData(static_cast<char *>(info), length);
/* Free the temporary buffer that was allocated in the previous step. */
free(info);
this->callback.OnReceiveData(static_cast<char *>(info), length);
if (length == 0) {
/* Next step: no more data available: request is finished. */
@ -177,13 +183,13 @@ void NetworkHTTPRequest::WinHttpCallback(DWORD code, void *info, DWORD length)
case WINHTTP_CALLBACK_STATUS_REQUEST_ERROR:
Debug(net, 0, "HTTP request failed: {}", GetLastErrorAsString());
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
break;
default:
Debug(net, 0, "HTTP request failed: unexepected callback code 0x{:x}", code);
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
return;
}
}
@ -227,7 +233,7 @@ void NetworkHTTPRequest::Connect()
if (this->connection == nullptr) {
Debug(net, 0, "HTTP request failed: {}", GetLastErrorAsString());
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
return;
}
@ -237,7 +243,7 @@ void NetworkHTTPRequest::Connect()
Debug(net, 0, "HTTP request failed: {}", GetLastErrorAsString());
this->finished = true;
this->callback->OnFailure();
this->callback.OnFailure();
return;
}
@ -258,14 +264,17 @@ void NetworkHTTPRequest::Connect()
*/
bool NetworkHTTPRequest::Receive()
{
if (this->callback->IsCancelled()) {
if (this->callback.cancelled && !this->finished) {
Debug(net, 1, "HTTP request failed: cancelled by user");
this->finished = true;
this->callback->OnFailure();
return true;
this->callback.OnFailure();
/* Fall-through, as we are waiting for IsQueueEmpty() to happen. */
}
return this->finished;
/* Only return true if the queue was also dequeued. */
if (!this->finished) return false;
if (!this->callback.IsQueueEmpty()) return false;
return true;
}
/**
@ -279,6 +288,9 @@ NetworkHTTPRequest::~NetworkHTTPRequest()
WinHttpCloseHandle(this->request);
WinHttpCloseHandle(this->connection);
}
std::lock_guard<std::mutex> lock(_http_callback_mutex);
_http_callbacks.erase(std::remove(_http_callbacks.begin(), _http_callbacks.end(), &this->callback), _http_callbacks.end());
}
/* static */ void NetworkHTTPSocketHandler::Connect(const std::string &uri, HTTPCallback *callback, const std::string data)
@ -292,6 +304,25 @@ NetworkHTTPRequest::~NetworkHTTPRequest()
/* static */ void NetworkHTTPSocketHandler::HTTPReceive()
{
/* Process all callbacks. */
{
std::lock_guard<std::mutex> lock(_http_callback_mutex);
{
std::lock_guard<std::mutex> lock(_new_http_callback_mutex);
if (!_new_http_callbacks.empty()) {
/* We delay adding new callbacks, as HandleQueue() below might add a new callback. */
_http_callbacks.insert(_http_callbacks.end(), _new_http_callbacks.begin(), _new_http_callbacks.end());
_new_http_callbacks.clear();
}
}
for (auto &callback : _http_callbacks) {
callback->HandleQueue();
}
}
/* Process all requests. */
{
std::lock_guard<std::mutex> lock(_new_http_requests_mutex);
if (!_new_http_requests.empty()) {

@ -613,12 +613,18 @@ void ClientNetworkContentSocketHandler::OnReceiveData(const char *data, size_t l
assert(data == nullptr || length != 0);
/* Ignore any latent data coming from a connection we closed. */
if (this->http_response_index == -2) return;
if (this->http_response_index == -2) {
if (data != nullptr) {
free(data);
}
return;
}
if (this->http_response_index == -1) {
if (data != nullptr) {
/* Append the rest of the response. */
this->http_response.insert(this->http_response.end(), data, data + length);
free(data);
return;
} else {
/* Make sure the response is properly terminated. */
@ -638,7 +644,9 @@ void ClientNetworkContentSocketHandler::OnReceiveData(const char *data, size_t l
/* Just received the data. */
this->OnDownloadProgress(this->curInfo, (int)length);
}
/* Nothing more to do now. */
free(data);
return;
}

@ -394,5 +394,7 @@ void NetworkSurveyHandler::OnReceiveData(const char *data, size_t)
if (data == nullptr) {
Debug(net, 1, "Survey: survey results sent");
this->loaded.notify_all();
} else {
free(data);
}
}

Loading…
Cancel
Save