orte/liborte/objectEntryTimer.c

   1 /*
   2  *  $Id: objectEntryTimer.c,v 0.0.0.1   2003/09/10
   3  *
   4  *  DEBUG:  section 12                  Timer function on object from eventEntry
   5  *
   6  *  -------------------------------------------------------------------
   7  *                                ORTE
   8  *                      Open Real-Time Ethernet
   9  *
  10  *                      Copyright (C) 2001-2006
  11  *  Department of Control Engineering FEE CTU Prague, Czech Republic
  12  *                      http://dce.felk.cvut.cz
  13  *                      http://www.ocera.org
  14  *
  15  *  Author:              Petr Smolik    petr@smoliku.cz
  16  *  Advisor:             Pavel Pisa
  17  *  Project Responsible: Zdenek Hanzalek
  18  *  --------------------------------------------------------------------
  19  *
  20  *  This program is free software; you can redistribute it and/or modify
  21  *  it under the terms of the GNU General Public License as published by
  22  *  the Free Software Foundation; either version 2 of the License, or
  23  *  (at your option) any later version.
  24  *
  25  *  This program is distributed in the hope that it will be useful,
  26  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  27  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  28  *  GNU General Public License for more details.
  29  *
  30  */
  31
  32 #include "orte_all.h"
  33
  34 /*****************************************************************************/
  35 int
  36 objectEntryPurgeTimer(ORTEDomain *d,void *vobjectEntryOID) {
  37   ObjectEntryOID   *objectEntryOID=(ObjectEntryOID*)vobjectEntryOID;
  38   GUID_RTPS        guid;
  39
  40   guid=objectEntryOID->guid;
  41   if ((d->guid.aid & 0x03) == MANAGER) {
  42     if ((guid.aid & 0x03) == MANAGER) {
  43       pthread_rwlock_wrlock(&d->writerManagers.lock);
  44       CSTWriterMakeGAP(d,&d->writerManagers,&guid);
  45       pthread_rwlock_unlock(&d->writerManagers.lock);
  46     }
  47     if (((guid.aid & 0x03) == MANAGEDAPPLICATION) &&
  48         (objectEntryOID->appMOM)) {
  49       pthread_rwlock_wrlock(&d->writerApplications.lock);
  50       CSTWriterMakeGAP(d,&d->writerApplications,&guid);
  51       pthread_rwlock_unlock(&d->writerApplications.lock);
  52     }
  53   }
  54   if ((d->guid.aid & 0x03) == MANAGEDAPPLICATION) {
  55     switch (guid.oid & 0x07) {
  56       case OID_APPLICATION:
  57         break;
  58       case OID_PUBLICATION:
  59         pthread_rwlock_wrlock(&d->writerPublications.lock);
  60         CSTWriterMakeGAP(d,&d->writerPublications,&guid);
  61         pthread_rwlock_unlock(&d->writerPublications.lock);
  62         break;
  63       case OID_SUBSCRIPTION:
  64         pthread_rwlock_wrlock(&d->writerSubscriptions.lock);
  65         CSTWriterMakeGAP(d,&d->writerSubscriptions,&guid);
  66         pthread_rwlock_unlock(&d->writerSubscriptions.lock);
  67         break;
  68     }
  69   }
  70   debug(12,3) ("purged: 0x%x-0x%x-0x%x object removed\n",
  71                objectEntryOID->objectEntryHID->hid,
  72                objectEntryOID->objectEntryAID->aid,
  73                objectEntryOID->oid);
  74   objectEntryDelete(d,objectEntryOID,ORTE_TRUE);
  75   objectEntryDump(&d->objectEntry);
  76
  77   debug(12,10) ("objectEntryPurgeTimer: finished\n");
  78   return 2;
  79 }
  80
  81 /*****************************************************************************/
  82 void
  83 removeSubscriptionsOnLocalPublications(ORTEDomain *d,ObjectEntryOID *robjectEntryOID) {
  84   CSTWriter       *cstWriter;
  85   CSTRemoteReader *cstRemoteReader;
  86   ObjectEntryOID  *objectEntryOID;
  87
  88   if ((!d) || (!robjectEntryOID)) return;
  89   pthread_rwlock_wrlock(&d->publications.lock);
  90   gavl_cust_for_each(CSTWriter,
  91                      &d->publications,cstWriter) {
  92     pthread_rwlock_wrlock(&cstWriter->lock);
  93     gavl_cust_for_each(ObjectEntryOID,
  94                        robjectEntryOID->objectEntryAID,
  95                        objectEntryOID) {
  96       cstRemoteReader=CSTRemoteReader_find(cstWriter,&objectEntryOID->guid);
  97       CSTWriterDestroyRemoteReader(d,cstRemoteReader);
  98     }
  99     pthread_rwlock_unlock(&cstWriter->lock);
 100   }
 101   pthread_rwlock_unlock(&d->publications.lock);
 102 }
 103
 104 /*****************************************************************************/
 105 void
 106 removePublicationsOnLocalSubscriptions(ORTEDomain *d,ObjectEntryOID *robjectEntryOID) {
 107   CSTReader       *cstReader;
 108   CSTRemoteWriter *cstRemoteWriter;
 109   ObjectEntryOID  *objectEntryOID;
 110
 111   if ((!d) || (!robjectEntryOID)) return;
 112   pthread_rwlock_wrlock(&d->subscriptions.lock);
 113   gavl_cust_for_each(CSTReader,
 114                      &d->subscriptions,cstReader) {
 115     pthread_rwlock_wrlock(&cstReader->lock);
 116     gavl_cust_for_each(ObjectEntryOID,
 117                        robjectEntryOID->objectEntryAID,
 118                        objectEntryOID) {
 119       cstRemoteWriter=CSTRemoteWriter_find(cstReader,&objectEntryOID->guid);
 120       if (cstRemoteWriter) {
 121         CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 122         if ((cstReader->cstRemoteWriterCounter==0) && (cstReader->createdByPattern)) {
 123           debug(12,2) ("scheduled: 0x%x-0x%x-0x%x for remove (patternSubscription)\n",
 124                       cstReader->guid.hid,cstReader->guid.aid,cstReader->guid.oid);
 125           ORTESubscriptionDestroyLocked(cstReader);
 126         }
 127       }
 128     }
 129     pthread_rwlock_unlock(&cstReader->lock);
 130   }
 131   pthread_rwlock_unlock(&d->subscriptions.lock);
 132 }
 133
 134 /*****************************************************************************/
 135 void
 136 removeApplication(ORTEDomain *d,ObjectEntryOID *robjectEntryOID) {
 137   GUID_RTPS        guid;
 138   ObjectEntryOID   *objectEntryOID;
 139   CSTRemoteWriter  *cstRemoteWriter;
 140   CSTRemoteReader  *cstRemoteReader;
 141
 142   if (!robjectEntryOID) return;
 143   if (!gavl_cmp_guid(&robjectEntryOID->guid,&d->guid)) return;
 144   debug(12,3) ("application removed - AID: 0%x\n",robjectEntryOID->guid.aid);
 145
 146   guid=robjectEntryOID->guid;
 147   //publication, subsription and application
 148   pthread_rwlock_wrlock(&d->writerPublications.lock);
 149   guid.oid=OID_READ_PUBL;
 150   cstRemoteReader=CSTRemoteReader_find(&d->writerPublications,&guid);
 151   CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 152   pthread_rwlock_unlock(&d->writerPublications.lock);
 153   pthread_rwlock_wrlock(&d->writerSubscriptions.lock);
 154   guid.oid=OID_READ_SUBS;
 155   cstRemoteReader=CSTRemoteReader_find(&d->writerSubscriptions,&guid);
 156   CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 157   pthread_rwlock_unlock(&d->writerSubscriptions.lock);
 158   pthread_rwlock_wrlock(&d->readerPublications.lock);
 159   guid.oid=OID_WRITE_PUBL;
 160   cstRemoteWriter=CSTRemoteWriter_find(&d->readerPublications,&guid);
 161   CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 162   pthread_rwlock_unlock(&d->readerPublications.lock);
 163   pthread_rwlock_wrlock(&d->readerSubscriptions.lock);
 164   guid.oid=OID_WRITE_SUBS;
 165   cstRemoteWriter=CSTRemoteWriter_find(&d->readerSubscriptions,&guid);
 166   CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 167   pthread_rwlock_unlock(&d->readerSubscriptions.lock);
 168   //destroy all services
 169   removePublicationsOnLocalSubscriptions(d,robjectEntryOID);
 170   removeSubscriptionsOnLocalPublications(d,robjectEntryOID);
 171   //destroy all object - the object will be disconneced in objectEntryDelete
 172   objectEntryOID=ObjectEntryOID_first(robjectEntryOID->objectEntryAID);
 173   while (objectEntryOID) {
 174     ObjectEntryOID *objectEntryOID_delete=objectEntryOID;
 175     objectEntryOID=ObjectEntryOID_next(robjectEntryOID->objectEntryAID,objectEntryOID);
 176     switch (objectEntryOID_delete->oid & 0x07) {
 177       case OID_PUBLICATION:
 178         pthread_rwlock_wrlock(&d->psEntry.publicationsLock);
 179         PublicationList_delete(&d->psEntry,objectEntryOID_delete);
 180         pthread_rwlock_unlock(&d->psEntry.publicationsLock);
 181         break;
 182       case OID_SUBSCRIPTION:
 183         pthread_rwlock_wrlock(&d->psEntry.subscriptionsLock);
 184         SubscriptionList_delete(&d->psEntry,objectEntryOID_delete);
 185         pthread_rwlock_unlock(&d->psEntry.subscriptionsLock);
 186         break;
 187     }
 188     objectEntryDelete(d,objectEntryOID_delete,ORTE_TRUE);
 189   }
 190 }
 191
 192 /*****************************************************************************/
 193 //Remove manager
 194 void
 195 removeManager(ORTEDomain *d,ObjectEntryOID *robjectEntryOID) {
 196   CSTRemoteWriter  *cstRemoteWriter;
 197   ObjectEntryAID   *objectEntryAID;
 198   GUID_RTPS        guid;
 199
 200   if (!robjectEntryOID) return;
 201   debug(12,3) ("manager removed\n");
 202
 203   guid=robjectEntryOID->guid;
 204   //exists another live Manager on going down node
 205   gavl_cust_for_each(ObjectEntryAID,
 206                      robjectEntryOID->objectEntryHID,objectEntryAID) {
 207     if (((objectEntryAID->aid & 0x03) == MANAGER) &&
 208         (objectEntryAID->aid!=robjectEntryOID->guid.aid))
 209       break;  //yes
 210   }
 211   if (!objectEntryAID) {  //not exists
 212     objectEntryAID=ObjectEntryAID_first(robjectEntryOID->objectEntryHID);
 213     while (objectEntryAID) {
 214       ObjectEntryAID *objectEntryAID_delete=objectEntryAID;
 215       objectEntryAID=ObjectEntryAID_next(robjectEntryOID->objectEntryHID,objectEntryAID);
 216       if ((objectEntryAID_delete->aid & 0x03) == MANAGEDAPPLICATION) {
 217         ObjectEntryOID   *objectEntryOID;
 218         objectEntryOID=ObjectEntryOID_find(objectEntryAID_delete,&guid.oid);
 219         if (gavl_cmp_guid(&objectEntryOID->guid,&d->guid)) { //!=
 220           removeApplication(d,objectEntryOID);
 221         }
 222       }
 223     }
 224   }
 225   pthread_rwlock_wrlock(&d->readerApplications.lock);
 226   pthread_rwlock_wrlock(&d->readerManagers.lock);
 227   guid.oid=OID_WRITE_APP;
 228   cstRemoteWriter=CSTRemoteWriter_find(&d->readerApplications,&guid);
 229   CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 230   guid.oid=OID_WRITE_MGR;
 231   cstRemoteWriter=CSTRemoteWriter_find(&d->readerManagers,&guid);
 232   CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 233   pthread_rwlock_unlock(&d->readerApplications.lock);
 234   pthread_rwlock_unlock(&d->readerManagers.lock);
 235   objectEntryDelete(d,robjectEntryOID,ORTE_TRUE);
 236 }
 237
 238
 239 /*****************************************************************************/
 240 int
 241 objectEntryExpirationTimer(ORTEDomain *d,void *vobjectEntryOID) {
 242   ObjectEntryOID   *objectEntryOID=(ObjectEntryOID*)vobjectEntryOID;
 243   ObjectEntryOID   *objectEntryOID1;
 244   ObjectEntryAID   *objectEntryAID;
 245   CSTWriter        *cstWriter;
 246   CSTReader        *cstReader;
 247   CSTRemoteWriter  *cstRemoteWriter;
 248   CSTRemoteReader  *cstRemoteReader;
 249   CSChange         *csChange;
 250   GUID_RTPS        guid;
 251
 252   //Manager, Manager expired
 253   guid=objectEntryOID->guid;
 254   //Event
 255   generateEvent(d,&guid,objectEntryOID->attributes,ORTE_FALSE);
 256   debug(12,3) ("expired: 0x%x-0x%x-0x%x removed\n",
 257                objectEntryOID->guid.hid,
 258                objectEntryOID->guid.aid,
 259                objectEntryOID->guid.oid);
 260   if (((d->guid.aid & 3) == MANAGER) &&
 261       ((guid.aid & 0x03) == MANAGER)) {
 262     pthread_rwlock_wrlock(&d->readerManagers.lock);
 263     pthread_rwlock_wrlock(&d->writerApplications.lock);
 264     pthread_rwlock_wrlock(&d->readerApplications.lock);
 265     guid.oid=OID_WRITE_APPSELF;
 266     cstRemoteWriter=CSTRemoteWriter_find(&d->readerManagers,&guid);
 267     CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 268     guid.oid=OID_WRITE_APP;
 269     cstRemoteWriter=CSTRemoteWriter_find(&d->readerApplications,&guid);
 270     CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 271     guid.oid=OID_READ_APP;
 272     cstRemoteReader=CSTRemoteReader_find(&d->writerApplications,&guid);
 273     CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 274     guid.oid=objectEntryOID->oid;  //restore oid
 275     //generate csChange for writerManager with alive=FALSE
 276     csChange=(CSChange*)MALLOC(sizeof(CSChange));
 277     CSChangeAttributes_init_head(csChange);
 278     csChange->guid=guid;
 279     csChange->alive=ORTE_FALSE;
 280     csChange->cdrCodec.buffer=NULL;
 281     CSTWriterAddCSChange(d,&d->writerManagers,csChange);
 282     gavl_cust_for_each(ObjectEntryAID,
 283                        objectEntryOID->objectEntryHID,objectEntryAID) {
 284       if (((objectEntryAID->aid & 0x03) == MANAGER) &&
 285           (objectEntryAID->aid!=objectEntryOID->guid.aid))
 286         break;  //yes
 287     }
 288     //if there is no another manager from expired node -> remove all app.
 289     if (!objectEntryAID) {
 290       gavl_cust_for_each(ObjectEntryAID,
 291                          objectEntryOID->objectEntryHID,objectEntryAID) {
 292         if ((objectEntryAID->aid & 0x03) == MANAGEDAPPLICATION) {
 293           if ((objectEntryOID1=ObjectEntryOID_find(objectEntryAID,&guid.oid))) {
 294             eventDetach(d,
 295                 objectEntryOID1->objectEntryAID,
 296                 &objectEntryOID1->expirationPurgeTimer,
 297                 0);
 298             eventAdd(d,
 299                 objectEntryOID1->objectEntryAID,
 300                 &objectEntryOID1->expirationPurgeTimer,
 301                 0,
 302                 "ExpirationTimer",
 303                 objectEntryExpirationTimer,
 304                 NULL,
 305                 objectEntryOID1,
 306                 NULL);
 307           }
 308         }
 309       }
 310     }
 311     pthread_rwlock_unlock(&d->readerApplications.lock);
 312     pthread_rwlock_unlock(&d->writerApplications.lock);
 313     pthread_rwlock_unlock(&d->readerManagers.lock);
 314   }
 315   //Manager, Application expired
 316   if (((d->guid.aid & 3) == MANAGER) &&
 317       ((guid.aid & 0x03) == MANAGEDAPPLICATION)) {
 318      pthread_rwlock_wrlock(&d->writerApplicationSelf.lock);
 319      pthread_rwlock_wrlock(&d->writerManagers.lock);
 320      pthread_rwlock_wrlock(&d->writerApplications.lock);
 321      pthread_rwlock_wrlock(&d->readerApplications.lock);
 322      guid.oid=OID_WRITE_APPSELF;  /* local app */
 323      cstRemoteWriter=CSTRemoteWriter_find(&d->readerApplications,&guid);
 324      CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 325      guid.oid=OID_WRITE_APP;      /* remote app */
 326      cstRemoteWriter=CSTRemoteWriter_find(&d->readerApplications,&guid);
 327      CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 328      guid.oid=OID_READ_APP;
 329      cstRemoteReader=CSTRemoteReader_find(&d->writerApplications,&guid);
 330      CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 331      guid.oid=OID_READ_MGR;
 332      cstRemoteReader=CSTRemoteReader_find(&d->writerManagers,&guid);
 333      CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 334      if (objectEntryOID->appMOM) {
 335         guid.oid=objectEntryOID->oid;  //restore oid
 336         //generate csChange for writerApplication with alive=FALSE
 337         csChange=(CSChange*)MALLOC(sizeof(CSChange));
 338         parameterUpdateCSChange(csChange,d->appParams,ORTE_TRUE);
 339         csChange->guid=guid;
 340         csChange->alive=ORTE_FALSE;
 341         csChange->cdrCodec.buffer=NULL;
 342         CSTWriterAddCSChange(d,&d->writerApplications,csChange);
 343         //increment vargAppsSequenceNumber and make csChange
 344         SeqNumberInc(d->appParams->vargAppsSequenceNumber,
 345                      d->appParams->vargAppsSequenceNumber);
 346         appSelfParamChanged(d,ORTE_FALSE,ORTE_FALSE,ORTE_TRUE,ORTE_TRUE);
 347      } else {
 348        objectEntryDelete(d,objectEntryOID,ORTE_TRUE);
 349        objectEntryOID=NULL;
 350      }
 351      pthread_rwlock_unlock(&d->writerApplicationSelf.lock);
 352      pthread_rwlock_unlock(&d->writerManagers.lock);
 353      pthread_rwlock_unlock(&d->writerApplications.lock);
 354      pthread_rwlock_unlock(&d->readerApplications.lock);
 355   }
 356   //Application
 357   if ((d->guid.aid & 0x03) == MANAGEDAPPLICATION) {
 358     switch (guid.oid & 0x07) {
 359       case OID_APPLICATION:
 360         if ((guid.aid & 0x03) == MANAGER) {                  //Manager
 361           removeManager(d,objectEntryOID);
 362           objectEntryOID=NULL;
 363         }
 364         if ((guid.aid & 0x03) == MANAGEDAPPLICATION) {       //Application
 365           removeApplication(d,objectEntryOID);
 366           objectEntryOID=NULL;
 367          }
 368         break;
 369       case OID_PUBLICATION:
 370         pthread_rwlock_wrlock(&d->subscriptions.lock);
 371         gavl_cust_for_each(CSTReader,&d->subscriptions,cstReader) {
 372           pthread_rwlock_wrlock(&cstReader->lock);
 373           cstRemoteWriter=CSTRemoteWriter_find(cstReader,&guid);
 374           if (cstRemoteWriter) {
 375             CSTReaderDestroyRemoteWriter(d,cstRemoteWriter);
 376             if ((cstReader->cstRemoteWriterCounter==0) && (cstReader->createdByPattern)) {
 377               debug(12,2) ("scheduled: 0x%x-0x%x-0x%x for remove (patternSubscription)\n",
 378                           cstReader->guid.hid,cstReader->guid.aid,cstReader->guid.oid);
 379               ORTESubscriptionDestroyLocked(cstReader);
 380             }
 381           }
 382           pthread_rwlock_unlock(&cstReader->lock);
 383         }
 384         pthread_rwlock_unlock(&d->subscriptions.lock);
 385         pthread_rwlock_wrlock(&d->publications.lock);
 386         cstWriter=CSTWriter_find(&d->publications,&guid);
 387         if (cstWriter) {
 388           CSTWriterDelete(d,cstWriter);
 389           CSTWriter_delete(&d->publications,cstWriter);
 390           FREE(cstWriter);
 391         }
 392         pthread_rwlock_unlock(&d->publications.lock);
 393         pthread_rwlock_wrlock(&d->psEntry.publicationsLock);
 394         PublicationList_delete(&d->psEntry,objectEntryOID);
 395         pthread_rwlock_unlock(&d->psEntry.publicationsLock);
 396         if (!objectEntryOID->privateCreated) { //not local object cann't be purged
 397           objectEntryDelete(d,objectEntryOID,ORTE_TRUE);
 398           objectEntryOID=NULL;
 399         }
 400         break;
 401       case OID_SUBSCRIPTION:
 402         pthread_rwlock_wrlock(&d->publications.lock);
 403         gavl_cust_for_each(CSTWriter,&d->publications,cstWriter) {
 404           cstRemoteReader=CSTRemoteReader_find(cstWriter,&guid);
 405           CSTWriterDestroyRemoteReader(d,cstRemoteReader);
 406         }
 407         pthread_rwlock_unlock(&d->publications.lock);
 408         pthread_rwlock_wrlock(&d->subscriptions.lock);
 409         cstReader=CSTReader_find(&d->subscriptions,&guid);
 410         if (cstReader) {
 411           CSTReaderDelete(d,cstReader);
 412           CSTReader_delete(&d->subscriptions,cstReader);
 413           FREE(cstReader);
 414         }
 415         pthread_rwlock_unlock(&d->subscriptions.lock);
 416         pthread_rwlock_wrlock(&d->psEntry.subscriptionsLock);
 417         SubscriptionList_delete(&d->psEntry,objectEntryOID);
 418         pthread_rwlock_unlock(&d->psEntry.subscriptionsLock);
 419         if (!objectEntryOID->privateCreated) { //local object cann't be purged immediately
 420           objectEntryDelete(d,objectEntryOID,ORTE_TRUE);
 421           objectEntryOID=NULL;
 422         }
 423         break;
 424     }
 425   }
 426   if (objectEntryOID) {
 427     eventDetach(d,
 428             objectEntryOID->objectEntryAID,
 429             &objectEntryOID->expirationPurgeTimer,
 430             0);
 431     eventAdd(d,
 432             objectEntryOID->objectEntryAID,
 433             &objectEntryOID->expirationPurgeTimer,
 434             0,
 435             "PurgeTimer",
 436             objectEntryPurgeTimer,
 437             NULL,
 438             objectEntryOID,
 439             &d->domainProp.baseProp.purgeTime);
 440     debug(12,3) ("expired: 0x%x-0x%x-0x%x marked for remove(%ds)\n",
 441                  objectEntryOID->objectEntryHID->hid,
 442                  objectEntryOID->objectEntryAID->aid,
 443                  objectEntryOID->oid,
 444                  d->domainProp.baseProp.purgeTime.seconds);
 445   }
 446   objectEntryDump(&d->objectEntry);
 447   debug(12,3) ("expired: finished\n");
 448   if (!objectEntryOID) return 2;
 449   return 0;
 450 }