src/port-internal.cc

   1 /*
   2  * Copyright (c) 2017 - 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 <aul_rpc_port.h>
  18 #include <dlog.h>
  19 #include <poll.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <sys/socket.h>
  23 #include <unistd.h>
  24 #include <uuid/uuid.h>
  25
  26 #include <chrono>
  27 #include <utility>
  28
  29 #include "include/rpc-port.h"
  30 #include "log-private.hh"
  31 #include "message-sending-thread-internal.hh"
  32 #include "port-internal.hh"
  33
  34 namespace rpc_port {
  35 namespace internal {
  36 namespace {
  37
  38 constexpr const int QUEUE_SIZE_MAX = 1024 * 1024 * 10;
  39 constexpr const int MAX_RETRY_CNT = 10;
  40 constexpr const int MAX_TIMEOUT = 1000;
  41 constexpr const int MIN_TIMEOUT = 50;
  42
  43 }  // namespace
  44
  45 Port::DelayMessage::DelayMessage(const char* msg, int index, int size)
  46     : message_(msg, msg + size), index_(index), size_(size) {
  47 }
  48
  49 void Port::DelayMessage::SetIndex(int index) {
  50   index_ += index;
  51 }
  52
  53 int Port::DelayMessage::GetSize() {
  54   return size_ - index_;
  55 }
  56
  57 int Port::DelayMessage::GetOriginalSize() {
  58   return size_;
  59 }
  60
  61 char* Port::DelayMessage::GetMessage() {
  62   char* ptr = reinterpret_cast<char*>(message_.data());
  63   ptr += index_;
  64   return ptr;
  65 }
  66
  67 Port::Port(int fd, std::string id)
  68     : fd_(fd), id_(std::move(id)), instance_(""), seq_(0) {
  69   char uuid[37];
  70   uuid_t u;
  71   uuid_generate(u);
  72   uuid_unparse(u, uuid);
  73   instance_ = std::string(uuid) + ":" + id_;
  74 }
  75
  76 Port::Port(int fd, std::string id, std::string instance)
  77     : fd_(fd), id_(std::move(id)), instance_(std::move(instance)), seq_(0) {}
  78
  79 Port::~Port() {
  80   std::lock_guard<std::recursive_mutex> lock(mutex_);
  81   ClearQueue();
  82   Disconnect();
  83 }
  84
  85 void Port::Disconnect() {
  86   IgnoreIOEvent();
  87
  88   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
  89   if (fd_ > 0) {
  90     _W("Close fd(%d)", fd_);
  91     close(fd_);
  92     fd_ = -1;
  93   }
  94 }
  95
  96 int Port::SetPrivateSharing(const char* paths[], unsigned int size) {
  97   int ret = aul_rpc_port_set_private_sharing(id_.c_str(), paths, size);
  98   if (ret != 0)
  99     return RPC_PORT_ERROR_IO_ERROR;
 100   return RPC_PORT_ERROR_NONE;
 101 }
 102
 103 int Port::SetPrivateSharing(const char* path) {
 104   const char* file_list[1] = {path};
 105   int ret = aul_rpc_port_set_private_sharing(id_.c_str(), file_list, 1);
 106   if (ret != 0)
 107     return RPC_PORT_ERROR_IO_ERROR;
 108   return RPC_PORT_ERROR_NONE;
 109 }
 110
 111 int Port::UnsetPrivateSharing() {
 112   int ret = aul_rpc_port_unset_private_sharing(id_.c_str());
 113   if (ret != 0)
 114     return RPC_PORT_ERROR_IO_ERROR;
 115   return RPC_PORT_ERROR_NONE;
 116 }
 117
 118 int Port::Read(void* buf, unsigned int size) {
 119   unsigned int left = size;
 120   ssize_t nb;
 121   int bytes_read = 0;
 122   char* buffer = static_cast<char*>(buf);
 123   int max_timeout = MAX_TIMEOUT * MAX_RETRY_CNT;
 124   int timeout = MIN_TIMEOUT;
 125   int fd;
 126
 127   {
 128     std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 129     fd = fd_;
 130   }
 131
 132   if (fd < 0 || fd >= sysconf(_SC_OPEN_MAX)) {
 133     _E("Invalid fd(%d)", fd);
 134     return RPC_PORT_ERROR_IO_ERROR;
 135   }
 136
 137   while (left) {
 138     {
 139       std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 140       nb = read(fd_, buffer, left);
 141     }
 142
 143     if (nb == 0) {
 144       _E("read_socket: ...read EOF, socket closed %d: nb %zd\n", fd, nb);
 145       return RPC_PORT_ERROR_IO_ERROR;
 146     } else if (nb == -1) {
 147       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
 148         bool can_read = false;
 149         while (!can_read && max_timeout > 0) {
 150           auto start = std::chrono::steady_clock::now();
 151           can_read = CanRead(timeout);
 152           auto end = std::chrono::steady_clock::now();
 153           auto elapsed_time =
 154               std::chrono::duration_cast<std::chrono::milliseconds>(
 155                   end - start);
 156
 157           max_timeout -= elapsed_time.count();
 158
 159           timeout *= 2;
 160           if (timeout > MAX_TIMEOUT)
 161             timeout = MAX_TIMEOUT;
 162         }
 163
 164         if (!can_read) {
 165           _E("read_socket: ...timed out fd %d: errno %d", fd, errno);
 166           return RPC_PORT_ERROR_IO_ERROR;
 167         }
 168
 169         continue;
 170       }
 171
 172       _E("read_socket: ...error fd %d: errno %d\n", fd, errno);
 173       return RPC_PORT_ERROR_IO_ERROR;
 174     }
 175
 176     left -= nb;
 177     buffer += nb;
 178     bytes_read += nb;
 179     timeout = MIN_TIMEOUT;
 180   }
 181
 182   return RPC_PORT_ERROR_NONE;
 183 }
 184
 185 bool Port::CanRead(int timeout) {
 186   struct pollfd fds[1];
 187   fds[0].fd = fd_;
 188   fds[0].events = POLLIN;
 189   fds[0].revents = 0;
 190   int ret = poll(fds, 1, timeout);
 191   if (ret <= 0) {
 192     _W("poll() is failed. fd(%d), error(%s)",
 193         fd_, ret == 0 ? "timed out" : std::to_string(-errno).c_str());
 194     return false;
 195   }
 196
 197   return true;
 198 }
 199
 200 bool Port::CanWrite() {
 201   struct pollfd fds[1];
 202   fds[0].fd = fd_;
 203   fds[0].events = POLLOUT;
 204   fds[0].revents = 0;
 205   int ret = poll(fds, 1, 100);
 206   if (ret <= 0) {
 207     _W("poll() is failed. fd(%d), error(%s)",
 208         fd_, ret == 0 ? "timed out" : std::to_string(-errno).c_str());
 209     return false;
 210   }
 211
 212   return true;
 213 }
 214
 215 int Port::Write(const void* buf, unsigned int size) {
 216   int sent_bytes = 0;
 217   int ret;
 218   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 219
 220   if (queue_.empty()) {
 221     ret = Write(buf, size, &sent_bytes);
 222     if (ret == PORT_STATUS_ERROR_NONE)
 223       return RPC_PORT_ERROR_NONE;
 224     else if (ret == PORT_STATUS_ERROR_IO_ERROR)
 225       return RPC_PORT_ERROR_IO_ERROR;
 226   }
 227
 228   if (delayed_message_size_ > QUEUE_SIZE_MAX) {
 229     _E("cache fail : delayed_message_size (%d), count(%zu)",
 230                             delayed_message_size_, queue_.size());
 231     return RPC_PORT_ERROR_IO_ERROR;
 232   }
 233
 234   ret = PushDelayedMessage(
 235           std::make_shared<DelayMessage>(static_cast<const char*>(buf),
 236           sent_bytes, size));
 237
 238   if (CanWrite()) {
 239     while (!queue_.empty()) {
 240       int port_status = PopDelayedMessage();
 241       if (port_status != PORT_STATUS_ERROR_NONE) {
 242         if (port_status == PORT_STATUS_ERROR_IO_ERROR)
 243           return RPC_PORT_ERROR_IO_ERROR;
 244
 245         break;
 246       }
 247     }
 248   }
 249
 250   return ret;
 251 }
 252
 253 int Port::Write(const void* buf, unsigned int size, int* sent_bytes) {
 254   unsigned int left = size;
 255   ssize_t nb;
 256   int retry_cnt = 0;
 257   const char* buffer = static_cast<const char*>(buf);
 258
 259   if (fd_ < 0 || fd_ >= sysconf(_SC_OPEN_MAX)) {
 260     _E("Invalid fd(%d)", fd_);
 261     return PORT_STATUS_ERROR_IO_ERROR;
 262   }
 263
 264   while (left && (retry_cnt < MAX_RETRY_CNT)) {
 265     nb = send(fd_, buffer, left, MSG_NOSIGNAL);
 266     if (nb == -1) {
 267       if (errno == EINTR) {
 268         LOGI("write_socket: EINTR continue ...");
 269         retry_cnt++;
 270         continue;
 271       }
 272
 273     if (errno == EAGAIN || errno == EWOULDBLOCK)
 274       return PORT_STATUS_ERROR_RESOURCE_UNAVAILABLE;
 275
 276       _E("write_socket: ...error fd %d: errno %d\n", fd_, errno);
 277       return PORT_STATUS_ERROR_IO_ERROR;
 278     }
 279
 280     left -= nb;
 281     buffer += nb;
 282     *sent_bytes += nb;
 283   }
 284
 285   if (left != 0) {
 286     _E("error fd %d: retry_cnt %d", fd_, retry_cnt);
 287     return PORT_STATUS_ERROR_IO_ERROR;
 288   }
 289
 290   return PORT_STATUS_ERROR_NONE;
 291 }
 292
 293 gboolean Port::OnEventReceived(GIOChannel* io, GIOCondition condition,
 294                                gpointer data) {
 295   auto* ptr = static_cast<std::weak_ptr<Port>*>(data);
 296   auto port = ptr->lock();
 297   if (port == nullptr) {
 298     _E("port is destructed");
 299     return G_SOURCE_REMOVE;
 300   }
 301
 302   std::lock_guard<std::recursive_mutex> lock(port->rw_mutex_);
 303   if (port->source_id_ == 0) {
 304     _E("GSource is destroyed");
 305     return G_SOURCE_REMOVE;
 306   }
 307
 308   if (port->queue_.empty()) {
 309     port->IgnoreIOEvent();
 310     return G_SOURCE_CONTINUE;
 311   }
 312
 313   port->PopDelayedMessage();
 314   return G_SOURCE_CONTINUE;
 315 }
 316
 317 void Port::ClearQueue() {
 318   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 319
 320   while (queue_.empty() == false)
 321     queue_.pop();
 322
 323   IgnoreIOEvent();
 324   delayed_message_size_ = 0;
 325 }
 326
 327 void Port::IgnoreIOEvent() {
 328   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 329   if (source_id_ != 0) {
 330     GSource* source = g_main_context_find_source_by_id(
 331         MessageSendingThread::GetInst().GetContext(), source_id_);
 332     if (source != nullptr && !g_source_is_destroyed(source))
 333       g_source_destroy(source);
 334
 335     source_id_ = 0;
 336   }
 337
 338   if (channel_ != nullptr) {
 339     g_io_channel_unref(channel_);
 340     channel_ = nullptr;
 341   }
 342 }
 343
 344 int Port::ListenIOEvent() {
 345   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 346   channel_ = g_io_channel_unix_new(fd_);
 347   if (channel_ == nullptr) {
 348     _E("Failed to create GIOChannel");
 349     return RPC_PORT_ERROR_OUT_OF_MEMORY;
 350   }
 351
 352   GSource* source = g_io_create_watch(channel_,
 353       static_cast<GIOCondition>(G_IO_OUT));
 354   if (source == nullptr) {
 355     _E("Failed to create GSource");
 356     IgnoreIOEvent();
 357     return RPC_PORT_ERROR_OUT_OF_MEMORY;
 358   }
 359
 360   auto* ptr = new (std::nothrow) std::weak_ptr<Port>(shared_from_this());
 361   g_source_set_callback(source, reinterpret_cast<GSourceFunc>(OnEventReceived),
 362                         static_cast<gpointer>(ptr), [](gpointer ptr) {
 363                           auto* port = static_cast<std::weak_ptr<Port>*>(ptr);
 364                           delete port;
 365                         });
 366   g_source_set_priority(source, G_PRIORITY_DEFAULT);
 367   source_id_ = g_source_attach(source,
 368       MessageSendingThread::GetInst().GetContext());
 369   g_source_unref(source);
 370
 371   return RPC_PORT_ERROR_NONE;
 372 }
 373
 374 int Port::PopDelayedMessage() {
 375   int sent_bytes = 0;
 376   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 377   auto dm = queue_.front();
 378
 379   int ret = Write(dm->GetMessage(), dm->GetSize(), &sent_bytes);
 380   if (ret == PORT_STATUS_ERROR_RESOURCE_UNAVAILABLE) {
 381     dm->SetIndex(sent_bytes);
 382   } else if (ret == PORT_STATUS_ERROR_IO_ERROR) {
 383     ClearQueue();
 384   } else {
 385     delayed_message_size_ -= dm->GetOriginalSize();
 386     queue_.pop();
 387   }
 388
 389   _W("cache : count(%zu), delayed_message_size(%d), ret(%d), sent_bytes(%d)",
 390       queue_.size(), delayed_message_size_, ret, sent_bytes);
 391   return ret;
 392 }
 393
 394 int Port::PushDelayedMessage(std::shared_ptr<DelayMessage> dm) {
 395   std::lock_guard<std::recursive_mutex> lock(rw_mutex_);
 396   if (queue_.empty()) {
 397     int ret = ListenIOEvent();
 398     if (ret != RPC_PORT_ERROR_NONE)
 399       return ret;
 400   }
 401
 402   delayed_message_size_ += dm->GetOriginalSize();
 403   queue_.push(dm);
 404
 405   _W("cache : count(%zu), delayed_message_size(%d)",
 406       queue_.size(), delayed_message_size_);
 407   return RPC_PORT_ERROR_NONE;
 408 }
 409
 410 }  // namespace internal
 411 }  // namespace rpc_port