src/slave_rpc.c

   1 /*
   2  * Copyright 2013  Samsung Electronics Co., Ltd
   3  *
   4  * Licensed under the Flora License, Version 1.1 (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://floralicense.org/license/
   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 <stdio.h>
  18 #include <string.h> /* strerror */
  19 #include <errno.h> /* errno */
  20 #include <unistd.h> /* pid_t */
  21 #include <stdlib.h> /* free */
  22 #include <assert.h>
  23
  24 #include <Eina.h>
  25 #include <Ecore.h>
  26
  27 #include <dlog.h>
  28
  29 #include <packet.h>
  30 #include <com-core_packet.h>
  31 #include <livebox-errno.h>
  32
  33 #include "debug.h"
  34 #include "slave_life.h"
  35 #include "slave_rpc.h"
  36 #include "client_life.h"
  37 #include "package.h"
  38 #include "fault_manager.h"
  39 #include "util.h"
  40 #include "conf.h"
  41
  42 struct slave_rpc {
  43         Ecore_Timer *pong_timer;
  44         int handle;
  45
  46         unsigned long ping_count;
  47         unsigned long next_ping_count;
  48         Eina_List *pending_list;
  49 };
  50
  51 struct command {
  52         /* create_command, destroy_command will care these varaibles */
  53         char *pkgname;
  54         struct packet *packet;
  55         struct slave_node *slave;
  56         int ttl; /* If it fails to handle this, destroy this */
  57
  58         /* Don't need to care these data */
  59         void (*ret_cb)(struct slave_node *slave, const struct packet *packet, void *cbdata);
  60         void *cbdata;
  61 };
  62
  63 static struct info {
  64         Eina_List *command_list;
  65         Ecore_Timer *command_consuming_timer;
  66 } s_info = {
  67         .command_list = NULL,
  68         .command_consuming_timer = NULL,
  69 };
  70
  71 #define DEFAULT_CMD_TTL 3
  72
  73 static inline void prepend_command(struct command *command);
  74
  75 static inline struct command *create_command(struct slave_node *slave, const char *pkgname, struct packet *packet)
  76 {
  77         struct command *command;
  78
  79         command = calloc(1, sizeof(*command));
  80         if (!command) {
  81                 ErrPrint("Heap: %s\n", strerror(errno));
  82                 return NULL;
  83         }
  84
  85         if (pkgname) {
  86                 command->pkgname = strdup(pkgname);
  87                 if (!command->pkgname) {
  88                         ErrPrint("Heap: %s\n", strerror(errno));
  89                         DbgFree(command);
  90                         return NULL;
  91                 }
  92         }
  93
  94         command->slave = slave_ref(slave); /*!< To prevent from destroying of the slave while communicating with the slave */
  95         command->packet = packet_ref(packet);
  96         command->ttl = DEFAULT_CMD_TTL;
  97
  98         return command;
  99 }
 100
 101 static inline void destroy_command(struct command *command)
 102 {
 103         slave_unref(command->slave);
 104         packet_unref(command->packet);
 105         DbgFree(command->pkgname);
 106         DbgFree(command);
 107 }
 108
 109 static inline struct command *pop_command(void)
 110 {
 111         struct command *command;
 112
 113         command = eina_list_nth(s_info.command_list, 0);
 114         if (!command)
 115                 return NULL;
 116
 117         s_info.command_list = eina_list_remove(s_info.command_list, command);
 118         return command;
 119 }
 120
 121 static int slave_async_cb(pid_t pid, int handle, const struct packet *packet, void *data)
 122 {
 123         struct command *command = data;
 124
 125         if (!command) {
 126                 ErrPrint("Command is NIL\n");
 127                 return LB_STATUS_SUCCESS;
 128         }
 129
 130         /*!
 131          * \note
 132          * command->packet is not valid from here.
 133          */
 134         if (!slave_is_activated(command->slave)) {
 135                 ErrPrint("Slave is not activated (accidently dead)\n");
 136                 if (command->ret_cb)
 137                         command->ret_cb(command->slave, packet, command->cbdata);
 138                 goto out;
 139         }
 140
 141         if (!packet) {
 142                 DbgPrint("packet == NULL\n");
 143                 if (command->ret_cb)
 144                         command->ret_cb(command->slave, packet, command->cbdata);
 145
 146                 /*
 147                  * \NOTE
 148                  * Slave will be deactivated from dead monitor if it lost its connections.
 149                  * So we don't need to care it again from here.
 150
 151                 command->slave = slave_deactivated_by_fault(command->slave);
 152
 153                  */
 154                 goto out;
 155         }
 156
 157         if (command->ret_cb)
 158                 command->ret_cb(command->slave, packet, command->cbdata);
 159
 160 out:
 161         destroy_command(command);
 162         return LB_STATUS_SUCCESS;
 163 }
 164
 165 static Eina_Bool command_consumer_cb(void *data)
 166 {
 167         struct command *command;
 168         struct slave_rpc *rpc;
 169
 170         command = pop_command();
 171         if (!command) {
 172                 s_info.command_consuming_timer = NULL;
 173                 return ECORE_CALLBACK_CANCEL;
 174         }
 175
 176         if (!slave_is_activated(command->slave)) {
 177                 ErrPrint("Slave is not activated: %s(%d)\n",
 178                                 slave_name(command->slave), slave_pid(command->slave));
 179                 goto errout;
 180         }
 181
 182         if (command->pkgname) {
 183                 struct pkg_info *info;
 184
 185                 info = package_find(command->pkgname);
 186                 if (info && package_is_fault(info)) {
 187                         ErrPrint("info: %p (%s) is fault package\n", info, command->pkgname);
 188                         goto errout;
 189                 }
 190         }
 191
 192         rpc = slave_data(command->slave, "rpc");
 193         if (!rpc || rpc->handle < 0) {
 194                 ErrPrint("Slave has no rpc info\n");
 195                 goto errout;
 196         }
 197
 198         if (packet_type(command->packet) == PACKET_REQ_NOACK) {
 199                 if (com_core_packet_send_only(rpc->handle, command->packet) == 0) {
 200                         /* Keep a slave alive, while processing events */
 201                         slave_give_more_ttl(command->slave);
 202                         destroy_command(command);
 203                         return ECORE_CALLBACK_RENEW;
 204                 }
 205         } else if (packet_type(command->packet) == PACKET_REQ) {
 206                 if (com_core_packet_async_send(rpc->handle, command->packet, 0.0f, slave_async_cb, command) == 0) {
 207                         /* Keep a slave alive, while processing events */
 208                         slave_give_more_ttl(command->slave);
 209                         return ECORE_CALLBACK_RENEW;
 210                 }
 211         }
 212
 213         /*!
 214          * \WARN
 215          * What happens at here?
 216          * We are failed to send a packet!!!
 217          * Let's try to send this again
 218          */
 219         /*!
 220          * \todo
 221          * Do we need to handle this error?
 222          * Close current connection and make new one?
 223          * how about pended command lists?
 224          */
 225         DbgPrint("Packet type: %d\n", packet_type(command->packet));
 226         DbgPrint("Packet: %p\n", command->packet);
 227         DbgPrint("Handle: %d\n", rpc->handle);
 228         DbgPrint("PID: %d\n", slave_pid(command->slave));
 229         DbgPrint("Name: %s\n", slave_name(command->slave));
 230         DbgPrint("Package: %s\n", command->pkgname);
 231         command->ttl--;
 232         if (command->ttl == 0) {
 233                 DbgPrint("Discard packet (%d)\n", command->ttl);
 234                 destroy_command(command);
 235         } else {
 236                 DbgPrint("Send again (%d)\n", command->ttl);
 237                 prepend_command(command);
 238         }
 239         return ECORE_CALLBACK_RENEW;
 240
 241 errout:
 242         if (command->ret_cb)
 243                 command->ret_cb(command->slave, NULL, command->cbdata);
 244
 245         destroy_command(command);
 246         return ECORE_CALLBACK_RENEW;
 247 }
 248
 249 static inline void prepend_command(struct command *command)
 250 {
 251         s_info.command_list = eina_list_prepend(s_info.command_list, command);
 252
 253         if (s_info.command_consuming_timer)
 254                 return;
 255
 256         s_info.command_consuming_timer = ecore_timer_add(PACKET_TIME, command_consumer_cb, NULL);
 257         if (!s_info.command_consuming_timer) {
 258                 ErrPrint("Failed to add command consumer\n");
 259                 s_info.command_list = eina_list_remove(s_info.command_list, command);
 260                 destroy_command(command);
 261         }
 262 }
 263
 264 static inline void push_command(struct command *command)
 265 {
 266         s_info.command_list = eina_list_append(s_info.command_list, command);
 267
 268         if (s_info.command_consuming_timer)
 269                 return;
 270
 271         s_info.command_consuming_timer = ecore_timer_add(PACKET_TIME, command_consumer_cb, NULL);
 272         if (!s_info.command_consuming_timer) {
 273                 ErrPrint("Failed to add command consumer\n");
 274                 s_info.command_list = eina_list_remove(s_info.command_list, command);
 275                 destroy_command(command);
 276         }
 277 }
 278
 279 static int slave_deactivate_cb(struct slave_node *slave, void *data)
 280 {
 281         struct slave_rpc *rpc;
 282         struct command *command;
 283         Eina_List *l;
 284         Eina_List *n;
 285
 286         rpc = slave_data(slave, "rpc");
 287         if (!rpc) {
 288                 /*!
 289                  * \note
 290                  * Return negative value will remove this callback from the event list of the slave
 291                  */
 292                 return LB_STATUS_ERROR_INVALID;
 293         }
 294
 295         if (rpc->pong_timer) {
 296                 ecore_timer_del(rpc->pong_timer);
 297                 rpc->pong_timer = NULL;
 298         } else {
 299                 ErrPrint("slave has no pong timer\n");
 300         }
 301
 302         if (rpc->handle < 0) {
 303                 EINA_LIST_FREE(rpc->pending_list, command) {
 304                         assert(command->slave == slave);
 305                         if (command->ret_cb)
 306                                 command->ret_cb(command->slave, NULL, command->cbdata);
 307                         destroy_command(command);
 308                 }
 309         } else {
 310                 EINA_LIST_FOREACH_SAFE(s_info.command_list, l, n, command) {
 311                         if (command->slave == slave) {
 312                                 s_info.command_list = eina_list_remove(s_info.command_list, command);
 313                                 if (command->ret_cb)
 314                                         command->ret_cb(command->slave, NULL, command->cbdata);
 315                                 destroy_command(command);
 316                         }
 317                 }
 318         }
 319
 320         /*!
 321          * \note
 322          * Reset handle
 323          */
 324         DbgPrint("Reset handle for %d (%d)\n", slave_pid(slave), rpc->handle);
 325         rpc->handle = -1;
 326
 327         /*!
 328          * \todo
 329          * Make statistics table
 330          */
 331         rpc->ping_count = 0;
 332         rpc->next_ping_count = 1;
 333         return LB_STATUS_SUCCESS;
 334 }
 335
 336 static Eina_Bool ping_timeout_cb(void *data)
 337 {
 338         struct slave_rpc *rpc;
 339         struct slave_node *slave = data;
 340
 341         rpc = slave_data(slave, "rpc");
 342         if (!rpc) {
 343                 ErrPrint("Slave RPC is not valid (%s)\n", slave_name(slave));
 344                 return ECORE_CALLBACK_CANCEL;
 345         }
 346
 347         /*!
 348          * \note
 349          * Clear the pong_timer
 350          */
 351         rpc->pong_timer = NULL;
 352
 353         if (!slave_is_activated(slave)) {
 354                 ErrPrint("Slave is not activated (%s)\n", slave_name(slave));
 355                 return ECORE_CALLBACK_CANCEL;
 356         }
 357
 358         /*!
 359          * Dead callback will handling this
 360          */
 361         DbgPrint("Slave PING TIMEOUT: %s(%d) : %p\n", slave_name(slave), slave_pid(slave), slave);
 362         slave = slave_deactivated_by_fault(slave);
 363         DbgPrint("Slave: %p\n", slave);
 364         return ECORE_CALLBACK_CANCEL;
 365 }
 366
 367 HAPI int slave_rpc_async_request(struct slave_node *slave, const char *pkgname, struct packet *packet, void (*ret_cb)(struct slave_node *slave, const struct packet *packet, void *data), void *data, int urgent)
 368 {
 369         struct command *command;
 370         struct slave_rpc *rpc;
 371
 372         command = create_command(slave, pkgname, packet);
 373         if (!command) {
 374                 ErrPrint("Failed to create command\n");
 375
 376                 if (ret_cb)
 377                         ret_cb(slave, NULL, data);
 378
 379                 packet_unref(packet);
 380                 return LB_STATUS_ERROR_MEMORY;
 381         }
 382
 383         command->ret_cb = ret_cb;
 384         command->cbdata = data;
 385         packet_unref(packet);
 386
 387         rpc = slave_data(slave, "rpc");
 388         if (!rpc) {
 389                 ErrPrint("Slave has no RPC\n");
 390                 if (ret_cb)
 391                         ret_cb(slave, NULL, data);
 392                 destroy_command(command);
 393                 return LB_STATUS_ERROR_FAULT;
 394         }
 395
 396         if (rpc->handle < 0) {
 397                 DbgPrint("RPC handle is not ready to use it\n");
 398                 if (slave_is_secured(slave) && !slave_is_activated(slave)) {
 399                         int ret;
 400                         DbgPrint("Activate slave forcely\n");
 401                         ret = slave_activate(slave);
 402                         if (ret < 0 && ret != LB_STATUS_ERROR_ALREADY) {
 403
 404                                 if (ret_cb)
 405                                         ret_cb(slave, NULL, data);
 406
 407                                 destroy_command(command);
 408                                 return ret;
 409                         }
 410                 }
 411
 412                 if (urgent)
 413                         rpc->pending_list = eina_list_prepend(rpc->pending_list, command);
 414                 else
 415                         rpc->pending_list = eina_list_append(rpc->pending_list, command);
 416
 417                 return LB_STATUS_SUCCESS;
 418         }
 419
 420         if (urgent)
 421                 prepend_command(command);
 422         else
 423                 push_command(command);
 424
 425         return LB_STATUS_SUCCESS;
 426 }
 427
 428 HAPI int slave_rpc_request_only(struct slave_node *slave, const char *pkgname, struct packet *packet, int urgent)
 429 {
 430         struct command *command;
 431         struct slave_rpc *rpc;
 432
 433         command = create_command(slave, pkgname, packet);
 434         if (!command) {
 435                 ErrPrint("Failed to create a command\n");
 436                 packet_unref(packet);
 437                 return LB_STATUS_ERROR_MEMORY;
 438         }
 439
 440         command->ret_cb = NULL;
 441         command->cbdata = NULL;
 442         packet_unref(packet);
 443
 444         rpc = slave_data(slave, "rpc");
 445         if (!rpc) {
 446                 ErrPrint("Slave has no RPC\n");
 447                 destroy_command(command);
 448                 return LB_STATUS_ERROR_FAULT;
 449         }
 450
 451         if (rpc->handle < 0) {
 452                 DbgPrint("RPC handle is not ready to use it\n");
 453
 454                 if (slave_is_secured(slave) && !slave_is_activated(slave)) {
 455                         int ret;
 456
 457                         DbgPrint("Activate slave forcely\n");
 458                         ret = slave_activate(slave);
 459                         if (ret < 0 && ret != LB_STATUS_ERROR_ALREADY) {
 460                                 destroy_command(command);
 461                                 return ret;
 462                         }
 463                 }
 464
 465                 if (urgent)
 466                         rpc->pending_list = eina_list_prepend(rpc->pending_list, command);
 467                 else
 468                         rpc->pending_list = eina_list_append(rpc->pending_list, command);
 469
 470                 return LB_STATUS_SUCCESS;
 471         }
 472
 473         if (urgent)
 474                 prepend_command(command);
 475         else
 476                 push_command(command);
 477
 478         return LB_STATUS_SUCCESS;
 479 }
 480
 481 HAPI int slave_rpc_update_handle(struct slave_node *slave, int handle)
 482 {
 483         struct slave_rpc *rpc;
 484         struct command *command;
 485
 486         rpc = slave_data(slave, "rpc");
 487         if (!rpc)
 488                 return LB_STATUS_ERROR_INVALID;
 489
 490         DbgPrint("SLAVE: New handle assigned for %d, %d\n", slave_pid(slave), handle);
 491         rpc->handle = handle;
 492         if (rpc->pong_timer)
 493                 ecore_timer_del(rpc->pong_timer);
 494
 495         rpc->pong_timer = ecore_timer_add(DEFAULT_PING_TIME, ping_timeout_cb, slave);
 496         if (!rpc->pong_timer)
 497                 ErrPrint("Failed to add ping timer\n");
 498
 499         /*!
 500          * \note
 501          * slave_activated will call the activated callback.
 502          * activated callback will try to recover the normal instances state.
 503          * so the reset_fault should be called after slave_activated function.
 504          */
 505         slave_activated(slave);
 506
 507         EINA_LIST_FREE(rpc->pending_list, command) {
 508                 push_command(command);
 509         }
 510
 511         return LB_STATUS_SUCCESS;
 512 }
 513
 514 HAPI int slave_rpc_init(struct slave_node *slave)
 515 {
 516         struct slave_rpc *rpc;
 517
 518         rpc = calloc(1, sizeof(*rpc));
 519         if (!rpc) {
 520                 ErrPrint("Heap: %s\n", strerror(errno));
 521                 return LB_STATUS_ERROR_MEMORY;
 522         }
 523
 524         if (slave_set_data(slave, "rpc", rpc) < 0) {
 525                 DbgFree(rpc);
 526                 return LB_STATUS_ERROR_MEMORY;
 527         }
 528
 529         slave_event_callback_add(slave, SLAVE_EVENT_DEACTIVATE, slave_deactivate_cb, NULL);
 530
 531         rpc->ping_count = 0;
 532         rpc->next_ping_count = 1;
 533         rpc->handle = -1;
 534
 535         return LB_STATUS_SUCCESS;
 536 }
 537
 538 HAPI int slave_rpc_fini(struct slave_node *slave)
 539 {
 540         struct slave_rpc *rpc;
 541
 542         rpc = slave_del_data(slave, "rpc");
 543         if (!rpc)
 544                 return LB_STATUS_ERROR_INVALID;
 545
 546         slave_event_callback_del(slave, SLAVE_EVENT_DEACTIVATE, slave_deactivate_cb, NULL);
 547
 548         if (rpc->pong_timer)
 549                 ecore_timer_del(rpc->pong_timer);
 550
 551         DbgFree(rpc);
 552         return LB_STATUS_SUCCESS;
 553 }
 554
 555 HAPI int slave_rpc_ping(struct slave_node *slave)
 556 {
 557         struct slave_rpc *rpc;
 558
 559         rpc = slave_data(slave, "rpc");
 560         if (!rpc) {
 561                 ErrPrint("Slave RPC is not valid\n");
 562                 return LB_STATUS_ERROR_INVALID;
 563         }
 564
 565         if (!slave_is_activated(slave)) {
 566                 ErrPrint("Slave is not activated\n");
 567                 return LB_STATUS_ERROR_FAULT;
 568         }
 569
 570         rpc->ping_count++;
 571         if (rpc->ping_count != rpc->next_ping_count) {
 572                 ErrPrint("Ping count is not correct\n");
 573                 rpc->next_ping_count = rpc->ping_count;
 574         }
 575         rpc->next_ping_count++;
 576
 577         ecore_timer_reset(rpc->pong_timer);
 578         return LB_STATUS_SUCCESS;
 579 }
 580
 581 HAPI int slave_rpc_ping_freeze(struct slave_node *slave)
 582 {
 583         struct slave_rpc *rpc;
 584
 585         rpc = slave_data(slave, "rpc");
 586         if (!rpc) {
 587                 ErrPrint("Slave RPC is not valid\n");
 588                 return LB_STATUS_ERROR_INVALID;
 589         }
 590
 591         if (!slave_is_activated(slave)) {
 592                 ErrPrint("Slave is not activated\n");
 593                 return LB_STATUS_ERROR_FAULT;
 594         }
 595
 596         ecore_timer_freeze(rpc->pong_timer);
 597         return LB_STATUS_SUCCESS;
 598 }
 599
 600 HAPI int slave_rpc_ping_thaw(struct slave_node *slave)
 601 {
 602         struct slave_rpc *rpc;
 603
 604         rpc = slave_data(slave, "rpc");
 605         if (!rpc) {
 606                 ErrPrint("Slave RPC is not valid\n");
 607                 return LB_STATUS_ERROR_INVALID;
 608         }
 609
 610         if (!slave_is_activated(slave)) {
 611                 ErrPrint("Slave is not activated\n");
 612                 return LB_STATUS_ERROR_FAULT;
 613         }
 614
 615         ecore_timer_thaw(rpc->pong_timer);
 616         return LB_STATUS_SUCCESS;
 617 }
 618
 619 HAPI void slave_rpc_request_update(const char *pkgname, const char *id, const char *cluster, const char *category)
 620 {
 621         struct slave_node *slave;
 622         struct pkg_info *info;
 623         struct packet *packet;
 624
 625         info = package_find(pkgname);
 626         if (!info) {
 627                 ErrPrint("Failed to find a package\n");
 628                 return;
 629         }
 630
 631         slave = package_slave(info);
 632         if (!slave) {
 633                 ErrPrint("Failed to find a slave for %s\n", pkgname);
 634                 return;
 635         }
 636
 637         packet = packet_create_noack("update_content", "ssss", pkgname, id, cluster, category);
 638         if (!packet) {
 639                 ErrPrint("Failed to create a new param\n");
 640                 return;
 641         }
 642
 643         (void)slave_rpc_request_only(slave, pkgname, packet, 0);
 644 }
 645
 646 HAPI int slave_rpc_handle(struct slave_node *slave)
 647 {
 648         struct slave_rpc *rpc;
 649
 650         rpc = slave_data(slave, "rpc");
 651         if (!rpc) {
 652                 DbgPrint("Slave RPC is not initiated\n");
 653                 return -EINVAL;
 654         }
 655
 656         return rpc->handle;
 657 }
 658
 659 /* End of a file */