schedcl.cpp

OPCSERVER源代码

    upt->bandwidth0 += percent * period;
   }
 upt->bandwidth = (upt->bandwidth0 + (1 << 11)) / (1 << 12);
#endif
#endif

#if 0 == SATURATE_PERCENT
  if (upt->bandwidth > 100) upt->bandwidth = 100;
#endif

SkipBW:
 if (upt->bandwidth >= 2)
   {
    if (statperiod < loSTATPERIOD) statperiod = loSTATPERIOD;
    statperiod += statperiod >> 1;
    if (statperiod <= nextsleep) nextsleep = statperiod >> 1;
//       UL_WARNING((LOGID, "UPDATE %u%% -> %u", (int)upt->bandwidth, nextsleep));
   }
 if (loMAXSLEEP <  nextsleep &&
     loMAXSLEEP < (nextsleep >>= 1)) nextsleep = loMAXSLEEP;
/* if (nextsleep > 20000)
   {
    UL_WARNING((LOGID, "UPDATE %u", nextsleep));
   }*/
 upt->nextsleep = nextsleep;

 return;
}

DWORD lo_adjust_bandwidth(DWORD bandwidth, loMilliSec bandw_time, loMilliSec curtime)
{
 if (-1 != bandwidth)
   {
    curtime -= bandw_time;
    curtime -= loSTATPERIOD >> 4;
    if (curtime < ~curtime) /* no overflow */
      {
       curtime  +=  loSTATPERIOD;
       bandwidth = (loSTATPERIOD * bandwidth + (curtime >> 1)) / curtime;
      }
   }
 return bandwidth;
}

static void unadvise_all(LightOPCServer *cli);

/* Be careful looking on "#if LO_TIMEBACK": here are 4(!) 
   different handlig for this minor problem. 
In short words, under some circumance (e.g. runnning VDM)
the Win2k may produce delays that are shorter than requested.
For example, Sleep(20) may awake after 15ms. Then Sleep(5) 
returns immediately (i.e. there would be a few hundreds of Sleep(5) 
for real delay of 5ms). 

I don't know is it problem of Sleep() or WaitFor...() or GetTickCount().
Is it problem of waiting or rounding of system clocks...
Anyway we have a problem.

Basically we have to increase delay by the timer granularity 
on the next sleep. We have following choices:

LO_TIMEBACK == 0 : ignore problem. Call cond_wait(...,delay) till the 
time elapsed. It's a kind of empty loop with too much context switches.

Otherwise we can detect we have been awoked before requsted and no 
other events pending.

LO_TIMEBACK > 0 : run the _one_ empty update cycle and then increase
delay (round-up) for some (LO_TIMEBACK ms) time:
        a) rounding-up by nanosecond recalculation of timer granularity;
        b) simply add the granularity (millisecond accuracy) 
           to the NEXTSLEEP.

LO_TIMEBACK < 0 : assume we have awoken at the time; 
perform update cycle prematurely as if it just in time (round-down);
then increase NEXTSLEEP by the difference of the real and 
assumed time [TimeShift].

The last way minimizes average updaterate error.
*/

