TAO/orbsvcs/ImplRepo_Service/Locator_Repository.cpp

   1 #include "orbsvcs/Log_Macros.h"
   2 #include "Locator_Repository.h"
   3 #include "ImR_Locator_i.h"
   4
   5 #include "utils.h"
   6 #include "tao/ORB_Core.h"
   7 #include "tao/default_ports.h"
   8 #include "ace/Read_Buffer.h"
   9 #include "ace/OS_NS_stdio.h"
  10 #include "ace/OS_NS_strings.h"
  11 #include "ace/OS_NS_ctype.h"
  12 #include "ace/OS_NS_unistd.h"
  13 #include "ace/Vector_T.h"
  14
  15 Locator_Repository::Locator_Repository (const Options& opts,
  16                                         CORBA::ORB_ptr orb)
  17 : opts_ (opts),
  18   orb_(CORBA::ORB::_duplicate(orb)),
  19   registered_(false)
  20 {
  21 }
  22
  23 Locator_Repository::~Locator_Repository ()
  24 {
  25   teardown_multicast();
  26 }
  27
  28 void
  29 Locator_Repository::shutdown ()
  30 {
  31   // default impl - no op
  32 }
  33
  34 int
  35 Locator_Repository::init (PortableServer::POA_ptr root_poa,
  36                           PortableServer::POA_ptr imr_poa,
  37                           const char* this_ior)
  38 {
  39   this->imr_ior_ = this_ior;
  40   int err = init_repo(imr_poa);
  41   if (err != 0)
  42     {
  43       return err;
  44     }
  45
  46   // Activate the two poa managers
  47   PortableServer::POAManager_var poaman =
  48     root_poa->the_POAManager ();
  49   poaman->activate ();
  50   poaman = imr_poa->the_POAManager ();
  51   poaman->activate ();
  52
  53   return err;
  54 }
  55
  56 int
  57 Locator_Repository::report_ior (PortableServer::POA_ptr )
  58 {
  59   if (this->registered_)
  60     {
  61       ORBSVCS_ERROR_RETURN ((LM_ERROR,
  62         ACE_TEXT ("ERROR: Repository already reported IOR\n")), -1);
  63     }
  64
  65   if (this->opts_.debug () > 0)
  66     {
  67       ORBSVCS_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) ImR: report_ior <%C>\n"),
  68         this->imr_ior_.in ()));
  69     }
  70
  71   // Register the ImR for use with INS
  72   CORBA::Object_var obj = this->orb_->resolve_initial_references ("AsyncIORTable");
  73   IORTable::Table_var ior_table = IORTable::Table::_narrow (obj.in ());
  74   ACE_ASSERT (! CORBA::is_nil (ior_table.in ()));
  75
  76   ior_table->rebind ("ImplRepoService", this->imr_ior_.in());
  77   ior_table->rebind ("ImR", this->imr_ior_.in());
  78
  79   // Set up multicast support (if enabled)
  80   if (this->opts_.multicast ())
  81     {
  82       ACE_Reactor* reactor = this->orb_->orb_core ()->reactor ();
  83       if (this->setup_multicast (reactor, this->imr_ior_.in ()) != 0)
  84         return -1;
  85     }
  86
  87   // We write the ior file last so that the tests can know we are ready.
  88   if (this->opts_.ior_filename ().length () > 0)
  89     {
  90       FILE* orig_fp = ACE_OS::fopen(this->opts_.ior_filename ().c_str(),
  91                                     ACE_TEXT("r"));
  92
  93       bool write_data = true;
  94       if (orig_fp != 0)
  95         {
  96           ACE_Read_Buffer reader (orig_fp, false);
  97
  98           char* string = reader.read ();
  99
 100           if (string != 0)
 101             {
 102               write_data =
 103                 (ACE_OS::strcasecmp (string, this->imr_ior_.in ()) != 0);
 104               reader.alloc ()->free (string);
 105             }
 106           ACE_OS::fclose (orig_fp);
 107         }
 108
 109       if (write_data)
 110         {
 111           FILE* fp = ACE_OS::fopen (this->opts_.ior_filename ().c_str (),
 112                                     ACE_TEXT("w"));
 113           if (fp == 0)
 114             {
 115               ORBSVCS_ERROR_RETURN ((LM_ERROR,
 116                 ACE_TEXT ("(%P|%t) ImR: Could not open file: %s\n"),
 117                 this->opts_.ior_filename ().c_str ()), -1);
 118             }
 119           ACE_OS::fprintf (fp, "%s", this->imr_ior_.in ());
 120           ACE_OS::fclose (fp);
 121         }
 122     }
 123
 124   registered_ = true;
 125
 126   return 0;
 127 }
 128
 129 int
 130 Locator_Repository::recover_ior ()
 131 {
 132   if (this->registered_)
 133     {
 134       ORBSVCS_ERROR_RETURN ((LM_ERROR,
 135          ACE_TEXT ("ERROR: Repository already registered IOR. ")
 136          ACE_TEXT ("recover_ior should not be called.\n")), -1);
 137     }
 138
 139   if (this->opts_.debug () > 0)
 140     {
 141       ORBSVCS_DEBUG ((LM_INFO, ACE_TEXT ("recover_ior <%C>\n"),
 142                   this->opts_.ior_filename ().c_str()));
 143     }
 144
 145   // Load the IOR from the specified file if it is available.
 146   const ACE_TString& combined_ior_file = this->opts_.ior_filename ();
 147
 148   // Check if the file exists. If not, then return 1 indicating
 149   // we cannot recover our state.
 150   if (ACE_OS::access (combined_ior_file.c_str (), F_OK) != 0)
 151     return -1;
 152
 153   try {
 154     ACE_TString combined_ior = ACE_TEXT ("file://") + combined_ior_file;
 155
 156     CORBA::Object_var combined_obj =
 157       this->orb_->string_to_object (combined_ior.c_str());
 158
 159     if (!CORBA::is_nil (combined_obj.in ()))
 160       {
 161         // Convert the object back into an IOR string to store in the
 162         // imr_ior_ attribute.
 163         this->imr_ior_ = this->orb_->object_to_string (combined_obj.in ());
 164       }
 165   }
 166   catch (const CORBA::Exception& ex)
 167     {
 168       ex._tao_print_exception ("ImR: Attempting to read combined_ior for ImR_Locator\n");
 169       return -1;
 170     }
 171
 172   // Register the ImR for use with INS
 173   CORBA::Object_var obj = this->orb_->resolve_initial_references ("AsyncIORTable");
 174   IORTable::Table_var ior_table = IORTable::Table::_narrow (obj.in ());
 175   ACE_ASSERT (! CORBA::is_nil (ior_table.in ()));
 176
 177   ior_table->bind ("ImplRepoService", this->imr_ior_.in());
 178   ior_table->bind ("ImR", this->imr_ior_.in());
 179
 180   // Set up multicast support (if enabled)
 181   if (this->opts_.multicast ())
 182     {
 183       ACE_Reactor* reactor = this->orb_->orb_core ()->reactor ();
 184       if (this->setup_multicast (reactor, this->imr_ior_.in ()) != 0)
 185         return -1;
 186     }
 187
 188   registered_ = true;
 189
 190   return 0;
 191 }
 192
 193 int
 194 Locator_Repository::setup_multicast (ACE_Reactor* reactor, const char* ior)
 195 {
 196   ACE_ASSERT (reactor != 0);
 197 #if defined (ACE_HAS_IP_MULTICAST)
 198
 199   TAO_ORB_Core* core = TAO_ORB_Core_instance ();
 200   // See if the -ORBMulticastDiscoveryEndpoint option was specified.
 201   ACE_CString mde (core->orb_params ()->mcast_discovery_endpoint ());
 202
 203   if (mde.length () != 0)
 204     {
 205       if (this->ior_multicast_.init (ior,
 206                                      mde.c_str (), TAO_SERVICEID_IMPLREPOSERVICE) == -1)
 207         {
 208           return -1;
 209         }
 210     }
 211   else
 212     {
 213       // Port can be specified as param, env var, or default
 214       CORBA::UShort port =
 215         core->orb_params ()->service_port (TAO::MCAST_IMPLREPOSERVICE);
 216       if (port == 0)
 217         {
 218           // Check environment var. for multicast port.
 219           const char* port_number = ACE_OS::getenv ("ImplRepoServicePort");
 220
 221           if (port_number != 0)
 222             port = static_cast<CORBA::UShort> (ACE_OS::atoi (port_number));
 223         }
 224       if (port == 0)
 225         port = TAO_DEFAULT_IMPLREPO_SERVER_REQUEST_PORT;
 226
 227       if (this->ior_multicast_.init (ior, port,
 228                                      ACE_DEFAULT_MULTICAST_ADDR, TAO_SERVICEID_IMPLREPOSERVICE) == -1)
 229         {
 230           return -1;
 231         }
 232     }
 233
 234   // Register event handler for the ior multicast.
 235   if (reactor->register_handler (&this->ior_multicast_,
 236                                  ACE_Event_Handler::READ_MASK) == -1)
 237     {
 238       if (this->opts_.debug() > 0)
 239         ORBSVCS_DEBUG ((LM_DEBUG, "ImR: cannot register Event handler\n"));
 240       return -1;
 241     }
 242 #else /* ACE_HAS_IP_MULTICAST*/
 243   ACE_UNUSED_ARG (reactor);
 244   ACE_UNUSED_ARG (ior);
 245 #endif /* ACE_HAS_IP_MULTICAST*/
 246   return 0;
 247 }
 248
 249 void
 250 Locator_Repository::teardown_multicast ()
 251 {
 252   ACE_Reactor* r = ior_multicast_.reactor ();
 253   if (r != 0) {
 254     r->remove_handler (&ior_multicast_, ACE_Event_Handler::READ_MASK);
 255     ior_multicast_.reactor (0);
 256   }
 257 }
 258
 259 bool
 260 Locator_Repository::multicast() const
 261 {
 262   return this->ior_multicast_.reactor () != 0;
 263 }
 264
 265 ACE_CString
 266 Locator_Repository::lcase (const ACE_CString& s)
 267 {
 268   ACE_CString ret(s);
 269   for (size_t i = 0; i < ret.length (); ++i)
 270     {
 271       ret[i] = static_cast<char> (ACE_OS::ace_tolower (s[i]));
 272     }
 273   return ret;
 274 }
 275
 276 int
 277 Locator_Repository::unregister_if_address_reused (const ACE_CString& fqname,
 278                                                   const char* partial_ior,
 279                                                   ImR_Locator_i* imr_locator)
 280
 281 {
 282   if (this->opts_.debug() > 0)
 283   {
 284     ORBSVCS_DEBUG ((LM_DEBUG,
 285                     ACE_TEXT ("(%P|%t) ImR: checking reuse address ")
 286                     ACE_TEXT ("for server <%C> ior <%C>\n"),
 287                     fqname.c_str(),
 288                     partial_ior));
 289   }
 290
 291   ACE_CString key;
 292   Server_Info_Ptr si;
 293
 294   Server_Info::fqname_to_key (fqname.c_str(), key);
 295   servers ().find (key, si);
 296   ACE_CString poa_name;
 297   ACE_CString server_id;
 298   if (si.null())
 299     {
 300       Server_Info::parse_id (fqname.c_str(), server_id, poa_name);
 301     }
 302   else
 303     {
 304       server_id = si->active_info ()->server_id;
 305       poa_name = si->active_info ()->poa_name;
 306     }
 307
 308   Locator_Repository::SIMap::ENTRY* sientry = 0;
 309   Locator_Repository::SIMap::ITERATOR siit (servers ());
 310   for (; siit.next (sientry); siit.advance() )
 311   {
 312     Server_Info *info = sientry->int_id_->active_info ();
 313
 314     if (this->opts_.debug() > 0)
 315     {
 316       ORBSVCS_DEBUG ((LM_DEBUG,
 317                       ACE_TEXT ("(%P|%t) ImR: iterating - registered server")
 318                       ACE_TEXT ("<%C:%C> key <%C> ior <%C>\n"), info->server_id.c_str(),
 319                       info->poa_name.c_str (), info->key_name_.c_str(), info->partial_ior.c_str ()));
 320     }
 321     bool same_server = info->server_id == server_id;
 322     if (same_server && (server_id.length () == 0))
 323       {
 324         same_server = info->poa_name == poa_name;
 325       }
 326     if (info->partial_ior == partial_ior && !same_server)
 327       {
 328         if (this->opts_.debug() > 0)
 329           {
 330             ORBSVCS_DEBUG ((LM_DEBUG,
 331                             ACE_TEXT ("(%P|%t) ImR: reuse address <%C> so remove server <%C>\n"),
 332                             info->partial_ior.c_str (), info->poa_name.c_str ()));
 333           }
 334         imr_locator->pinger ().remove_server (info->key_name_.c_str(), info->pid);
 335         AsyncAccessManager_ptr aam = imr_locator->find_aam (info->key_name_.c_str ());
 336         if (!aam.is_nil())
 337           {
 338             aam->server_is_shutting_down ();
 339           }
 340         info->reset_runtime ();
 341       }
 342   }
 343
 344   return 0;
 345 }
 346
 347 int
 348 Locator_Repository::add_server
 349   (const ACE_CString& fqname,
 350    const ImplementationRepository::StartupOptions & options)
 351 {
 352   Server_Info *si = 0;
 353   ACE_NEW_RETURN (si,
 354                   Server_Info (fqname,
 355                                options.activator.in (),
 356                                options.command_line.in (),
 357                                options.environment,
 358                                options.working_directory.in (),
 359                                options.activation,
 360                                options.start_limit),
 361                   -1);
 362   return this->add_server_i (si);
 363 }
 364
 365 int
 366 Locator_Repository::add_server
 367   (const ACE_CString& fqname,
 368    const ACE_CString& partial_ior,
 369    const ACE_CString& ior,
 370    ImplementationRepository::ServerObject_ptr srvobj)
 371 {
 372   Server_Info *si = 0;
 373   ACE_NEW_RETURN (si,
 374                   Server_Info (fqname, "", "",
 375                                ImplementationRepository::EnvironmentList (),
 376                                "", ImplementationRepository::NORMAL,
 377                                1, partial_ior, ior, srvobj),
 378                   -1);
 379   return this->add_server_i (si);
 380 }
 381
 382 int
 383 Locator_Repository::add_server_i (Server_Info *si)
 384 {
 385   Server_Info_Ptr info(si);
 386
 387   int err = sync_load ();
 388   if (err != 0)
 389     {
 390       return err;
 391     }
 392   err = servers ().bind (si->key_name_, info);
 393   if (err != 0)
 394     {
 395       return err;
 396     }
 397
 398   this->persistent_update(info, true);
 399   return 0;
 400 }
 401
 402 int
 403 Locator_Repository::add_activator (const ACE_CString& name,
 404                         const CORBA::Long token,
 405                         const ACE_CString& ior,
 406                         ImplementationRepository::Activator_ptr act)
 407 {
 408   int err = sync_load ();
 409   if (err != 0)
 410     {
 411       return err;
 412     }
 413
 414   Activator_Info_Ptr info (new Activator_Info (name, token, ior, act));
 415
 416   err = activators ().bind (lcase (name), info);
 417   if (err != 0)
 418     {
 419       return err;
 420     }
 421   this->persistent_update(info, true);
 422   return 0;
 423 }
 424
 425 int
 426 Locator_Repository::update_server (const Server_Info_Ptr& info)
 427 {
 428   return this->persistent_update(info, false);
 429 }
 430
 431 int
 432 Locator_Repository::update_activator (const Activator_Info_Ptr& info)
 433 {
 434   return this->persistent_update(info, false);
 435 }
 436
 437 void
 438 Locator_Repository::notify_remote_access (const char *,
 439                                           ImplementationRepository::AAM_Status)
 440 {
 441 }
 442
 443 Server_Info_Ptr
 444 Locator_Repository::find_by_poa (const ACE_CString & name)
 445 {
 446   Locator_Repository::SIMap::ENTRY* sientry = 0;
 447   Locator_Repository::SIMap::ITERATOR siit (servers ());
 448   for (; siit.next (sientry); siit.advance() )
 449   {
 450     Server_Info_Ptr& info = sientry->int_id_;
 451     if (info->poa_name == name)
 452       {
 453         return info;
 454       }
 455   }
 456   return Server_Info_Ptr();
 457 }
 458
 459 Server_Info_Ptr
 460 Locator_Repository::get_active_server (const ACE_CString& name, int pid)
 461 {
 462   sync_load ();
 463   ACE_CString key;
 464   Server_Info_Ptr si;
 465   if (name.length() == 0)
 466     {
 467       return si;
 468     }
 469   Server_Info::fqname_to_key (name.c_str(), key);
 470   servers ().find (key, si);
 471   if (si.null())
 472     {
 473       if (this->opts_.debug() > 5)
 474         {
 475           ORBSVCS_DEBUG ((LM_DEBUG,
 476                           ACE_TEXT ("(%P|%t) ImR: get_active_server could not find <%C>\n"),
 477                           name.c_str()));
 478         }
 479       si = find_by_poa (key);
 480       if (si.null())
 481         {
 482           if (name.find ("JACORB:") == ACE_CString::npos)
 483             {
 484               ACE_CString jo_key ("JACORB:");
 485               ACE_CString::size_type const pos = name.find (':');
 486               if (pos == ACE_CString::npos)
 487                 {
 488                   jo_key += name;
 489                 }
 490               else
 491                 {
 492                   jo_key += name.substring (0, pos);
 493                   jo_key += '/';
 494                   jo_key += name.substring (pos+1);
 495                 }
 496               return this->get_active_server (jo_key, pid);
 497             }
 498           else
 499             {
 500               return si;
 501             }
 502         }
 503     }
 504
 505   if (pid != 0 && si->pid != 0 && si->pid != pid)
 506     {
 507       if (this->opts_.debug() > 5)
 508         {
 509           ORBSVCS_DEBUG ((LM_DEBUG,
 510                           ACE_TEXT ("(%P|%t) ImR: get_active_server could not")
 511                           ACE_TEXT (" find <%C> pid <%d> != <%d>\n"),
 512                           name.c_str(), pid, si->pid));
 513         }
 514       si.reset ();
 515     }
 516   return si;
 517 }
 518
 519 int
 520 Locator_Repository::remove_server (const ACE_CString& name,
 521                                    ImR_Locator_i* imr_locator)
 522 {
 523   int const err = sync_load ();
 524   if (err != 0)
 525     {
 526       return err;
 527     }
 528   Server_Info_Ptr si;
 529   this->servers().find (name, si);
 530   int const ret = this->servers().unbind (name);
 531   if (ret != 0)
 532     {
 533       return ret;
 534     }
 535
 536   if (!si->alt_info_.null ())
 537     {
 538       // name is a peer to another and must be removed from other list
 539       bool found = false;
 540       for (CORBA::ULong i = 0; i < si->alt_info_->peers.length(); i++)
 541         {
 542           if (!found && si->poa_name == si->alt_info_->peers[i])
 543             {
 544               found = true;
 545               continue;
 546             }
 547           if (found)
 548             {
 549               si->alt_info_->peers[i-1] = si->alt_info_->peers[i];
 550             }
 551         }
 552       si->alt_info_->peers.length (si->alt_info_->peers.length() - 1);
 553     }
 554   else if (si->peers.length () > 0)
 555     {
 556       for (CORBA::ULong i = 0; i < si->peers.length(); i++)
 557         {
 558           ACE_CString key;
 559           ACE_CString const peer (si->peers[i]);
 560           Server_Info::gen_key (si->server_id, peer, key);
 561
 562           Server_Info_Ptr si2;
 563           this->servers().find (key, si2);
 564           imr_locator->destroy_poa (si2->poa_name);
 565           this->servers ().unbind (key);
 566           this->persistent_remove (key, false);
 567         }
 568     }
 569   return persistent_remove (name, false);
 570 }
 571
 572 int
 573 Locator_Repository::link_peers (Server_Info_Ptr base,
 574                                 const CORBA::StringSeq p)
 575 {
 576   sync_load ();
 577   CORBA::ULong const len = base->peers.length();
 578   base->peers.length (len + p.length());
 579   for (CORBA::ULong i = 0; i < p.length(); i++)
 580     {
 581       base->peers[len + i] =  p[i];
 582       Server_Info *si = 0;
 583       ACE_CString peer(p[i]);
 584       ACE_NEW_RETURN (si,
 585                       Server_Info (base->server_id, peer, base->is_jacorb, base),
 586                       -1);
 587       Server_Info_Ptr sip(si);
 588       servers ().bind (si->key_name_, sip);
 589       this->persistent_update (sip, true);
 590     }
 591
 592   this->persistent_update (base, true);
 593
 594   return 0;
 595 }
 596
 597 // -------------------------------------------------------------------------------------
 598
 599 Activator_Info_Ptr
 600 Locator_Repository::get_activator (const ACE_CString& name)
 601 {
 602   sync_load ();
 603
 604   Activator_Info_Ptr activator (0);
 605   activators ().find (lcase (name), activator);
 606   return activator;
 607 }
 608
 609 bool
 610 Locator_Repository::has_activator (const ACE_CString& name)
 611 {
 612   Activator_Info_Ptr activator (0);
 613   return activators().find (lcase (name), activator) == 0;
 614 }
 615
 616 int
 617 Locator_Repository::remove_activator (const ACE_CString& name)
 618 {
 619   int const err = sync_load ();
 620   if (err != 0)
 621     {
 622       return err;
 623     }
 624
 625   int const ret = activators().unbind (lcase(name));
 626   if (ret != 0)
 627     {
 628       return ret;
 629     }
 630
 631   Locator_Repository::SIMap::ENTRY* sientry = 0;
 632   Locator_Repository::SIMap::ITERATOR siit (servers ());
 633   for (; siit.next (sientry); siit.advance() )
 634   {
 635     Server_Info *info = sientry->int_id_->active_info ();
 636
 637     if (info->death_notify && info->activator == name)
 638       {
 639         info->death_notify = false;
 640       }
 641   }
 642
 643   return persistent_remove(name, true);
 644 }
 645
 646 Locator_Repository::SIMap&
 647 Locator_Repository::servers ()
 648 {
 649   return server_infos_;
 650 }
 651
 652 const Locator_Repository::SIMap&
 653 Locator_Repository::servers () const
 654 {
 655   return server_infos_;
 656 }
 657
 658 Locator_Repository::AIMap&
 659 Locator_Repository::activators ()
 660 {
 661   return activator_infos_;
 662 }
 663
 664 const Locator_Repository::AIMap&
 665 Locator_Repository::activators () const
 666 {
 667   return activator_infos_;
 668 }
 669
 670 int
 671 Locator_Repository::sync_load ()
 672 {
 673   // nothing more to do for default server/activator load
 674   return 0;
 675 }
 676
 677 bool
 678 Locator_Repository::registered () const
 679 {
 680   return this->registered_;
 681 }
 682
 683 //--------------------------------------------------------------------------
 684
 685 No_Backing_Store::No_Backing_Store (const Options& opts,
 686                                     CORBA::ORB_ptr orb)
 687   : Locator_Repository(opts, orb)
 688 {
 689 }
 690
 691 No_Backing_Store::~No_Backing_Store ()
 692 {
 693 }
 694
 695 const ACE_TCHAR*
 696 No_Backing_Store::repo_mode () const
 697 {
 698   return ACE_TEXT ("Disabled");
 699 }
 700
 701 int
 702 No_Backing_Store::init_repo (PortableServer::POA_ptr )
 703 {
 704   // nothing more to do for no backing store init
 705   return 0;
 706 }
 707
 708 int
 709 No_Backing_Store::persistent_update (const Server_Info_Ptr& , bool )
 710 {
 711   // nothing more to do for no backing store update
 712   return 0;
 713 }
 714
 715 int
 716 No_Backing_Store::persistent_update (const Activator_Info_Ptr& , bool )
 717 {
 718   // nothing more to do for no backing store update
 719   return 0;
 720 }
 721
 722 int
 723 No_Backing_Store::persistent_remove (const ACE_CString& , bool )
 724 {
 725   // nothing more to do for no backing store remove
 726   return 0;
 727 }