src/server/worker_thread.cc

   1 /*
   2  * Copyright (c) 2021 Samsung Electronics Co., Ltd All Rights Reserved
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include "worker_thread.hh"
  18
  19 #include <sqlite3.h>
  20 #include <sys/types.h>
  21 #include <malloc.h>
  22
  23 #include <tzplatform_config.h>
  24
  25 #include "abstract_parcelable.hh"
  26 #include "cynara_checker.hh"
  27 #include "db_change_observer.hh"
  28 #include "request_handler_factory.hh"
  29 #include "server/database/db_handle_provider.hh"
  30 #include "server/database/update_pending_cache_handler.hh"
  31 #include "utils/logging.hh"
  32
  33 #include "pkgmgrinfo_debug.h"
  34
  35 #ifdef LOG_TAG
  36 #undef LOG_TAG
  37 #endif
  38 #define LOG_TAG "PKGMGR_INFO"
  39
  40 #ifndef SQLITE_ENABLE_MEMORY_MANAGEMENT
  41 #define SQLITE_ENABLE_MEMORY_MANAGEMENT
  42 #endif
  43
  44 namespace {
  45
  46 uid_t globaluser_uid = -1;
  47
  48 uid_t GetGlobalUID() {
  49   if (globaluser_uid == (uid_t)-1)
  50     globaluser_uid = tzplatform_getuid(TZ_SYS_GLOBALAPP_USER);
  51
  52   return globaluser_uid;
  53 }
  54
  55 uid_t ConvertUID(uid_t uid) {
  56   if (uid < REGULAR_USER)
  57     return GetGlobalUID();
  58   else
  59     return uid;
  60 }
  61
  62 const char PRIVILEGE_PACKAGE_MANAGER_ADMIN[] =
  63     "http://tizen.org/privilege/packagemanager.admin";
  64
  65 std::vector<std::string> GetPrivileges(pkgmgr_common::ReqType type) {
  66   std::vector<std::string> ret;
  67   if (type == pkgmgr_common::SET_CERT_INFO)
  68     ret.emplace_back(PRIVILEGE_PACKAGE_MANAGER_ADMIN);
  69   else if (type == pkgmgr_common::SET_PKG_INFO)
  70     ret.emplace_back(PRIVILEGE_PACKAGE_MANAGER_ADMIN);
  71
  72   return ret;
  73 }
  74
  75 }  // namespace
  76
  77 namespace pkgmgr_server {
  78
  79 WorkerThread::WorkerThread(unsigned int num) : stop_all_(false) {
  80   threads_.reserve(num);
  81   for (unsigned int i = 0; i < num; ++i)
  82     threads_.emplace_back([this]() -> void { this->Run(); });
  83
  84   LOG(DEBUG) << num << " Worker threads are created";
  85 }
  86
  87 WorkerThread::~WorkerThread() {
  88   stop_all_ = true;
  89   cv_.notify_all();
  90
  91   for (auto& t : threads_)
  92     t.join();
  93 }
  94
  95 bool WorkerThread::PushQueue(std::shared_ptr<PkgRequest> req) {
  96   {
  97     std::unique_lock<std::mutex> u(lock_);
  98     queue_.push(req);
  99     cv_.notify_one();
 100   }
 101   return true;
 102 }
 103
 104 std::shared_ptr<PkgRequest> WorkerThread::PopQueue() {
 105   SetMemoryTrimTimer();
 106   std::unique_lock<std::mutex> u(lock_);
 107   cv_.wait(u, [this] { return !this->queue_.empty() || stop_all_; });
 108   if (stop_all_ && queue_.empty())
 109     return nullptr;
 110
 111   auto req = queue_.front();
 112   queue_.pop();
 113   return req;
 114 }
 115
 116 void WorkerThread::Run() {
 117   RequestHandlerFactory factory;
 118   LOG(DEBUG) << "Initialize request handlers";
 119   while (true) {
 120     std::shared_ptr<PkgRequest> req = PopQueue();
 121     if (req == nullptr)
 122       return;
 123
 124     if (!req->GetPrivilegeChecked()) {
 125       if (!req->ReceiveData()) {
 126         LOG(ERROR) << "Fail to receive data";
 127         continue;
 128       }
 129
 130       pkgmgr_common::ReqType type = req->GetRequestType();
 131       if (pkgmgr_common::IsDbWriteRequest(type))
 132         StopDbChangeListening();
 133       std::vector<std::string> privileges = GetPrivileges(type);
 134       if (!CynaraChecker::GetInst().CheckPrivilege(this, req, privileges))
 135         continue;
 136     }
 137
 138     auto type = req->GetRequestType();
 139     LOG(WARNING) << "Request type: " << pkgmgr_common::ReqTypeToString(type)
 140         << " pid: " << req->GetSenderPID() << " tid: " << req->GetSenderTID();
 141     auto handler = factory.GetRequestHandler(type);
 142     if (handler == nullptr)
 143       continue;
 144
 145     try {
 146       handler->PreExec();
 147       handler->SetUID(ConvertUID(req->GetSenderUID()));
 148       handler->SetPID(req->GetSenderPID());
 149       if (!handler->HandleRequest(req->DetachData(), req->GetSize(),
 150           locale_.GetObject()))
 151         LOG(ERROR) << "Failed to handle request";
 152
 153       if (req->SendData(handler->ExtractResult()) == false)
 154         LOG(ERROR) << "Failed to send response pid: " << req->GetSenderPID();
 155       else
 156         LOG(WARNING) << "Success response pid: " << req->GetSenderPID()
 157             << " tid: " << req->GetSenderTID();
 158     } catch (const std::exception& err) {
 159       LOG(ERROR) << "Exception occurred: " << err.what()
 160                  << ", pid: " << req->GetSenderPID();
 161       SendError(req);
 162     } catch (...) {
 163       LOG(ERROR) << "Exception occurred pid: " << req->GetSenderPID();
 164       SendError(req);
 165     }
 166
 167     handler->PostExec();
 168   }
 169 }
 170
 171 void WorkerThread::SetMemoryTrimTimer() {
 172   std::lock_guard<std::recursive_mutex> lock(mutex_);
 173   if (timer_ > 0)
 174     g_source_remove(timer_);
 175
 176   timer_ = g_timeout_add_seconds_full(G_PRIORITY_LOW, 10,
 177       TrimMemory, this, NULL);
 178 }
 179
 180 gboolean WorkerThread::TrimMemory(void* data) {
 181   LOG(DEBUG) << "Trim memory";
 182   auto* h = static_cast<WorkerThread*>(data);
 183   {
 184     std::lock_guard<std::recursive_mutex> lock(h->mutex_);
 185     h->timer_ = 0;
 186   }
 187
 188   auto crashed_writer_pids =
 189       database::DBHandleProvider::CrashedWriteRequestPIDs();
 190   if (!crashed_writer_pids.empty()) {
 191     database::UpdatePendingCacheHandler db(getuid(), std::move(crashed_writer_pids), {});
 192     db.SetLocale(h->locale_.GetObject());
 193     db.Execute();
 194   }
 195
 196   sqlite3_release_memory(-1);
 197   malloc_trim(0);
 198   return G_SOURCE_REMOVE;
 199 }
 200
 201 void WorkerThread::SetLocale(std::string locale) {
 202   LOG(DEBUG) << "Change locale : " << locale_.GetObject()
 203       << " -> "  << locale;
 204   locale_.SetObject(std::move(locale));
 205 }
 206
 207 void WorkerThread::SendError(const std::shared_ptr<PkgRequest>& req) {
 208   pkgmgr_common::parcel::AbstractParcelable parcelable(
 209       0, pkgmgr_common::parcel::ParcelableType::Unknown, PMINFO_R_ERROR);
 210   tizen_base::Parcel p;
 211   p.WriteParcelable(parcelable);
 212   req->SendData(p);
 213 }
 214
 215 void WorkerThread::StopDbChangeListening() {
 216   auto& db_observer = pkgmgr_common::DbChangeObserver::GetInst();
 217
 218   if (db_observer.GetDisposed())
 219     return;
 220
 221   LOG(WARNING) << "Try stop listening db change";
 222   db_observer.StopListening();
 223 }
 224
 225 }  // namespace pkgmgr_server