orte/liborte/ORTEDomainMgr.c

   1 /*
   2  *  $Id: ORTEDomainMgr.c,v 0.0.0.1      2003/09/12
   3  *
   4  *  DEBUG:  section 29                  Domain manager
   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.h"
  23
  24 /*****************************************************************************/
  25 ORTEDomain *
  26 ORTEDomainMgrCreate(int domain, ORTEDomainProp *prop,
  27                     ORTEDomainAppEvents *events,Boolean suspended) {
  28   ORTEDomain        *d;
  29   ObjectEntryOID    *objectEntryOID;
  30   AppParams         *appParams;
  31   CSTWriterParams   cstWriterParams;
  32   CSTReaderParams   cstReaderParams;
  33   char              iflocal[MAX_INTERFACES*MAX_STRING_IPADDRESS_LENGTH];
  34   char              sIPAddress[MAX_STRING_IPADDRESS_LENGTH];
  35   char              sbuff[128];
  36   int               i;
  37   u_int16_t         port=0;
  38
  39   debug(29,10) ("ORTEDomainMgrCreate: start\n");
  40   //Create domainApplication
  41   d=MALLOC(sizeof(ORTEDomain));
  42   if (!d) return NULL;  //no memory
  43   //initialization local values
  44   d->domain=domain;
  45   d->taskRecvMetatraffic.terminate=ORTE_TRUE;
  46   d->taskSend.terminate=ORTE_TRUE;
  47   d->taskRecvMetatraffic.sock.port=0;
  48   d->taskSend.sock.port=0;
  49   //init structure objectEntry
  50   ObjectEntryHID_init_root_field(&d->objectEntry);
  51   pthread_rwlock_init(&d->objectEntry.objRootLock,NULL);
  52   htimerRoot_init_queue(&d->objectEntry);
  53   pthread_rwlock_init(&d->objectEntry.htimRootLock,NULL);
  54   sem_init(&d->objectEntry.htimSendSem, 0, 0);
  55
  56   //create domainProp
  57   if (prop!=NULL) {
  58     memcpy(&d->domainProp,prop,sizeof(ORTEDomainProp));
  59   } else {
  60     ORTEDomainPropDefaultGet(&d->domainProp);
  61   }
  62
  63   //print local IP addresses
  64   iflocal[0]=0;
  65   if (d->domainProp.IFCount) {
  66     for(i=0;i<d->domainProp.IFCount;i++)
  67       strcat(iflocal,IPAddressToString(d->domainProp.IFProp[i].ipAddress,sIPAddress));
  68     debug(29,2) ("ORTEDomainMgrCreate: localIPAddres(es) %s\n",iflocal);
  69   } else{
  70     debug(29,2) ("ORTEDomainMgrCreate: no activ interface card\n");
  71   }
  72
  73   //DomainEvents
  74   if (events!=NULL) {
  75     memcpy(&d->domainEvents,events,sizeof(ORTEDomainAppEvents));
  76   } else {
  77     memset(&d->domainEvents,0,sizeof(ORTEDomainAppEvents));
  78   }
  79
  80   //local buffers
  81   d->mbRecvMetatraffic.cdrStream.buffer=
  82       (u_int8_t*)MALLOC(d->domainProp.recvBuffSize);
  83   d->mbRecvUserdata.cdrStream.buffer=NULL;
  84   d->mbSend.cdrStream.buffer=
  85       (u_int8_t*)MALLOC(d->domainProp.sendBuffSize);
  86   if ((!d->mbRecvMetatraffic.cdrStream.buffer) ||
  87       (!d->mbSend.cdrStream.buffer)) {    //no memory
  88     FREE(d->mbRecvMetatraffic.cdrStream.buffer);
  89     FREE(d->mbSend.cdrStream.buffer);
  90     FREE(d);
  91     return NULL;
  92   }
  93   d->mbRecvMetatraffic.cdrStream.bufferPtr=d->mbRecvMetatraffic.cdrStream.buffer;
  94   d->mbRecvMetatraffic.cdrStream.length=0;
  95   d->mbRecvUserdata.cdrStream.bufferPtr=d->mbRecvUserdata.cdrStream.buffer;
  96   d->mbRecvUserdata.cdrStream.length=0;
  97   d->mbSend.cdrStream.bufferPtr=d->mbSend.cdrStream.buffer;
  98   d->mbSend.cdrStream.length=0;
  99
 100   //Sockets
 101   sock_init_udp(&d->taskRecvMetatraffic.sock);
 102   sock_init_udp(&d->taskSend.sock);
 103   if (d->domainProp.multicast.enabled) {
 104     Domain2PortMulticastMetatraffic(d->domain,port);
 105   } else {
 106     Domain2Port(d->domain,port);
 107   }
 108   sock_bind(&d->taskRecvMetatraffic.sock,port);    //receiving port
 109   debug(29,2) ("ORTEDomainMgrCreate: bind on port(Recv): %u\n",
 110                d->taskRecvMetatraffic.sock.port);
 111   sock_bind(&d->taskSend.sock,0);       //give me sending port
 112   debug(29,2) ("ORTEDomainAppCreate: bind on port(Send): %u\n",
 113                d->taskSend.sock.port);
 114   if (d->domainProp.multicast.enabled) {
 115     struct ip_mreq mreq;
 116     //ttl
 117     if(sock_setsockopt(&d->taskSend.sock,IP_MULTICAST_TTL,
 118         &d->domainProp.multicast.ttl,sizeof(d->domainProp.multicast.ttl))>=0) {
 119       debug(29,2) ("ORTEDomainAppCreate: ttl set on: %u\n",
 120            d->domainProp.multicast.ttl);
 121     }
 122     // join multicast group
 123     mreq.imr_multiaddr.s_addr=htonl(d->domainProp.multicast.ipAddress);
 124     mreq.imr_interface.s_addr=htonl(INADDR_ANY);
 125     if(sock_setsockopt(&d->taskRecvUserdata.sock,IP_ADD_MEMBERSHIP,
 126         (void *) &mreq, sizeof(mreq))>=0) {
 127       debug(29,2) ("ORTEDomainAppCreate: listening to mgroup %s\n",
 128            IPAddressToString(d->domainProp.multicast.ipAddress,sIPAddress));
 129     }
 130   }
 131   if ((d->taskRecvMetatraffic.sock.fd<0) || (d->taskSend.sock.fd<0) ||
 132       (port!=d->taskRecvMetatraffic.sock.port))  {
 133     printf("Error creating socket(s).\n");
 134     sock_cleanup(&d->taskRecvMetatraffic.sock);
 135     sock_cleanup(&d->taskSend.sock);
 136     FREE(d->mbRecvMetatraffic.cdrStream.buffer);
 137     FREE(d->mbSend.cdrStream.buffer);
 138     FREE(d);
 139     return NULL;
 140   }
 141
 142   //Generates local GUID
 143   if (d->domainProp.IFCount>0)
 144     d->guid.hid=d->domainProp.IFProp[0].ipAddress;
 145   else
 146     d->guid.hid=StringToIPAddress("127.0.0.1");
 147   d->guid.aid=(d->taskSend.sock.port<<8)+MANAGER;
 148   d->guid.oid=OID_APP;
 149   debug(29,2) ("ORTEDomainMgrCreate: GUID: %#10.8x,%#10.8x,%#10.8x\n",
 150                d->guid.hid,d->guid.aid,d->guid.oid);
 151
 152   //create HEADER of message for sending task
 153   RTPSHeaderCreate(d->mbSend.cdrStream.buffer,d->guid.hid,d->guid.aid);
 154   d->mbSend.cdrStream.bufferPtr=
 155       d->mbSend.cdrStream.buffer+RTPS_HEADER_LENGTH;
 156   d->mbSend.cdrStream.length=RTPS_HEADER_LENGTH;
 157   d->mbSend.needSend=ORTE_FALSE;
 158   d->mbSend.containsInfoReply=ORTE_FALSE;
 159
 160   //Self object data & fellow managers object data
 161   appParams=(AppParams*)MALLOC(sizeof(AppParams));
 162   AppParamsInit(appParams);
 163   appParams->expirationTime=d->domainProp.baseProp.expirationTime;
 164   VENDOR_ID_OCERA(appParams->vendorId);
 165   appParams->hostId=d->guid.hid;
 166   appParams->appId=d->guid.aid;
 167   appParams->metatrafficUnicastPort=d->taskRecvMetatraffic.sock.port;
 168   appParams->userdataUnicastPort=0;  //Manager support only metatraffic
 169   if (d->domainProp.multicast.enabled) {
 170     //multicast
 171     for(i=0;i<d->domainProp.IFCount;i++)
 172       appParams->metatrafficMulticastIPAddressList[i]=d->domainProp.IFProp[i].ipAddress;
 173     appParams->metatrafficMulticastIPAddressCount=d->domainProp.IFCount;
 174   } else {
 175     //unicast
 176     if (d->domainProp.IFCount) {
 177       for(i=0;i<d->domainProp.IFCount;i++)
 178         appParams->unicastIPAddressList[i]=d->domainProp.IFProp[i].ipAddress;
 179       appParams->unicastIPAddressCount=d->domainProp.IFCount;
 180     } else {
 181       appParams->unicastIPAddressList[0]=StringToIPAddress("127.0.0.1");
 182       appParams->unicastIPAddressCount=1;
 183     }
 184   }
 185   //managerKeyList
 186   if (!d->domainProp.keys) {
 187     appParams->managerKeyList[0]=StringToIPAddress("127.0.0.1");
 188     for(i=0;i<d->domainProp.IFCount;i++)
 189       appParams->managerKeyList[i+1]=d->domainProp.IFProp[i].ipAddress;
 190     appParams->managerKeyCount=d->domainProp.IFCount+1;
 191   } else {
 192     appParams->managerKeyCount=i=0;
 193     while (getStringPart(d->domainProp.keys,':',i,sbuff)) {
 194       appParams->managerKeyList[appParams->managerKeyCount++]=
 195           StringToIPAddress(sbuff);
 196     }
 197   }
 198   d->appParams=appParams;
 199   //insert object, doesn't need to be locked
 200   d->objectEntryOID=objectEntryAdd(d,&d->guid,(void*)appParams);
 201   d->objectEntryOID->private=ORTE_TRUE;
 202
 203   //CST objects
 204   //  writerApplicationSelf (WAS)
 205   NTPTIME_ZERO(cstWriterParams.waitWhileDataUnderwayTime);
 206   cstWriterParams.refreshPeriod=d->domainProp.baseProp.refreshPeriod;
 207   cstWriterParams.repeatAnnounceTime=d->domainProp.baseProp.repeatAnnounceTime;
 208   NTPTIME_ZERO(cstWriterParams.delayResponceTime);
 209   cstWriterParams.HBMaxRetries=d->domainProp.baseProp.HBMaxRetries;
 210   cstWriterParams.fullAcknowledge=ORTE_FALSE;
 211   CSTWriterInit(d,&d->writerApplicationSelf,d->objectEntryOID,
 212       OID_WRITE_APPSELF,&cstWriterParams,NULL);
 213   //  add to WAS remote writer(s)
 214   i=0;
 215   while (getStringPart(d->domainProp.mgrs,':',i,sbuff)>0) {
 216     GUID_RTPS guid;
 217     IPAddress ipAddress=StringToIPAddress(sbuff);
 218     guid.hid=ipAddress;
 219     guid.aid=AID_UNKNOWN;
 220     guid.oid=OID_APP;
 221     if (!objectEntryFind(d,&guid)) {
 222       appParams=(AppParams*)MALLOC(sizeof(AppParams));
 223       AppParamsInit(appParams);
 224       appParams->hostId=guid.hid;
 225       appParams->appId=guid.aid;
 226       appParams->metatrafficUnicastPort=d->appParams->metatrafficUnicastPort;
 227       appParams->userdataUnicastPort=0;  //Manager support only metatraffic
 228       appParams->unicastIPAddressList[0]=ipAddress;
 229       appParams->unicastIPAddressCount=1;
 230       objectEntryOID=objectEntryAdd(d,&guid,(void*)appParams);
 231       CSTWriterAddRemoteReader(d,&d->writerApplicationSelf,objectEntryOID,
 232           OID_READ_MGR);
 233       debug(29,2) ("ORTEDomainAppCreate: add fellow manager (%s)\n",
 234                   IPAddressToString(ipAddress,sIPAddress));
 235     }
 236   }
 237   //  readerManagers
 238   cstReaderParams.delayResponceTimeMin=d->domainProp.baseProp.delayResponceTimeACKMin;
 239   cstReaderParams.delayResponceTimeMax=d->domainProp.baseProp.delayResponceTimeACKMax;
 240   cstReaderParams.ACKMaxRetries=d->domainProp.baseProp.ACKMaxRetries;
 241   cstReaderParams.repeatActiveQueryTime=iNtpTime;  //RM cann't repeatly send ACKf
 242   cstReaderParams.fullAcknowledge=ORTE_FALSE;      //never will send ACK
 243   CSTReaderInit(d,&d->readerManagers,d->objectEntryOID,
 244       OID_READ_MGR,&cstReaderParams,NULL);
 245   //  readerApplications
 246   cstReaderParams.delayResponceTimeMin=d->domainProp.baseProp.delayResponceTimeACKMin;
 247   cstReaderParams.delayResponceTimeMax=d->domainProp.baseProp.delayResponceTimeACKMax;
 248   cstReaderParams.ACKMaxRetries=d->domainProp.baseProp.ACKMaxRetries;
 249   cstReaderParams.repeatActiveQueryTime=d->domainProp.baseProp.repeatActiveQueryTime;
 250   cstReaderParams.fullAcknowledge=ORTE_TRUE;
 251   CSTReaderInit(d,&d->readerApplications,d->objectEntryOID,
 252       OID_READ_APP,&cstReaderParams,NULL);
 253   //  writerApplications
 254   NTPTIME_ZERO(cstWriterParams.waitWhileDataUnderwayTime);
 255   cstWriterParams.refreshPeriod=iNtpTime;  //only WAS,WM can refresh csChange(s)
 256   cstWriterParams.repeatAnnounceTime=d->domainProp.baseProp.repeatAnnounceTime;
 257   NtpTimeAssembFromMs(cstWriterParams.delayResponceTime,0,20);
 258   cstWriterParams.HBMaxRetries=d->domainProp.baseProp.HBMaxRetries;
 259   cstWriterParams.fullAcknowledge=ORTE_FALSE;
 260   CSTWriterInit(d,&d->writerApplications,d->objectEntryOID,
 261       OID_WRITE_APP,&cstWriterParams,NULL);
 262   //  writerManagers
 263   NTPTIME_ZERO(cstWriterParams.waitWhileDataUnderwayTime);
 264   cstWriterParams.refreshPeriod=d->domainProp.baseProp.refreshPeriod;
 265   cstWriterParams.repeatAnnounceTime=d->domainProp.baseProp.repeatAnnounceTime;
 266   NtpTimeAssembFromMs(cstWriterParams.delayResponceTime,0,20);
 267   cstWriterParams.HBMaxRetries=d->domainProp.baseProp.HBMaxRetries;
 268   cstWriterParams.fullAcknowledge=ORTE_TRUE;
 269   CSTWriterInit(d,&d->writerManagers,d->objectEntryOID,
 270       OID_WRITE_MGR,&cstWriterParams,NULL);
 271
 272   //add csChange for WAS
 273   appSelfParamChanged(d,ORTE_FALSE,ORTE_FALSE,ORTE_FALSE,ORTE_TRUE);
 274
 275   //Start threads
 276   if (!suspended) {
 277     ORTEDomainStart(d,ORTE_TRUE,ORTE_FALSE,ORTE_TRUE);
 278   }
 279
 280   debug(29,10) ("ORTEDomainMgrCreate: finished\n");
 281   return d;
 282 }
 283
 284 /*****************************************************************************/
 285 Boolean
 286 ORTEDomainMgrDestroy(ORTEDomain *d) {
 287
 288   debug(29,10) ("ORTEDomainMgrDestroy: start\n");
 289   pthread_rwlock_wrlock(&d->objectEntry.objRootLock);
 290   pthread_rwlock_wrlock(&d->objectEntry.htimRootLock);
 291   appSelfParamChanged(d,ORTE_TRUE,ORTE_TRUE,ORTE_FALSE,ORTE_FALSE);
 292   pthread_rwlock_unlock(&d->objectEntry.htimRootLock);
 293   pthread_rwlock_unlock(&d->objectEntry.objRootLock);
 294   //Stoping threads
 295   if(!d->taskRecvMetatraffic.terminate) {
 296     d->taskRecvMetatraffic.terminate=ORTE_TRUE;
 297     ORTEDomainWakeUpReceivingThread(d,
 298         &d->taskSend.sock,d->taskRecvMetatraffic.sock.port);
 299     pthread_join(d->taskRecvMetatraffic.thread,NULL);
 300   }
 301   if (!d->taskSend.terminate) {
 302     d->taskSend.terminate=ORTE_TRUE;
 303     ORTEDomainWakeUpSendingThread(&d->objectEntry);
 304     pthread_join(d->taskSend.thread,NULL);
 305   }
 306   debug(29,8) ("ORTEDomainMgrDestroy: threads stoped and destroyed\n");
 307
 308   objectEntryDump(&d->objectEntry);
 309
 310   //Sockets
 311   sock_cleanup(&d->taskRecvMetatraffic.sock);
 312   sock_cleanup(&d->taskSend.sock);
 313
 314   //Semas
 315   sem_destroy(&d->objectEntry.htimSendSem);
 316
 317   //rwLocks
 318   pthread_rwlock_destroy(&d->objectEntry.objRootLock);
 319   pthread_rwlock_destroy(&d->objectEntry.htimRootLock);
 320
 321   //CSTReaders and CSTWriters
 322   CSTReaderDelete(d,&d->readerManagers);
 323   CSTReaderDelete(d,&d->readerApplications);
 324   CSTWriterDelete(d,&d->writerManagers);
 325   CSTWriterDelete(d,&d->writerApplications);
 326   CSTWriterDelete(d,&d->writerApplicationSelf);
 327
 328   //objects in objectsEntry
 329   objectEntryDeleteAll(d,&d->objectEntry);
 330
 331   FREE(d->mbRecvMetatraffic.cdrStream.buffer);
 332   FREE(d->mbSend.cdrStream.buffer);
 333   FREE(d);
 334   debug(29,10) ("ORTEDomainMgrDestroy: finished\n");
 335   return ORTE_TRUE;
 336 }