1.1 --- a/core/Database.cpp Tue Aug 20 23:59:11 2013 -0700
1.2 +++ b/core/Database.cpp Wed Aug 21 00:01:28 2013 -0700
1.3 @@ -1,5 +1,5 @@
1.4 /**
1.5 - * vim: set ts=4 :
1.6 + * vim: set ts=4 sw=4 tw=99 noet :
1.7 * =============================================================================
1.8 * SourceMod
1.9 * Copyright (C) 2004-2008 AlliedModders LLC. All rights reserved.
1.10 @@ -45,7 +45,10 @@
1.11 static bool s_OneTimeThreaderErrorMsg = false;
1.12
1.13 DBManager::DBManager()
1.14 -: m_ParseLevel(0), m_ParseState(0), m_pDefault(NULL)
1.15 + : m_Terminate(false),
1.16 + m_ParseLevel(0),
1.17 + m_ParseState(0),
1.18 + m_pDefault(NULL)
1.19 {
1.20 }
1.21
1.22 @@ -64,10 +67,6 @@
1.23
1.24 g_SourceMod.BuildPath(Path_SM, m_Filename, sizeof(m_Filename), "configs/databases.cfg");
1.25
1.26 - m_pConfigLock = g_pThreader->MakeMutex();
1.27 - m_pThinkLock = g_pThreader->MakeMutex();
1.28 - m_pQueueLock = g_pThreader->MakeMutex();
1.29 -
1.30 scripts->AddPluginsListener(this);
1.31 }
1.32
1.33 @@ -80,7 +79,7 @@
1.34 * This way the thread's search won't be searching through a
1.35 * potentially empty/corrupt list, which would be very bad.
1.36 */
1.37 - m_pConfigLock->Lock();
1.38 + ke::AutoLock lock(&m_ConfigLock);
1.39 if ((err = textparsers->ParseFile_SMC(m_Filename, this, &states)) != SMCError_Okay)
1.40 {
1.41 g_Logger.LogError("[SM] Detected parse error(s) in file \"%s\"", m_Filename);
1.42 @@ -90,16 +89,12 @@
1.43 g_Logger.LogError("[SM] Line %d: %s", states.line, txt);
1.44 }
1.45 }
1.46 - m_pConfigLock->Unlock();
1.47 }
1.48
1.49 void DBManager::OnSourceModShutdown()
1.50 {
1.51 KillWorkerThread();
1.52 scripts->RemovePluginsListener(this);
1.53 - m_pConfigLock->DestroyThis();
1.54 - m_pThinkLock->DestroyThis();
1.55 - m_pQueueLock->DestroyThis();
1.56 handlesys->RemoveType(m_DatabaseType, g_pCoreIdent);
1.57 handlesys->RemoveType(m_DriverType, g_pCoreIdent);
1.58 ClearConfigs();
1.59 @@ -502,12 +497,17 @@
1.60
1.61 void DBManager::KillWorkerThread()
1.62 {
1.63 - if (m_pWorker)
1.64 + if (m_Worker)
1.65 {
1.66 - m_pWorker->Stop(false);
1.67 - g_pThreader->DestroyWorker(m_pWorker);
1.68 - m_pWorker = NULL;
1.69 + {
1.70 + ke::AutoLock lock(&m_QueueEvent);
1.71 + m_Terminate = true;
1.72 + m_QueueEvent.Notify();
1.73 + }
1.74 + m_Worker->Join();
1.75 + m_Worker = NULL;
1.76 s_OneTimeThreaderErrorMsg = false;
1.77 + m_Terminate = false;
1.78 }
1.79 }
1.80
1.81 @@ -520,99 +520,95 @@
1.82 return false;
1.83 }
1.84
1.85 - if (!m_pWorker)
1.86 + if (!m_Worker)
1.87 {
1.88 - m_pWorker = g_pThreader->MakeWorker(this, true);
1.89 - if (!m_pWorker)
1.90 + m_Worker = new ke::Thread(this, "SM SQL Worker");
1.91 + if (!m_Worker->Succeeded())
1.92 {
1.93 if (!s_OneTimeThreaderErrorMsg)
1.94 {
1.95 g_Logger.LogError("[SM] Unable to create db threader (error unknown)");
1.96 s_OneTimeThreaderErrorMsg = true;
1.97 }
1.98 - return false;
1.99 - }
1.100 - if (!m_pWorker->Start())
1.101 - {
1.102 - if (!s_OneTimeThreaderErrorMsg)
1.103 - {
1.104 - g_Logger.LogError("[SM] Unable to start db threader (error unknown)");
1.105 - s_OneTimeThreaderErrorMsg = true;
1.106 - }
1.107 - g_pThreader->DestroyWorker(m_pWorker);
1.108 - m_pWorker = NULL;
1.109 + m_Worker = NULL;
1.110 return false;
1.111 }
1.112 }
1.113
1.114 /* Add to the queue */
1.115 {
1.116 - m_pQueueLock->Lock();
1.117 + ke::AutoLock lock(&m_QueueEvent);
1.118 Queue<IDBThreadOperation *> &queue = m_OpQueue.GetQueue(prio);
1.119 queue.push(op);
1.120 - m_pQueueLock->Unlock();
1.121 + m_QueueEvent.Notify();
1.122 }
1.123
1.124 - /* Make the thread */
1.125 - m_pWorker->MakeThread(this);
1.126 -
1.127 return true;
1.128 }
1.129
1.130 -void DBManager::OnWorkerStart(IThreadWorker *pWorker)
1.131 +void DBManager::Run()
1.132 {
1.133 - m_drSafety.clear();
1.134 - for (size_t i=0; i<m_drivers.size(); i++)
1.135 + // Initialize DB threadsafety.
1.136 + for (size_t i=0; i < m_drivers.size(); i++)
1.137 {
1.138 if (m_drivers[i]->IsThreadSafe())
1.139 - {
1.140 m_drSafety.push_back(m_drivers[i]->InitializeThreadSafety());
1.141 - } else {
1.142 + else
1.143 m_drSafety.push_back(false);
1.144 - }
1.145 }
1.146 -}
1.147
1.148 -void DBManager::OnWorkerStop(IThreadWorker *pWorker)
1.149 -{
1.150 + // Run actual worker thread logic.
1.151 + ThreadMain();
1.152 +
1.153 + // Shutdown DB threadsafety.
1.154 for (size_t i=0; i<m_drivers.size(); i++)
1.155 {
1.156 if (m_drSafety[i])
1.157 - {
1.158 m_drivers[i]->ShutdownThreadSafety();
1.159 - }
1.160 }
1.161 m_drSafety.clear();
1.162 }
1.163
1.164 -void DBManager::RunThread(IThreadHandle *pThread)
1.165 +void DBManager::ThreadMain()
1.166 {
1.167 - IDBThreadOperation *op = NULL;
1.168 + ke::AutoLock lock(&m_QueueEvent);
1.169 +
1.170 + while (true) {
1.171 + // The lock has been acquired. Grab everything we can out of the
1.172 + // queue. Since we want to flush the queue even if we're terminated,
1.173 + // we process all operations we can before checking to terminate.
1.174 + // There's no risk of starvation since the main thread blocks on us
1.175 + // terminating.
1.176 + while (true)
1.177 + {
1.178 + Queue<IDBThreadOperation *> &queue = m_OpQueue.GetLikelyQueue();
1.179 + if (queue.empty())
1.180 + break;
1.181
1.182 - /* Get something from the queue */
1.183 - {
1.184 - m_pQueueLock->Lock();
1.185 - Queue<IDBThreadOperation *> &queue = m_OpQueue.GetLikelyQueue();
1.186 - if (!queue.empty())
1.187 - {
1.188 - op = queue.first();
1.189 + IDBThreadOperation *op = queue.first();
1.190 queue.pop();
1.191 +
1.192 + // Unlock the queue when we run the query, so the main thread can
1.193 + // keep pumping events. We re-acquire the lock to check for more
1.194 + // items. It's okay if we terminate while unlocked; the main
1.195 + // thread would be blocked and we'd need to flush the queue
1.196 + // anyway, so after we've depleted the queue here, we'll just
1.197 + // reach the terminate at the top of the loop.
1.198 + {
1.199 + ke::AutoUnlock unlock(&m_QueueEvent);
1.200 + op->RunThreadPart();
1.201 +
1.202 + ke::AutoLock lock(&m_ThinkLock);
1.203 + m_ThinkQueue.push(op);
1.204 + }
1.205 }
1.206 - m_pQueueLock->Unlock();
1.207 - }
1.208 +
1.209 + if (m_Terminate)
1.210 + return;
1.211
1.212 - /* whoa. hi. did we get something? we should have. */
1.213 - if (!op)
1.214 - {
1.215 - /* wtf? */
1.216 - return;
1.217 + // Release the lock and wait for a signal.
1.218 + m_QueueEvent.Wait();
1.219 }
1.220 -
1.221 - op->RunThreadPart();
1.222 -
1.223 - m_pThinkLock->Lock();
1.224 - m_ThinkQueue.push(op);
1.225 - m_pThinkLock->Unlock();
1.226 }
1.227
1.228 void DBManager::RunFrame()
1.229 @@ -624,19 +620,16 @@
1.230 }
1.231
1.232 /* Dump one thing per-frame so the server stays sane. */
1.233 - m_pThinkLock->Lock();
1.234 - IDBThreadOperation *op = m_ThinkQueue.first();
1.235 - m_ThinkQueue.pop();
1.236 - m_pThinkLock->Unlock();
1.237 + IDBThreadOperation *op;
1.238 + {
1.239 + ke::AutoLock lock(&m_ThinkLock);
1.240 + op = m_ThinkQueue.first();
1.241 + m_ThinkQueue.pop();
1.242 + }
1.243 op->RunThinkPart();
1.244 op->Destroy();
1.245 }
1.246
1.247 -void DBManager::OnTerminate(IThreadHandle *pThread, bool cancel)
1.248 -{
1.249 - /* Do nothing */
1.250 -}
1.251 -
1.252 void DBManager::OnSourceModIdentityDropped(IdentityToken_t *pToken)
1.253 {
1.254 s_pAddBlock = pToken;
1.255 @@ -712,12 +705,12 @@
1.256
1.257 void DBManager::LockConfig()
1.258 {
1.259 - m_pConfigLock->Lock();
1.260 + m_ConfigLock.Lock();
1.261 }
1.262
1.263 void DBManager::UnlockConfig()
1.264 {
1.265 - m_pConfigLock->Unlock();
1.266 + m_ConfigLock.Unlock();
1.267 }
1.268
1.269 const char *DBManager::GetDefaultDriverName()
