orte/liborte/ORTEAppSendThread.c

   1 /*
   2  *  $Id: ORTEAppSendThread.c,v 0.0.0.1  2003/08/21
   3  *
   4  *  DEBUG:  section 24                  Sending thread
   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 void ORTESendData(ORTEDomain *d,ObjectEntryAID *objectEntryAID,Boolean meta) {
  26   struct sockaddr_in  des;
  27   ObjectEntryOID      *objectEntryOID;
  28   AppParams           *appParams;
  29   ObjectId            oid=OID_APP;
  30   int                 i;
  31
  32   objectEntryOID=ObjectEntryOID_find(objectEntryAID,&oid);
  33   if (objectEntryOID) {
  34     appParams=(AppParams*)objectEntryOID->attributes;
  35     for(i=0;i<appParams->unicastIPAddressCount;i++) {
  36       des.sin_family=AF_INET;
  37       des.sin_addr.s_addr = htonl(appParams->unicastIPAddressList[i]);
  38       if (meta) {
  39         des.sin_port = htons((u_int16_t)appParams->metatrafficUnicastPort);
  40         sock_sendto (
  41             &d->taskSend.sock,
  42             d->mbSend.cdrStream.buffer,
  43             d->mbSend.cdrStream.length,
  44             &des,
  45             sizeof(des));
  46       } else {
  47         des.sin_port = htons((u_int16_t)appParams->userdataUnicastPort);
  48         sock_sendto (
  49             &d->taskSend.sock,
  50             d->mbSend.cdrStreamDirect->buffer,
  51             d->mbSend.cdrStreamDirect->length,
  52             &des,
  53             sizeof(des));
  54       }
  55     }
  56   }
  57   //prepare buffer for next sending
  58   d->mbSend.cdrStream.length=RTPS_HEADER_LENGTH;
  59   d->mbSend.cdrStream.bufferPtr=d->mbSend.cdrStream.buffer+RTPS_HEADER_LENGTH;
  60   d->mbSend.needSend=ORTE_FALSE;
  61   d->mbSend.containsInfoReply=ORTE_FALSE;
  62 }
  63
  64 /*****************************************************************************/
  65 void ORTEAppSendThread(ORTEDomain *d) {
  66   struct timespec     wtime;
  67   NtpTime             actTime,nextExpire,whenExpire,sleepingTime;
  68   int32_t             s,ms;
  69
  70   debug(24,10) ("ORTEAppSendThread: start\n");
  71   d->objectEntry.htimNeedWakeUp=ORTE_TRUE;
  72
  73   while (!d->taskSend.terminate) {
  74     actTime=getActualNtpTime();
  75     pthread_rwlock_wrlock(&d->objectEntry.htimRootLock);
  76     if (htimerRoot_next_expire(&d->objectEntry,&nextExpire)==0) {
  77       NTPTIME_BUILD(whenExpire,300); //max time for sleeping (no. events)
  78       NtpTimeAdd(nextExpire,actTime,whenExpire);
  79     }
  80     pthread_rwlock_unlock(&d->objectEntry.htimRootLock);
  81     NtpTimeDisAssembToUs(wtime.tv_sec,wtime.tv_nsec,nextExpire);
  82     wtime.tv_nsec*=1000;  //conver to nano seconds
  83     NtpTimeSub(sleepingTime,nextExpire,actTime);
  84     NtpTimeDisAssembToMs(s,ms,sleepingTime);
  85     if (s<0) s=ms=0;
  86     debug(24,4) ("ORTEAppSendThread: sleeping for %lis %lims\n",s,ms);
  87     if (!((wtime.tv_sec==0) && (wtime.tv_nsec==0))) {
  88       pthread_mutex_timedlock(
  89           &d->objectEntry.htimSendMutex,
  90           &wtime);
  91     }
  92     debug(24,7) ("ORTEAppSendThread: fired\n");
  93     actTime=getActualNtpTime();
  94     pthread_rwlock_wrlock(&d->objectEntry.objRootLock);
  95     pthread_rwlock_wrlock(&d->objectEntry.htimRootLock);
  96     d->objectEntry.htimNeedWakeUp=ORTE_FALSE;
  97     htimerRoot_run_expired(d,&actTime);
  98     d->objectEntry.htimNeedWakeUp=ORTE_TRUE;
  99     pthread_rwlock_unlock(&d->objectEntry.htimRootLock);
 100     pthread_rwlock_unlock(&d->objectEntry.objRootLock);
 101   }
 102   pthread_rwlock_unlock(&d->objectEntry.htimRootLock);
 103   debug(24,10) ("ORTEAppSendThread: finished\n");
 104 }
 105