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("/tmp/");
 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         write(client, &helloMessage, sizeof(helloMessage));
 261
 262         if (forReverse)
 263         {
 264                 GError* pGError = null;
 265                 GSource* pGSource = null;
 266                 ;
 267                 GIOChannel* pChannel = g_io_channel_unix_new(client);
 268                 GMainContext* pGMainContext = g_main_context_default();
 269
 270                 g_io_channel_set_encoding(pChannel, null, &pGError);
 271                 g_io_channel_set_flags(pChannel, G_IO_FLAG_NONBLOCK, &pGError);
 272                 g_io_channel_set_close_on_unref(pChannel, TRUE);
 273
 274                 pGSource = g_io_create_watch(pChannel, (GIOCondition) (G_IO_IN | G_IO_ERR | G_IO_NVAL | G_IO_HUP));
 275                 g_source_set_callback(pGSource, (GSourceFunc) OnReadMessage, this, null);
 276                 g_source_attach(pGSource, pGMainContext);
 277
 278                 g_io_channel_unref(pChannel);
 279                 __pReverseSource = pGSource;
 280         }
 281         else
 282         {
 283                 ReleaseFd(client);
 284         }
 285
 286         return r;
 287
 288 CATCH:
 289         if (client != -1)
 290         {
 291                 close(client);
 292         }
 293
 294         return r;
 295 }
 296
 297 gboolean
 298 _IpcClient::OnReadMessage(GIOChannel* source, GIOCondition condition, gpointer data)
 299 {
 300
 301         _IpcClient* pIpcClient = (_IpcClient*) data;
 302
 303         return pIpcClient->HandleReceivedMessage(source, condition);
 304 }
 305
 306 gboolean
 307 _IpcClient::HandleReceivedMessage(GIOChannel* source, GIOCondition condition)
 308 {
 309         GError* pGError = null;
 310         GIOStatus status;
 311         IPC::Message* pMessage = null;
 312
 313         if (condition & G_IO_HUP)
 314         {
 315                 SysLog(NID_IO, "G_IO_HUP, the connection is closed.");
 316
 317                 g_source_destroy(__pReverseSource);
 318                 g_source_unref(__pReverseSource);
 319                 __pReverseSource = null;
 320
 321                 if (__pListener)
 322                 {
 323                         __pListener->OnIpcServerDisconnected(*this);
 324                 }
 325
 326                 return FALSE;
 327         }
 328         else if (condition & G_IO_IN)
 329         {
 330                 gsize readSize = 0;
 331                 const char* pStart = null;
 332                 const char* pEnd = null;
 333                 const char* pEndOfMessage = null;
 334
 335                 while (true)
 336                 {
 337                         pGError = null;
 338                         status = g_io_channel_read_chars(source, (char*) __messageBuffer, __MAX_MESSAGE_BUFFER_SIZE, &readSize, &pGError);
 339                         if (status == G_IO_STATUS_EOF || status == G_IO_STATUS_ERROR)
 340                         {
 341                                 if (status == G_IO_STATUS_EOF)
 342                                 {
 343                                         SysLog(NID_IO, "G_IO_STATUS_EOF, the connection is closed.");
 344                                 }
 345                                 else
 346                                 {
 347                                         SysLog(NID_IO, "G_IO_STATUS_ERROR, the connection is closed. ");
 348                                 }
 349
 350                                 pGError = null;
 351
 352                                 g_io_channel_shutdown(source, FALSE, &pGError);
 353
 354                                 g_source_destroy(__pReverseSource);
 355                                 g_source_unref(__pReverseSource);
 356                                 __pReverseSource = null;
 357
 358                                 if (__pListener)
 359                                 {
 360                                         __pListener->OnIpcServerDisconnected(*this);
 361                                 }
 362
 363                                 return FALSE;
 364                         }
 365
 366                         if (readSize == 0)
 367                         {
 368                                 break;
 369                         }
 370
 371                         if (__pending.empty())
 372                         {
 373                                 pStart = __messageBuffer;
 374                                 pEnd = pStart + readSize;
 375                         }
 376                         else
 377                         {
 378                                 __pending.append(__messageBuffer, readSize);
 379                                 pStart = __pending.data();
 380                                 pEnd = pStart + __pending.size();
 381                         }
 382
 383                         while (true)
 384                         {
 385                                 pEndOfMessage = IPC::Message::FindNext(pStart, pEnd);
 386                                 if (pEndOfMessage == null)
 387                                 {
 388                                         __pending.assign(pStart, pEnd - pStart);
 389                                         break;
 390                                 }
 391
 392                                 pMessage = new (std::nothrow) IPC::Message(pStart, pEndOfMessage - pStart);
 393                                 SysTryReturn(NID_IO, pMessage != null, FALSE, E_OUT_OF_MEMORY, "[E_OUT_OF_MEMORY] The memory is insufficient.");
 394
 395                                 if (__pListener)
 396                                 {
 397                                         __pListener->OnIpcResponseReceived(*this, *pMessage);
 398                                 }
 399
 400                                 delete pMessage;
 401
 402                                 pStart = pEndOfMessage;
 403                         }
 404                 }
 405         }
 406         else
 407         {
 408                 // empty statement.
 409         }
 410
 411         return TRUE;
 412 }
 413
 414 int
 415 _IpcClient::AcquireFd(void)
 416 {
 417         result r = E_SUCCESS;
 418         int fd = -1;
 419
 420         while (fd == -1)
 421         {
 422                 __pFdLock->Acquire();
 423                 if (__fds.size() == 0)
 424                 {
 425                         __pFdLock->Release();
 426                         r = MakeConnection(false);
 427                         SysTryReturn(NID_IO, !IsFailed(r), -1, r, "[%s] Failed to connect to the server.", GetErrorMessage(r));
 428
 429                         continue;
 430                 }
 431
 432                 fd = __fds.back();
 433                 __fds.pop_back();
 434
 435                 __pFdLock->Release();
 436         }
 437
 438         return fd;
 439 }
 440
 441 void
 442 _IpcClient::ReleaseFd(int fd)
 443 {
 444         __pFdLock->Acquire();
 445
 446         __fds.push_back(fd);
 447
 448         __pFdLock->Release();
 449 }
 450
 451 result
 452 _IpcClient::SendAsync(IPC::Message* pMessage)
 453 {
 454         result r = E_SUCCESS;
 455
 456         int fd = -1;
 457         char* pData = null;
 458         int remain = 0;
 459         int written = 0;
 460
 461         pData = (char*) pMessage->data();
 462         remain = pMessage->size();
 463
 464         fd = AcquireFd();
 465         SysTryReturnResult(NID_IO, fd != -1, E_SYSTEM, "Failed to get fd.");
 466
 467         while (remain > 0)
 468         {
 469                 written = write(fd, (char*) pData, remain);
 470                 if (written < 0)
 471                 {
 472                         ReleaseFd(fd);
 473
 474                         if (errno == EAGAIN)
 475                         {
 476                                 SysLogException(NID_IO, E_RESOURCE_UNAVAILABLE, "[E_RESOURCE_UNAVAILABLE] The socket buffer is full.");
 477                                 return E_RESOURCE_UNAVAILABLE;
 478                         }
 479                         else if (errno == EPIPE)
 480                         {
 481                                 SysLogException(NID_IO, E_INVALID_CONNECTION, "[E_INVALID_CONNECTION] The socket connection is closed.");
 482                                 return E_INVALID_CONNECTION;
 483                         }
 484
 485                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to send a request: %d, %s", errno, strerror(errno));
 486                         return E_SYSTEM;
 487                 }
 488
 489                 remain -= written;
 490                 pData += written;
 491         }
 492
 493         ReleaseFd(fd);
 494
 495         return r;
 496 }
 497
 498 result
 499 _IpcClient::SendSync(IPC::Message* pMessage)
 500 {
 501         result r = E_SUCCESS;
 502
 503         int messageId = 0;
 504         int fd = -1;
 505
 506         char* pData = null;
 507         int remain = 0;
 508         int written = 0;
 509         int readSize = 0;
 510         char buffer[1024];
 511         char* pEndOfMessage = null;
 512
 513         std::string message;
 514
 515         IPC::Message* pReply = null;
 516         MessageReplyDeserializer* pReplyDeserializer = null;
 517         IPC::SyncMessage* pSyncMessage = dynamic_cast <IPC::SyncMessage*>(pMessage);
 518         SysTryReturnResult(NID_IO, pSyncMessage != null, E_INVALID_ARG, "pMessage is not a sync message.");
 519
 520         messageId = SyncMessage::GetMessageId(*pSyncMessage);
 521
 522         fd = AcquireFd();
 523         SysTryReturnResult(NID_IO, fd != -1, E_SYSTEM, "Failed to get fd.");
 524
 525         pData = (char*) pSyncMessage->data();
 526         remain = pSyncMessage->size();
 527
 528         while (remain > 0)
 529         {
 530                 written = write(fd, (char*) pData, remain);
 531                 if (written < 0)
 532                 {
 533                         ReleaseFd(fd);
 534
 535                         if (errno == EAGAIN)
 536                         {
 537                                 SysLogException(NID_IO, E_RESOURCE_UNAVAILABLE, "[E_RESOURCE_UNAVAILABLE] The socket buffer is full.");
 538                                 return E_RESOURCE_UNAVAILABLE;
 539                         }
 540                         else if (errno == EPIPE)
 541                         {
 542                                 SysLogException(NID_IO, E_INVALID_CONNECTION, "[E_INVALID_CONNECTION] The socket connection is closed.");
 543                                 return E_INVALID_CONNECTION;
 544                         }
 545
 546                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] Failed to send a request: %d, %s", errno, strerror(errno));
 547                         return E_SYSTEM;
 548                 }
 549
 550                 remain -= written;
 551                 pData += written;
 552         }
 553
 554         // Wait reply
 555         struct pollfd pfd;
 556
 557         pfd.fd = fd;
 558         pfd.events = POLLIN | POLLRDHUP;
 559         pfd.revents = 0;
 560
 561         int ret = 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                         ReleaseFd(fd);
 576                         return E_SYSTEM;
 577                 }
 578
 579                 if (pfd.revents & POLLRDHUP)
 580                 {
 581                         SysLogException(NID_IO, E_SYSTEM, "[E_SYSTEM] POLLRDHUP");
 582
 583                         ReleaseFd(fd);
 584                         return E_SYSTEM;
 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                         if (pReply == null)
 602                         {
 603                                 SysLogException(NID_IO, E_OUT_OF_MEMORY, "[E_OUT_OF_MEMORY] The memory is insufficient.");
 604
 605                                 ReleaseFd(fd);
 606                                 return E_OUT_OF_MEMORY;
 607                         }
 608
 609                         break;
 610                 }
 611         }
 612
 613         pReplyDeserializer = pSyncMessage->GetReplyDeserializer();
 614         pReplyDeserializer->SerializeOutputParameters(*pReply);
 615
 616         delete pReply;
 617         delete pReplyDeserializer;
 618
 619         ReleaseFd(fd);
 620
 621         return r;
 622 }
 623
 624 result
 625 _IpcClient::Send(IPC::Message* pMessage)
 626 {
 627         result r = E_SUCCESS;
 628
 629         SysAssertf(__pFdLock != null, "Not yet constructed. Construct() should be called before use.\n");
 630
 631         if (pMessage->is_sync())
 632         {
 633                 r = SendSync(pMessage);
 634         }
 635         else
 636         {
 637                 r = SendAsync(pMessage);
 638         }
 639
 640         return r;
 641 }
 642
 643 result
 644 _IpcClient::SendRequest(IPC::Message* pMessage)
 645 {
 646         return Send(pMessage);
 647 }
 648
 649 result
 650 _IpcClient::SendRequest(const IPC::Message& message)
 651 {
 652         result r = E_SUCCESS;
 653
 654         SysAssertf(__pFdLock != null, "Not yet constructed. Construct() should be called before use.\n");
 655
 656         if (message.is_sync())
 657         {
 658                 r = SendSync(const_cast<IPC::Message*>(&message));
 659         }
 660         else
 661         {
 662                 r = SendAsync(const_cast<IPC::Message*>(&message));
 663         }
 664
 665         return r;
 666 }
 667
 668 } } //Tizen::Io