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 #include <sqlite3.h>
  19 #include <malloc.h>
  20
  21 #include "pkgmgrinfo_debug.h"
  22 #include "get_appinfo_request_handler.hh"
  23 #include "get_cert_request_handler.hh"
  24 #include "get_pkginfo_request_handler.hh"
  25 #include "get_depinfo_request_handler.hh"
  26 #include "query_request_handler.hh"
  27 #include "set_cert_request_handler.hh"
  28 #include "set_pkginfo_request_handler.hh"
  29 #include "abstract_parcelable.hh"
  30 #include "command_request_handler.hh"
  31
  32 #ifdef LOG_TAG
  33 #undef LOG_TAG
  34 #endif
  35 #define LOG_TAG "PKGMGR_INFO"
  36
  37 namespace pkgmgr_server {
  38
  39 WorkerThread::WorkerThread(int num) : stop_all_(false) {
  40   threads_.reserve(num);
  41   for (int i = 0; i < num; ++i)
  42     threads_.emplace_back([this]() -> void { this->Run(); });
  43   LOGD("%d Worker threads are created", num);
  44 }
  45
  46 WorkerThread::~WorkerThread() {
  47   stop_all_ = true;
  48   cv_.notify_all();
  49
  50   for (auto& t : threads_) t.join();
  51 }
  52
  53 bool WorkerThread::PushQueue(std::shared_ptr<PkgRequest> req) {
  54   {
  55     std::unique_lock<std::mutex> u(lock_);
  56     queue_.push(req);
  57     cv_.notify_one();
  58   }
  59   return true;
  60 }
  61
  62 void WorkerThread::Run() {
  63   std::unique_ptr<request_handler::AbstractRequestHandler> handler[pkgmgr_common::ReqType::MAX];
  64   handler[pkgmgr_common::ReqType::GET_PKG_INFO].reset(
  65       new request_handler::GetPkginfoRequestHandler());
  66   handler[pkgmgr_common::ReqType::GET_APP_INFO].reset(
  67       new request_handler::GetAppinfoRequestHandler());
  68   handler[pkgmgr_common::ReqType::SET_PKG_INFO].reset(
  69       new request_handler::SetPkginfoRequestHandler());
  70   handler[pkgmgr_common::ReqType::SET_CERT_INFO].reset(
  71       new request_handler::SetCertRequestHandler());
  72   handler[pkgmgr_common::ReqType::GET_CERT_INFO].reset(
  73       new request_handler::GetCertRequestHandler());
  74   handler[pkgmgr_common::ReqType::GET_PKG_DEP_INFO].reset(
  75       new request_handler::GetDepinfoRequestHandler());
  76   handler[pkgmgr_common::ReqType::QUERY].reset(
  77       new request_handler::QueryRequestHandler());
  78   handler[pkgmgr_common::ReqType::COMMAND].reset(
  79       new request_handler::CommandRequestHandler());
  80
  81   LOGD("Initialize request handlers");
  82   while (true) {
  83     std::shared_ptr<PkgRequest> req;
  84     {
  85       std::unique_lock<std::mutex> u(lock_);
  86       cv_.wait(u, [this] { return !this->queue_.empty() || stop_all_; });
  87       if (stop_all_ && queue_.empty()) return;
  88       req = PopQueue();
  89     }
  90
  91     if (req->ReceiveData() == false) {
  92       LOGE("Failed to ReceiveData");
  93       continue;
  94     }
  95     pkgmgr_common::ReqType type = req->GetRequestType();
  96     LOGD("Request type(%d), pid(%d)", static_cast<int>(type), req->GetSenderPID());
  97     if (type <= pkgmgr_common::ReqType::REQ_TYPE_NONE
  98             || type >= pkgmgr_common::ReqType::MAX) {
  99       LOGE("Request type is invalid (%d)", static_cast<int>(type));
 100
 101       pkgmgr_common::parcel::AbstractParcelable parcelable(
 102           0, pkgmgr_common::parcel::ParcelableType::Unknown, PMINFO_R_ERROR);
 103       tizen_base::Parcel p;
 104       p.WriteParcelable(parcelable);
 105       std::vector<uint8_t> raw = p.GetRaw();
 106       req->SendData(&raw[0], raw.size());
 107       continue;
 108     }
 109
 110     try {
 111       handler[type]->SetPID(req->GetSenderPID());
 112       if (!handler[type]->HandleRequest(req->GetData(), req->GetSize(),
 113                                         locale_.GetObject()))
 114         LOGE("Failed to handle request");
 115     } catch (const std::exception& err) {
 116       LOGE("Exception occurred (%s)", err.what());
 117       pkgmgr_common::parcel::AbstractParcelable parcelable(
 118           0, pkgmgr_common::parcel::ParcelableType::Unknown, PMINFO_R_ERROR);
 119       tizen_base::Parcel p;
 120       p.WriteParcelable(parcelable);
 121       std::vector<uint8_t> raw = p.GetRaw();
 122       req->SendData(&raw[0], raw.size());
 123       continue;
 124     } catch (...) {
 125       LOGE("Exception occurred");
 126       pkgmgr_common::parcel::AbstractParcelable parcelable(
 127           0, pkgmgr_common::parcel::ParcelableType::Unknown, PMINFO_R_ERROR);
 128       tizen_base::Parcel p;
 129       p.WriteParcelable(parcelable);
 130       std::vector<uint8_t> raw = p.GetRaw();
 131       req->SendData(&raw[0], raw.size());
 132       continue;
 133     }
 134
 135     std::vector<uint8_t> result_data = handler[type]->ExtractResult();
 136     req->SendData(result_data.data(), result_data.size());
 137   }
 138 }
 139
 140 std::shared_ptr<PkgRequest> WorkerThread::PopQueue() {
 141   static int cnt = 0;
 142   if (++cnt % 50 == 0) { /* trim memory */
 143     sqlite3_release_memory(-1);
 144     malloc_trim(0);
 145   }
 146
 147   auto req = queue_.front();
 148   queue_.pop();
 149   return req;
 150 }
 151
 152 void WorkerThread::SetLocale(std::string locale) {
 153   LOGD("Change locale (%s) -> (%s)", locale_.GetObject().c_str(),
 154        locale.c_str());
 155   locale_.SetObject(std::move(locale));
 156 }
 157
 158 }  // namespace pkgmgr_server