|
| 1 | +#include "threading.h" |
| 2 | +using namespace std::chrono_literals; |
| 3 | + |
| 4 | +#include <iomanip> // put_time |
| 5 | +#include <windows.h> |
| 6 | +void logMessageWithTime(std::string msg) { |
| 7 | + auto now = std::chrono::system_clock::now(); |
| 8 | + auto in_time_t = std::chrono::system_clock::to_time_t(now); |
| 9 | + |
| 10 | + std::chrono::milliseconds ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()); |
| 11 | + std::size_t fractional_seconds = ms.count() % 1000; |
| 12 | + |
| 13 | + |
| 14 | + |
| 15 | + std::stringstream ss; |
| 16 | + ss << msg << " " << std::put_time(std::localtime(&in_time_t), "%H:%M:%S.") << fractional_seconds << "\n"; |
| 17 | + |
| 18 | + OutputDebugStringA(ss.str().c_str()); |
| 19 | +} |
| 20 | + |
| 21 | + |
| 22 | + |
| 23 | +void Worker::run() { |
| 24 | + while (true) { |
| 25 | + std::unique_lock l(taskLock); |
| 26 | + auto parentCon = parentConnection.lock(); |
| 27 | + |
| 28 | + if (!parentCon) {//If parent got deleted (went out of scope in SQF land) we want to exit too |
| 29 | + exiting = true; //#TODO going out of scope doesn't work |
| 30 | + while (!tasks.empty()) { |
| 31 | + std::shared_ptr<Task> task = tasks.front(); |
| 32 | + tasks.pop(); |
| 33 | + task->prom.set_value(false); |
| 34 | + } |
| 35 | + return; |
| 36 | + } |
| 37 | + |
| 38 | + if (tasks.empty()) { |
| 39 | + hasTasksCondition.wait(l); |
| 40 | + } |
| 41 | + |
| 42 | + if (!tasks.empty()) { |
| 43 | + std::shared_ptr<Task> task = tasks.front(); |
| 44 | + tasks.pop(); |
| 45 | + lastJob = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()); |
| 46 | + l.unlock(); |
| 47 | + logMessageWithTime("do task"); |
| 48 | + task->prom.set_value(task->job(workerConnection)); |
| 49 | + |
| 50 | + if (task->isInWorkList) { |
| 51 | + Threading::get().updateAsyncWorkLists(); |
| 52 | + } |
| 53 | + |
| 54 | + |
| 55 | + logMessageWithTime("task done"); |
| 56 | + } |
| 57 | + } |
| 58 | + |
| 59 | +} |
| 60 | + |
| 61 | +std::future<bool> Worker::pushTask(std::function<bool(mariadb::connection_ref)> task, bool intoWorkList) { |
| 62 | + logMessageWithTime("Worker::pushTask"); |
| 63 | + auto newTask = std::make_shared<Task>(); |
| 64 | + newTask->worker = shared_from_this(); |
| 65 | + newTask->job = std::move(task); |
| 66 | + newTask->isInWorkList = intoWorkList; |
| 67 | + auto fut = newTask->prom.get_future(); |
| 68 | + |
| 69 | + |
| 70 | + std::unique_lock l(taskLock); |
| 71 | + if (exiting) { |
| 72 | + newTask->prom.set_value(false); |
| 73 | + return fut; |
| 74 | + } |
| 75 | + tasks.emplace(newTask); |
| 76 | + l.unlock(); |
| 77 | + |
| 78 | + hasTasksCondition.notify_all(); |
| 79 | + logMessageWithTime("Worker::pushTask ret"); |
| 80 | + |
| 81 | + return fut; |
| 82 | +} |
| 83 | + |
| 84 | +std::future<bool> Threading::pushTask(mariadb::connection_ref con, std::function<bool(mariadb::connection_ref)> task, bool intoWorkList) { |
| 85 | + logMessageWithTime("Threading::pushTask"); |
| 86 | + auto found = workers.find(con->account()); |
| 87 | + if (found != workers.end()) { |
| 88 | + //Already have a worker on that account. |
| 89 | + return found->second->pushTask(std::move(task), intoWorkList); |
| 90 | + } |
| 91 | + logMessageWithTime("create worker"); |
| 92 | + auto newWorker = std::make_shared<Worker>(); |
| 93 | + newWorker->parentConnection = con; |
| 94 | + newWorker->workerConnection = mariadb::connection::create(con->account()); |
| 95 | + newWorker->myThread = std::make_shared<std::thread>([newWorker]() {newWorker->run(); }); |
| 96 | + workers[con->account()] = newWorker; |
| 97 | + |
| 98 | + return newWorker->pushTask(std::move(task), intoWorkList); |
| 99 | +} |
| 100 | + |
| 101 | + |
| 102 | +void Threading::doCleanup() { |
| 103 | + if (lastCleanup + 30s > std::chrono::system_clock::now()) return; |
| 104 | + logMessageWithTime("Threading::doCleanup"); |
| 105 | + |
| 106 | + |
| 107 | + //Will only be called from mainthread so noone can insert stuff now. |
| 108 | + for (auto& [acc,worker] : workers) { |
| 109 | + std::unique_lock l(worker->taskLock); |
| 110 | + if (!worker->tasks.empty()) continue; //Is still working on tasks |
| 111 | + logMessageWithTime("worker cleanup check"); |
| 112 | + auto lastTask = std::chrono::system_clock::from_time_t(worker->lastJob); |
| 113 | + if (lastTask + 60s < std::chrono::system_clock::now()) { //no tasks for 60s, kill worker |
| 114 | + logMessageWithTime("worker do cleanup"); |
| 115 | + worker->parentConnection.reset(); //no parent will mean it exits next iteration |
| 116 | + worker->hasTasksCondition.notify_all(); //force iteration |
| 117 | + if (worker->myThread->joinable()) worker->myThread->join(); //wait for thread to exit |
| 118 | + logMessageWithTime("worker cleaned up"); |
| 119 | + } |
| 120 | + workers.erase(acc); |
| 121 | + return; |
| 122 | + } |
| 123 | + lastCleanup = std::chrono::system_clock::now(); |
| 124 | +} |
| 125 | + |
| 126 | +void Threading::pushAsyncWork(ref<GameDataDBAsyncResult> work) { |
| 127 | + std::unique_lock l(asyncWorkMutex); |
| 128 | + asyncWork.emplace_back(work); |
| 129 | + |
| 130 | +} |
| 131 | + |
| 132 | +void Threading::updateAsyncWorkLists() { |
| 133 | + std::unique_lock l(asyncWorkMutex); |
| 134 | + |
| 135 | + |
| 136 | + |
| 137 | + auto p = std::stable_partition(asyncWork.begin(), asyncWork.end(), |
| 138 | + [&](const auto& x) { |
| 139 | + auto res = x->data->fut.wait_for(std::chrono::nanoseconds(0)); |
| 140 | + |
| 141 | + return res != std::future_status::ready; }); |
| 142 | + // range insert with move |
| 143 | + completedAsyncTasks.insert(completedAsyncTasks.end(), std::make_move_iterator(p), |
| 144 | + std::make_move_iterator(asyncWork.end())); |
| 145 | + // erase the moved-from elements. |
| 146 | + asyncWork.erase(p, asyncWork.end()); |
| 147 | + //#TODO we can just delete completed tasks that don't have callback or callbackargs |
| 148 | + //better do it here than in mainthread |
| 149 | + hasCompletedAsyncWork = !completedAsyncTasks.empty(); |
| 150 | + |
| 151 | +} |
0 commit comments