Differential D1921 Diff 9354 source/ps/UserReport.cpp

Changeset View

Standalone View

source/ps/UserReport.cpp

Show All 16 Lines

#include "precompiled.h"		#include "precompiled.h"

#include "UserReport.h"		#include "UserReport.h"

#include "lib/timer.h"		#include "lib/timer.h"
#include "lib/utf8.h"		#include "lib/utf8.h"
#include "lib/external_libraries/curl.h"		#include "lib/external_libraries/curl.h"
#include "lib/external_libraries/libsdl.h"
#include "lib/external_libraries/zlib.h"		#include "lib/external_libraries/zlib.h"
#include "lib/file/archive/stream.h"		#include "lib/file/archive/stream.h"
#include "lib/os_path.h"		#include "lib/os_path.h"
#include "lib/sysdep/sysdep.h"		#include "lib/sysdep/sysdep.h"
#include "ps/ConfigDB.h"		#include "ps/ConfigDB.h"
#include "ps/Filesystem.h"		#include "ps/Filesystem.h"
#include "ps/Profiler2.h"		#include "ps/Profiler2.h"

		#include <condition_variable>
#include <fstream>		#include <fstream>
#include <mutex>		#include <mutex>
#include <string>		#include <string>
#include <thread>		#include <thread>

#define DEBUG_UPLOADS 0		#define DEBUG_UPLOADS 0

/*		/*
▲ Show 20 Lines • Show All 89 Lines • ▼ Show 20 Lines	#endif
ua += " (http://play0ad.com/)";		ua += " (http://play0ad.com/)";
m_Headers = curl_slist_append(m_Headers, ua.c_str());		m_Headers = curl_slist_append(m_Headers, ua.c_str());
// Override the default application/x-www-form-urlencoded type since we're not using that type		// Override the default application/x-www-form-urlencoded type since we're not using that type
m_Headers = curl_slist_append(m_Headers, "Content-Type: application/octet-stream");		m_Headers = curl_slist_append(m_Headers, "Content-Type: application/octet-stream");
// Disable the Accept header because it's a waste of a dozen bytes		// Disable the Accept header because it's a waste of a dozen bytes
m_Headers = curl_slist_append(m_Headers, "Accept: ");		m_Headers = curl_slist_append(m_Headers, "Accept: ");
curl_easy_setopt(m_Curl, CURLOPT_HTTPHEADER, m_Headers);		curl_easy_setopt(m_Curl, CURLOPT_HTTPHEADER, m_Headers);


// Set up the worker thread:

// Use SDL semaphores since OS X doesn't implement sem_init
m_WorkerSem = SDL_CreateSemaphore(0);
ENSURE(m_WorkerSem);

m_WorkerThread = std::thread(RunThread, this);		m_WorkerThread = std::thread(RunThread, this);
}		}

~CUserReporterWorker()		~CUserReporterWorker()
{		{
// Clean up resources

SDL_DestroySemaphore(m_WorkerSem);

curl_slist_free_all(m_Headers);		curl_slist_free_all(m_Headers);
curl_easy_cleanup(m_Curl);		curl_easy_cleanup(m_Curl);
}		}

/**		/**
* Called by main thread, when the online reporting is enabled/disabled.		* Called by main thread, when the online reporting is enabled/disabled.
*/		*/
void SetEnabled(bool enabled)		void SetEnabled(bool enabled)
{		{
std::lock_guard<std::mutex> lock(m_WorkerMutex);		std::lock_guard<std::mutex> lock(m_WorkerMutex);
if (enabled != m_Enabled)		if (enabled != m_Enabled)
{		{
m_Enabled = enabled;		m_Enabled = enabled;

// Wake up the worker thread		// Wake up the worker thread
SDL_SemPost(m_WorkerSem);		m_WorkerCV.notify_all();
}		}
}		}

/**		/**
* Called by main thread to request shutdown.		* Called by main thread to request shutdown.
* Returns true if we've shut down successfully.		* Returns true if we've shut down successfully.
* Returns false if shutdown is taking too long (we might be blocked on a		* Returns false if shutdown is taking too long (we might be blocked on a
* sync network operation) - you mustn't destroy this object, just leak it		* sync network operation) - you mustn't destroy this object, just leak it
* and terminate.		* and terminate.
*/		*/
bool Shutdown()		bool Shutdown()
{		{
{		{
std::lock_guard<std::mutex> lock(m_WorkerMutex);		std::lock_guard<std::mutex> lock(m_WorkerMutex);
m_Shutdown = true;		m_Shutdown = true;
}		}

// Wake up the worker thread		// Wake up the worker thread
SDL_SemPost(m_WorkerSem);		m_WorkerCV.notify_all();

// Wait for it to shut down cleanly		// Wait for it to shut down cleanly
// TODO: should have a timeout in case of network hangs		// TODO: should have a timeout in case of network hangs
m_WorkerThread.join();		m_WorkerThread.join();

return true;		return true;
}		}

Show All 12 Lines	#endif
void Submit(const shared_ptr<CUserReport>& report)		void Submit(const shared_ptr<CUserReport>& report)
{		{
{		{
std::lock_guard<std::mutex> lock(m_WorkerMutex);		std::lock_guard<std::mutex> lock(m_WorkerMutex);
m_ReportQueue.push_back(report);		m_ReportQueue.push_back(report);
}		}

// Wake up the worker thread		// Wake up the worker thread
SDL_SemPost(m_WorkerSem);		m_WorkerCV.notify_all();
}		}

