src/com-core_thread.c

   1 /*
   2  * Copyright (c) 2000 - 2013 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
  18 #define _GNU_SOURCE
  19 #include <stdio.h>
  20 #include <errno.h>
  21 #include <stdlib.h>
  22 #include <unistd.h>
  23 #include <fcntl.h>
  24 #include <string.h>
  25 #include <sys/ioctl.h>
  26 #include <pthread.h>
  27
  28 #include <glib.h>
  29
  30 #include <dlog.h>
  31
  32 #include "dlist.h"
  33 #include "secure_socket.h"
  34 #include "debug.h"
  35 #include "com-core.h"
  36 #include "com-core_internal.h"
  37 #include "util.h"
  38
  39 int errno;
  40 #define EVENT_READY 'a'
  41 #define EVENT_TERM 'e'
  42
  43 static struct {
  44         struct dlist *tcb_list;
  45         struct dlist *server_list;
  46 } s_info = {
  47         .tcb_list = NULL,
  48         .server_list = NULL,
  49 };
  50
  51 /*!
  52  * \brief Representing the Server Object
  53  */
  54 struct server {
  55         int (*service_cb)(int fd, void *data);
  56         void *data;
  57
  58         guint id;
  59         int handle;
  60 };
  61
  62 /*!
  63  * \brief This is used to holds a packet
  64  */
  65 struct chunk {
  66         char *data;
  67         int offset;
  68         int size;
  69         pid_t pid;
  70         int fd;
  71 };
  72
  73 /*!
  74  * \brief Thread Control Block
  75  */
  76 struct tcb {
  77         pthread_t thid;
  78         int handle;
  79         struct dlist *chunk_list;
  80         int evt_pipe[PIPE_MAX];
  81         int ctrl_pipe[PIPE_MAX];
  82         pthread_mutex_t chunk_lock;
  83         guint id; /*!< g_io_watch */
  84
  85         int server_handle;
  86
  87         int (*service_cb)(int fd, void *data);
  88         void *data;
  89 };
  90
  91 static ssize_t write_safe(int fd, const void *data, size_t bufsz)
  92 {
  93         int ret;
  94         int again;
  95
  96         do {
  97                 again = 0;
  98                 ret = write(fd, data, bufsz);
  99                 if (ret < 0) {
 100                         ret = -errno;
 101                         switch (ret) {
 102                         case -EAGAIN:
 103                         case -EINTR:
 104                                 again = 1;
 105                                 ErrPrint("Interrupted[%d] Again[%d]\n", fd, -ret);
 106                                 break;
 107                         default:
 108                                 ErrPrint("Failed to write: %s (%d)\n", strerror(-ret), -ret);
 109                                 return ret;
 110                         }
 111                 }
 112         } while (again);
 113
 114         return ret;
 115 }
 116
 117 /*!
 118  * \NOTE
 119  * Running thread: Main
 120  */
 121 static inline void server_destroy(struct server *server)
 122 {
 123         dlist_remove_data(s_info.server_list, server);
 124
 125         if (server->id > 0) {
 126                 g_source_remove(server->id);
 127         }
 128
 129         if (server->handle > 0) {
 130                 secure_socket_destroy_handle(server->handle);
 131         }
 132
 133         free(server);
 134 }
 135
 136 /*!
 137  * \NOTE
 138  * Running thread: Main
 139  */
 140 static inline struct server *server_create(int handle, int (*service_cb)(int fd, void *data), void *data)
 141 {
 142         struct server *server;
 143
 144         server = malloc(sizeof(*server));
 145         if (!server) {
 146                 ErrPrint("Heap: %s\n", strerror(errno));
 147                 return NULL;
 148         }
 149
 150         server->handle = handle;
 151         server->service_cb = service_cb;
 152         server->data = data;
 153
 154         s_info.server_list = dlist_append(s_info.server_list, server);
 155         return server;
 156 }
 157
 158 /*!
 159  * \NOTE
 160  * Running thread: Main
 161  */
 162 static inline void destroy_chunk(struct chunk *chunk)
 163 {
 164         free(chunk->data);
 165         free(chunk);
 166 }
 167
 168 /*!
 169  * \NOTE
 170  * Running thread: Main
 171  */
 172 static inline void terminate_thread(struct tcb *tcb)
 173 {
 174         int status;
 175         struct dlist *l;
 176         struct dlist *n;
 177         void *res = NULL;
 178         struct chunk *chunk;
 179
 180         if (write_safe(tcb->ctrl_pipe[PIPE_WRITE], &tcb, sizeof(tcb)) != sizeof(tcb)) {
 181                 ErrPrint("Unable to write CTRL pipe (%d)\n", sizeof(tcb));
 182         }
 183
 184         secure_socket_destroy_handle(tcb->handle);
 185
 186         status = pthread_join(tcb->thid, &res);
 187         if (status != 0) {
 188                 ErrPrint("Join: %s\n", strerror(status));
 189         } else {
 190                 ErrPrint("Thread returns: %d\n", (int)res);
 191         }
 192
 193         dlist_foreach_safe(tcb->chunk_list, l, n, chunk) {
 194                 /*!
 195                  * Discarding all packets
 196                  */
 197                 DbgPrint("Discarding chunks\n");
 198                 tcb->chunk_list = dlist_remove(tcb->chunk_list, l);
 199                 destroy_chunk(chunk);
 200         }
 201 }
 202
 203 /*!
 204  * \NOTE
 205  * Running thread: Main
 206  */
 207 static inline void chunk_remove(struct tcb *tcb, struct chunk *chunk)
 208 {
 209         char event_ch;
 210
 211         /* Consuming the event */
 212         if (read(tcb->evt_pipe[PIPE_READ], &event_ch, sizeof(event_ch)) != sizeof(event_ch)) {
 213                 ErrPrint("Failed to get readsize\n");
 214                 return;
 215         }
 216
 217         CRITICAL_SECTION_BEGIN(&tcb->chunk_lock);
 218
 219         dlist_remove_data(tcb->chunk_list, chunk);
 220
 221         CRITICAL_SECTION_END(&tcb->chunk_lock);
 222
 223         destroy_chunk(chunk);
 224 }
 225
 226 /*!
 227  * \NOTE
 228  * Running thread: Other
 229  */
 230 static inline int chunk_append(struct tcb *tcb, struct chunk *chunk)
 231 {
 232         char event_ch = EVENT_READY;
 233         int ret;
 234
 235         CRITICAL_SECTION_BEGIN(&tcb->chunk_lock);
 236
 237         tcb->chunk_list = dlist_append(tcb->chunk_list, chunk);
 238
 239         CRITICAL_SECTION_END(&tcb->chunk_lock);
 240
 241         ret = write_safe(tcb->evt_pipe[PIPE_WRITE], &event_ch, sizeof(event_ch));
 242         if (ret < 0) {
 243                 CRITICAL_SECTION_BEGIN(&tcb->chunk_lock);
 244
 245                 dlist_remove_data(tcb->chunk_list, chunk);
 246
 247                 CRITICAL_SECTION_END(&tcb->chunk_lock);
 248                 return ret;
 249         }
 250
 251         if (ret != sizeof(event_ch)) {
 252                 ErrPrint("Failed to trigger reader\n");
 253         }
 254
 255         /* Take a breathe */
 256         pthread_yield();
 257         return 0;
 258 }
 259
 260 /*!
 261  * \NOTE
 262  * Running thread: Main
 263  */
 264 static inline int wait_event(struct tcb *tcb, double timeout)
 265 {
 266         fd_set set;
 267         int ret;
 268
 269         FD_ZERO(&set);
 270         FD_SET(tcb->evt_pipe[PIPE_READ], &set);
 271
 272         if (timeout > 0.0f) {
 273                 struct timeval tv;
 274                 tv.tv_sec = (unsigned long)timeout;
 275                 tv.tv_usec = (timeout - (unsigned long)timeout) * 1000000u;
 276                 ret = select(tcb->evt_pipe[PIPE_READ] + 1, &set, NULL, NULL, &tv);
 277         } else if (timeout == 0.0f) {
 278                 ret = select(tcb->evt_pipe[PIPE_READ] + 1, &set, NULL, NULL, NULL);
 279         } else {
 280                 ErrPrint("Invalid timeout: %lf (it must be greater than 0.0)\n", timeout);
 281                 return -EINVAL;
 282         }
 283
 284         if (ret < 0) {
 285                 ret = -errno;
 286                 if (errno == EINTR) {
 287                         DbgPrint("Select receives INTR\n");
 288                         return -EAGAIN;
 289                 }
 290
 291                 ErrPrint("Error: %s\n", strerror(errno));
 292                 return ret;
 293         } else if (ret == 0) {
 294                 ErrPrint("Timeout expired\n");
 295                 return -ETIMEDOUT;
 296         }
 297
 298         if (!FD_ISSET(tcb->evt_pipe[PIPE_READ], &set)) {
 299                 ErrPrint("Unexpected handle is toggled\n");
 300                 return -EINVAL;
 301         }
 302
 303         return 0;
 304 }
 305
 306 /*!
 307  * \NOTE
 308  * Running thread: Main
 309  */
 310 static inline struct chunk *create_chunk(int size)
 311 {
 312         struct chunk *chunk;
 313
 314         chunk = malloc(sizeof(*chunk));
 315         if (!chunk) {
 316                 ErrPrint("Heap: %s\n", strerror(errno));
 317                 return NULL;
 318         }
 319
 320         chunk->data = malloc(size);
 321         if (!chunk->data) {
 322                 ErrPrint("Heap: %s\n", strerror(errno));
 323                 free(chunk);
 324                 return NULL;
 325         }
 326
 327         chunk->pid = (pid_t)-1;
 328         chunk->size = size;
 329         chunk->offset = 0;
 330         chunk->fd = -1;
 331         return chunk;
 332 }
 333
 334 /*!
 335  * \NOTE
 336  * Running thread: Other
 337  */
 338 static void *client_cb(void *data)
 339 {
 340         struct tcb *tcb = data;
 341         struct chunk *chunk;
 342         int ret = 0;
 343         fd_set set;
 344         int readsize;
 345         char event_ch;
 346         int fd;
 347
 348         DbgPrint("Thread is created for %d (server: %d)\n", tcb->handle, tcb->server_handle);
 349         /*!
 350          * \NOTE
 351          * Read all data from the socket as possible as it can do
 352          */
 353         while (1) {
 354                 FD_ZERO(&set);
 355                 FD_SET(tcb->handle, &set);
 356                 FD_SET(tcb->ctrl_pipe[PIPE_READ], &set);
 357
 358                 fd = tcb->handle > tcb->ctrl_pipe[PIPE_READ] ? tcb->handle : tcb->ctrl_pipe[PIPE_READ];
 359
 360                 ret = select(fd + 1, &set, NULL, NULL, NULL);
 361                 if (ret < 0) {
 362                         if (errno == EINTR) {
 363                                 DbgPrint("Select receives INTR\n");
 364                                 continue;
 365                         }
 366                         ret = -errno;
 367                         /*!< Error */
 368                         ErrPrint("Error: %s\n", strerror(errno));
 369                         break;
 370                 } else if (ret == 0) {
 371                         ErrPrint("What happens? [%d]\n", tcb->handle);
 372                         continue;
 373                 }
 374
 375                 if (FD_ISSET(tcb->ctrl_pipe[PIPE_READ], &set)) {
 376                         DbgPrint("Thread is canceled\n");
 377                         ret = -ECANCELED;
 378                         break;
 379                 }
 380
 381                 if (!FD_ISSET(tcb->handle, &set)) {
 382                         ErrPrint("Unexpected handle is toggled\n");
 383                         ret = -EINVAL;
 384                         break;
 385                 }
 386
 387                 readsize = 0;
 388                 ret = ioctl(tcb->handle, FIONREAD, &readsize);
 389                 if (ret < 0) {
 390                         ErrPrint("ioctl: %s\n", strerror(errno));
 391                         break;
 392                 }
 393
 394                 if (readsize <= 0) {
 395                         ErrPrint("Available data: %d\n", readsize);
 396                         ret = -ECONNRESET;
 397                         break;
 398                 }
 399
 400                 chunk = create_chunk(readsize);
 401                 if (!chunk) {
 402                         ErrPrint("Failed to create a new chunk: %d\n", readsize);
 403                         ret = -ENOMEM;
 404                         break;
 405                 }
 406
 407                 ret = secure_socket_recv_with_fd(tcb->handle, chunk->data, chunk->size, &chunk->pid, &chunk->fd);
 408                 if (ret <= 0) {
 409                         destroy_chunk(chunk);
 410                         if (ret == -EAGAIN) {
 411                                 DbgPrint("Retry to get data\n");
 412                                 continue;
 413                         }
 414
 415                         DbgPrint("Recv returns: %d\n", ret);
 416                         break;
 417                 }
 418
 419                 /* Update chunk size */
 420                 chunk->size = ret;
 421
 422                 /*!
 423                  * Count of chunk elements are same with PIPE'd data
 424                  */
 425                 if (chunk_append(tcb, chunk) < 0) {
 426                         destroy_chunk(chunk);
 427                         break;
 428                 }
 429         }
 430
 431         DbgPrint("Client CB is terminated (%d)\n", tcb->handle);
 432         /* Wake up main thread to get disconnected event */
 433         event_ch = EVENT_TERM;
 434
 435         if (write_safe(tcb->evt_pipe[PIPE_WRITE], &event_ch, sizeof(event_ch)) != sizeof(event_ch)) {
 436                 ErrPrint("%d byte is not written\n", sizeof(event_ch));
 437         }
 438
 439         return (void *)(unsigned long)ret;
 440 }
 441
 442 /*!
 443  * \NOTE
 444  * Running thread: Main
 445  */
 446 static inline void tcb_destroy(struct tcb *tcb)
 447 {
 448         int status;
 449
 450         dlist_remove_data(s_info.tcb_list, tcb);
 451
 452         if (tcb->id > 0) {
 453                 g_source_remove(tcb->id);
 454         }
 455
 456         CLOSE_PIPE(tcb->evt_pipe);
 457         CLOSE_PIPE(tcb->ctrl_pipe);
 458
 459         status = pthread_mutex_destroy(&tcb->chunk_lock);
 460         if (status != 0) {
 461                 ErrPrint("Failed to destroy mutex: %s\n", strerror(status));
 462         }
 463
 464         free(tcb);
 465 }
 466
 467 /*!
 468  * \NOTE
 469  * Running thread: Main
 470  */
 471 static gboolean evt_pipe_cb(GIOChannel *src, GIOCondition cond, gpointer data)
 472 {
 473         int pipe_read;
 474         struct tcb *tcb = data;
 475         int ret;
 476
 477         pipe_read = g_io_channel_unix_get_fd(src);
 478
 479         if (tcb->evt_pipe[PIPE_READ] != pipe_read) {
 480                 ErrPrint("Closed handle (%d <> %d)\n", tcb->evt_pipe[PIPE_READ], pipe_read);
 481                 goto errout;
 482         }
 483
 484         if (!(cond & G_IO_IN)) {
 485                 ErrPrint("PIPE is not valid\n");
 486                 goto errout;
 487         }
 488
 489         if ((cond & G_IO_ERR) || (cond & G_IO_HUP) || (cond & G_IO_NVAL)) {
 490                 ErrPrint("PIPE is not valid\n");
 491                 goto errout;
 492         }
 493
 494         ret = tcb->service_cb(tcb->handle, tcb->data);
 495         if (ret < 0) {
 496                 DbgPrint("Service callback returns %d < 0\n", ret);
 497                 goto errout;
 498         }
 499
 500         return TRUE;
 501
 502 errout:
 503         DbgPrint("Disconnecting\n");
 504         invoke_disconn_cb_list(tcb->handle, 0, 0, 0);
 505         terminate_thread(tcb);
 506         tcb_destroy(tcb);
 507         return FALSE;
 508 }
 509
 510 /*!
 511  * \NOTE
 512  * Running thread: Main
 513  */
 514 static inline struct tcb *tcb_create(int client_fd, int is_sync, int (*service_cb)(int fd, void *data), void *data)
 515 {
 516         struct tcb *tcb;
 517         int status;
 518
 519         tcb = malloc(sizeof(*tcb));
 520         if (!tcb) {
 521                 ErrPrint("Error: %s\n", strerror(errno));
 522                 return NULL;
 523         }
 524
 525         tcb->handle = client_fd;
 526         tcb->chunk_list = NULL;
 527         tcb->service_cb = service_cb;
 528         tcb->data = data;
 529         tcb->id = 0;
 530
 531         status = pthread_mutex_init(&tcb->chunk_lock, NULL);
 532         if (status != 0) {
 533                 ErrPrint("Error: %s\n", strerror(status));
 534                 free(tcb);
 535                 return NULL;
 536         }
 537
 538         if (pipe2(tcb->evt_pipe, O_CLOEXEC) < 0) {
 539                 ErrPrint("Error: %s\n", strerror(errno));
 540                 status = pthread_mutex_destroy(&tcb->chunk_lock);
 541                 if (status != 0) {
 542                         ErrPrint("Error: %s\n", strerror(status));
 543                 }
 544                 free(tcb);
 545                 return NULL;
 546         }
 547
 548         if (pipe2(tcb->ctrl_pipe, O_CLOEXEC) < 0) {
 549                 ErrPrint("Error: %s\n", strerror(errno));
 550
 551                 CLOSE_PIPE(tcb->evt_pipe);
 552
 553                 status = pthread_mutex_destroy(&tcb->chunk_lock);
 554                 if (status != 0) {
 555                         ErrPrint("Error: %s\n", strerror(status));
 556                 }
 557
 558                 free(tcb);
 559                 return NULL;
 560         }
 561
 562         DbgPrint("[%d] New TCB created: R(%d), W(%d)\n", client_fd, tcb->evt_pipe[PIPE_READ], tcb->evt_pipe[PIPE_WRITE]);
 563         return tcb;
 564 }
 565
 566 /*!
 567  * \NOTE
 568  * Running thread: Main
 569  */
 570 static gboolean accept_cb(GIOChannel *src, GIOCondition cond, gpointer data)
 571 {
 572         int socket_fd;
 573         int fd;
 574         int ret;
 575         struct tcb *tcb;
 576         GIOChannel *gio;
 577         struct server *server = data;
 578         pthread_attr_t attr;
 579         pthread_attr_t *pattr = NULL;
 580
 581         socket_fd = g_io_channel_unix_get_fd(src);
 582         if (!(cond & G_IO_IN)) {
 583                 ErrPrint("Accept socket closed\n");
 584                 server_destroy(server);
 585                 return FALSE;
 586         }
 587
 588         if ((cond & G_IO_ERR) || (cond & G_IO_HUP) || (cond & G_IO_NVAL)) {
 589                 DbgPrint("Socket connection is lost\n");
 590                 server_destroy(server);
 591                 return FALSE;
 592         }
 593
 594         fd = secure_socket_get_connection_handle(socket_fd);
 595         if (fd < 0) {
 596                 ErrPrint("Failed to get client fd from socket\n");
 597                 server_destroy(server);
 598                 return FALSE;
 599         }
 600
 601         if (fcntl(fd, F_SETFD, FD_CLOEXEC) < 0) {
 602                 ErrPrint("Error: %s\n", strerror(errno));
 603         }
 604
 605         if (fcntl(fd, F_SETFL, O_NONBLOCK) < 0) {
 606                 ErrPrint("Error: %s\n", strerror(errno));
 607         }
 608
 609         tcb = tcb_create(fd, 0, server->service_cb, server->data);
 610         if (!tcb) {
 611                 ErrPrint("Failed to create a TCB\n");
 612                 secure_socket_destroy_handle(fd);
 613                 server_destroy(server);
 614                 return FALSE;
 615         }
 616
 617         tcb->server_handle = socket_fd;
 618
 619         s_info.tcb_list = dlist_append(s_info.tcb_list, tcb);
 620
 621         gio = g_io_channel_unix_new(tcb->evt_pipe[PIPE_READ]);
 622         if (!gio) {
 623                 ErrPrint("Failed to get gio\n");
 624                 secure_socket_destroy_handle(tcb->handle);
 625                 tcb_destroy(tcb);
 626                 server_destroy(server);
 627                 return FALSE;
 628         }
 629
 630         g_io_channel_set_close_on_unref(gio, FALSE);
 631
 632         tcb->id = g_io_add_watch(gio, G_IO_IN | G_IO_HUP | G_IO_ERR | G_IO_NVAL, (GIOFunc)evt_pipe_cb, tcb);
 633         if (tcb->id == 0) {
 634                 GError *err = NULL;
 635                 ErrPrint("Failed to add IO Watch\n");
 636                 g_io_channel_shutdown(gio, TRUE, &err);
 637                 if (err) {
 638                         ErrPrint("Shutdown: %s\n", err->message);
 639                         g_error_free(err);
 640                 }
 641                 g_io_channel_unref(gio);
 642                 secure_socket_destroy_handle(tcb->handle);
 643                 tcb_destroy(tcb);
 644                 server_destroy(server);
 645                 return FALSE;
 646         }
 647         g_io_channel_unref(gio);
 648
 649         invoke_con_cb_list(tcb->handle, tcb->handle, 0, NULL, 0);
 650
 651         ret = pthread_attr_init(&attr);
 652         if (ret == 0) {
 653                 pattr = &attr;
 654
 655                 ret = pthread_attr_setscope(pattr, PTHREAD_SCOPE_SYSTEM);
 656                 if (ret != 0) {
 657                         ErrPrint("setscope: %s\n", strerror(ret));
 658                 }
 659
 660                 ret = pthread_attr_setinheritsched(pattr, PTHREAD_EXPLICIT_SCHED);
 661                 if (ret != 0) {
 662                         ErrPrint("setinheritsched: %s\n", strerror(ret));
 663                 }
 664         } else {
 665                 ErrPrint("attr_init: %s\n", strerror(ret));
 666         }
 667         ret = pthread_create(&tcb->thid, pattr, client_cb, tcb);
 668         if (pattr) {
 669                 pthread_attr_destroy(pattr);
 670         }
 671         if (ret != 0) {
 672                 ErrPrint("Thread creation failed: %s\n", strerror(ret));
 673                 invoke_disconn_cb_list(tcb->handle, 0, 0, 0);
 674                 secure_socket_destroy_handle(tcb->handle);
 675                 tcb_destroy(tcb);
 676                 server_destroy(server);
 677                 return FALSE;
 678         }
 679
 680         return TRUE;
 681 }
 682
 683 /*!
 684  * \NOTE
 685  * Running thread: Main
 686  */
 687 EAPI int com_core_thread_client_create(const char *addr, int is_sync, int (*service_cb)(int fd, void *data), void *data)
 688 {
 689         GIOChannel *gio;
 690         int client_fd;
 691         struct tcb *tcb;
 692         int ret;
 693         pthread_attr_t attr;
 694         pthread_attr_t *pattr = NULL;
 695
 696         client_fd = secure_socket_create_client(addr);
 697         if (client_fd < 0) {
 698                 return client_fd;
 699         }
 700
 701         if (fcntl(client_fd, F_SETFD, FD_CLOEXEC) < 0) {
 702                 ErrPrint("Error: %s\n", strerror(errno));
 703         }
 704
 705         if (fcntl(client_fd, F_SETFL, O_NONBLOCK) < 0) {
 706                 ErrPrint("Error: %s\n", strerror(errno));
 707         }
 708
 709         tcb = tcb_create(client_fd, is_sync, service_cb, data);
 710         if (!tcb) {
 711                 ErrPrint("Failed to create a new TCB\n");
 712                 secure_socket_destroy_handle(client_fd);
 713                 return -EFAULT;
 714         }
 715
 716         tcb->server_handle = -1;
 717
 718         s_info.tcb_list = dlist_append(s_info.tcb_list, tcb);
 719
 720         gio = g_io_channel_unix_new(tcb->evt_pipe[PIPE_READ]);
 721         if (!gio) {
 722                 ErrPrint("Failed to get gio\n");
 723                 secure_socket_destroy_handle(tcb->handle);
 724                 tcb_destroy(tcb);
 725                 return -EIO;
 726         }
 727
 728         g_io_channel_set_close_on_unref(gio, FALSE);
 729
 730         tcb->id = g_io_add_watch(gio, G_IO_IN | G_IO_HUP | G_IO_ERR | G_IO_NVAL, (GIOFunc)evt_pipe_cb, tcb);
 731         if (tcb->id == 0) {
 732                 GError *err = NULL;
 733                 ErrPrint("Failed to add IO Watch\n");
 734                 g_io_channel_shutdown(gio, TRUE, &err);
 735                 if (err) {
 736                         ErrPrint("Shutdown: %s\n", err->message);
 737                         g_error_free(err);
 738                 }
 739                 g_io_channel_unref(gio);
 740                 secure_socket_destroy_handle(tcb->handle);
 741                 tcb_destroy(tcb);
 742                 return -EIO;
 743         }
 744
 745         g_io_channel_unref(gio);
 746
 747         invoke_con_cb_list(tcb->handle, tcb->handle, 0, NULL, 0);
 748
 749         ret = pthread_attr_init(&attr);
 750         if (ret == 0) {
 751                 pattr = &attr;
 752
 753                 ret = pthread_attr_setscope(pattr, PTHREAD_SCOPE_SYSTEM);
 754                 if (ret != 0) {
 755                         ErrPrint("setscope: %s\n", strerror(ret));
 756                 }
 757
 758                 ret = pthread_attr_setinheritsched(pattr, PTHREAD_EXPLICIT_SCHED);
 759                 if (ret != 0) {
 760                         ErrPrint("setinheritsched: %s\n", strerror(ret));
 761                 }
 762         } else {
 763                 ErrPrint("attr_init: %s\n", strerror(ret));
 764         }
 765         ret = pthread_create(&tcb->thid, pattr, client_cb, tcb);
 766         if (pattr) {
 767                 pthread_attr_destroy(pattr);
 768         }
 769         if (ret != 0) {
 770                 ErrPrint("Thread creation failed: %s\n", strerror(ret));
 771                 invoke_disconn_cb_list(tcb->handle, 0, 0, 0);
 772                 secure_socket_destroy_handle(tcb->handle);
 773                 tcb_destroy(tcb);
 774                 return -EFAULT;
 775         }
 776
 777         return tcb->handle;
 778 }
 779
 780 /*!
 781  * \NOTE
 782  * Running thread: Main
 783  */
 784 EAPI int com_core_thread_server_create(const char *addr, int is_sync, const char *label, int (*service_cb)(int fd, void *data), void *data)
 785 {
 786         GIOChannel *gio;
 787         int fd;
 788         struct server *server;
 789
 790         fd = secure_socket_create_server_with_permission(addr, label);
 791         if (fd < 0) {
 792                 return fd;
 793         }
 794
 795         if (fcntl(fd, F_SETFD, FD_CLOEXEC) < 0) {
 796                 ErrPrint("fcntl: %s\n", strerror(errno));
 797         }
 798
 799         if (!is_sync && fcntl(fd, F_SETFL, O_NONBLOCK) < 0) {
 800                 ErrPrint("fcntl: %s\n", strerror(errno));
 801         }
 802
 803         server = server_create(fd, service_cb, data);
 804         if (!server) {
 805                 secure_socket_destroy_handle(fd);
 806                 return -ENOMEM;
 807         }
 808
 809         DbgPrint("Create new IO channel for socket FD: %d\n", fd);
 810         gio = g_io_channel_unix_new(server->handle);
 811         if (!gio) {
 812                 ErrPrint("Failed to create new io channel\n");
 813                 server_destroy(server);
 814                 return -EIO;
 815         }
 816
 817         g_io_channel_set_close_on_unref(gio, FALSE);
 818
 819         server->id = g_io_add_watch(gio, G_IO_IN | G_IO_ERR | G_IO_HUP | G_IO_NVAL, (GIOFunc)accept_cb, server);
 820         if (server->id == 0) {
 821                 GError *err = NULL;
 822                 ErrPrint("Failed to add IO watch\n");
 823                 g_io_channel_shutdown(gio, TRUE, &err);
 824                 if (err) {
 825                         ErrPrint("Shutdown: %s\n", err->message);
 826                         g_error_free(err);
 827                 }
 828                 g_io_channel_unref(gio);
 829                 server_destroy(server);
 830                 return -EIO;
 831         }
 832
 833         g_io_channel_unref(gio);
 834         return server->handle;
 835 }
 836
 837 /*!
 838  * \NOTE
 839  * Running thread: Main
 840  */
 841 static inline struct tcb *find_tcb_by_handle(int handle)
 842 {
 843         struct dlist *l;
 844         struct tcb *tcb;
 845
 846         dlist_foreach(s_info.tcb_list, l, tcb) {
 847                 if (tcb->handle == handle) {
 848                         return tcb;
 849                 }
 850         }
 851
 852         return NULL;
 853 }
 854
 855 EAPI int com_core_thread_send_with_fd(int handle, const char *buffer, int size, double timeout, int fd)
 856 {
 857         int writesize;
 858         int ret;
 859         struct tcb *tcb;
 860
 861         fd_set set;
 862
 863         tcb = find_tcb_by_handle(handle);
 864         if (!tcb) {
 865                 ErrPrint("TCB is not found\n");
 866                 return -EINVAL;
 867         }
 868
 869         writesize = 0;
 870         while (size > 0) {
 871                 FD_ZERO(&set);
 872                 FD_SET(tcb->handle, &set);
 873
 874                 if (timeout > 0.0f) {
 875                         struct timeval tv;
 876
 877                         tv.tv_sec = (unsigned long)timeout;
 878                         tv.tv_usec = (timeout - (unsigned long)timeout) * 1000000u;
 879
 880                         ret = select(tcb->handle + 1, NULL, &set, NULL, &tv);
 881                 } else if (timeout == 0.0f) {
 882                         ret = select(tcb->handle + 1, NULL, &set, NULL, NULL);
 883                 } else {
 884                         ErrPrint("Invalid timeout: %lf (it must be greater than 0.0)\n", timeout);
 885                         return -EINVAL;
 886                 }
 887
 888                 if (ret < 0) {
 889                         ret = -errno;
 890                         if (errno == EINTR) {
 891                                 DbgPrint("Select receives INTR\n");
 892                                 continue;
 893                         }
 894
 895                         ErrPrint("Error: %s\n", strerror(errno));
 896                         return ret;
 897                 } else if (ret == 0) {
 898                         ErrPrint("Timeout expired\n");
 899                         break;
 900                 }
 901
 902                 if (!FD_ISSET(tcb->handle, &set)) {
 903                         ErrPrint("Unexpected handle is toggled\n");
 904                         return -EINVAL;
 905                 }
 906
 907                 ret = secure_socket_send_with_fd(tcb->handle, buffer + writesize, size, fd);
 908                 if (ret < 0) {
 909                         if (ret == -EAGAIN) {
 910                                 DbgPrint("Retry to send data (%d:%d)\n", writesize, size);
 911                                 continue;
 912                         }
 913                         DbgPrint("Failed to send: %d\n", ret);
 914                         return ret;
 915                 } else if (ret == 0) {
 916                         DbgPrint("Disconnected? : Send bytes: 0\n");
 917                         return 0;
 918                 }
 919
 920                 fd = -1;    /* Send only once if it is fd */
 921                 size -= ret;
 922                 writesize += ret;
 923         }
 924
 925         return writesize;
 926 }
 927
 928 /*!
 929  * \NOTE
 930  * Running thread: Main
 931  */
 932 EAPI int com_core_thread_send(int handle, const char *buffer, int size, double timeout)
 933 {
 934         return com_core_thread_send_with_fd(handle, buffer, size, timeout, -1);
 935 }
 936
 937 EAPI int com_core_thread_recv_with_fd(int handle, char *buffer, int size, int *sender_pid, double timeout, int *fd)
 938 {
 939         int readsize;
 940         int ret;
 941         struct chunk *chunk;
 942         struct dlist *l;
 943         struct tcb *tcb;
 944         int _sender_pid;
 945         int _fd;
 946
 947         tcb = find_tcb_by_handle(handle);
 948         if (!tcb) {
 949                 ErrPrint("TCB is not exists\n");
 950                 return -EINVAL;
 951         }
 952
 953         if (!sender_pid) {
 954                 sender_pid = &_sender_pid;
 955         }
 956
 957         if (!fd) {
 958                 fd = &_fd;
 959         }
 960
 961         *fd = -1;
 962         readsize = 0;
 963         while (readsize < size) {
 964                 l = dlist_nth(tcb->chunk_list, 0);
 965                 chunk = dlist_data(l);
 966                 /*!
 967                  * \note
 968                  * Pumping up the pipe data
 969                  * This is the first time to use a chunk
 970                  */
 971                 if (!chunk) {
 972                         ret = wait_event(tcb, timeout);
 973                         if (ret == -EAGAIN) {
 974                                 /* Log is printed from wait_event */
 975                                 continue;
 976                         } else if (ret == -ECONNRESET) {
 977                                 DbgPrint("Connection is lost\n");
 978                                 break;
 979                         } else if (ret < 0) {
 980                                 /* Log is printed from wait_event */
 981                                 return ret;
 982                         }
 983
 984                         l = dlist_nth(tcb->chunk_list, 0);
 985                         chunk = dlist_data(l);
 986                         if (!chunk) {
 987                                 char event_ch;
 988
 989                                 /* Consuming the event */
 990                                 if (read(tcb->evt_pipe[PIPE_READ], &event_ch, sizeof(event_ch)) != sizeof(event_ch)) {
 991                                         ErrPrint("Failed to get readsize: %s\n", strerror(errno));
 992                                 } else if (event_ch == EVENT_READY) {
 993                                         ErrPrint("Failed to get a new chunk\n");
 994                                 } else if (event_ch == EVENT_TERM) {
 995                                         DbgPrint("Disconnected\n");
 996                                 }
 997
 998                                 break;
 999                         }
1000                 }
1001
1002                 ret = chunk->size - chunk->offset;
1003                 ret = ret > (size - readsize) ? (size - readsize) : ret;
1004                 memcpy(buffer + readsize, chunk->data + chunk->offset, ret);
1005                 readsize += ret;
1006                 chunk->offset += ret;
1007
1008                 *sender_pid = chunk->pid;
1009                 if (chunk->fd >= 0) {
1010                         *fd = chunk->fd;
1011                 }
1012
1013                 if (chunk->offset == chunk->size) {
1014                         chunk_remove(tcb, chunk);
1015                 }
1016         }
1017
1018         return readsize;
1019 }
1020
1021 /*!
1022  * \NOTE
1023  * Running thread: Main
1024  */
1025 EAPI int com_core_thread_recv(int handle, char *buffer, int size, int *sender_pid, double timeout)
1026 {
1027         return com_core_thread_recv_with_fd(handle, buffer, size, sender_pid, timeout, NULL);
1028 }
1029
1030 /*!
1031  * \NOTE
1032  * Running thread: Main
1033  */
1034 EAPI int com_core_thread_server_destroy(int handle)
1035 {
1036         struct dlist *l;
1037         struct dlist *n;
1038         struct tcb *tcb;
1039         struct server *server;
1040
1041         dlist_foreach_safe(s_info.tcb_list, l, n, tcb) {
1042                 if (tcb->server_handle != handle) {
1043                         continue;
1044                 }
1045
1046                 invoke_disconn_cb_list(handle, 0, 0, 0);
1047                 terminate_thread(tcb);
1048                 tcb_destroy(tcb);
1049                 return 0;
1050         }
1051
1052         dlist_foreach_safe(s_info.server_list, l, n, server) {
1053                 if (server->handle != handle) {
1054                         continue;
1055                 }
1056
1057                 invoke_disconn_cb_list(handle, 0, 0, 0);
1058                 server_destroy(server);
1059                 return 0;
1060         }
1061
1062         return -ENOENT;
1063 }
1064
1065 /*!
1066  * \NOTE
1067  * Running thread: Main
1068  */
1069 EAPI int com_core_thread_client_destroy(int handle)
1070 {
1071         struct tcb *tcb;
1072
1073         tcb = find_tcb_by_handle(handle);
1074         if (!tcb) {
1075                 return -ENOENT;
1076         }
1077
1078         invoke_disconn_cb_list(handle, 0, 0, 0);
1079         terminate_thread(tcb);
1080         tcb_destroy(tcb);
1081         return 0;
1082 }
1083
1084 /* End of a file */