cpp/src/IceStorm/TopicI.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/DisableWarnings.h>
  11 #include <IceStorm/TopicI.h>
  12 #include <IceStorm/Instance.h>
  13 #include <IceStorm/Subscriber.h>
  14 #include <IceStorm/TraceLevels.h>
  15 #include <IceStorm/NodeI.h>
  16 #include <IceStorm/Observers.h>
  17 #include <IceStorm/DB.h>
  18 #include <IceStorm/Util.h>
  19 #include <Ice/LoggerUtil.h>
  20 #include <algorithm>
  21
  22 using namespace std;
  23 using namespace IceStorm;
  24 using namespace IceStormElection;
  25
  26 using namespace IceDB;
  27
  28 namespace
  29 {
  30
  31 void
  32 halt(const Ice::CommunicatorPtr& com, const DatabaseException& ex)
  33 {
  34     {
  35         Ice::Error error(com->getLogger());
  36         error << "fatal exception: " << ex << "\n*** Aborting application ***";
  37     }
  38
  39     abort();
  40 }
  41
  42 //
  43 // The servant has a 1-1 association with a topic. It is used to
  44 // receive events from Publishers.
  45 //
  46 class PublisherI : public Ice::BlobjectArray
  47 {
  48 public:
  49
  50     PublisherI(const TopicImplPtr& topic, const InstancePtr& instance) :
  51         _topic(topic), _instance(instance)
  52     {
  53     }
  54
  55     ~PublisherI()
  56     {
  57         //cout << "~PublisherI" << endl;
  58     }
  59
  60     virtual bool
  61     ice_invoke(const pair<const Ice::Byte*, const Ice::Byte*>& inParams,
  62                Ice::ByteSeq&,
  63                const Ice::Current& current)
  64     {
  65         // The publish call does a cached read.
  66         EventDataPtr event = new EventData(
  67             current.operation,
  68             current.mode,
  69             Ice::ByteSeq(),
  70             current.ctx);
  71
  72         //
  73         // COMPILERBUG: gcc 4.0.1 doesn't like this.
  74         //
  75         //event->data.swap(Ice::ByteSeq(inParams.first, inParams.second));
  76         Ice::ByteSeq data(inParams.first, inParams.second);
  77         event->data.swap(data);
  78
  79         EventDataSeq v;
  80         v.push_back(event);
  81         _topic->publish(false, v);
  82
  83         return true;
  84     }
  85
  86 private:
  87
  88     const TopicImplPtr _topic;
  89     const InstancePtr _instance;
  90 };
  91
  92 //
  93 // The servant has a 1-1 association with a topic. It is used to
  94 // receive events from linked Topics.
  95 //
  96 class TopicLinkI : public TopicLink
  97 {
  98 public:
  99
 100     TopicLinkI(const TopicImplPtr& impl, const InstancePtr& instance) :
 101         _impl(impl), _instance(instance)
 102     {
 103     }
 104
 105     ~TopicLinkI()
 106     {
 107         //cout << "~TopicLinkI" << endl;
 108     }
 109
 110     virtual void
 111     forward(const EventDataSeq& v, const Ice::Current& current)
 112     {
 113         // The publish call does a cached read.
 114         _impl->publish(true, v);
 115     }
 116
 117 private:
 118
 119     const TopicImplPtr _impl;
 120     const InstancePtr _instance;
 121 };
 122
 123 class TopicI : public TopicInternal
 124 {
 125 public:
 126
 127     TopicI(const TopicImplPtr& impl, const InstancePtr& instance) :
 128         _impl(impl), _instance(instance)
 129     {
 130     }
 131
 132     ~TopicI()
 133     {
 134         //cout << "~TopicI" << endl;
 135     }
 136
 137     virtual string getName(const Ice::Current&) const
 138     {
 139         // Use cached reads.
 140         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
 141         return _impl->getName();
 142     }
 143
 144     virtual Ice::ObjectPrx getPublisher(const Ice::Current&) const
 145     {
 146         // Use cached reads.
 147         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
 148         return _impl->getPublisher();
 149     }
 150
 151     virtual Ice::ObjectPrx getNonReplicatedPublisher(const Ice::Current&) const
 152     {
 153         // Use cached reads.
 154         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
 155         return _impl->getNonReplicatedPublisher();
 156     }
 157
 158     virtual void subscribe(const QoS& qos, const Ice::ObjectPrx& obj, const Ice::Current& current)
 159     {
 160         while(true)
 161         {
 162             Ice::Long generation = -1;
 163             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 164             if(master)
 165             {
 166                 try
 167                 {
 168                     master->subscribe(qos, obj);
 169                 }
 170                 catch(const Ice::ConnectFailedException&)
 171                 {
 172                     _instance->node()->recovery(generation);
 173                     continue;
 174                 }
 175                 catch(const Ice::TimeoutException&)
 176                 {
 177                     _instance->node()->recovery(generation);
 178                     continue;
 179                 }
 180             }
 181             else
 182             {
 183                 FinishUpdateHelper unlock(_instance->node());
 184                 _impl->subscribe(qos, obj);
 185             }
 186             break;
 187         }
 188     }
 189
 190     virtual Ice::ObjectPrx subscribeAndGetPublisher(const QoS& qos, const Ice::ObjectPrx& obj,
 191                                                     const Ice::Current& current)
 192     {
 193         while(true)
 194         {
 195             Ice::Long generation = -1;
 196             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 197             if(master)
 198             {
 199                 try
 200                 {
 201                     return master->subscribeAndGetPublisher(qos, obj);
 202                 }
 203                 catch(const Ice::ConnectFailedException&)
 204                 {
 205                     _instance->node()->recovery(generation);
 206                     continue;
 207                 }
 208                 catch(const Ice::TimeoutException&)
 209                 {
 210                     _instance->node()->recovery(generation);
 211                     continue;
 212                 }
 213             }
 214             else
 215             {
 216                 FinishUpdateHelper unlock(_instance->node());
 217                 return _impl->subscribeAndGetPublisher(qos, obj);
 218             }
 219         }
 220     }
 221
 222     virtual void unsubscribe(const Ice::ObjectPrx& subscriber, const Ice::Current& current)
 223     {
 224         while(true)
 225         {
 226             Ice::Long generation = -1;
 227             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 228             if(master)
 229             {
 230                 try
 231                 {
 232                     master->unsubscribe(subscriber);
 233                 }
 234                 catch(const Ice::ConnectFailedException&)
 235                 {
 236                     _instance->node()->recovery(generation);
 237                     continue;
 238                 }
 239                 catch(const Ice::TimeoutException&)
 240                 {
 241                     _instance->node()->recovery(generation);
 242                     continue;
 243                 }
 244             }
 245             else
 246             {
 247                 FinishUpdateHelper unlock(_instance->node());
 248                 _impl->unsubscribe(subscriber);
 249             }
 250             break;
 251         }
 252     }
 253
 254     virtual TopicLinkPrx getLinkProxy(const Ice::Current&)
 255     {
 256         // Use cached reads.
 257         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
 258         return _impl->getLinkProxy();
 259     }
 260
 261     virtual void reap(const Ice::IdentitySeq& ids, const Ice::Current& current)
 262     {
 263         NodeIPtr node = _instance->node();
 264         if(!node->updateMaster(__FILE__, __LINE__))
 265         {
 266             throw ReapWouldBlock();
 267         }
 268         FinishUpdateHelper unlock(node);
 269         _impl->reap(ids);
 270     }
 271
 272     virtual void link(const TopicPrx& topic, Ice::Int cost, const Ice::Current& current)
 273     {
 274         while(true)
 275         {
 276             Ice::Long generation = -1;
 277             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 278             if(master)
 279             {
 280                 try
 281                 {
 282                     master->link(topic, cost);
 283                 }
 284                 catch(const Ice::ConnectFailedException&)
 285                 {
 286                     _instance->node()->recovery(generation);
 287                     continue;
 288                 }
 289                 catch(const Ice::TimeoutException&)
 290                 {
 291                     _instance->node()->recovery(generation);
 292                     continue;
 293                 }
 294             }
 295             else
 296             {
 297                 FinishUpdateHelper unlock(_instance->node());
 298                 _impl->link(topic, cost);
 299             }
 300             break;
 301         }
 302     }
 303
 304     virtual void unlink(const TopicPrx& topic, const Ice::Current& current)
 305     {
 306         while(true)
 307         {
 308             Ice::Long generation = -1;
 309             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 310             if(master)
 311             {
 312                 try
 313                 {
 314                     master->unlink(topic);
 315                 }
 316                 catch(const Ice::ConnectFailedException&)
 317                 {
 318                     _instance->node()->recovery(generation);
 319                     continue;
 320                 }
 321                 catch(const Ice::TimeoutException&)
 322                 {
 323                     _instance->node()->recovery(generation);
 324                     continue;
 325                 }
 326             }
 327             else
 328             {
 329                 FinishUpdateHelper unlock(_instance->node());
 330                 _impl->unlink(topic);
 331             }
 332             break;
 333         }
 334     }
 335
 336     virtual LinkInfoSeq getLinkInfoSeq(const Ice::Current&) const
 337     {
 338         // Use cached reads.
 339         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
 340         return _impl->getLinkInfoSeq();
 341     }
 342
 343     virtual void destroy(const Ice::Current& current)
 344     {
 345         while(true)
 346         {
 347             Ice::Long generation = -1;
 348             TopicPrx master = getMasterFor(current, generation, __FILE__, __LINE__);
 349             if(master)
 350             {
 351                 try
 352                 {
 353                     master->destroy();
 354                 }
 355                 catch(const Ice::ConnectFailedException&)
 356                 {
 357                     _instance->node()->recovery(generation);
 358                     continue;
 359                 }
 360                 catch(const Ice::TimeoutException&)
 361                 {
 362                     _instance->node()->recovery(generation);
 363                     continue;
 364                 }
 365             }
 366             else
 367             {
 368                 FinishUpdateHelper unlock(_instance->node());
 369                 _impl->destroy();
 370             }
 371             break;
 372         }
 373     }
 374
 375 private:
 376
 377     TopicPrx getMasterFor(const Ice::Current& cur, Ice::Long& generation, const char* file, int line) const
 378     {
 379         NodeIPtr node = _instance->node();
 380         Ice::ObjectPrx master;
 381         if(node)
 382         {
 383             master = _instance->node()->startUpdate(generation, file, line);
 384         }
 385         return (master) ? TopicPrx::uncheckedCast(master->ice_identity(cur.id)) : TopicPrx();
 386     }
 387
 388     const TopicImplPtr _impl;
 389     const InstancePtr _instance;
 390 };
 391
 392 }
 393
 394 namespace IceStorm
 395 {
 396 extern string identityToTopicName(const Ice::Identity& id);
 397 }
 398
 399 TopicImpl::TopicImpl(
 400     const InstancePtr& instance,
 401     const string& name,
 402     const Ice::Identity& id,
 403     const SubscriberRecordSeq& subscribers) :
 404     _instance(instance),
 405     _name(name),
 406     _id(id),
 407     _databaseCache(instance->databaseCache()),
 408     _destroyed(false)
 409 {
 410     try
 411     {
 412         __setNoDelete(true);
 413
 414         // TODO: If we want to improve the performance of the
 415         // non-replicated case we could allocate a null-topic impl here.
 416         _servant = new TopicI(this, instance);
 417
 418         //
 419         // Create a servant per topic to receive event data. If the
 420         // category is empty then we are in backwards compatibility
 421         // mode. In this case the servant's identity is
 422         // category=<topicname>, name=publish, otherwise the name is
 423         // <instancename>/<topicname>.publish. The same applies to the
 424         // link proxy.
 425         //
 426         // Activate the object and save a reference to give to publishers.
 427         //
 428         Ice::Identity pubid;
 429         Ice::Identity linkid;
 430         if(id.category.empty())
 431         {
 432             pubid.category = _name;
 433             pubid.name = "publish";
 434             linkid.category = _name;
 435             linkid.name = "link";
 436         }
 437         else
 438         {
 439             pubid.category = id.category;
 440             pubid.name = _name + ".publish";
 441             linkid.category = id.category;
 442             linkid.name = _name + ".link";
 443         }
 444
 445         _publisherPrx = _instance->publishAdapter()->add(new PublisherI(this, instance), pubid);
 446         _linkPrx = TopicLinkPrx::uncheckedCast(
 447             _instance->publishAdapter()->add(new TopicLinkI(this, instance), linkid));
 448
 449         //
 450         // Re-establish subscribers.
 451         //
 452         for(SubscriberRecordSeq::const_iterator p = subscribers.begin(); p != subscribers.end(); ++p)
 453         {
 454             Ice::Identity id = p->obj->ice_getIdentity();
 455             TraceLevelsPtr traceLevels = _instance->traceLevels();
 456             if(traceLevels->topic > 0)
 457             {
 458                 Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 459                 out << _name << " recreate " << _instance->communicator()->identityToString(id);
 460                 if(traceLevels->topic > 1)
 461                 {
 462                     out << " endpoints: " << IceStormInternal::describeEndpoints(p->obj);
 463                 }
 464             }
 465
 466             try
 467             {
 468                 //
 469                 // Create the subscriber object add it to the set of
 470                 // subscribers.
 471                 //
 472                 SubscriberPtr subscriber = Subscriber::create(_instance, *p);
 473                 _subscribers.push_back(subscriber);
 474             }
 475             catch(const Ice::Exception& ex)
 476             {
 477                 Ice::Warning out(traceLevels->logger);
 478                 out << _name << " recreate " << _instance->communicator()->identityToString(id);
 479                 if(traceLevels->topic > 1)
 480                 {
 481                     out << " endpoints: " << IceStormInternal::describeEndpoints(p->obj);
 482                 }
 483                 out << " failed: " << ex;
 484             }
 485         }
 486     }
 487     catch(...)
 488     {
 489         shutdown();
 490         __setNoDelete(false);
 491         throw;
 492     }
 493     __setNoDelete(false);
 494 }
 495
 496 TopicImpl::~TopicImpl()
 497 {
 498     //cout << "~TopicImpl" << endl;
 499 }
 500
 501 string
 502 TopicImpl::getName() const
 503 {
 504     // Immutable
 505     return _name;
 506 }
 507
 508 Ice::ObjectPrx
 509 TopicImpl::getPublisher() const
 510 {
 511     // Immutable
 512     if(_instance->publisherReplicaProxy())
 513     {
 514         return _instance->publisherReplicaProxy()->ice_identity(_publisherPrx->ice_getIdentity());
 515     }
 516     return _publisherPrx;
 517 }
 518
 519 Ice::ObjectPrx
 520 TopicImpl::getNonReplicatedPublisher() const
 521 {
 522     // If there is an adapter id configured then we're using icegrid
 523     // so create an indirect proxy, otherwise create a direct proxy.
 524     if(!_publisherPrx->ice_getAdapterId().empty())
 525     {
 526         return _instance->publishAdapter()->createIndirectProxy(_publisherPrx->ice_getIdentity());
 527     }
 528     else
 529     {
 530         return _instance->publishAdapter()->createDirectProxy(_publisherPrx->ice_getIdentity());
 531     }
 532 }
 533
 534 //
 535 // COMPILERFIX: For some reason with VC6 find reports an error.
 536 //
 537 #if defined(_MSC_VER) && (_MSC_VER < 1300)
 538 namespace
 539 {
 540 vector<SubscriberPtr>::iterator
 541 find(vector<SubscriberPtr>::iterator start, vector<SubscriberPtr>::iterator end, const Ice::Identity& ident)
 542 {
 543     while(start != end)
 544     {
 545         if(*start == ident)
 546         {
 547             return start;
 548         }
 549         ++start;
 550     }
 551     return end;
 552 }
 553 }
 554 #endif
 555
 556 namespace
 557 {
 558 void
 559 trace(Ice::Trace& out, const InstancePtr& instance, const vector<SubscriberPtr>& s)
 560 {
 561     out << '[';
 562     for(vector<SubscriberPtr>::const_iterator p = s.begin(); p != s.end(); ++p)
 563     {
 564         if(p != s.begin())
 565         {
 566             out << ",";
 567         }
 568         out << instance->communicator()->identityToString((*p)->id());
 569     }
 570     out << "]";
 571 }
 572 }
 573
 574 void
 575 TopicImpl::subscribe(const QoS& origQoS, const Ice::ObjectPrx& obj)
 576 {
 577     Ice::Identity id = obj->ice_getIdentity();
 578     TraceLevelsPtr traceLevels = _instance->traceLevels();
 579     QoS qos = origQoS;
 580     if(traceLevels->topic > 0)
 581     {
 582         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 583         out << _name << ": subscribe: " << _instance->communicator()->identityToString(id);
 584
 585         if(traceLevels->topic > 1)
 586         {
 587             out << " endpoints: " << IceStormInternal::describeEndpoints(obj)
 588                 << " QoS: ";
 589             for(QoS::const_iterator p = qos.begin(); p != qos.end() ; ++p)
 590             {
 591                 if(p != qos.begin())
 592                 {
 593                     out << ',';
 594                 }
 595                 out << '[' << p->first << "," << p->second << ']';
 596             }
 597             out << " subscriptions: ";
 598             trace(out, _instance, _subscribers);
 599         }
 600     }
 601
 602     string reliability = "oneway";
 603     {
 604         QoS::iterator p = qos.find("reliability");
 605         if(p != qos.end())
 606         {
 607             reliability = p->second;
 608             qos.erase(p);
 609         }
 610     }
 611
 612     Ice::ObjectPrx newObj = obj;
 613     if(reliability == "batch")
 614     {
 615         if(newObj->ice_isDatagram())
 616         {
 617             newObj = newObj->ice_batchDatagram();
 618         }
 619         else
 620         {
 621             newObj = newObj->ice_batchOneway();
 622         }
 623     }
 624     else if(reliability == "twoway")
 625     {
 626         newObj = newObj->ice_twoway();
 627     }
 628     else if(reliability == "twoway ordered")
 629     {
 630         qos["reliability"] = "ordered";
 631         newObj = newObj->ice_twoway();
 632     }
 633     else // reliability == "oneway"
 634     {
 635         if(reliability != "oneway" && traceLevels->subscriber > 0)
 636         {
 637             Ice::Trace out(traceLevels->logger, traceLevels->subscriberCat);
 638             out << reliability <<" mode not understood.";
 639         }
 640         if(!newObj->ice_isDatagram())
 641         {
 642             newObj = newObj->ice_oneway();
 643         }
 644     }
 645
 646     IceUtil::Mutex::Lock sync(_subscribersMutex);
 647     SubscriberRecord record;
 648     record.id = id;
 649     record.obj = newObj;
 650     record.theQoS = qos;
 651     record.topicName = _name;
 652     record.link = false;
 653     record.cost = 0;
 654
 655     LogUpdate llu;
 656
 657     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 658     if(p != _subscribers.end())
 659     {
 660         // If we already have this subscriber remove it from our
 661         // subscriber list and remove it from the database.
 662         (*p)->destroy();
 663         _subscribers.erase(p);
 664
 665         for(;;)
 666         {
 667             try
 668             {
 669                 DatabaseConnectionPtr connection = _databaseCache->getConnection();
 670                 TransactionHolder txn(connection);
 671
 672                 SubscriberRecordKey key;
 673                 key.topic = _id;
 674                 key.id =  record.id;
 675
 676                 SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
 677                 subscribersWrapper->erase(key);
 678
 679                 LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
 680                 llu = lluWrapper->get();
 681                 llu.iteration++;
 682                 lluWrapper->put(llu);
 683
 684                 txn.commit();
 685                 break;
 686             }
 687             catch(const DeadlockException&)
 688             {
 689                 continue;
 690             }
 691             catch(const DatabaseException& ex)
 692             {
 693                 halt(_instance->communicator(), ex);
 694             }
 695         }
 696         Ice::IdentitySeq ids;
 697         ids.push_back(id);
 698         _instance->observers()->removeSubscriber(llu, _name, ids);
 699     }
 700
 701     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 702     for(;;)
 703     {
 704         try
 705         {
 706             DatabaseConnectionPtr connection = _databaseCache->getConnection();
 707             TransactionHolder txn(connection);
 708
 709             SubscriberRecordKey key;
 710             key.topic = _id;
 711             key.id = subscriber->id();
 712
 713             SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
 714             subscribersWrapper->put(key, record);
 715
 716             // Update the LLU.
 717             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
 718             llu = lluWrapper->get();
 719             llu.iteration++;
 720             lluWrapper->put(llu);
 721
 722             txn.commit();
 723             break;
 724         }
 725         catch(const DeadlockException&)
 726         {
 727             continue;
 728         }
 729         catch(const DatabaseException& ex)
 730         {
 731             halt(_instance->communicator(), ex);
 732         }
 733     }
 734
 735     _subscribers.push_back(subscriber);
 736
 737     _instance->observers()->addSubscriber(llu, _name, record);
 738 }
 739
 740 Ice::ObjectPrx
 741 TopicImpl::subscribeAndGetPublisher(const QoS& qos, const Ice::ObjectPrx& obj)
 742 {
 743     Ice::Identity id = obj->ice_getIdentity();
 744
 745     TraceLevelsPtr traceLevels = _instance->traceLevels();
 746     if(traceLevels->topic > 0)
 747     {
 748         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 749         out << _name << ": subscribeAndGetPublisher: " << _instance->communicator()->identityToString(id);
 750
 751         if(traceLevels->topic > 1)
 752         {
 753             out << " endpoints: " << IceStormInternal::describeEndpoints(obj)
 754                 << " QoS: ";
 755             for(QoS::const_iterator p = qos.begin(); p != qos.end() ; ++p)
 756             {
 757                 if(p != qos.begin())
 758                 {
 759                     out << ',';
 760                 }
 761
 762             }
 763             out << " subscriptions: ";
 764             trace(out, _instance, _subscribers);
 765         }
 766     }
 767
 768     IceUtil::Mutex::Lock sync(_subscribersMutex);
 769
 770     SubscriberRecord record;
 771     record.id = id;
 772     record.obj = obj;
 773     record.theQoS = qos;
 774     record.topicName = _name;
 775     record.link = false;
 776     record.cost = 0;
 777
 778     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 779     if(p != _subscribers.end())
 780     {
 781         throw AlreadySubscribed();
 782     }
 783
 784     LogUpdate llu;
 785
 786     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 787     for(;;)
 788     {
 789         try
 790         {
 791             DatabaseConnectionPtr connection = _databaseCache->getConnection();
 792             TransactionHolder txn(connection);
 793
 794             SubscriberRecordKey key;
 795             key.topic = _id;
 796             key.id = subscriber->id();
 797
 798             SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
 799             subscribersWrapper->put(key, record);
 800
 801             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
 802             llu = lluWrapper->get();
 803             llu.iteration++;
 804             lluWrapper->put(llu);
 805
 806             txn.commit();
 807             break;
 808         }
 809         catch(const DeadlockException&)
 810         {
 811             continue;
 812         }
 813         catch(const DatabaseException& ex)
 814         {
 815             halt(_instance->communicator(), ex);
 816         }
 817     }
 818
 819     _subscribers.push_back(subscriber);
 820
 821     _instance->observers()->addSubscriber(llu, _name, record);
 822
 823     return subscriber->proxy();
 824 }
 825
 826 void
 827 TopicImpl::unsubscribe(const Ice::ObjectPrx& subscriber)
 828 {
 829     TraceLevelsPtr traceLevels = _instance->traceLevels();
 830     if(!subscriber)
 831     {
 832         if(traceLevels->topic > 0)
 833         {
 834             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 835             out << "unsubscribe with null subscriber.";
 836         }
 837         return;
 838     }
 839
 840     Ice::Identity id = subscriber->ice_getIdentity();
 841
 842     if(traceLevels->topic > 0)
 843     {
 844         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 845         out << _name << ": unsubscribe: " << _instance->communicator()->identityToString(id);
 846
 847         if(traceLevels->topic > 1)
 848         {
 849             out << " endpoints: " << IceStormInternal::describeEndpoints(subscriber);
 850             trace(out, _instance, _subscribers);
 851         }
 852     }
 853
 854     IceUtil::Mutex::Lock sync(_subscribersMutex);
 855     Ice::IdentitySeq ids;
 856     ids.push_back(id);
 857     removeSubscribers(ids);
 858 }
 859
 860 TopicLinkPrx
 861 TopicImpl::getLinkProxy()
 862 {
 863     // immutable
 864     if(_instance->publisherReplicaProxy())
 865     {
 866         return TopicLinkPrx::uncheckedCast(_instance->publisherReplicaProxy()->ice_identity(
 867                                                _linkPrx->ice_getIdentity()));
 868     }
 869     return _linkPrx;
 870 }
 871
 872 void
 873 TopicImpl::link(const TopicPrx& topic, Ice::Int cost)
 874 {
 875     TopicInternalPrx internal = TopicInternalPrx::uncheckedCast(topic);
 876     TopicLinkPrx link = internal->getLinkProxy();
 877
 878     TraceLevelsPtr traceLevels = _instance->traceLevels();
 879     if(traceLevels->topic > 0)
 880     {
 881         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 882         out << _name << ": link " << _instance->communicator()->identityToString(topic->ice_getIdentity())
 883             << " cost " << cost;
 884     }
 885
 886     IceUtil::Mutex::Lock sync(_subscribersMutex);
 887
 888     Ice::Identity id = topic->ice_getIdentity();
 889
 890     SubscriberRecord record;
 891     record.id = id;
 892     record.obj = link;
 893     record.theTopic = topic;
 894     record.topicName = _name;
 895     record.link = true;
 896     record.cost = cost;
 897
 898     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 899     if(p != _subscribers.end())
 900     {
 901         string name = identityToTopicName(id);
 902         LinkExists ex;
 903         ex.name = name;
 904         throw ex;
 905     }
 906
 907     LogUpdate llu;
 908
 909     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 910
 911     for(;;)
 912     {
 913         try
 914         {
 915             DatabaseConnectionPtr connection = _databaseCache->getConnection();
 916             TransactionHolder txn(connection);
 917
 918             SubscriberRecordKey key;
 919             key.topic = _id;
 920             key.id = id;
 921
 922             SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
 923             subscribersWrapper->put(key, record);
 924
 925             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
 926             llu = lluWrapper->get();
 927             llu.iteration++;
 928             lluWrapper->put(llu);
 929
 930             txn.commit();
 931             break;
 932         }
 933         catch(const DeadlockException&)
 934         {
 935             continue;
 936         }
 937         catch(const DatabaseException& ex)
 938         {
 939             halt(_instance->communicator(), ex);
 940         }
 941     }
 942
 943     _subscribers.push_back(subscriber);
 944
 945     _instance->observers()->addSubscriber(llu, _name, record);
 946 }
 947
 948 void
 949 TopicImpl::unlink(const TopicPrx& topic)
 950 {
 951     IceUtil::Mutex::Lock sync(_subscribersMutex);
 952     if(_destroyed)
 953     {
 954         throw Ice::ObjectNotExistException(__FILE__, __LINE__);
 955     }
 956
 957     Ice::Identity id = topic->ice_getIdentity();
 958
 959     vector<SubscriberPtr>::const_iterator p = find(_subscribers.begin(), _subscribers.end(), id);
 960     if(p == _subscribers.end())
 961     {
 962         string name = identityToTopicName(id);
 963         TraceLevelsPtr traceLevels = _instance->traceLevels();
 964         if(traceLevels->topic > 0)
 965         {
 966             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 967             out << _name << ": unlink " << name << " failed - not linked";
 968         }
 969
 970         NoSuchLink ex;
 971         ex.name = name;
 972         throw ex;
 973     }
 974
 975     TraceLevelsPtr traceLevels = _instance->traceLevels();
 976     if(traceLevels->topic > 0)
 977     {
 978         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 979         out << _name << " unlink " << _instance->communicator()->identityToString(id);
 980     }
 981
 982     Ice::IdentitySeq ids;
 983     ids.push_back(id);
 984     removeSubscribers(ids);
 985 }
 986
 987 void
 988 TopicImpl::reap(const Ice::IdentitySeq& ids)
 989 {
 990     IceUtil::Mutex::Lock sync(_subscribersMutex);
 991
 992     TraceLevelsPtr traceLevels = _instance->traceLevels();
 993     if(traceLevels->topic > 0)
 994     {
 995         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 996         out << _name << ": reap ";
 997         for(Ice::IdentitySeq::const_iterator p = ids.begin(); p != ids.end() ; ++p)
 998         {
 999             if(p != ids.begin())
1000             {
1001                 out << ",";
1002             }
1003             out << _instance->communicator()->identityToString(*p);
1004         }
1005     }
1006
1007     removeSubscribers(ids);
1008 }
1009
1010 void
1011 TopicImpl::shutdown()
1012 {
1013     IceUtil::Mutex::Lock sync(_subscribersMutex);
1014     _servant = 0;
1015
1016     // Shutdown each subscriber. This waits for the event queues to drain.
1017     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
1018     {
1019         (*p)->shutdown();
1020     }
1021 }
1022
1023 LinkInfoSeq
1024 TopicImpl::getLinkInfoSeq() const
1025 {
1026     IceUtil::Mutex::Lock sync(_subscribersMutex);
1027
1028     LinkInfoSeq seq;
1029     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
1030     {
1031         SubscriberRecord record = (*p)->record();
1032         if(record.link && !(*p)->errored())
1033         {
1034             LinkInfo info;
1035             info.name = identityToTopicName(record.theTopic->ice_getIdentity());
1036             info.cost = record.cost;
1037             info.theTopic = record.theTopic;
1038             seq.push_back(info);
1039         }
1040     }
1041     return seq;
1042 }
1043
1044 void
1045 TopicImpl::destroy()
1046 {
1047     IceUtil::Mutex::Lock sync(_subscribersMutex);
1048
1049     if(_destroyed)
1050     {
1051         throw Ice::ObjectNotExistException(__FILE__, __LINE__);
1052     }
1053     _destroyed = true;
1054
1055     TraceLevelsPtr traceLevels = _instance->traceLevels();
1056     if(traceLevels->topic > 0)
1057     {
1058         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1059         out << _name << ": destroy";
1060     }
1061
1062     // destroyInternal clears out the topic content.
1063     LogUpdate llu = {0,0};
1064     _instance->observers()->destroyTopic(destroyInternal(llu, true), _name);
1065 }
1066
1067 TopicContent
1068 TopicImpl::getContent() const
1069 {
1070     IceUtil::Mutex::Lock sync(_subscribersMutex);
1071
1072     TopicContent content;
1073     content.id = _id;
1074     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
1075     {
1076         // Don't return errored subscribers (subscribers that have
1077         // errored out, but not reaped due to a failure with the
1078         // master). This means we can avoid the reaping step later.
1079         if(!(*p)->errored())
1080         {
1081             content.records.push_back((*p)->record());
1082         }
1083     }
1084     return content;
1085 }
1086
1087 void
1088 TopicImpl::update(const SubscriberRecordSeq& records)
1089 {
1090     IceUtil::Mutex::Lock sync(_subscribersMutex);
1091
1092     // We do this with two scans. The first runs through the subscribers
1093     // that we have and removes those not in the init list. The second
1094     // runs through the init list and add the ones that don't
1095     // exist.
1096
1097     {
1098         vector<SubscriberPtr>::iterator p = _subscribers.begin();
1099         while(p != _subscribers.end())
1100         {
1101             SubscriberRecordSeq::const_iterator q;
1102             for(q = records.begin(); q != records.end(); ++q)
1103             {
1104                 if((*p)->id() == q->id)
1105                 {
1106                     break;
1107                 }
1108             }
1109             // The subscriber doesn't exist in the incoming subscriber
1110             // set so destroy it.
1111             if(q == records.end())
1112             {
1113                 (*p)->destroy();
1114                 p = _subscribers.erase(p);
1115             }
1116             else
1117             {
1118                 // Otherwise reset the reaped status if necessary.
1119                 (*p)->resetIfReaped();
1120                 ++p;
1121             }
1122         }
1123     }
1124
1125     for(SubscriberRecordSeq::const_iterator p = records.begin(); p != records.end(); ++p)
1126     {
1127         vector<SubscriberPtr>::iterator q;
1128         for(q = _subscribers.begin(); q != _subscribers.end(); ++q)
1129         {
1130             if((*q)->id() == p->id)
1131             {
1132                 break;
1133             }
1134         }
1135         if(q == _subscribers.end())
1136         {
1137             SubscriberPtr subscriber = Subscriber::create(_instance, *p);
1138             _subscribers.push_back(subscriber);
1139         }
1140     }
1141 }
1142
1143 bool
1144 TopicImpl::destroyed() const
1145 {
1146     IceUtil::Mutex::Lock sync(_subscribersMutex);
1147     return _destroyed;
1148 }
1149
1150 Ice::Identity
1151 TopicImpl::id() const
1152 {
1153     // immutable
1154     return _id;
1155 }
1156
1157 TopicPrx
1158 TopicImpl::proxy() const
1159 {
1160     // immutable
1161     Ice::ObjectPrx prx;
1162     if(_instance->topicReplicaProxy())
1163     {
1164         prx = _instance->topicReplicaProxy()->ice_identity(_id);
1165     }
1166     else
1167     {
1168         prx = _instance->topicAdapter()->createProxy(_id);
1169     }
1170     return TopicPrx::uncheckedCast(prx);
1171 }
1172
1173 namespace
1174 {
1175
1176 class TopicInternal_reapI : public AMI_TopicInternal_reap
1177 {
1178 public:
1179
1180     TopicInternal_reapI(const InstancePtr& instance, Ice::Long generation) :
1181         _instance(instance), _generation(generation)
1182     {
1183     }
1184
1185     virtual void ice_response()
1186     {
1187     }
1188
1189     virtual void ice_exception(const Ice::Exception& ex)
1190     {
1191         TraceLevelsPtr traceLevels = _instance->traceLevels();
1192         if(traceLevels->topic > 0)
1193         {
1194             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1195             out << "exception when calling `reap' on the master replica: " << ex;
1196         }
1197         _instance->node()->recovery(_generation);
1198     }
1199
1200 private:
1201
1202     const InstancePtr _instance;
1203     const Ice::Long _generation;
1204 };
1205
1206 }
1207
1208 void
1209 TopicImpl::publish(bool forwarded, const EventDataSeq& events)
1210 {
1211     TopicInternalPrx masterInternal;
1212     Ice::Long generation = -1;
1213     Ice::IdentitySeq reap;
1214     {
1215         // Use cached reads.
1216         CachedReadHelper unlock(_instance->node(), __FILE__, __LINE__);
1217
1218         //
1219         // Copy of the subscriber list so that event publishing can occur
1220         // in parallel.
1221         //
1222         vector<SubscriberPtr> copy;
1223         {
1224             IceUtil::Mutex::Lock sync(_subscribersMutex);
1225             copy = _subscribers;
1226         }
1227
1228         //
1229         // Queue each event, gathering a list of those subscribers that
1230         // must be reaped.
1231         //
1232         for(vector<SubscriberPtr>::const_iterator p = copy.begin(); p != copy.end(); ++p)
1233         {
1234             if(!(*p)->queue(forwarded, events) && (*p)->reap())
1235             {
1236                 reap.push_back((*p)->id());
1237             }
1238         }
1239
1240         // If there are no subscribers in error then we're done.
1241         if(reap.empty())
1242         {
1243             return;
1244         }
1245         if(!unlock.getMaster())
1246         {
1247             IceUtil::Mutex::Lock sync(_subscribersMutex);
1248             removeSubscribers(reap);
1249             return;
1250         }
1251         masterInternal = TopicInternalPrx::uncheckedCast(unlock.getMaster()->ice_identity(_id));
1252         generation = unlock.generation();
1253     }
1254
1255
1256     // Tell the master to reap this set of subscribers. This is an
1257     // AMI invocation so it shouldn't block the caller (in the
1258     // typical case) we do it outside of the mutex lock for
1259     // performance reasons.
1260     //
1261     // We must release the cached lock before calling this as the AMI
1262     // call may raise an exception in the caller (that is directly
1263     // call ice_exception) which calls recover() on the node which
1264     // would result in a deadlock since the node is locked.
1265     masterInternal->reap_async(new TopicInternal_reapI(_instance, generation), reap);
1266 }
1267
1268 void
1269 TopicImpl::observerAddSubscriber(const LogUpdate& llu, const SubscriberRecord& record)
1270 {
1271     IceUtil::Mutex::Lock sync(_subscribersMutex);
1272
1273     TraceLevelsPtr traceLevels = _instance->traceLevels();
1274     if(traceLevels->topic > 0)
1275     {
1276         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1277         out << _name << ": add replica observer: " << _instance->communicator()->identityToString(record.id);
1278
1279         if(traceLevels->topic > 1)
1280         {
1281             out << " endpoints: " << IceStormInternal::describeEndpoints(record.obj)
1282                 << " QoS: ";
1283             for(QoS::const_iterator p = record.theQoS.begin(); p != record.theQoS.end() ; ++p)
1284             {
1285                 if(p != record.theQoS.begin())
1286                 {
1287                     out << ',';
1288                 }
1289                 out << '[' << p->first << "," << p->second << ']';
1290             }
1291         }
1292         out << " llu: " << llu.generation << "/" << llu.iteration;
1293     }
1294
1295     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
1296     if(p != _subscribers.end())
1297     {
1298         // If the subscriber is already in the database display a
1299         // diagnostic.
1300         TraceLevelsPtr traceLevels = _instance->traceLevels();
1301         if(traceLevels->topic > 0)
1302         {
1303             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1304             out << _instance->communicator()->identityToString(record.id) << ": already subscribed";
1305         }
1306         return;
1307     }
1308
1309     SubscriberPtr subscriber = Subscriber::create(_instance, record);
1310     for(;;)
1311     {
1312         try
1313         {
1314             DatabaseConnectionPtr connection = _databaseCache->getConnection();
1315             TransactionHolder txn(connection);
1316
1317             SubscriberRecordKey key;
1318             key.topic = _id;
1319             key.id = subscriber->id();
1320
1321             SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
1322             subscribersWrapper->put(key, record);
1323
1324             // Update the LLU.
1325             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
1326             lluWrapper->put(llu);
1327
1328             txn.commit();
1329             break;
1330         }
1331         catch(const DeadlockException&)
1332         {
1333             continue;
1334         }
1335         catch(const DatabaseException& ex)
1336         {
1337             halt(_instance->communicator(), ex);
1338         }
1339     }
1340
1341     _subscribers.push_back(subscriber);
1342 }
1343
1344 void
1345 TopicImpl::observerRemoveSubscriber(const LogUpdate& llu, const Ice::IdentitySeq& ids)
1346 {
1347     TraceLevelsPtr traceLevels = _instance->traceLevels();
1348     if(traceLevels->topic > 0)
1349     {
1350         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1351         out << _name << ": remove replica observer: ";
1352         for(Ice::IdentitySeq::const_iterator id = ids.begin(); id != ids.end(); ++id)
1353         {
1354             if(id != ids.begin())
1355             {
1356                 out << ",";
1357             }
1358             out << _instance->communicator()->identityToString(*id);
1359         }
1360         out << " llu: " << llu.generation << "/" << llu.iteration;
1361     }
1362
1363     IceUtil::Mutex::Lock sync(_subscribersMutex);
1364
1365     // Remove the subscriber from the subscribers list. If the
1366     // subscriber had a local failure and was removed from the
1367     // subscriber list it could already be gone. That's not a problem.
1368     for(Ice::IdentitySeq::const_iterator id = ids.begin(); id != ids.end(); ++id)
1369     {
1370         vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), *id);
1371         if(p != _subscribers.end())
1372         {
1373             (*p)->destroy();
1374             _subscribers.erase(p);
1375         }
1376     }
1377
1378     // Next remove from the database.
1379     for(;;)
1380     {
1381         try
1382         {
1383             DatabaseConnectionPtr connection = _databaseCache->getConnection();
1384             TransactionHolder txn(connection);
1385
1386             for(Ice::IdentitySeq::const_iterator id = ids.begin(); id != ids.end(); ++id)
1387             {
1388                 SubscriberRecordKey key;
1389                 key.topic = _id;
1390                 key.id = *id;
1391
1392                 SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
1393                 subscribersWrapper->erase(key);
1394             }
1395
1396             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
1397             lluWrapper->put(llu);
1398
1399             txn.commit();
1400             break;
1401         }
1402         catch(const DeadlockException&)
1403         {
1404             continue;
1405         }
1406         catch(const DatabaseException& ex)
1407         {
1408             halt(_instance->communicator(), ex);
1409         }
1410     }
1411 }
1412
1413 void
1414 TopicImpl::observerDestroyTopic(const LogUpdate& llu)
1415 {
1416     IceUtil::Mutex::Lock sync(_subscribersMutex);
1417
1418     if(_destroyed)
1419     {
1420         return;
1421     }
1422     _destroyed = true;
1423
1424     TraceLevelsPtr traceLevels = _instance->traceLevels();
1425     if(traceLevels->topic > 0)
1426     {
1427         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
1428         out << _name << ": destroyed";
1429         out << " llu: " << llu.generation << "/" << llu.iteration;
1430     }
1431     destroyInternal(llu, false);
1432 }
1433
1434 Ice::ObjectPtr
1435 TopicImpl::getServant() const
1436 {
1437     return _servant;
1438 }
1439
1440 LogUpdate
1441 TopicImpl::destroyInternal(const LogUpdate& origLLU, bool master)
1442 {
1443     _instance->publishAdapter()->remove(_linkPrx->ice_getIdentity());
1444     _instance->publishAdapter()->remove(_publisherPrx->ice_getIdentity());
1445
1446     // Destroy each of the subscribers.
1447     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
1448     {
1449         (*p)->destroy();
1450     }
1451     _subscribers.clear();
1452
1453     // Clear out the database records related to this topic.
1454     LogUpdate llu;
1455     for(;;)
1456     {
1457         try
1458         {
1459             DatabaseConnectionPtr connection = _databaseCache->getConnection();
1460             TransactionHolder txn(connection);
1461
1462             // Erase all subscriber records and the topic record.
1463             SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
1464             subscribersWrapper->eraseTopic(_id);
1465
1466             // Update the LLU.
1467             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
1468             if(master)
1469             {
1470                 llu = lluWrapper->get();
1471                 llu.iteration++;
1472             }
1473             else
1474             {
1475                 llu = origLLU;
1476             }
1477             lluWrapper->put(llu);
1478
1479             txn.commit();
1480             break;
1481         }
1482         catch(const DeadlockException&)
1483         {
1484             continue;
1485         }
1486         catch(const DatabaseException& ex)
1487         {
1488             halt(_instance->communicator(), ex);
1489         }
1490     }
1491
1492     _instance->topicAdapter()->remove(_id);
1493
1494     _servant = 0;
1495
1496     return llu;
1497 }
1498
1499 void
1500 TopicImpl::removeSubscribers(const Ice::IdentitySeq& ids)
1501 {
1502     Ice::IdentitySeq removed;
1503
1504     // First remove the subscriber from the subscribers list. Its
1505     // possible that some of these subscribers have already been
1506     // removed (consider, for example, a concurrent reap call from two
1507     // replicas on the same subscriber). To avoid sending unnecessary
1508     // observer updates keep track of the observers that are actually
1509     // removed.
1510     for(Ice::IdentitySeq::const_iterator id = ids.begin(); id != ids.end(); ++id)
1511     {
1512         vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), *id);
1513         if(p != _subscribers.end())
1514         {
1515             (*p)->destroy();
1516             _subscribers.erase(p);
1517             removed.push_back(*id);
1518         }
1519     }
1520
1521     // If there is no further work to do we are done.
1522     if(removed.empty())
1523     {
1524         return;
1525     }
1526
1527     // Next update the database and send the notification to any
1528     // slaves.
1529     LogUpdate llu;
1530     for(;;)
1531     {
1532         try
1533         {
1534             DatabaseConnectionPtr connection = _databaseCache->getConnection();
1535             TransactionHolder txn(connection);
1536
1537             for(Ice::IdentitySeq::const_iterator id = ids.begin(); id != ids.end(); ++id)
1538             {
1539                 SubscriberRecordKey key;
1540                 key.topic = _id;
1541                 key.id = *id;
1542
1543                 SubscribersWrapperPtr subscribersWrapper = _databaseCache->getSubscribers(connection);
1544                 subscribersWrapper->erase(key);
1545             }
1546
1547             LLUWrapperPtr lluWrapper = _databaseCache->getLLU(connection);
1548             llu = lluWrapper->get();
1549             llu.iteration++;
1550             lluWrapper->put(llu);
1551
1552             txn.commit();
1553             break;
1554         }
1555         catch(const DeadlockException&)
1556         {
1557             continue;
1558         }
1559         catch(const DatabaseException& ex)
1560         {
1561             halt(_instance->communicator(), ex);
1562         }
1563     }
1564
1565     _instance->observers()->removeSubscriber(llu, _name, ids);
1566 }