src/io/FIo_IpcClient.cpp

   1 //
   2 // Copyright (c) 2012 Samsung Electronics Co., Ltd.
   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 /**
  18  * @file        FIo_IpcClient.cpp
  19  * @brief       This is the implementation file for the _IpcClient class.
  20  *
  21  */
  22
  23 #include <stdio.h>
  24 #include <errno.h>
  25 #include <sys/types.h>
  26 #include <sys/socket.h>
  27 #include <sys/un.h>
  28 #include <stdio.h>
  29 #include <stdlib.h>
  30 #include <unistd.h>
  31 #include <poll.h>
  32 #include <pthread.h>
  33 #include <fcntl.h>
  34
  35 #include <iostream>
  36 #include <queue>
  37 #include <map>
  38
  39 #include <FBaseRtMutex.h>
  40 #include <FBaseSysLog.h>
  41 #include <FBase_StringConverter.h>
  42 #include "FIo_IpcClient.h"
  43 #include "FIo_IIpcClientEventListener.h"
  44
  45 using namespace IPC;
  46 using namespace std;
  47 using namespace Tizen::App;
  48 using namespace Tizen::Base;
  49 using namespace Tizen::Base::Runtime;
  50
  51 namespace Tizen { namespace Io
  52 {
  53
  54 _IpcClient::_IpcClient(void)
  55         : __pReverseSource(null)
  56         , __pFdLock(null)
  57         , __pListener(null)
  58 {
  59         __messageBuffer[0] = '\0';
  60 }
  61
  62 _IpcClient::~_IpcClient(void)
  63 {
  64         int fd = 0;
  65
  66         if (__pReverseSource != null)
  67         {
  68                 g_source_destroy(__pReverseSource);
  69                 g_source_unref(__pReverseSource);
  70                 __pReverseSource = null;
  71         }
  72
  73         while (__fds.size() > 0)
  74         {
  75                 fd = __fds.back();
  76                 __fds.pop_back();
  77
  78                 close(fd);
  79         }
  80
  81         delete __pFdLock;
  82 }
  83
  84 result
  85 _IpcClient::Construct(const String& name, const _IIpcClientEventListener* pListener)
  86 {
  87         SysAssertf(__pFdLock == null, "Already constructed. Calling Construct() twice or more on a same instance is not allowed for this class.");
  88         result r = E_SUCCESS;
  89         Mutex* pMutex = null;
  90
  91         __name = name;
  92         __pListener = const_cast <_IIpcClientEventListener*>(pListener);
  93
  94         pMutex = new (std::nothrow) Mutex();
  95         SysTryReturnResult(NID_IO, pMutex != null, E_OUT_OF_MEMORY, "The memory is insufficient.");
  96
  97         r = pMutex->Create();
  98         SysTryCatch(NID_IO, !IsFailed(r), , r, "[%s] Failed to create SourceLock.", GetErrorMessage(r));
  99
 100         __pFdLock = pMutex;
 101
 102         r = MakeConnection();
 103         SysTryCatch(NID_IO, !IsFailed(r), , r, "[%s] Failed to connect to server.", GetErrorMessage(r));
 104
 105         if (__pListener)
 106         {
 107                 r = MakeConnection(true);
 108                 SysTryCatch(NID_IO, !IsFailed(r), , r, "[%s] Failed to connect to server.", GetErrorMessage(r));
 109         }
 110
 111         return E_SUCCESS;
 112
 113 CATCH:
 114         __name.Clear();
 115         __pListener = null;
 116         __pFdLock = null;
 117
 118         delete pMutex;
 119
 120         return r;
 121 }
 122
 123 String
 124 _IpcClient::GetName(void) const
 125 {
 126         return __name;
 127 }
 128
 129 struct HelloMessage
 130 {
 131         int reverse;
 132 };
 133
 134 result
 135 _IpcClient::MakeConnection(bool forReverse)
 136 {
 137         result r = E_SUCCESS;
 138
 139         struct sockaddr_un server;
 140         socklen_t serverLen = 0;
 141         int client = -1;
 142         int ret = 0;
 143         int retry = 0;
 144         HelloMessage helloMessage = {0};
 145         std::string socketName;
 146         char* pSocketName = null;
 147         size_t socketNameLength = 0;
 148         int flags = 0;
 149
 150         if (forReverse)
 151         {
 152                 helloMessage.reverse = 1;
 153         }
 154         else
 155         {
 156                 helloMessage.reverse = 0;
 157         }
 158
 159         pSocketName = _StringConverter::CopyToCharArrayN(__name);
 160         SysTryReturnResult(NID_IO, pSocketName != null, E_OUT_OF_MEMORY, "The memory is insufficient.");
 161
 162         socketName.append("/var/run/osp/");
 163         socketName.append(pSocketName);
 164
 165         delete[] pSocketName;
 166
 167         socketNameLength = socketName.size() + 1;
 168         SysTryReturnResult(NID_IO, socketNameLength < 108, E_INVALID_ARG, "Server name is too long.");
 169
 170         client = socket(AF_UNIX, SOCK_STREAM, 0);
 171         SysTryCatch(NID_IO, client != -1, r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to create a socket : %s.", strerror(errno));
 172
 173         flags = fcntl(client, F_GETFL, 0);
 174         ret = fcntl(client, F_SETFL, flags | O_NONBLOCK);
 175         SysTryCatch(NID_IO, ret >= 0 , r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to set file status flags (%d, %s).",
 176                                            errno, strerror(errno));
 177
 178         bzero(&server, sizeof(server));
 179         server.sun_family = AF_UNIX;
 180         strncpy(server.sun_path, socketName.c_str(), socketNameLength);
 181         serverLen = sizeof(server);
 182
 183         // Retry if the server is not ready
 184         retry = 5;
 185         while (retry > 0)
 186         {
 187                 ret = connect(client, (struct sockaddr*) &server, serverLen);
 188                 if (ret < 0 && errno == ENOENT)
 189                 {
 190                         SysLog(NID_IO, "The server is not ready. %d", retry);
 191
 192                         usleep(1000 * 1000);
 193
 194                         --retry;
 195                 }
 196                 else
 197                 {
 198                         break;
 199                 }
 200         }
 201
 202         if (ret < 0)
 203         {
 204                 SysTryCatch(NID_IO, errno == EINPROGRESS, r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to connect to server(%s) : %s",
 205                                    socketName.c_str(), strerror(errno));
 206
 207                 fd_set rset;
 208                 fd_set wset;
 209                 struct timeval timeout;
 210                 int length = 0;
 211                 int error = 0;
 212                 socklen_t socketLength = 0;
 213
 214                 FD_ZERO(&rset);
 215                 FD_SET(client, &rset);
 216                 wset = rset;
 217                 timeout.tv_sec = 10;
 218                 timeout.tv_usec = 0;
 219
 220                 while (true)
 221                 {
 222                         ret = select(client+1, &rset, &wset, NULL, &timeout);
 223                         if (ret < 0)
 224                         {
 225                                 SysTryCatch(NID_IO, errno == EINTR , r = E_SYSTEM, E_SYSTEM, "[E_TIMEOUT] Failed to connect due to system error.");
 226
 227                                 continue;
 228                         }
 229                         else if (ret == 0)
 230                         {
 231                                 r = E_TIMEOUT;
 232                                 SysLogException(NID_IO, E_TIMEOUT, "[E_TIMEOUT] Failed to connect due to timeout.");
 233                                 goto CATCH;
 234                         }
 235                         else
 236                         {
 237                                 break;
 238                         }
 239                 }
 240
 241                 if (FD_ISSET(client, &rset) || FD_ISSET(client, &wset))
 242                 {
 243                         length = sizeof(error);
 244                         ret = getsockopt(client, SOL_SOCKET, SO_ERROR, &error, &socketLength);
 245                         SysTryCatch(NID_IO, ret >= 0 , r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to connect to server(%s) : %s",
 246                                            socketName.c_str(), strerror(errno));
 247                 }
 248                 else
 249                 {
 250                         r = E_SYSTEM;
 251                         SysLogException(NID_IO, E_SYSTEM, "[E_TIMEOUT] Failed to connect due to system error.");
 252                         goto CATCH;
 253                 }
 254         }
 255
 256         ret = fcntl(client, F_SETFL, flags);
 257         SysTryCatch(NID_IO, ret >= 0, r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to set file status flags (%d, %s).",
 258                                            errno, strerror(errno));
 259
 260         ret = write(client, &helloMessage, sizeof(helloMessage));
 261         SysTryCatch(NID_IO, ret >= 0, r = E_SYSTEM, E_SYSTEM, "[E_SYSTEM] Failed to send hello message (%d, %s).",
 262                         errno, strerror(errno));
 263
 264         if (forReverse)
 265         {
 266                 GError* pGError = null;
 267                 GSource* pGSource = null;
 268                 ;
 269                 GIOChannel* pChannel = g_io_channel_unix_new(client);
 270                 GMainContext* pGMainContext = g_main_context_default();
 271
 272                 g_io_channel_set_encoding(pChannel, null, &pGError);
 273                 g_io_channel_set_flags(pChannel, G_IO_FLAG_NONBLOCK, &pGError);
 274                 g_io_channel_set_close_on_unref(pChannel, TRUE);
 275
 276                 pGSource = g_io_create_watch(pChannel, (GIOCondition) (G_IO_IN | G_IO_ERR | G_IO_NVAL | G_IO_HUP));
 277                 g_source_set_callback(pGSource, (GSourceFunc) OnReadMessage, this, null);
 278                 g_source_attach(pGSource, pGMainContext);
 279
 280                 g_io_channel_unref(pChannel);
 281                 __pReverseSource = pGSource;
 282         }
 283         else
 284         {
 285                 ReleaseFd(client);
 286         }
 287
 288         return r;
 289
 290 CATCH:
 291         if (client != -1)
 292         {
 293                 close(client);
 294         }
 295
 296         return r;
 297 }
 298
 299 gboolean
 300 _IpcClient::OnReadMessage(GIOChannel* source, GIOCondition condition, gpointer data)
 301 {
 302
 303         _IpcClient* pIpcClient = (_IpcClient*) data;
 304
 305         return pIpcClient->HandleReceivedMessage(source, condition);
 306 }
 307
 308 gboolean
 309 _IpcClient::HandleReceivedMessage(GIOChannel* source, GIOCondition condition)
 310 {
 311         GError* pGError = null;
 312         GIOStatus status;
 313         IPC::Message* pMessage = null;
 314
 315         if (condition & G_IO_HUP)
 316         {
 317                 SysLog(NID_IO, "G_IO_HUP, the connection is closed.");
 318
 319                 g_source_destroy(__pReverseSource);
 320                 g_source_unref(__pReverseSource);
 321                 __pReverseSource = null;
 322
 323                 if (__pListener)
 324                 {
 325                         __pListener->OnIpcServerDisconnected(*this);
 326                 }
 327
 328                 return FALSE;
 329         }
 330         else if (condition & G_IO_IN)
 331         {
 332                 gsize readSize = 0;
 333                 const char* pStart = null;
 334                 const char* pEnd = null;
 335                 const char* pEndOfMessage = null;
 336
 337                 while (true)
 338                 {
 339                         pGError = null;
 340                         status = g_io_channel_read_chars(source, (char*) __messageBuffer, __MAX_MESSAGE_BUFFER_SIZE, &readSize, &pGError);
 341                         if (status == G_IO_STATUS_EOF || status == G_IO_STATUS_ERROR)
 342                         {
 343                                 if (status == G_IO_STATUS_EOF)
 344                                 {
 345                                         SysLog(NID_IO, "G_IO_STATUS_EOF, the connection is closed.");
 346                                 }
 347                                 else
 348                                 {
 349                                         SysLog(NID_IO, "G_IO_STATUS_ERROR, the connection is closed. ");
 350                                 }
 351
 352                                 pGError = null;
 353
 354                                 g_io_channel_shutdown(source, FALSE, &pGError);
 355
 356                                 g_source_destroy(__pReverseSource);
 357                                 g_source_unref(__pReverseSource);
 358                                 __pReverseSource = null;
 359
 360                                 if (__pListener)
 361                                 {
 362                                         __pListener->OnIpcServerDisconnected(*this);
 363                                 }
 364
 365                                 return FALSE;
 366                         }
 367
 368                         if (readSize == 0)
 369                         {
 370                                 break;
 371                         }
 372
 373                         if (__pending.empty())
 374                         {
 375                                 pStart = __messageBuffer;
 376                                 pEnd = pStart + readSize;
 377                         }
 378                         else
 379                         {
 380                                 __pending.append(__messageBuffer, readSize);
 381                                 pStart = __pending.data();
 382                                 pEnd = pStart + __pending.size();
 383                         }
 384
 385                         while (true)
 386                         {
 387                                 pEndOfMessage = IPC::Message::FindNext(pStart, pEnd);
 388                                 if (pEndOfMessage == null)
 389                                 {
 390                                         __pending.assign(pStart, pEnd - pStart);
 391                                         break;
 392                                 }
 393
 394                                 pMessage = new (std::nothrow) IPC::Message(pStart, pEndOfMessage - pStart);
 395                                 SysTryReturn(NID_IO, pMessage != null, FALSE, E_OUT_OF_MEMORY, "[E_OUT_OF_MEMORY] The memory is insufficient.");
 396
 397                                 if (__pListener)
 398                                 {
 399                                         __pListener->OnIpcResponseReceived(*this, *pMessage);
 400                                 }
 401
 402                                 delete pMessage;
 403
 404                                 pStart = pEndOfMessage;
 405                         }
 406                 }
 407         }
 408         else
 409         {
 410                 // empty statement.
 411         }
 412
 413         return TRUE;
 414 }
 415
 416 int
 417 _IpcClient::AcquireFd(void)
 418 {
 419         result r = E_SUCCESS;
 420         int fd = -1;
 421
 422         while (fd == -1)
 423         {
 424                 __pFdLock->Acquire();
 425                 if (__fds.size() == 0)
 426                 {
 427                         __pFdLock->Release();
 428                         r = MakeConnection(false);
 429                         SysTryReturn(NID_IO, !IsFailed(r), -1, r, "[%s] Failed to connect to the server.", GetErrorMessage(r));
 430
 431                         continue;
 432                 }
 433
 434                 fd = __fds.back();
 435                 __fds.pop_back();
 436
 437                 __pFdLock->Release();
 438         }
 439
 440         return fd;
 441 }
 442
 443 void
 444 _IpcClient::ReleaseFd(int fd)
 445 {
 446         __pFdLock->Acquire();
 447
 448         __fds.push_back(fd);
 449
 450         __pFdLock->Release();
 451 }
 452
 453 result
 454 _IpcClient::SendAsync(IPC::Message* pMessage)
 455 {
 456         result r = E_SUCCESS;
 457
 458         int fd = -1;
 459         char* pData = null;
 460         int remain = 0;
 461         int written = 0;
 462
 463         pData = (char*) pMessage->data();
 464         remain = pMessage->size();
 465
 466         fd = AcquireFd();
 467         SysTryReturnResult(NID_IO, fd != -1, E_SYSTEM, "Failed to get fd.");
 468
 469         while (remain > 0)
 470         {
 471                 written = write(fd, (char*) pData, remain);
 472                 if (written < 0)
 473                 {
 474                         ReleaseFd(fd);
 475
 476                         if (errno == EAGAIN)
 477                         {
 478                                 SysLogException(NID_IO, E_RESOURCE_UNAVAILABLE, "[E_RESOURCE_UNAVAILABLE] The socket buffer is full.");
 479                                 return E_RESOURCE_UNAVAILABLE;
 480                         }
 481                         else if (errno == EPIPE)
 482                         {
 483                                 SysLogException(NID_IO, E_INVALID_CONNECTION, "[E_INVALID_CONNECTION] The socket connection is closed.");
 484                                 return E_INVALID_CONNECTION;
 485                         }
 486
 487                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to send a request: %d, %s", errno, strerror(errno));
 488                         return E_SYSTEM;
 489                 }
 490
 491                 remain -= written;
 492                 pData += written;
 493         }
 494
 495         ReleaseFd(fd);
 496
 497         return r;
 498 }
 499
 500 result
 501 _IpcClient::SendSync(IPC::Message* pMessage)
 502 {
 503         result r = E_SUCCESS;
 504
 505         int messageId = 0;
 506         int fd = -1;
 507         int ret = 0;
 508
 509         char* pData = null;
 510         int remain = 0;
 511         int written = 0;
 512         int readSize = 0;
 513         char buffer[1024];
 514         char* pEndOfMessage = null;
 515
 516         std::string message;
 517
 518         IPC::Message* pReply = null;
 519         IPC::SyncMessage* pSyncMessage = dynamic_cast <IPC::SyncMessage*>(pMessage);
 520         SysTryReturnResult(NID_IO, pSyncMessage != null, E_INVALID_ARG, "pMessage is not a sync message.");
 521
 522         MessageReplyDeserializer* pReplyDeserializer = pSyncMessage->GetReplyDeserializer();
 523         messageId = SyncMessage::GetMessageId(*pSyncMessage);
 524
 525         fd = AcquireFd();
 526         if (fd < 0)
 527         {
 528                 SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to get an fd");
 529
 530                 delete pReplyDeserializer;
 531                 return E_SYSTEM;
 532         }
 533
 534         pData = (char*) pSyncMessage->data();
 535         remain = pSyncMessage->size();
 536
 537         while (remain > 0)
 538         {
 539                 written = write(fd, (char*) pData, remain);
 540                 if (written < 0)
 541                 {
 542                         SysTryCatch(NID_IO, errno != EAGAIN, r = E_RESOURCE_UNAVAILABLE, E_RESOURCE_UNAVAILABLE, "[E_RESOURCE_UNAVAILABLE] The socket buffer is full.");
 543
 544                         SysTryCatch(NID_IO, errno != EPIPE, r = E_INVALID_CONNECTION, E_INVALID_CONNECTION, "[E_INVALID_CONNECTION] The socket connection is closed.");
 545
 546                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to send a request: %d, %s", errno, strerror(errno));
 547
 548                         r = E_SYSTEM;
 549                         goto CATCH;
 550                 }
 551
 552                 remain -= written;
 553                 pData += written;
 554         }
 555
 556         // Wait reply
 557         struct pollfd pfd;
 558
 559         pfd.fd = fd;
 560         pfd.events = POLLIN | POLLRDHUP;
 561         pfd.revents = 0;
 562
 563         while (true)
 564         {
 565                 ret = poll(&pfd, 1, -1);
 566                 if (ret < 0)
 567                 {
 568                         if (errno == EINTR)
 569                         {
 570                                 continue;
 571                         }
 572
 573                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to poll (%d, %s).", errno, strerror(errno));
 574
 575                         r = E_SYSTEM;
 576                         goto CATCH;
 577                 }
 578
 579                 if (pfd.revents & POLLRDHUP)
 580                 {
 581                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] POLLRDHUP");
 582
 583                         r = E_SYSTEM;
 584                         goto CATCH;
 585                 }
 586
 587                 if (pfd.revents & POLLIN)
 588                 {
 589                         readSize = read(fd, buffer, 1024);
 590                 }
 591
 592                 if (readSize > 0)
 593                 {
 594                         message.append(buffer, readSize);
 595                 }
 596
 597                 pEndOfMessage = (char*) IPC::Message::FindNext(message.data(), message.data() + message.size());
 598                 if (pEndOfMessage)
 599                 {
 600                         pReply = new (std::nothrow) IPC::Message(message.data(), pEndOfMessage - message.data());
 601                         SysTryCatch(NID_IO, pReply != null, r = E_OUT_OF_MEMORY, E_OUT_OF_MEMORY, "[E_OUT_OF_MEMORY] The memory is insufficient.");
 602
 603                         break;
 604                 }
 605         }
 606
 607         pReplyDeserializer->SerializeOutputParameters(*pReply);
 608         delete pReply;
 609
 610 CATCH:
 611         delete pReplyDeserializer;
 612
 613         ReleaseFd(fd);
 614
 615         return r;
 616 }
 617
 618 result
 619 _IpcClient::Send(IPC::Message* pMessage)
 620 {
 621         result r = E_SUCCESS;
 622
 623         SysAssertf(__pFdLock != null, "Not yet constructed. Construct() should be called before use.\n");
 624
 625         if (pMessage->is_sync())
 626         {
 627                 r = SendSync(pMessage);
 628         }
 629         else
 630         {
 631                 r = SendAsync(pMessage);
 632         }
 633
 634         return r;
 635 }
 636
 637 result
 638 _IpcClient::SendRequest(IPC::Message* pMessage)
 639 {
 640         return Send(pMessage);
 641 }
 642
 643 result
 644 _IpcClient::SendRequest(const IPC::Message& message)
 645 {
 646         result r = E_SUCCESS;
 647
 648         SysAssertf(__pFdLock != null, "Not yet constructed. Construct() should be called before use.\n");
 649
 650         if (message.is_sync())
 651         {
 652                 r = SendSync(const_cast<IPC::Message*>(&message));
 653         }
 654         else
 655         {
 656                 r = SendAsync(const_cast<IPC::Message*>(&message));
 657         }
 658
 659         return r;
 660 }
 661
 662 } } //Tizen::Io