src/iop.c

   1 /**************************************************************************/
   2 /* ---------------------------------------------------------------------- */
   3 /* Copyright (C) 2006 - 2008 FRESCOR consortium partners:                 */
   4 /*                                                                        */
   5 /*   Universidad de Cantabria,              SPAIN                         */
   6 /*   University of York,                    UK                            */
   7 /*   Scuola Superiore Sant'Anna,            ITALY                         */
   8 /*   Kaiserslautern University,             GERMANY                       */
   9 /*   Univ. Politécnica  Valencia,           SPAIN                        */
  10 /*   Czech Technical University in Prague,  CZECH REPUBLIC                */
  11 /*   ENEA                                   SWEDEN                        */
  12 /*   Thales Communication S.A.              FRANCE                        */
  13 /*   Visual Tools S.A.                      SPAIN                         */
  14 /*   Rapita Systems Ltd                     UK                            */
  15 /*   Evidence                               ITALY                         */
  16 /*                                                                        */
  17 /*   See http://www.frescor.org for a link to partners' websites          */
  18 /*                                                                        */
  19 /*          FRESCOR project (FP6/2005/IST/5-034026) is funded             */
  20 /*       in part by the European Union Sixth Framework Programme          */
  21 /*       The European Union is not liable of any use that may be          */
  22 /*       made of this code.                                               */
  23 /*                                                                        */
  24 /*                                                                        */
  25 /*  This file is part of FORB (Frescor Object Request Broker)             */
  26 /*                                                                        */
  27 /* FORB is free software; you can redistribute it and/or modify it        */
  28 /* under terms of the GNU General Public License as published by the      */
  29 /* Free Software Foundation; either version 2, or (at your option) any    */
  30 /* later version.  FORB is distributed in the hope that it will be        */
  31 /* useful, but WITHOUT ANY WARRANTY; without even the implied warranty    */
  32 /* of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU    */
  33 /* General Public License for more details. You should have received a    */
  34 /* copy of the GNU General Public License along with FORB; see file       */
  35 /* COPYING. If not, write to the Free Software Foundation, 675 Mass Ave,  */
  36 /* Cambridge, MA 02139, USA.                                              */
  37 /*                                                                        */
  38 /* As a special exception, including FORB header files in a file,         */
  39 /* instantiating FORB generics or templates, or linking other files       */
  40 /* with FORB objects to produce an executable application, does not       */
  41 /* by itself cause the resulting executable application to be covered     */
  42 /* by the GNU General Public License. This exception does not             */
  43 /* however invalidate any other reasons why the executable file might be  */
  44 /* covered by the GNU Public License.                                     */
  45 /**************************************************************************/
  46
  47 /**
  48  * @file   iop.c
  49  * @author Michal Sojka <sojkam1@fel.cvut.cz>
  50  * @date   Mon Sep  1 21:51:58 2008
  51  *
  52  * @brief  Implementation of Inter-ORB protocol.
  53  *
  54  *
  55  */
  56 #include "iop.h"
  57 #include <forb/cdr.h>
  58 #include <forb/object.h>
  59 #include "proto.h"
  60 #include <ul_log.h>
  61 #include "exec_req.h"
  62 #include <errno.h>
  63 #include "peer.h"
  64 #include "discovery.h"
  65
  66 /** Version of the protocol */
  67 #define VER_MAJOR 0
  68 #define VER_MINOR 0
  69
  70 #define VER(major, minor) ((major)<<8 || (minor))
  71
  72 extern UL_LOG_CUST(ulogd_forb_iop);
  73
  74
  75 CORBA_boolean
  76 forb_iop_prepend_message_header(FORB_CDR_Codec *codec, forb_iop_message_type mt)
  77 {
  78         CORBA_boolean ret;
  79         forb_iop_message_header mh;
  80         mh.proto_version.major = VER_MAJOR;
  81         mh.proto_version.minor = VER_MINOR;
  82         mh.message_type = mt;
  83         mh.flags = (codec->data_endian == LittleEndian) ? forb_iop_LITTLE_ENDIAN : 0;
  84         mh.message_size = FORB_CDR_data_size(codec);
  85         ret = FORB_CDR_buffer_prepend(codec, forb_iop_MESSAGE_HEADER_SIZE);
  86         if (ret) {
  87                 ret = forb_iop_message_header_serialize(codec, &mh);
  88         }
  89         return ret;
  90 }
  91
  92 CORBA_boolean
  93 forb_iop_prepare_request(forb_request_t *req,
  94                          unsigned *index,
  95                          CORBA_Environment *env)
  96 {
  97         CORBA_boolean ret;
  98         forb_iop_request_header rh;
  99
 100         rh.request_id = req->request_id;
 101         rh.iface = req->interface;
 102         rh.objkey = forb_object_to_key(req->obj);
 103         rh.method_index = req->method_ind;
 104         rh.source = forb_object_to_forb(req->obj)->server_id;
 105         ret = forb_iop_request_header_serialize(&req->cdr_request, &rh);
 106         *index = req->cdr_request.wptr;
 107         if (ret) {
 108                 /* Request body is 8 byte aligned */
 109                 ret = FORB_CDR_put_align(&req->cdr_request, 8);
 110                 char str[50];
 111                 ul_logdeb("preparing request: id=%d  dest=%s  iface=%s method=%d\n", req->request_id,
 112                           forb_server_id_to_string(str, &req->obj->server, sizeof(str)),
 113                           rh.iface, rh.method_index);
 114         }
 115         return ret;
 116 }
 117
 118 static CORBA_boolean
 119 forb_iop_prepare_hello(FORB_CDR_Codec *codec,
 120                        const forb_server_id *server_id,
 121                        const void *src_addr,
 122                        CORBA_boolean (*serialize_addr)(FORB_CDR_Codec *codec, const void *addr),
 123                        const CORBA_char *orb_id)
 124 {
 125         if (!forb_server_id_serialize(codec, server_id)) return CORBA_FALSE;
 126         if (serialize_addr) {
 127                 if (!serialize_addr(codec, src_addr)) return CORBA_FALSE;
 128         }
 129         if (!CORBA_string_serialize(codec, &orb_id)) return CORBA_FALSE;
 130         /* All headers must be 8 byte aligned so align the length of
 131          * this message */
 132         if (!FORB_CDR_put_align(codec, 8)) return CORBA_FALSE;
 133         if (!forb_iop_prepend_message_header(codec, forb_iop_HELLO)) return CORBA_FALSE;
 134         return CORBA_TRUE;
 135 }
 136
 137 int forb_iop_send_hello_to(forb_peer_t *peer)
 138 {
 139         forb_port_t *port = peer->port;
 140         FORB_CDR_Codec codec;
 141         int ret;
 142         FORB_CDR_codec_init_static(&codec, port->forb->orb);
 143         if (!FORB_CDR_buffer_init(&codec, 1024, 0))
 144                 return -1;
 145         if (!FORB_CDR_buffer_reset(&codec, forb_iop_MESSAGE_HEADER_SIZE)) {
 146                 ret = -1;
 147                 goto free;
 148         }
 149         if (!forb_iop_prepare_hello(&codec, &port->forb->server_id,
 150                                     port->desc.addr,
 151                                     port->desc.proto->serialize_addr,
 152                                     port->forb->attr.orb_id)) {
 153                 ret = -1;
 154                 goto free;
 155         }
 156         ret = forb_proto_send(peer, &codec);
 157         if (ret > 0) ret = 0;
 158 free:
 159         FORB_CDR_codec_release_buffer(&codec);
 160         return ret;
 161 }
 162
 163 int
 164 forb_iop_redistribute_hello_to(forb_peer_t *dest, forb_peer_t *peer)
 165 {
 166         forb_port_t *port = peer->port;
 167         FORB_CDR_Codec codec;
 168         int ret;
 169         FORB_CDR_codec_init_static(&codec, port->forb->orb);
 170         if (!FORB_CDR_buffer_init(&codec, 1024, 0))
 171                 return -1;
 172         if (!FORB_CDR_buffer_reset(&codec, forb_iop_MESSAGE_HEADER_SIZE)) {
 173                 ret = -1;
 174                 goto free;
 175         }
 176         if (!forb_iop_prepare_hello(&codec, &peer->server_id,
 177                                     peer->addr,
 178                                     peer->port->desc.proto->serialize_addr,
 179                                     peer->orb_id)) {
 180                 ret = -1;
 181                 goto free;
 182         }
 183         ul_logdeb("redistributing hello of %s (%s) to %s (%s)\n",
 184                   ""/*TODO:id*/, peer->orb_id, "", dest->orb_id);
 185         ret = forb_proto_send(dest, &codec);
 186         if (ret > 0) ret = 0;
 187 free:
 188         FORB_CDR_codec_release_buffer(&codec);
 189         return ret;
 190 }
 191
 192 bool
 193 forb_iop_process_message_header(forb_iop_message_header *mh, FORB_CDR_Codec *codec)
 194 {
 195         /* FIXME: If we have multiple protocol versions, use different
 196          * type (independent from version) for return value instead of
 197          * mh */
 198         forb_iop_version_deserialize(codec, &mh->proto_version);
 199         switch (VER(mh->proto_version.major, mh->proto_version.minor)) {
 200                 case VER(VER_MAJOR, VER_MINOR):
 201                         forb_iop_message_type_deserialize(codec, &mh->message_type);
 202                         forb_iop_message_flags_deserialize(codec, &mh->flags);
 203                         /* Check whwther the type is meaningfull */
 204                         switch (mh->message_type) {
 205                                 case forb_iop_REQUEST:
 206                                 case forb_iop_REPLY:
 207                                 case forb_iop_HELLO:
 208                                         break;
 209                                 default:
 210                                         ul_logerr("rcvd wrong message type: %d\n",
 211                                                   mh->message_type);
 212                                         return false;
 213                         }
 214                         codec->data_endian = (mh->flags && forb_iop_LITTLE_ENDIAN) ?
 215                                 LittleEndian : BigEndian;
 216                         CORBA_unsigned_long_deserialize(codec, &mh->message_size);
 217                         return true;
 218                         break;
 219                 default:
 220                         ul_logerr("rcvd wrong protocol versio: %d.%d\n",
 221                                   mh->proto_version.major, mh->proto_version.minor);
 222                         return false;
 223         }
 224 }
 225
 226 static inline CORBA_boolean
 227 forb_exception_serialize(FORB_CDR_Codec *codec, struct forb_env *env)
 228 {
 229         return CORBA_long_serialize(codec, &env->major);
 230 }
 231
 232 static inline CORBA_boolean
 233 forb_exception_deserialize(FORB_CDR_Codec *codec, struct forb_env *env)
 234 {
 235         /* TODO: Declare exceptions in IDL and don't typecast here. */
 236         return CORBA_long_deserialize(codec, (CORBA_long*)&env->major);
 237 }
 238
 239 void
 240 forb_iop_send_reply(forb_t *forb,
 241            forb_server_id *dest,
 242            FORB_CDR_Codec *codec,
 243            CORBA_long request_id,
 244            struct forb_env *env)
 245 {
 246         forb_iop_reply_header reply_header;
 247         CORBA_boolean ret;
 248         forb_peer_t *peer;
 249         fosa_abs_time_t timeout;
 250
 251         reply_header.request_id = request_id;
 252         reply_header.flags = 0;
 253         if (forb_exception_occurred(env)) {
 254                 reply_header.flags |= forb_iop_FLAG_EXCEPTION;
 255                 FORB_CDR_buffer_reset(codec, forb_iop_MESSAGE_HEADER_SIZE +
 256                                              forb_iop_REPLY_HEADER_SIZE);
 257                 forb_exception_serialize(codec, env);
 258         }
 259         /* All headers must be 8 byte aligned so align the length of
 260          * this message */
 261         if (!FORB_CDR_put_align(codec, 8)) {
 262                 ul_logerr("Not enough space for tail align\n");
 263                 return;         /* FIXME: handle error (goto above)*/
 264         }
 265         /* forb_iop_REPLY_HEADER_SIZE equals to 8 even if the real
 266          * header is shorter. We want reply data to be 8 byte
 267          * aligned */
 268         ret = FORB_CDR_buffer_prepend(codec, forb_iop_REPLY_HEADER_SIZE);
 269         if (!ret) {
 270                 goto err;
 271         }
 272         ret = forb_iop_reply_header_serialize(codec, &reply_header);
 273         if (!ret) {
 274                 goto err;
 275         }
 276         forb_iop_prepend_message_header(codec, forb_iop_REPLY);
 277
 278         fosa_clock_get_time(FOSA_CLOCK_ABSOLUTE, &timeout);
 279         timeout = fosa_abs_time_incr(timeout,
 280                                      fosa_msec_to_rel_time(1000));
 281
 282         peer = forb_get_next_hop(forb, dest, &timeout);
 283         if (!peer) {
 284                 char str[60];
 285                 forb_server_id_to_string(str, dest, sizeof(str));
 286                 ul_logerr("Reply destination not found: %s\n", str);
 287                 goto err;
 288         } else {
 289                 char str[60];
 290                 forb_server_id_to_string(str, dest, sizeof(str));
 291                 ul_logdeb("sending reply: dest=%s, id=%u\n", str, request_id);
 292         }
 293         forb_proto_send(peer, codec);
 294         forb_peer_put(peer);
 295 err:
 296         ;
 297 }
 298
 299 static void
 300 process_request(forb_port_t *port, FORB_CDR_Codec *codec, uint32_t message_size)
 301 {
 302         forb_iop_request_header request_header;
 303         CORBA_boolean ret;
 304         forb_object obj;
 305         size_t n;
 306         forb_t *forb = port->forb;
 307         struct forb_env env;
 308         FORB_CDR_Codec reply_codec;
 309         forb_exec_req_t *exec_req;
 310         uint32_t req_size, data_size, header_size;
 311         char str[32];
 312         forb_peer_t *peer;
 313
 314         data_size = FORB_CDR_data_size(codec);
 315         ret = forb_iop_request_header_deserialize(codec, &request_header);
 316         if (!ret) {
 317                 ul_logerr("Malformed request recevied\n");
 318                 env.major = FORB_EX_COMM_FAILURE;
 319                 goto out;
 320         }
 321         ret = FORB_CDR_get_align(codec, 8);
 322         if (!ret) {
 323                 ul_logerr("Malformed request recevied\n");
 324                 env.major = FORB_EX_COMM_FAILURE;
 325                 goto out;
 326         }
 327
 328         header_size = data_size - FORB_CDR_data_size(codec);
 329         req_size = message_size - header_size;
 330
 331         ul_logdeb("rcvd request: src=%s, id=%u, iface=%s, method=%hd\n",
 332                   forb_server_id_to_string(str, &request_header.source, sizeof(str)),
 333                   request_header.request_id,
 334                   request_header.iface,
 335                   request_header.method_index);
 336
 337         if (port->new_peer) {
 338                 /* Request from a new peer was reported by the underlaying protocol */
 339                 peer = forb_peer_find(port->forb, &request_header.source);
 340                 if (peer) {
 341                         /* We already know this peer */
 342                         /* TODO: Can it be in FORB_PEER_WANTED state?
 343                          * If yes, we cannot simply igore this. */
 344                         ul_logmsg("new_peer was already known\n");
 345                         forb_peer_put(port->new_peer);
 346                         port->new_peer = NULL;
 347                 } else {
 348                         ul_logdeb("discovered new_peer from incomming connection\n");
 349                         peer = port->new_peer;
 350                         port->new_peer = NULL;
 351                         forb_new_peer_discovered(port, peer, request_header.source,
 352                                                  peer->addr, peer->orb_id);
 353                 }
 354
 355         }
 356
 357         obj = forb_key_to_object(forb, request_header.objkey);
 358         if (!obj) {
 359                 ul_logerr("Nonexistent object key\n");
 360                 env.major = FORB_EX_OBJECT_NOT_EXIST;
 361                 goto send_execption;
 362         }
 363
 364         n = strlen(request_header.iface);
 365         ret = strncmp(request_header.iface, obj->interface->name, n);
 366         forb_free(request_header.iface);
 367         request_header.iface = NULL;
 368         if (ret != 0) {
 369                 env.major = FORB_EX_INV_OBJREF;
 370                 ul_logerr("Object reference has incorrect type\n");
 371                 goto send_execption;
 372         }
 373
 374         if (request_header.method_index >= obj->interface->num_methods) {
 375                 env.major = FORB_EX_INV_IDENT;
 376                 ul_logerr("To high method number\n");
 377                 goto send_execption;
 378         }
 379
 380         if (!obj->executor) {
 381                 env.major = FORB_EX_NO_IMPLEMENT;
 382                 ul_logerr("No executor for object\n");
 383                 goto send_execption;
 384
 385         }
 386
 387         /* Enqueue execution request */
 388         exec_req = forb_malloc(sizeof(*exec_req));
 389         if (exec_req) {
 390                 memset(exec_req, 0, sizeof(exec_req));
 391                 exec_req->request_type = FORB_EXEC_REQ_REMOTE;
 392                 exec_req->request_id = request_header.request_id;
 393                 exec_req->source = request_header.source;
 394                 exec_req->obj = obj;
 395                 exec_req->method_index = request_header.method_index;
 396                 /* Copy the request to exec_req */
 397                 FORB_CDR_codec_init_static(&exec_req->codec, codec->orb);
 398                 ret = FORB_CDR_buffer_init(&exec_req->codec,
 399                                       req_size,
 400                                       0);
 401                 if (!ret) {
 402                         env.major = FORB_EX_NO_MEMORY;
 403                         ul_logerr("No memory for executor request bufer of size %d (header_size=%d)\n",
 404                                   req_size, header_size);
 405                         goto send_execption;
 406                 }
 407                 exec_req->codec.data_endian = codec->data_endian;
 408                 FORB_CDR_buffer_gets(codec, exec_req->codec.buffer, req_size);
 409                 /* TODO: Use better data structure for incomming
 410                    buffer to achieve zero-copy behaviour. */
 411                 forb_exec_req_ins_tail(obj->executor, exec_req);
 412         } else {
 413                 env.major = FORB_EX_NO_MEMORY;
 414                 ul_logerr("No memory for executor request\n");
 415                 goto send_execption;
 416         }
 417
 418 out:
 419         return;
 420
 421 send_execption:
 422         FORB_CDR_codec_init_static(&reply_codec, codec->orb);
 423         ret = FORB_CDR_buffer_init(&reply_codec, 4096,
 424                               forb_iop_MESSAGE_HEADER_SIZE +
 425                               forb_iop_REPLY_HEADER_SIZE);
 426         if (!ret) {
 427                 ul_logerr("No memory for exception reply buffer\n");
 428                 return;
 429         }
 430
 431         forb_iop_send_reply(port->forb, &request_header.source,
 432                             &reply_codec, request_header.request_id, &env);
 433         FORB_CDR_codec_release_buffer(&reply_codec);
 434         /* TODO: relese exec_req etc. */
 435 }
 436
 437 static void
 438 process_reply(forb_port_t *port, FORB_CDR_Codec *codec)
 439 {
 440         forb_iop_reply_header rh;
 441         forb_t *forb = port->forb;
 442         forb_request_t *req;
 443
 444         forb_iop_reply_header_deserialize(codec, &rh);
 445         /* Reply data are 8 byte aligned */
 446         FORB_CDR_get_align(codec, 8);
 447         ul_logdeb("rcvd reply: id=%u\n", rh.request_id);
 448         req = forb_request_find(forb, &rh.request_id);
 449         if (!req) {
 450                 ul_logerr("Received reply to unknown request_id %u\n", rh.request_id);
 451                 return;
 452         }
 453         forb_request_delete(forb, req); /* Deregister request from forb */
 454
 455         if (rh.flags & forb_iop_FLAG_EXCEPTION) {
 456                 forb_exception_deserialize(codec, req->env);
 457         } else {
 458                 req->cdr_reply = codec;
 459         }
 460
 461         /* Tell the stub where to signal that reply processing is
 462          * finished */
 463         req->reply_processed = &port->reply_processed;
 464
 465         /* Resume the stub waiting in forb_wait_for_reply() */
 466         forb_syncobj_signal(&req->reply_ready);
 467
 468         /* Wait for stub to process the results from the codec's buffer */
 469         forb_syncobj_wait(&port->reply_processed);
 470 }
 471
 472 /**
 473  * Process incomming HELLO messages.
 474  *
 475  * For every incomming HELLO message the peer table is searched
 476  * whether it already contains a record for that peer or not. If not,
 477  * the new peer is added to the table and another hello message is
 478  * sent so that the new peer discovers us quickly.
 479  *
 480  * @param port Port, where hello was received
 481  * @param codec Buffer with the hello message
 482  */
 483 static void
 484 process_hello(forb_port_t *port, FORB_CDR_Codec *codec)
 485 {
 486         forb_server_id server_id;
 487         void *addr = NULL;
 488         forb_peer_t *peer;
 489         forb_t *forb = port->forb;
 490         CORBA_string peer_orb_id = NULL;
 491
 492 /*      printf("Hello received at port %p\n", port); */
 493
 494         forb_server_id_deserialize(codec, &server_id);
 495         if (port->desc.proto->deserialize_addr) {
 496                 port->desc.proto->deserialize_addr(codec, &addr);
 497         }
 498         {
 499                 char str[60], addrstr[60];
 500                 if (port->desc.proto->addr2str) {
 501                         port->desc.proto->addr2str(addrstr, sizeof(addrstr), addr);
 502                 } else
 503                         addrstr[0] = 0;
 504                 ul_logdeb("rcvd hello from %s (addr %s)\n",
 505                           forb_server_id_to_string(str, &server_id, sizeof(str)),
 506                           addrstr);
 507         }
 508         CORBA_string_deserialize(codec, &peer_orb_id);
 509         if (forb_server_id_cmp(&server_id, &forb->server_id) != 0) {
 510                 peer = forb_peer_find(forb, &server_id);
 511                 if (peer && peer->state == FORB_PEER_DISCOVERED) {
 512                         /* TODO: Update last hello receive time */
 513                         if (port->new_peer) {
 514                                 ul_logdeb("peer already discovered but not connected - replacing\n");
 515                                 forb_peer_disconnected(peer);
 516                                 forb_peer_put(peer);
 517                                 peer = port->new_peer;
 518                                 port->new_peer = NULL;
 519                                 forb_new_peer_discovered(port, peer, server_id, addr, peer_orb_id);
 520                         } else {
 521                                 ul_logdeb("peer already discovered - ignoring\n");
 522                                 if (addr)
 523                                         forb_free(addr);
 524                                 if (peer_orb_id)
 525                                         forb_free(peer_orb_id);
 526                                 forb_peer_put(peer);
 527                         }
 528                 } else {
 529                         if (port->new_peer) {
 530                                 if (peer) {
 531                                         ul_logerr("Unahandled case - FORB_PEER_WANTED && port->new_peer\n");
 532                                         forb_peer_put(peer);
 533                                 }
 534                                 peer = port->new_peer;
 535                                 port->new_peer = NULL;
 536                         }
 537
 538                         forb_new_peer_discovered(port, peer, server_id, addr, peer_orb_id);
 539                 }
 540         }
 541 }
 542
 543 static void
 544 process_message(forb_port_t *port, const forb_iop_message_header *mh,
 545                 FORB_CDR_Codec *codec)
 546 {
 547         CORBA_long data_size = FORB_CDR_data_size(codec);
 548         /* TODO: Check destination address, whether the message is for
 549          * us or should be routed. */
 550
 551         /* TODO: If there is some processing error, skip the rest of
 552          * the message according mh.message_size. */
 553         switch (mh->message_type) {
 554                 case forb_iop_REQUEST:
 555                         process_request(port, codec, mh->message_size);
 556                         break;
 557                 case forb_iop_REPLY:
 558                         process_reply(port, codec);
 559                         break;
 560                 case forb_iop_HELLO:
 561                         process_hello(port, codec);
 562                         break;
 563                 default:
 564                         ul_logmsg("rcvd unknown message type\n");
 565                         break;
 566         }
 567         if (port->new_peer) {
 568                 /* If for some reaseon the new_peer was not processed so free it here. */
 569                 ul_logmsg("Forgotten new_peer\n");
 570                 forb_peer_put(port->new_peer);
 571                 port->new_peer = NULL;
 572         }
 573
 574         FORB_CDR_get_align(codec, 8);
 575
 576         if (FORB_CDR_data_size(codec) != data_size - mh->message_size) {
 577                 size_t processed = data_size - FORB_CDR_data_size(codec);
 578                 ul_logmsg("Message of type %d handled incorrectly (size=%d, processed=%zu); fixing\n",
 579                           mh->message_type, mh->message_size, processed);
 580                 ;
 581                 codec->rptr += mh->message_size - processed;
 582         }
 583 }
 584
 585 /**
 586  * Thread run for every port to receive FORB messages from that port.
 587  *
 588  * @param arg Pointer to ::forb_port_t typecasted to void *.
 589  *
 590  * @return Always NULL
 591  */
 592 void *forb_iop_receiver_thread(void *arg)
 593 {
 594         forb_port_t *port = arg;
 595         const forb_proto_t *proto = port->desc.proto;
 596         FORB_CDR_Codec *c = &port->codec;
 597         ssize_t rcvd;
 598         size_t len;
 599         forb_iop_message_header mh;
 600         bool header_received = false;
 601
 602         while (!port->finish) {
 603                 if (c->rptr == c->wptr) {
 604                         /* The buffer is empty now - start writing from begining*/
 605                         FORB_CDR_buffer_reset(c, 0);
 606                 }
 607                 /* TODO: If there is not enough space for reception,
 608                  * we should shift the already received data to the
 609                  * beginning of the buffer. */
 610                 rcvd = proto->recv(port,
 611                                    &c->buffer[c->wptr],
 612                                    c->wptr_max - c->wptr);
 613                 if (rcvd < 0) {
 614                         ul_logerr("recv returned error %zd (%s), exiting\n", rcvd, strerror(errno));
 615                         return NULL;
 616                 }
 617                 c->wptr += rcvd;
 618                 c->wptr_last = c->wptr;
 619
 620                 /* While there are some data in the buffer, process them. */
 621                 while (FORB_CDR_data_size(c) > 0) {
 622                         len = FORB_CDR_data_size(c);
 623                         /* Wait for and then process message header */
 624                         if (!header_received) {
 625                                 if (len >= forb_iop_MESSAGE_HEADER_SIZE) {
 626                                         if (c->rptr % 8 != 0) {
 627                                                 ul_logerr("Header doesn't start at 8 byte bounday\n");
 628                                         }
 629                                         header_received = forb_iop_process_message_header(&mh, c);
 630                                         if (!header_received) {
 631                                                 ul_logerr("Wrong header received\n");
 632                                                 /* TODO: We should probably reset the buffer here */
 633                                         }
 634                                         len = FORB_CDR_data_size(c);
 635                                 } else {
 636                                         break; /* Wait for more data to arrive*/
 637                                 }
 638                         }
 639
 640                         /* Wait for and then process the message body */
 641                         if (header_received) {
 642                                 if (len >= mh.message_size) {
 643                                         process_message(port, &mh, c);
 644                                         /* Wait for the next message */
 645                                         header_received = false;
 646                                 } else {
 647                                         break; /* Wait for more data to arrive*/
 648                                 }
 649                         }
 650                 }
 651         }
 652         return NULL;
 653 }
 654
 655 static void
 656 discovery_cleanup(void *codec)
 657 {
 658         FORB_CDR_codec_release_buffer((FORB_CDR_Codec*)codec);
 659         /* TODO: Broadcast some kind of bye bye message */
 660 }
 661
 662
 663 /**
 664  * Thread run for every port to broadcast HELLO messages. These
 665  * messages are used for a FORB to discover all peers (and in future
 666  * also to detect their disconnection).
 667  *
 668  * @param arg Pointer to ::forb_port_t typecasted to void *.
 669  *
 670  * @return Always NULL
 671  */
 672 void *forb_iop_discovery_thread(void *arg)
 673 {
 674         forb_port_t *port = arg;
 675         const forb_proto_t *proto = port->desc.proto;
 676         FORB_CDR_Codec codec;
 677         fosa_abs_time_t hello_time;
 678         fosa_rel_time_t hello_interval = fosa_msec_to_rel_time(1000*proto->hello_interval);
 679         int ret;
 680
 681         FORB_CDR_codec_init_static(&codec, port->forb->orb);
 682         FORB_CDR_buffer_init(&codec, 1024, 0);
 683
 684         pthread_cleanup_push(discovery_cleanup, &codec);
 685
 686         /* Next hello interval is now */
 687         fosa_clock_get_time(FOSA_CLOCK_ABSOLUTE, &hello_time);
 688
 689         while (!port->finish) {
 690                 /* Wait for next hello interval or until somebody
 691                  * signals us. */
 692                 ret = forb_syncobj_timedwait(&port->hello, &hello_time);
 693                 /* sem_timedwait would be more appropriate */
 694                 if (ret == FOSA_ETIMEDOUT) {
 695                         hello_time = fosa_abs_time_incr(hello_time, hello_interval);
 696                 } else if (ret != 0) {
 697                         ul_logerr("hello syncobj error: %s\n", strerror(ret));
 698                 }
 699
 700                 if (port->finish) break;
 701
 702                 FORB_CDR_buffer_reset(&codec, forb_iop_MESSAGE_HEADER_SIZE);
 703                 forb_iop_prepare_hello(&codec, &port->forb->server_id, port->desc.addr,
 704                                        proto->serialize_addr, port->forb->attr.orb_id);
 705 /*              printf("Broadcasting hello from port %p\n", port);  */
 706                 proto->broadcast(port, &codec.buffer[codec.rptr],
 707                                  FORB_CDR_data_size(&codec));
 708         }
 709
 710         pthread_cleanup_pop(1);
 711         return NULL;
 712 }
 713
 714 /**
 715  * Sends REQUEST message to another FORB.
 716  *
 717  * The request @a req has to be prepared by
 718  * forb_iop_prepare_request(). Then, this function adds a message
 719  * header, connects to the destination FORB and sends the request.
 720  *
 721  * If no exception is reported, then the caller must wait for response
 722  * by calling forb_wait_for_reply().
 723  *
 724  * @param req A request prepared by forb_iop_prepare_request()
 725  * @param env Environment for returning exceptions
 726  */
 727 void
 728 forb_request_send(forb_request_t *req, unsigned index, CORBA_Environment *env)
 729 {
 730         CORBA_boolean ret;
 731         forb_peer_t *peer;
 732         ssize_t size;
 733         size_t len;
 734         fosa_abs_time_t timeout;
 735         forb_t *forb = forb_object_to_forb(req->obj);
 736         forb_exec_req_t *exec_req;
 737
 738         if (!forb) {
 739                 env->major = FORB_EX_INTERNAL;
 740                 return;
 741         }
 742
 743         req->env = env;     /* Remember, where to return exceptions */
 744
 745         /* All headers must be 8 byte aligned so align the length of
 746          * this message */
 747         if (!FORB_CDR_put_align(&req->cdr_request, 8)) {
 748                 env->major = FORB_EX_INTERNAL;
 749                 ul_logerr("Not enough space for tail align\n");
 750                 return;
 751         }
 752
 753         /* Local invocation case, destination of a message is only
 754          * a different executor thread */
 755         if (forb_object_is_local(req->obj)) {
 756                 exec_req = forb_malloc(sizeof(*exec_req));
 757                 memset(exec_req, 0, sizeof(exec_req));
 758                 exec_req->request_type = FORB_EXEC_REQ_LOCAL;
 759                 exec_req->input_request = req;
 760                 exec_req->obj = req->obj;
 761                 exec_req->method_index = req->method_ind;
 762                 exec_req->interface = req->interface;
 763                 req->cdr_request.rptr = index;
 764                 exec_req->codec = req->cdr_request;
 765                 exec_req->request_id = req->request_id;
 766                 forb_exec_req_ins_tail(forb_object_get_executor(exec_req->obj), exec_req);
 767                 return;
 768         }
 769
 770         ret = forb_iop_prepend_message_header(&req->cdr_request, forb_iop_REQUEST);
 771         if (!ret) {
 772                 /* This should never happen */
 773                 env->major = FORB_EX_INTERNAL;
 774                 return;
 775         }
 776
 777         fosa_clock_get_time(FOSA_CLOCK_ABSOLUTE, &timeout);
 778         timeout = fosa_abs_time_incr(timeout,
 779                                      fosa_msec_to_rel_time(1000));
 780         peer = forb_get_next_hop(forb, &req->obj->server, &timeout);
 781         if (!peer) {
 782                 char str[50];
 783                 ul_logerr("Cannot find peer to send request for server %s\n",
 784                           forb_server_id_to_string(str, &req->obj->server, sizeof(str)));
 785                 env->major = FORB_EX_COMM_FAILURE;
 786                 return;
 787         }
 788         /* Register the request with forb so we can match incomming
 789          * reply to this request. */
 790         ret = forb_request_insert(forb, req);
 791         if (ret <= 0) {
 792                 ul_logerr("Insert request error %d\n", ret);
 793                 env->major = FORB_EX_INTERNAL;
 794                 goto err_peer_put;
 795         }
 796
 797         {
 798                 char str[50];
 799                 ul_logdeb("sending request: id=%d  dest=%s\n", req->request_id,
 800                           forb_server_id_to_string(str, &req->obj->server, sizeof(str)));
 801         }
 802         len = FORB_CDR_data_size(&req->cdr_request);
 803         fosa_mutex_lock(&peer->send_lock);
 804         size = forb_proto_send(peer, &req->cdr_request);
 805         fosa_mutex_unlock(&peer->send_lock);
 806         if (size <= 0 || size != len) {
 807                 env->major = FORB_EX_COMM_FAILURE;
 808                 /* Request is deleted when the stub calls forb_request_destroy() */
 809         }
 810  err_peer_put:
 811         forb_peer_put(peer);
 812 }