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