void LightOPCServer::client_scheduler(void)
{
#if LO_TIMEBACK > 0 /* round up */
 loMilliSec TimeBack = 0;
#endif
#if LO_TIMEBACK < 0 /* round down */
 loMilliSec TimeShift = 0;
#else
# define    TimeShift  (0)
#endif
 loUpdTimes upt;
 loDCL_ABSTIME(abst);
// if (!loIS_VALID(this)) return;
 UL_NOTICE((LOG_SR("thr_async(%p) started"), this));
 loThrControl_accept(&async);
 memset(&upt, 0, sizeof(upt));
 upt.nextsleep = loMAXSLEEP;
 lo_MILLISECABS(upt.finish, abst);

 lw_mutex_lock(&async.lk);

 if (async.tstate >= 0)
   {
#if LO_USE_BOTHMODEL
 if (0 == ole_initialized &&
     (/*se->driver.*/ldFlags & loDf_BOTHMODEL))
   {
    HRESULT oli = CoInitializeEx(NULL, COINIT_MULTITHREADED);
    if (SUCCEEDED(oli)) ole_initialized = 1;
    else UL_WARNING((LOGID, "%!l client_scheduler(%p)::CoInitialize() FAILED",
                   oli, this));
   }
#endif

 for(;;)
   {
//if (async.state) UL_ERROR((LOG_SR("SHC  %d"), async.state));
#if 0
 UL_DEBUG((LOGID, "WA:%d %u + %ums =%u", 
   upt.datasent, bandwidth_time, upt.nextsleep, bandwidth_time + upt.nextsleep));
#endif
    if (0 == async.tstate && /* don't sleep when a request pending! */
        !(shuttingdown && shutdown_conn.request))
      {
#if LO_TIMEBACK > 0
       if (TimeBack)
         {
          if ((bandwidth_time - TimeBack) > LO_TIMEBACK) TimeBack = 0;
#if 1
          /* millisec round-up */
          lw_conds_timedwait(&async.cond, &async.lk, 
                             loABSREL(abst, upt.nextsleep + lo_timegran_ms));
          UL_DEBUG((LOGID, "WAIT AA: %u", lo_timegran_ms));
#else
          /* nanosec round-up */
          unsigned add_sleep;
          ULONGLONG upt_nextsleep = upt.nextsleep;
          upt_nextsleep *= 10000;
          if (add_sleep = (unsigned)(upt_nextsleep % lo_timegran_100ns))
            add_sleep = (lo_timegran_100ns - add_sleep + 9999) / 10000 & 0x0ff;
               /* add no more than 1/4 sec */
          lw_conds_timedwait(&async.cond, &async.lk, 
                             loABSREL(abst, upt.nextsleep add_sleep));
          UL_DEBUG((LOGID, "WAIT AA: %u", add_sleep));
#endif
         }
       else 
#endif /*LO_TIMEBACK > 0*/
       lw_conds_timedwait(&async.cond, &async.lk, 
                         loABSREL(abst, upt.nextsleep + TimeShift));
      }
    LO_HEAPCHECK(this);
    if (async.tstate < 0) break;
    lo_MILLISEC(upt.awoke);
#if 1
// if (bandwidth_time + upt.nextsleep >= upt.awoke)
   UL_DEBUG((LOGID, "WAIT WA:%d:%d %u + %ums =%u %+d", 
   upt.datasent, async.tstate, 
   bandwidth_time, upt.nextsleep, bandwidth_time + upt.nextsleep,   
   upt.awoke - (bandwidth_time + upt.nextsleep)));
#endif

    if (!async.tstate && !lo_NO_TIMEBACK &&
        upt.nextsleep > upt.awoke - bandwidth_time)
#if LO_TIMEBACK < 0
      {
        /* round down */
        UL_TRACE((LOGID, "WAITing problem was:%d now:%d", 
          TimeShift, upt.nextsleep - (upt.awoke - bandwidth_time)));
        TimeShift = upt.nextsleep - (upt.awoke - bandwidth_time);
        if (TimeShift > lo_timegran_ms) TimeShift = 0/*lo_timegran_ms*/;
        upt.awoke += TimeShift;
      }
    else TimeShift = 0;
#elif LO_TIMEBACK > 0
      {
        /* round up */
        TimeBack = upt.awoke;
        UL_TRACE((LOGID, "WAITing problem"));
      }
#else
    UL_WARNING((LOGID, "WAITing problem")); /* no solution */
#endif /*LO_TIMEBACK*/

    if (shuttingdown && shutdown_conn.request)
      {
       loShutdownConn sc = shutdown_conn;
       loShutdownConn_init(&shutdown_conn);
       lw_mutex_unlock(&async.lk);
       UL_NOTICE((LOG_SR("Going to IOPCShutdown")));
       loShutdownConn_call(&sc);
       lw_mutex_lock(&async.lk);
      }
    if (loThrControl_outof01(&async))
      {
       UL_DEBUG((LOG_SR("Going to shutdown (%p) ..."), this));
       loShutdownConn_clear(&shutdown_conn);
       lw_mutex_unlock(&async.lk); /* MUST be called outside of lock !!!*/
       UL_NOTICE((LOG_SR("Unadvising groups thr_async(%p) ..."), this));
       unadvise_all(this);
       otrk.ot_disconnect_all(0);
       clear_all();
       otrk.ot_disconnect_all(0);
       loQueueBcast_abort(&q_ret);
       UL_NOTICE((LOG_SR("Self terminating thr_async(%p) ..."), this));
       selfdestroy(); /* at this point <this> is no longer valid */
       UL_NOTICE((LOGID, "Self terminating thr_async(%p) Ok", this));
       return;
      }
/* on stopped we have to sleep
   on shutdown we have to sleep or to break */
    else
      {
       if (1 == async.tstate) /* no timeout occured */
         {
          async.tstate = 0; /* clear signal */
          if (q_req.req)
            {
             lo_rio_requests(this); /* Do requests first ==> low delay */
             if (loThrControl_outof01(&async)) continue;
            }
         }
       lw_mutex_unlock(&async.lk);

       upt.datasent = 0;
       upt.overload = 0;
       upt.updated = 0;
       upt.nextsleep = loHUGESLEEP;

       lo_MILLISEC(upt.beginupd);  upt.beginupd += TimeShift;
       update_sheduler(otrk.ot_stopped? 0: this, se, &upt);

       lw_mutex_lock(&async.lk);
#if 0
       if (q_req.req)
         {
          lo_rio_requests(this);  /* it does loThrControl_outof01(&async); */
         }
#endif
       if (upt.datasent) se->driver.ldCurrentTime(&ctxt.cactx, &ostatus.ftLastUpdateTime);

       lo_MILLISECABS(bandwidth_time, abst);  bandwidth_time += TimeShift;
       sched_timings(&upt, bandwidth_time);
       ostatus.dwBandWidth = /*stopped? (DWORD)-1: */(DWORD)upt.bandwidth;
      }
#if 0
    if (shuttingdown && upt.nextsleep > 4096) upt.nextsleep = 4096;
#endif
   } /* end of for(;;) */
 } /* if (async.tstate >= 0) */

{
#if LO_USE_BOTHMODEL
 int oleini = ole_initialized; ole_initialized = 0;
#endif

 loThrControl_finish(&async);
 lw_mutex_unlock(&async.lk);

 UL_NOTICE((LOGID, "thr_async(%p) finished", this));
#if LO_USE_BOTHMODEL
 if (oleini) CoUninitialize();
#endif
}
}

static void unadvise_all(LightOPCServer *cli)
{
 unsigned gndx;
 LightOPCGroup **grpl;

 cli->lock_read();
 gndx = cli->grl.gl_count;
 grpl = (LightOPCGroup**)cli->grl.gl_list;
 while(gndx)
   {
    LightOPCGroup *grp;
    if ((grp = grpl[--gndx]) && grp->advise_present)
      grp->clear_advise();
   }
 cli->unlock();
}
/* end of shedcl.cpp */