cpp/src/IceGrid/NodeI.cpp

   1 // **********************************************************************
   2 //
   3 // Copyright (c) 2003-2011 ZeroC, Inc. All rights reserved.
   4 //
   5 // This copy of Ice is licensed to you under the terms described in the
   6 // ICE_LICENSE file included in this distribution.
   7 //
   8 // **********************************************************************
   9
  10 #include <IceUtil/Timer.h>
  11 #include <IceUtil/FileUtil.h>
  12 #include <Ice/Ice.h>
  13 #include <IcePatch2/Util.h>
  14 #include <IcePatch2/ClientUtil.h>
  15 #include <IceGrid/NodeI.h>
  16 #include <IceGrid/Activator.h>
  17 #include <IceGrid/ServerI.h>
  18 #include <IceGrid/ServerAdapterI.h>
  19 #include <IceGrid/Util.h>
  20 #include <IceGrid/TraceLevels.h>
  21 #include <IceGrid/NodeSessionManager.h>
  22
  23 using namespace std;
  24 using namespace IcePatch2;
  25 using namespace IceGrid;
  26
  27 namespace
  28 {
  29
  30 class LogPatcherFeedback : public IcePatch2::PatcherFeedback
  31 {
  32 public:
  33
  34     LogPatcherFeedback(const TraceLevelsPtr& traceLevels, const string& dest) :
  35         _traceLevels(traceLevels),
  36         _startedPatch(false),
  37         _lastProgress(0),
  38         _dest(dest)
  39     {
  40     }
  41
  42     void
  43     setPatchingPath(const string& path)
  44     {
  45         _path = path;
  46         _startedPatch = false;
  47         _lastProgress = 0;
  48     }
  49
  50     virtual bool
  51     noFileSummary(const string& reason)
  52     {
  53         if(_traceLevels->patch > 0)
  54         {
  55             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
  56             out << _dest << ": can't load summary file (will perform a thorough patch):\n" << reason;
  57         }
  58         return true;
  59     }
  60
  61     virtual bool
  62     checksumStart()
  63     {
  64         if(_traceLevels->patch > 0)
  65         {
  66             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
  67             out << _dest << ": started checksum calculation";
  68         }
  69         return true;
  70     }
  71
  72     virtual bool
  73     checksumProgress(const string& path)
  74     {
  75         if(_traceLevels->patch > 2)
  76         {
  77             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
  78             out << _dest << ": calculating checksum for " << getBasename(path);
  79         }
  80         return true;
  81     }
  82
  83     virtual bool
  84     checksumEnd()
  85     {
  86         if(_traceLevels->patch > 0)
  87         {
  88             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
  89             out << _dest << ": finished checksum calculation";
  90         }
  91         return true;
  92     }
  93
  94     virtual bool
  95     fileListStart()
  96     {
  97         if(_traceLevels->patch > 0)
  98         {
  99             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
 100             out << _dest << ": getting list of file to patch";
 101         }
 102         return true;
 103     }
 104
 105     virtual bool
 106     fileListProgress(Ice::Int percent)
 107     {
 108         return true;
 109     }
 110
 111     virtual bool
 112     fileListEnd()
 113     {
 114         if(_traceLevels->patch > 0)
 115         {
 116             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
 117             out << _dest << ": getting list of file to patch completed";
 118         }
 119         return true;
 120     }
 121
 122     virtual bool
 123     patchStart(const string& path, Ice::Long size, Ice::Long totalProgress, Ice::Long totalSize)
 124     {
 125         if(_traceLevels->patch > 1 && totalSize > (1024 * 1024))
 126         {
 127             int progress = static_cast<int>(static_cast<double>(totalProgress) / totalSize * 100.0);
 128             progress /= 5;
 129             progress *= 5;
 130             if(progress != _lastProgress)
 131             {
 132                 _lastProgress = progress;
 133                 Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
 134                 out << _dest << ": downloaded " << progress << "% (" << totalProgress << '/' << totalSize << ')';
 135                 if(!_path.empty())
 136                 {
 137                     out << " of " << _path;
 138                 }
 139             }
 140         }
 141         else if(_traceLevels->patch > 0)
 142         {
 143             if(!_startedPatch)
 144             {
 145                 Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
 146                 int roundedSize = static_cast<int>(static_cast<double>(totalSize) / 1024);
 147                 if(roundedSize == 0 && totalSize > 0)
 148                 {
 149                     roundedSize = 1;
 150                 }
 151                 out << _dest << ": downloading " << (_path.empty() ? string("") : (_path + " ")) << roundedSize
 152                     << "KB ";
 153                 _startedPatch = true;
 154             }
 155         }
 156
 157         return true;
 158     }
 159
 160     virtual bool
 161     patchProgress(Ice::Long progress, Ice::Long size, Ice::Long totalProgress, Ice::Long totalSize)
 162     {
 163         return true;
 164     }
 165
 166     virtual bool
 167     patchEnd()
 168     {
 169         return true;
 170     }
 171
 172     void
 173     finishPatch()
 174     {
 175         if(_traceLevels->patch > 0)
 176         {
 177             Ice::Trace out(_traceLevels->logger, _traceLevels->patchCat);
 178             out << _dest << ": downloading completed";
 179         }
 180     }
 181
 182 private:
 183
 184     const TraceLevelsPtr _traceLevels;
 185     bool _startedPatch;
 186     int _lastProgress;
 187     string _path;
 188     string _dest;
 189 };
 190
 191 class NodeUp : public NodeI::Update, public AMI_NodeObserver_nodeUp
 192 {
 193 public:
 194
 195     NodeUp(const NodeIPtr& node, const NodeObserverPrx& observer, NodeDynamicInfo info) :
 196         NodeI::Update(node, observer), _info(info)
 197     {
 198     }
 199
 200     virtual bool
 201     send()
 202     {
 203         try
 204         {
 205             _observer->nodeUp_async(this, _info);
 206         }
 207         catch(const Ice::LocalException&)
 208         {
 209             return false;
 210         }
 211         return true;
 212     }
 213
 214     virtual void
 215     ice_response()
 216     {
 217         finished(true);
 218     }
 219
 220     virtual void
 221     ice_exception(const Ice::Exception&)
 222     {
 223         finished(false);
 224     }
 225
 226 private:
 227
 228     NodeDynamicInfo _info;
 229 };
 230
 231 class UpdateServer : public NodeI::Update, public AMI_NodeObserver_updateServer
 232 {
 233 public:
 234
 235     UpdateServer(const NodeIPtr& node, const NodeObserverPrx& observer, ServerDynamicInfo info) :
 236         NodeI::Update(node, observer), _info(info)
 237     {
 238     }
 239
 240     virtual bool
 241     send()
 242     {
 243         try
 244         {
 245             _observer->updateServer_async(this, _node->getName(), _info);
 246         }
 247         catch(const Ice::LocalException&)
 248         {
 249             return false;
 250         }
 251         return true;
 252     }
 253
 254     virtual void
 255     ice_response()
 256     {
 257         finished(true);
 258     }
 259
 260     virtual void
 261     ice_exception(const Ice::Exception&)
 262     {
 263         finished(false);
 264     }
 265
 266 private:
 267
 268     ServerDynamicInfo _info;
 269 };
 270
 271 class UpdateAdapter : public NodeI::Update, public AMI_NodeObserver_updateAdapter
 272 {
 273 public:
 274
 275     UpdateAdapter(const NodeIPtr& node, const NodeObserverPrx& observer, AdapterDynamicInfo info) :
 276         NodeI::Update(node, observer), _info(info)
 277     {
 278     }
 279
 280     virtual bool
 281     send()
 282     {
 283         try
 284         {
 285             _observer->updateAdapter_async(this, _node->getName(), _info);
 286         }
 287         catch(const Ice::LocalException&)
 288         {
 289             return false;
 290         }
 291         return true;
 292     }
 293
 294     virtual void
 295     ice_response()
 296     {
 297         finished(true);
 298     }
 299
 300     virtual void
 301     ice_exception(const Ice::Exception&)
 302     {
 303         finished(false);
 304     }
 305
 306 private:
 307
 308     AdapterDynamicInfo _info;
 309 };
 310
 311 }
 312
 313 NodeI::Update::Update(const NodeIPtr& node, const NodeObserverPrx& observer) : _node(node), _observer(observer)
 314 {
 315 }
 316
 317 NodeI::Update::~Update()
 318 {
 319 }
 320
 321 void
 322 NodeI::Update::finished(bool success)
 323 {
 324     _node->dequeueUpdate(_observer, this, !success);
 325 }
 326
 327 NodeI::NodeI(const Ice::ObjectAdapterPtr& adapter,
 328              NodeSessionManager& sessions,
 329              const ActivatorPtr& activator,
 330              const IceUtil::TimerPtr& timer,
 331              const TraceLevelsPtr& traceLevels,
 332              const NodePrx& proxy,
 333              const string& name,
 334              const UserAccountMapperPrx& mapper) :
 335     _communicator(adapter->getCommunicator()),
 336     _adapter(adapter),
 337     _sessions(sessions),
 338     _activator(activator),
 339     _timer(timer),
 340     _traceLevels(traceLevels),
 341     _name(name),
 342     _proxy(proxy),
 343     _redirectErrToOut(false),
 344     _allowEndpointsOverride(false),
 345     _waitTime(0),
 346     _userAccountMapper(mapper),
 347     _platform("IceGrid.Node", _communicator, _traceLevels),
 348     _fileCache(new FileCache(_communicator)),
 349     _serial(1),
 350     _consistencyCheckDone(false)
 351 {
 352     Ice::PropertiesPtr props = _communicator->getProperties();
 353
 354     const_cast<string&>(_dataDir) = _platform.getDataDir();
 355     const_cast<string&>(_serversDir) = _dataDir + "/servers";
 356     const_cast<string&>(_tmpDir) = _dataDir + "/tmp";
 357     const_cast<string&>(_instanceName) = _communicator->getDefaultLocator()->ice_getIdentity().category;
 358     const_cast<Ice::Int&>(_waitTime) = props->getPropertyAsIntWithDefault("IceGrid.Node.WaitTime", 60);
 359     const_cast<string&>(_outputDir) = props->getProperty("IceGrid.Node.Output");
 360     const_cast<bool&>(_redirectErrToOut) = props->getPropertyAsInt("IceGrid.Node.RedirectErrToOut") > 0;
 361     const_cast<bool&>(_allowEndpointsOverride) = props->getPropertyAsInt("IceGrid.Node.AllowEndpointsOverride") > 0;
 362
 363     //
 364     // Parse the properties override property.
 365     //
 366     vector<string> overrides = props->getPropertyAsList("IceGrid.Node.PropertiesOverride");
 367     if(!overrides.empty())
 368     {
 369         for(vector<string>::iterator p = overrides.begin(); p != overrides.end(); ++p)
 370         {
 371             if(p->find("--") != 0)
 372             {
 373                 *p = "--" + *p;
 374             }
 375         }
 376
 377         Ice::PropertiesPtr p = Ice::createProperties();
 378         p->parseCommandLineOptions("", overrides);
 379         Ice::PropertyDict propDict = p->getPropertiesForPrefix("");
 380         for(Ice::PropertyDict::const_iterator q = propDict.begin(); q != propDict.end(); ++q)
 381         {
 382             _propertiesOverride.push_back(createProperty(q->first, q->second));
 383         }
 384     }
 385 }
 386
 387 NodeI::~NodeI()
 388 {
 389 }
 390
 391 void
 392 NodeI::loadServer_async(const AMD_Node_loadServerPtr& amdCB,
 393                         const InternalServerDescriptorPtr& descriptor,
 394                         const string& replicaName,
 395                         const Ice::Current& current)
 396 {
 397     ServerCommandPtr command;
 398     {
 399         Lock sync(*this);
 400         ++_serial;
 401
 402         Ice::Identity id = createServerIdentity(descriptor->id);
 403
 404         //
 405         // Check if we already have a servant for this server. If that's
 406         // the case, the server is already loaded and we just need to
 407         // update it.
 408         //
 409         while(true)
 410         {
 411             bool added = false;
 412             ServerIPtr server;
 413             try
 414             {
 415                 server = ServerIPtr::dynamicCast(_adapter->find(id));
 416                 if(!server)
 417                 {
 418                     ServerPrx proxy = ServerPrx::uncheckedCast(_adapter->createProxy(id));
 419                     server = new ServerI(this, proxy, _serversDir, descriptor->id, _waitTime);
 420                     _adapter->add(server, id);
 421                     added = true;
 422                 }
 423             }
 424             catch(const Ice::ObjectAdapterDeactivatedException&)
 425             {
 426                 //
 427                 // We throw an object not exist exception to avoid
 428                 // dispatch warnings. The registry will consider the
 429                 // node has being unreachable upon receival of this
 430                 // exception (like any other Ice::LocalException). We
 431                 // could also have disabled dispatch warnings but they
 432                 // can still useful to catch other issues.
 433                 //
 434                 throw Ice::ObjectNotExistException(__FILE__, __LINE__, current.id, current.facet, current.operation);
 435             }
 436
 437             try
 438             {
 439                 command = server->load(amdCB, descriptor, replicaName);
 440             }
 441             catch(const Ice::ObjectNotExistException&)
 442             {
 443                 assert(!added);
 444                 continue;
 445             }
 446             catch(const Ice::Exception&)
 447             {
 448                 if(added)
 449                 {
 450                     try
 451                     {
 452                         _adapter->remove(id);
 453                     }
 454                     catch(const Ice::ObjectAdapterDeactivatedException&)
 455                     {
 456                         // IGNORE
 457                     }
 458                 }
 459                 throw;
 460             }
 461             break;
 462         }
 463     }
 464     if(command)
 465     {
 466         command->execute();
 467     }
 468 }
 469
 470 void
 471 NodeI::destroyServer_async(const AMD_Node_destroyServerPtr& amdCB,
 472                            const string& serverId,
 473                            const string& uuid,
 474                            int revision,
 475                            const string& replicaName,
 476                            const Ice::Current& current)
 477 {
 478     ServerCommandPtr command;
 479     {
 480         Lock sync(*this);
 481         ++_serial;
 482
 483         ServerIPtr server;
 484         try
 485         {
 486             server = ServerIPtr::dynamicCast(_adapter->find(createServerIdentity(serverId)));
 487         }
 488         catch(const Ice::ObjectAdapterDeactivatedException&)
 489         {
 490             //
 491             // We throw an object not exist exception to avoid
 492             // dispatch warnings. The registry will consider the node
 493             // has being unreachable upon receival of this exception
 494             // (like any other Ice::LocalException). We could also
 495             // have disabled dispatch warnings but they can still
 496             // useful to catch other issues.
 497             //
 498             throw Ice::ObjectNotExistException(__FILE__, __LINE__, current.id, current.facet, current.operation);
 499         }
 500
 501         if(!server)
 502         {
 503             server = new ServerI(this, 0, _serversDir, serverId, _waitTime);
 504         }
 505
 506         //
 507         // Destroy the server object if it's loaded.
 508         //
 509         try
 510         {
 511             command = server->destroy(amdCB, uuid, revision, replicaName);
 512         }
 513         catch(const Ice::ObjectNotExistException&)
 514         {
 515             amdCB->ice_response();
 516             return;
 517         }
 518     }
 519     if(command)
 520     {
 521         command->execute();
 522     }
 523 }
 524
 525 void
 526 NodeI::patch_async(const AMD_Node_patchPtr& amdCB,
 527                    const PatcherFeedbackPrx& feedback,
 528                    const string& application,
 529                    const string& server,
 530                    const InternalDistributionDescriptorPtr& appDistrib,
 531                    bool shutdown,
 532                    const Ice::Current&)
 533 {
 534     amdCB->ice_response();
 535
 536     {
 537         Lock sync(*this);
 538         while(_patchInProgress.find(application) != _patchInProgress.end())
 539         {
 540             wait();
 541         }
 542         _patchInProgress.insert(application);
 543     }
 544
 545
 546     set<ServerIPtr> servers;
 547     bool patchApplication = !appDistrib->icepatch.empty();
 548     if(server.empty())
 549     {
 550         //
 551         // Patch all the servers from the application.
 552         //
 553         servers = getApplicationServers(application);
 554     }
 555     else
 556     {
 557         ServerIPtr svr;
 558         try
 559         {
 560             svr = ServerIPtr::dynamicCast(_adapter->find(createServerIdentity(server)));
 561         }
 562         catch(const Ice::ObjectAdapterDeactivatedException&)
 563         {
 564         }
 565
 566         if(svr)
 567         {
 568             if(appDistrib->icepatch.empty() || !svr->dependsOnApplicationDistrib())
 569             {
 570                 //
 571                 // Don't patch the application if the server doesn't
 572                 // depend on it.
 573                 //
 574                 patchApplication = false;
 575                 servers.insert(svr);
 576             }
 577             else
 578             {
 579                 //
 580                 // If the server to patch depends on the application,
 581                 // we need to shutdown all the application servers
 582                 // that depend on the application.
 583                 //
 584                 servers = getApplicationServers(application);
 585             }
 586         }
 587     }
 588
 589     set<ServerIPtr>::iterator s = servers.begin();
 590     while(s != servers.end())
 591     {
 592         if(!appDistrib->icepatch.empty() && (*s)->dependsOnApplicationDistrib())
 593         {
 594             ++s;
 595         }
 596         else if((*s)->getDistribution() && (server.empty() || server == (*s)->getId()))
 597         {
 598             ++s;
 599         }
 600         else
 601         {
 602             //
 603             // Exclude servers which don't depend on the application distribution
 604             // or don't have a distribution.
 605             //
 606             servers.erase(s++);
 607         }
 608     }
 609
 610     string failure;
 611     if(!servers.empty())
 612     {
 613         try
 614         {
 615             set<ServerIPtr>::iterator s = servers.begin();
 616             vector<string> running;
 617             while(s != servers.end())
 618             {
 619                 try
 620                 {
 621                     if(!(*s)->startPatch(shutdown))
 622                     {
 623                         running.push_back((*s)->getId());
 624                         servers.erase(s++);
 625                     }
 626                     else
 627                     {
 628                         ++s;
 629                     }
 630                 }
 631                 catch(const Ice::ObjectNotExistException&)
 632                 {
 633                     servers.erase(s++);
 634                 }
 635             }
 636
 637             if(!running.empty())
 638             {
 639                 if(running.size() == 1)
 640                 {
 641                     throw "server `" + toString(running) + "' is active";
 642                 }
 643                 else
 644                 {
 645                     throw "servers `" + toString(running, ", ") + "' are active";
 646                 }
 647             }
 648
 649             for(s = servers.begin(); s != servers.end(); ++s)
 650             {
 651                 (*s)->waitForPatch();
 652             }
 653
 654             //
 655             // Patch the application.
 656             //
 657             FileServerPrx icepatch;
 658             if(patchApplication)
 659             {
 660                 assert(!appDistrib->icepatch.empty());
 661                 icepatch = FileServerPrx::checkedCast(_communicator->stringToProxy(appDistrib->icepatch));
 662                 if(!icepatch)
 663                 {
 664                     throw "proxy `" + appDistrib->icepatch + "' is not a file server.";
 665                 }
 666                 patch(icepatch, "distrib/" + application, appDistrib->directories);
 667             }
 668
 669             //
 670             // Patch the server(s).
 671             //
 672             for(s = servers.begin(); s != servers.end(); ++s)
 673             {
 674                 InternalDistributionDescriptorPtr dist = (*s)->getDistribution();
 675                 if(dist && (server.empty() || (*s)->getId() == server))
 676                 {
 677                     icepatch = FileServerPrx::checkedCast(_communicator->stringToProxy(dist->icepatch));
 678                     if(!icepatch)
 679                     {
 680                         throw "proxy `" + dist->icepatch + "' is not a file server.";
 681                     }
 682                     patch(icepatch, "servers/" + (*s)->getId() + "/distrib", dist->directories);
 683
 684                     if(!server.empty())
 685                     {
 686                         break; // No need to continue.
 687                     }
 688                 }
 689             }
 690         }
 691         catch(const Ice::LocalException& e)
 692         {
 693             ostringstream os;
 694             os << e;
 695             failure = os.str();
 696         }
 697         catch(const string& e)
 698         {
 699             failure = e;
 700         }
 701         catch(const char* e)
 702         {
 703             failure = e;
 704         }
 705
 706         for(set<ServerIPtr>::const_iterator s = servers.begin(); s != servers.end(); ++s)
 707         {
 708             (*s)->finishPatch();
 709         }
 710     }
 711
 712     {
 713         Lock sync(*this);
 714         _patchInProgress.erase(application);
 715         notifyAll();
 716     }
 717
 718     try
 719     {
 720         if(failure.empty())
 721         {
 722             feedback->finished();
 723         }
 724         else
 725         {
 726             feedback->failed(failure);
 727         }
 728     }
 729     catch(const Ice::LocalException&)
 730     {
 731     }
 732 }
 733
 734 void
 735 NodeI::registerWithReplica(const InternalRegistryPrx& replica, const Ice::Current&)
 736 {
 737     _sessions.create(replica);
 738 }
 739
 740 void
 741 NodeI::replicaInit(const InternalRegistryPrxSeq& replicas, const Ice::Current&)
 742 {
 743     _sessions.replicaInit(replicas);
 744 }
 745
 746 void
 747 NodeI::replicaAdded(const InternalRegistryPrx& replica, const Ice::Current&)
 748 {
 749     _sessions.replicaAdded(replica);
 750 }
 751
 752 void
 753 NodeI::replicaRemoved(const InternalRegistryPrx& replica, const Ice::Current&)
 754 {
 755     _sessions.replicaRemoved(replica);
 756 }
 757
 758 std::string
 759 NodeI::getName(const Ice::Current&) const
 760 {
 761     return _name;
 762 }
 763
 764 std::string
 765 NodeI::getHostname(const Ice::Current&) const
 766 {
 767     return _platform.getHostname();
 768 }
 769
 770 LoadInfo
 771 NodeI::getLoad(const Ice::Current&) const
 772 {
 773     return _platform.getLoadInfo();
 774 }
 775
 776 int
 777 NodeI::getProcessorSocketCount(const Ice::Current&) const
 778 {
 779     return _platform.getProcessorSocketCount();
 780 }
 781
 782 void
 783 NodeI::shutdown(const Ice::Current&) const
 784 {
 785     _activator->shutdown();
 786 }
 787
 788 Ice::Long
 789 NodeI::getOffsetFromEnd(const string& filename, int count, const Ice::Current&) const
 790 {
 791     return _fileCache->getOffsetFromEnd(getFilePath(filename), count);
 792 }
 793
 794 bool
 795 NodeI::read(const string& filename, Ice::Long pos, int size, Ice::Long& newPos, Ice::StringSeq& lines,
 796             const Ice::Current&) const
 797 {
 798     return _fileCache->read(getFilePath(filename), pos, size, newPos, lines);
 799 }
 800
 801 void
 802 NodeI::shutdown()
 803 {
 804     IceUtil::Mutex::Lock sync(_serversLock);
 805     for(map<string, set<ServerIPtr> >::const_iterator p = _serversByApplication.begin();
 806         p != _serversByApplication.end(); ++p)
 807     {
 808         for(set<ServerIPtr>::const_iterator q = p->second.begin(); q != p->second.end(); ++q)
 809         {
 810             (*q)->shutdown();
 811         }
 812     }
 813     _serversByApplication.clear();
 814 }
 815
 816 Ice::CommunicatorPtr
 817 NodeI::getCommunicator() const
 818 {
 819     return _communicator;
 820 }
 821
 822 Ice::ObjectAdapterPtr
 823 NodeI::getAdapter() const
 824 {
 825     return _adapter;
 826 }
 827
 828 ActivatorPtr
 829 NodeI::getActivator() const
 830 {
 831     return _activator;
 832 }
 833
 834 IceUtil::TimerPtr
 835 NodeI::getTimer() const
 836 {
 837     return _timer;
 838 }
 839
 840 TraceLevelsPtr
 841 NodeI::getTraceLevels() const
 842 {
 843     return _traceLevels;
 844 }
 845
 846 UserAccountMapperPrx
 847 NodeI::getUserAccountMapper() const
 848 {
 849     return _userAccountMapper;
 850 }
 851
 852 PlatformInfo&
 853 NodeI::getPlatformInfo() const
 854 {
 855     return _platform;
 856 }
 857
 858 FileCachePtr
 859 NodeI::getFileCache() const
 860 {
 861     return _fileCache;
 862 }
 863
 864 NodePrx
 865 NodeI::getProxy() const
 866 {
 867     return _proxy;
 868 }
 869
 870 const PropertyDescriptorSeq&
 871 NodeI::getPropertiesOverride() const
 872 {
 873     return _propertiesOverride;
 874 }
 875
 876 string
 877 NodeI::getOutputDir() const
 878 {
 879     return _outputDir;
 880 }
 881
 882 bool
 883 NodeI::getRedirectErrToOut() const
 884 {
 885     return _redirectErrToOut;
 886 }
 887
 888 bool
 889 NodeI::allowEndpointsOverride() const
 890 {
 891     return _allowEndpointsOverride;
 892 }
 893
 894 NodeSessionPrx
 895 NodeI::registerWithRegistry(const InternalRegistryPrx& registry)
 896 {
 897     return registry->registerNode(_platform.getInternalNodeInfo(), _proxy, _platform.getLoadInfo());
 898 }
 899
 900 void
 901 NodeI::checkConsistency(const NodeSessionPrx& session)
 902 {
 903     //
 904     // Only do the consistency check on the startup. This ensures that servers can't
 905     // be removed by a bogus master when the master session is re-established.
 906     //
 907     if(_consistencyCheckDone)
 908     {
 909         return;
 910     }
 911     _consistencyCheckDone = true;
 912
 913     //
 914     // We use a serial number to keep track of the concurrent changes
 915     // on the node. When a server is loaded/destroyed the serial is
 916     // incremented. This allows to ensure that the list of servers
 917     // returned by the registry is consistent with the servers
 918     // currently deployed on the node: if the serial didn't change
 919     // after getting the list of servers from the registry, we have
 920     // the accurate list of servers that should be deployed on the
 921     // node.
 922     //
 923     unsigned long serial = 0;
 924     Ice::StringSeq servers;
 925     vector<ServerCommandPtr> commands;
 926     while(true)
 927     {
 928         {
 929             Lock sync(*this);
 930             if(serial == _serial)
 931             {
 932                 _serial = 1; // We can reset the serial number.
 933                 commands = checkConsistencyNoSync(servers);
 934                 break;
 935             }
 936             serial = _serial;
 937         }
 938         assert(session);
 939         try
 940         {
 941             servers = session->getServers();
 942         }
 943         catch(const Ice::LocalException&)
 944         {
 945             return; // The connection with the session was lost.
 946         }
 947         sort(servers.begin(), servers.end());
 948     }
 949
 950     for_each(commands.begin(), commands.end(), IceUtil::voidMemFun(&ServerCommand::execute));
 951 }
 952
 953 void
 954 NodeI::addObserver(const NodeSessionPrx& session, const NodeObserverPrx& observer)
 955 {
 956     IceUtil::Mutex::Lock sync(_observerMutex);
 957     assert(_observers.find(session) == _observers.end());
 958     _observers.insert(make_pair(session, observer));
 959
 960     _observerUpdates.erase(observer); // Remove any updates from the previous session.
 961
 962     ServerDynamicInfoSeq serverInfos;
 963     AdapterDynamicInfoSeq adapterInfos;
 964     for(map<string, ServerDynamicInfo>::const_iterator p = _serversDynamicInfo.begin();
 965         p != _serversDynamicInfo.end(); ++p)
 966     {
 967         assert(p->second.state != Destroyed && (p->second.state != Inactive || !p->second.enabled));
 968         serverInfos.push_back(p->second);
 969     }
 970
 971     for(map<string, AdapterDynamicInfo>::const_iterator q = _adaptersDynamicInfo.begin();
 972         q != _adaptersDynamicInfo.end(); ++q)
 973     {
 974         assert(q->second.proxy);
 975         adapterInfos.push_back(q->second);
 976     }
 977
 978     NodeDynamicInfo info;
 979     info.info = _platform.getNodeInfo();
 980     info.servers = serverInfos;
 981     info.adapters = adapterInfos;
 982     queueUpdate(observer, new NodeUp(this, observer, info));
 983 }
 984
 985 void
 986 NodeI::removeObserver(const NodeSessionPrx& session)
 987 {
 988     IceUtil::Mutex::Lock sync(_observerMutex);
 989     _observers.erase(session);
 990 }
 991
 992 void
 993 NodeI::observerUpdateServer(const ServerDynamicInfo& info)
 994 {
 995     IceUtil::Mutex::Lock sync(_observerMutex);
 996
 997     if(info.state == Destroyed || (info.state == Inactive && info.enabled))
 998     {
 999         _serversDynamicInfo.erase(info.id);
1000     }
1001     else
1002     {
1003         _serversDynamicInfo[info.id] = info;
1004     }
1005
1006     //
1007     // Send the update and make sure we don't send the update twice to
1008     // the same observer (it's possible for the observer to be
1009     // registered twice if a replica is removed and added right away
1010     // after).
1011     //
1012     set<NodeObserverPrx> sent;
1013     for(map<NodeSessionPrx, NodeObserverPrx>::const_iterator p = _observers.begin(); p != _observers.end(); ++p)
1014     {
1015         if(sent.find(p->second) == sent.end())
1016         {
1017             queueUpdate(p->second, new UpdateServer(this, p->second, info));
1018         }
1019     }
1020 }
1021
1022 void
1023 NodeI::observerUpdateAdapter(const AdapterDynamicInfo& info)
1024 {
1025     IceUtil::Mutex::Lock sync(_observerMutex);
1026
1027     if(info.proxy)
1028     {
1029         _adaptersDynamicInfo[info.id] = info;
1030     }
1031     else
1032     {
1033         _adaptersDynamicInfo.erase(info.id);
1034     }
1035
1036     //
1037     // Send the update and make sure we don't send the update twice to
1038     // the same observer (it's possible for the observer to be
1039     // registered twice if a replica is removed and added right away
1040     // after).
1041     //
1042     set<NodeObserverPrx> sent;
1043     for(map<NodeSessionPrx, NodeObserverPrx>::const_iterator p = _observers.begin(); p != _observers.end(); ++p)
1044     {
1045         if(sent.find(p->second) == sent.end())
1046         {
1047             queueUpdate(p->second, new UpdateAdapter(this, p->second, info));
1048         }
1049     }
1050 }
1051
1052 void
1053 NodeI::queueUpdate(const NodeObserverPrx& proxy, const UpdatePtr& update)
1054 {
1055     //Lock sync(*this); Called within the synchronization
1056     map<NodeObserverPrx, deque<UpdatePtr> >::iterator p = _observerUpdates.find(proxy);
1057     if(p == _observerUpdates.end())
1058     {
1059         if(update->send())
1060         {
1061             _observerUpdates[proxy].push_back(update);
1062         }
1063     }
1064     else
1065     {
1066         p->second.push_back(update);
1067     }
1068 }
1069
1070 void
1071 NodeI::dequeueUpdate(const NodeObserverPrx& proxy, const UpdatePtr& update, bool all)
1072 {
1073     IceUtil::Mutex::Lock sync(_observerMutex);
1074     map<NodeObserverPrx, deque<UpdatePtr> >::iterator p = _observerUpdates.find(proxy);
1075     if(p == _observerUpdates.end() || p->second.front().get() != update.get())
1076     {
1077         return;
1078     }
1079
1080     p->second.pop_front();
1081
1082     if(all || (!p->second.empty() && !p->second.front()->send()))
1083     {
1084         p->second.clear();
1085     }
1086
1087     if(p->second.empty())
1088     {
1089         _observerUpdates.erase(p);
1090     }
1091 }
1092
1093 void
1094 NodeI::addServer(const ServerIPtr& server, const string& application)
1095 {
1096     IceUtil::Mutex::Lock sync(_serversLock);
1097     map<string, set<ServerIPtr> >::iterator p = _serversByApplication.find(application);
1098     if(p == _serversByApplication.end())
1099     {
1100         map<string, set<ServerIPtr> >::value_type v(application, set<ServerIPtr>());
1101         p = _serversByApplication.insert(p, v);
1102     }
1103     p->second.insert(server);
1104 }
1105
1106 void
1107 NodeI::removeServer(const ServerIPtr& server, const std::string& application)
1108 {
1109     IceUtil::Mutex::Lock sync(_serversLock);
1110     map<string, set<ServerIPtr> >::iterator p = _serversByApplication.find(application);
1111     if(p != _serversByApplication.end())
1112     {
1113         p->second.erase(server);
1114         if(p->second.empty())
1115         {
1116             _serversByApplication.erase(p);
1117
1118             string appDir = _dataDir + "/distrib/" + application;
1119             if(IceUtilInternal::directoryExists(appDir))
1120             {
1121                 try
1122                 {
1123                     IcePatch2::removeRecursive(appDir);
1124                 }
1125                 catch(const string& msg)
1126                 {
1127                     Ice::Warning out(_traceLevels->logger);
1128                     out << "removing application directory `" << appDir << "' failed:\n" << msg;
1129                 }
1130             }
1131         }
1132     }
1133 }
1134
1135 Ice::Identity
1136 NodeI::createServerIdentity(const string& name) const
1137 {
1138     Ice::Identity id;
1139     id.category = _instanceName + "-Server";
1140     id.name = name;
1141     return id;
1142 }
1143
1144 string
1145 NodeI::getServerAdminCategory() const
1146 {
1147     return _instanceName + "-NodeRouter";
1148 }
1149
1150 vector<ServerCommandPtr>
1151 NodeI::checkConsistencyNoSync(const Ice::StringSeq& servers)
1152 {
1153     vector<ServerCommandPtr> commands;
1154
1155     //
1156     // Check if the servers directory doesn't contain more servers
1157     // than the registry really knows.
1158     //
1159     Ice::StringSeq contents;
1160     try
1161     {
1162         contents = readDirectory(_serversDir);
1163     }
1164     catch(const string& msg)
1165     {
1166         Ice::Error out(_traceLevels->logger);
1167         out << "couldn't read directory `" << _serversDir << "':\n" << msg;
1168         return commands;
1169     }
1170
1171     vector<string> remove;
1172     set_difference(contents.begin(), contents.end(), servers.begin(), servers.end(), back_inserter(remove));
1173
1174     //
1175     // Remove the extra servers if possible.
1176     //
1177     try
1178     {
1179         vector<string>::iterator p = remove.begin();
1180         while(p != remove.end())
1181         {
1182             ServerIPtr server = ServerIPtr::dynamicCast(_adapter->find(createServerIdentity(*p)));
1183             if(server)
1184             {
1185                 //
1186                 // If the server is loaded, we invoke on it to destroy it.
1187                 //
1188                 try
1189                 {
1190                     ServerCommandPtr command = server->destroy(0, "", 0, "Master");
1191                     if(command)
1192                     {
1193                         commands.push_back(command);
1194                     }
1195                     p = remove.erase(p);
1196                     continue;
1197                 }
1198                 catch(const Ice::LocalException& ex)
1199                 {
1200                     Ice::Error out(_traceLevels->logger);
1201                     out << "server `" << *p << "' destroy failed:\n" << ex;
1202                 }
1203                 catch(const string&)
1204                 {
1205                     assert(false);
1206                 }
1207             }
1208
1209             try
1210             {
1211                 if(canRemoveServerDirectory(*p))
1212                 {
1213                     //
1214                     // If the server directory can be removed and we
1215                     // either remove it or back it up before to remove it.
1216                     //
1217                     removeRecursive(_serversDir + "/" + *p);
1218                     p = remove.erase(p);
1219                     continue;
1220                 }
1221             }
1222             catch(const string& msg)
1223             {
1224                 Ice::Warning out(_traceLevels->logger);
1225                 out << "removing server directory `" << _serversDir << "/" << *p << "' failed:\n" << msg;
1226             }
1227
1228             *p = _serversDir + "/" + *p;
1229             ++p;
1230         }
1231     }
1232     catch(const Ice::ObjectAdapterDeactivatedException&)
1233     {
1234         //
1235         // Just return the server commands, we'll finish the
1236         // consistency check next time the node is started.
1237         //
1238         return commands;
1239     }
1240
1241     if(!remove.empty())
1242     {
1243         Ice::Warning out(_traceLevels->logger);
1244         out << "server directories containing data not created or written by IceGrid were not removed:\n";
1245         out << toString(remove);
1246     }
1247     return commands;
1248 }
1249
1250 NodeSessionPrx
1251 NodeI::getMasterNodeSession() const
1252 {
1253     return _sessions.getMasterNodeSession();
1254 }
1255
1256 bool
1257 NodeI::canRemoveServerDirectory(const string& name)
1258 {
1259     //
1260     // Check if there's files which we didn't create.
1261     //
1262     Ice::StringSeq c = readDirectory(_serversDir + "/" + name);
1263     set<string> contents(c.begin(), c.end());
1264     contents.erase("dbs");
1265     contents.erase("config");
1266     contents.erase("distrib");
1267     contents.erase("revision");
1268     if(!contents.empty())
1269     {
1270         return false;
1271     }
1272
1273     c = readDirectory(_serversDir + "/" + name + "/config");
1274     Ice::StringSeq::const_iterator p;
1275     for(p = c.begin() ; p != c.end(); ++p)
1276     {
1277         if(p->find("config") != 0)
1278         {
1279             return false;
1280         }
1281     }
1282
1283     c = readDirectory(_serversDir + "/" + name + "/dbs");
1284     for(p = c.begin() ; p != c.end(); ++p)
1285     {
1286         try
1287         {
1288             Ice::StringSeq files = readDirectory(_serversDir + "/" + name + "/dbs/" + *p);
1289             files.erase(remove(files.begin(), files.end(), "DB_CONFIG"), files.end());
1290             files.erase(remove(files.begin(), files.end(), "__Freeze"), files.end());
1291             if(!files.empty())
1292             {
1293                 return false;
1294             }
1295         }
1296         catch(const string&)
1297         {
1298             return false;
1299         }
1300     }
1301
1302     return true;
1303 }
1304
1305 void
1306 NodeI::patch(const FileServerPrx& icepatch, const string& dest, const vector<string>& directories)
1307 {
1308     IcePatch2::PatcherFeedbackPtr feedback = new LogPatcherFeedback(_traceLevels, dest);
1309     IcePatch2::createDirectory(_dataDir + "/" + dest);
1310     PatcherPtr patcher = new Patcher(icepatch, feedback, _dataDir + "/" + dest, false, 100, 1);
1311     bool aborted = !patcher->prepare();
1312     if(!aborted)
1313     {
1314         if(directories.empty())
1315         {
1316             aborted = !patcher->patch("");
1317             dynamic_cast<LogPatcherFeedback*>(feedback.get())->finishPatch();
1318         }
1319         else
1320         {
1321             for(vector<string>::const_iterator p = directories.begin(); p != directories.end(); ++p)
1322             {
1323                 dynamic_cast<LogPatcherFeedback*>(feedback.get())->setPatchingPath(*p);
1324                 if(!patcher->patch(*p))
1325                 {
1326                     aborted = true;
1327                     break;
1328                 }
1329                 dynamic_cast<LogPatcherFeedback*>(feedback.get())->finishPatch();
1330             }
1331         }
1332     }
1333     if(!aborted)
1334     {
1335         patcher->finish();
1336     }
1337
1338     //
1339     // Update the files owner/group
1340     //
1341 }
1342
1343 set<ServerIPtr>
1344 NodeI::getApplicationServers(const string& application) const
1345 {
1346     IceUtil::Mutex::Lock sync(_serversLock);
1347     set<ServerIPtr> servers;
1348     map<string, set<ServerIPtr> >::const_iterator p = _serversByApplication.find(application);
1349     if(p != _serversByApplication.end())
1350     {
1351         servers = p->second;
1352     }
1353     return servers;
1354 }
1355
1356
1357
1358 string
1359 NodeI::getFilePath(const string& filename) const
1360 {
1361     string file;
1362     if(filename == "stderr")
1363     {
1364         file = _communicator->getProperties()->getProperty("Ice.StdErr");
1365         if(file.empty())
1366         {
1367             throw FileNotAvailableException("Ice.StdErr configuration property is not set");
1368         }
1369     }
1370     else if(filename == "stdout")
1371     {
1372         file = _communicator->getProperties()->getProperty("Ice.StdOut");
1373         if(file.empty())
1374         {
1375             throw FileNotAvailableException("Ice.StdOut configuration property is not set");
1376         }
1377     }
1378     else
1379     {
1380         throw FileNotAvailableException("unknown file");
1381     }
1382     return file;
1383 }