/**		/**
* Called by the main thread every frame, so we can check		* Called by the main thread every frame, so we can check
* retransmission timers.		* retransmission timers.
*/		*/
void Update()		void Update()
{		{
double now = timer_Time();		double now = timer_Time();
if (now > m_LastUpdateTime + TIMER_CHECK_INTERVAL)		if (now > m_LastUpdateTime + TIMER_CHECK_INTERVAL)
{		{
// Wake up the worker thread		// Wake up the worker thread
SDL_SemPost(m_WorkerSem);		m_WorkerCV.notify_all();

m_LastUpdateTime = now;		m_LastUpdateTime = now;
}		}
}		}

private:		private:
static void RunThread(CUserReporterWorker* data)		static void RunThread(CUserReporterWorker* data)
{		{
Show All 14 Lines	std::wstring proxy;
PROFILE2("get proxy config");		PROFILE2("get proxy config");
if (sys_get_proxy_config(wstring_from_utf8(m_URL), proxy) == INFO::OK)		if (sys_get_proxy_config(wstring_from_utf8(m_URL), proxy) == INFO::OK)
curl_easy_setopt(m_Curl, CURLOPT_PROXY, utf8_from_wstring(proxy).c_str());		curl_easy_setopt(m_Curl, CURLOPT_PROXY, utf8_from_wstring(proxy).c_str());
}		}

SetStatus("waiting");		SetStatus("waiting");

/*		/*
* We use a semaphore to let the thread be woken up when it has		* We use a condition_variable to let the thread be woken up when it has
* work to do. Various actions from the main thread can wake it:		* work to do. Various actions from the main thread can wake it:
* * SetEnabled()		* * SetEnabled()
* * Shutdown()		* * Shutdown()
* * Submit()		* * Submit()
* * Retransmission timeouts, once every several seconds		* * Retransmission timeouts, once every several seconds
*		*
* If multiple actions have triggered wakeups, we might respond to		* If multiple actions have triggered wakeups, we might respond to
* all of those actions after the first wakeup, which is okay (we'll do		* all of those actions after the first wakeup, which is okay (we'll do
* nothing during the subsequent wakeups). We should never hang due to		* nothing during the subsequent wakeups). We should never hang due to
* processing fewer actions than wakeups.		* processing fewer actions than wakeups.
*		*
* Retransmission timeouts are triggered via the main thread - we can't simply		* Retransmission timeouts are triggered via the main thread.
* use SDL_SemWaitTimeout because on Linux it's implemented as an inefficient
* busy-wait loop, and we can't use a manual busy-wait with a long delay time
* because we'd lose responsiveness. So the main thread pings the worker
* occasionally so it can check its timer.
*/		*/

// Wait until the main thread wakes us up		// Wait until the main thread wakes us up
while (true)		while (true)
{		{
g_Profiler2.RecordRegionEnter("semaphore wait");		g_Profiler2.RecordRegionEnter("condition_variable wait");

ENSURE(SDL_SemWait(m_WorkerSem) == 0);		std::unique_lock<std::mutex> lock(m_WorkerMutex);
		StanUnsubmitted Not Done Inline Actions Why not use a scope here like you did everywhere else ? Stan: Why not use a scope here like you did everywhere else ?
		wraitiiAuthorUnsubmitted Done Inline Actions Can't because of how condition variables work. wraitii: Can't because of how condition variables work.
		m_WorkerCV.wait(lock, [this] { return m_Enabled \|\| m_Shutdown; });
		lock.unlock();

g_Profiler2.RecordRegionLeave();		g_Profiler2.RecordRegionLeave();

// Handle shutdown requests as soon as possible		// Handle shutdown requests as soon as possible
if (GetShutdown())		if (GetShutdown())
return;		return;

// If we're not enabled, ignore this wakeup		// If we're not enabled, ignore this wakeup
▲ Show 20 Lines • Show All 188 Lines • ▼ Show 20 Lines	static size_t SendCallback(char* bufptr, size_t size, size_t nmemb, void* userp)

return amount;		return amount;
}		}

private:		private:
// Thread-related members:		// Thread-related members:
std::thread m_WorkerThread;		std::thread m_WorkerThread;
std::mutex m_WorkerMutex;		std::mutex m_WorkerMutex;
SDL_sem* m_WorkerSem;		std::condition_variable m_WorkerCV;
		StanUnsubmitted Not Done Inline Actions Don't care much but maybe it should be Cv for CamelCase Stan: Don't care much but maybe it should be Cv for CamelCase

// Shared by main thread and worker thread:		// Shared by main thread and worker thread:
// These variables are all protected by m_WorkerMutex		// These variables are all protected by m_WorkerMutex
std::deque<shared_ptr<CUserReport> > m_ReportQueue;		std::deque<shared_ptr<CUserReport> > m_ReportQueue;
bool m_Enabled;		bool m_Enabled;
bool m_Shutdown;		bool m_Shutdown;
std::string m_Status;		std::string m_Status;

▲ Show 20 Lines • Show All 150 Lines • Show Last 20 Lines