fn shares(&self) -> usize {
        if let ControllerStatus::Disabled(shares) = self.state.get() {
            return shares;
        }

        let queue = self.queue.borrow();
        let mut expected = 0.0;
        let mut processed = 0.0;
        let now = Instant::now();

        for (exp, source) in queue.iter() {
            let remaining_time = exp.saturating_duration_since(now);
            trace!(
                "Remaining time for this source: {:#?}, total_units {}",
                remaining_time,
                source.total_units()
            );
            let time_fraction =
                1.0 - (remaining_time.as_secs_f64() / source.expected_duration().as_secs_f64());
            if remaining_time.as_nanos() == 0 && now.saturating_duration_since(*exp).as_secs() > 5 {
                // already too late, bump it up hard
                self.last_shares.set(1000);
                return 1000;
            }
            expected += source.total_units() as f64 * time_fraction;
            processed += source.processed_units() as f64;
        }

        // so little time has passed we can't really make any useful prediction
        if expected < 0.01 {
            return self.last_shares.get();
        }

        let error = 1.0 - processed / expected;
        let accumulated_error = self.accumulated_error.get();
        let acc = accumulated_error + error;
        self.accumulated_error.set(acc);
        let delta_error = error - self.last_error.get();
        self.last_error.set(error);

        // How did we pick our constants:
        //  * As with any stable PI controller we want the bulk of our gain to come from
        //    P.
        //  * As we normalize the maximum error to 1 we know that the gain should be at
        //    most 1000
        //  * physically, Ki can be expressed as Kp / Tau where Tau is a time constant,
        //    roughly equivalent to how many periods need to pass for the integral term
        //    to generate the same gain as the proportional term. We set that to 6 so
        //    the controller is not too sluggish, which is around 1.5 seconds on the
        //    default 250ms adjustment period.
        //
        //  We can then write X + X/6 = 1000, and solving for X we have the constants
        // below
        let kp = 850.0;
        let ki = kp / 6.0;
        let dshares = ki * error + kp * delta_error;

        let mut shares = (dshares + self.last_shares.get() as f64) as isize;
        shares = std::cmp::min(shares, 1000);
        shares = std::cmp::max(shares, self.min_shares.get() as isize);
        let shares = shares as usize;

        trace!(
            "processed: {}. expected: {} error: {}, delta_error {} , kp term {}, ki term {}, \
             shares: {}",
            processed,
            expected,
            error,
            delta_error,
            ki * error,
            kp * delta_error,
            shares
        );
        self.last_shares.set(shares);
        shares
    }
/// PI controller for shares.
///
/// We are not dealing with the derivative constant here: it is too risky
/// given the generic nature of the processes under control.
///
/// The variable we are controlling is the speed at which the units are
/// processed. Also, because we don't know what units are and different
/// items in the queue may have different magnitudes we need to work
/// with normalized units.
///
/// The error is the difference between our effective speed and the desired
/// speed:
///
///    e(t) = units_expected/delta_t -  units_processed/ delta_t,
///
///  and because we are normalizing:
///
///    e(t) = 1/delta_t * (1 - units_processed / units_expected)
///
///  The controller output is then:
///
///    u(t) = Kp * e(t) + Ki + Integral{0, t} e(t)
///
///  There are a couple of practical problems with that.
///
///  * The first is that the output of the controller would be zero if we
///    there are no error
///  * The second is that our integral term would accumulate errors that may
///    not be comparable as we accumulate artifacts of the delta_t
///    calculation (as we'll never in practice keep delta_t constant)
///
///  The way we'll solve this is by expressing an alternate error E(T) which
/// is  essentially the integral of e(t) in time:
///
///    E(t) = 1 - units_processed / units_expected.
///
///  That is easy to compute, as it is essentially the total count of units
/// for  all items in the queue, both processed and expected.
///
///  We can now express our output function as the derivative of U(t), the
/// output  function for the integral of the error:
///
///    u(t) = d(U(t)) / dt = Kp * dE(t) /dt + Ki * E(t)
///
///  Now we are calculating how many shares should be added or removed to
/// the  last output, and not how much the shares should be. It also
/// eliminates any  dependency on time when calculating the error which
/// increases resiliency.
PI controller for shares.
We are not dealing with the derivative constant here: it is too risky
given the generic nature of the processes under control.

The variable we are controlling is the speed at which the units are
processed. Also, because we don't know what units are and different
items in the queue may have different magnitudes we need to work
with normalized units.

The error is the difference between our effective speed and the desired
speed.



and because we are normalizing.



The controller output is then.



There are a couple of practical problems with that.

The first is that the output of the controller would be zero if we
there are no error
The second is that our integral term would accumulate errors that may
not be comparable as we accumulate artifacts of the delta_t
calculation (as we'll never in practice keep delta_t constant)

The way we'll solve this is by expressing an alternate error E(T) which
is  essentially the integral of e(t) in time.

E(t) = 1 - units_processed / units_expected.

That is easy to compute, as it is essentially the total count of units
for  all items in the queue, both processed and expected.

We can now express our output function as the derivative of U(t), the
output  function for the integral of the error.



Now we are calculating how many shares should be added or removed to
the  last output, and not how much the shares should be. It also
eliminates any  dependency on time when calculating the error which
increases resiliency.
PI controller for shares . We are not dealing with the derivative constant here : it is too risky given the generic nature of the processes under control . The variable we are controlling is the speed at which the units are processed . Also because we don ' t know what units are and different items in the queue may have different magnitudes we need to work with normalized units . The error is the difference between our effective speed and the desired speed . and because we are normalizing . The controller output is then . There are a couple of practical problems with that . The first is that the output of the controller would be zero if we there are no error The second is that our integral term would accumulate errors that may not be comparable as we accumulate artifacts of the delta_t calculation ( as we ' ll never in practice keep delta_t constant ) The way we ' ll solve this is by expressing an alternate error E ( T ) which is essentially the integral of e ( t ) in time . E ( t ) = 1 - units_processed / units_expected . That is easy to compute as it is essentially the total count of units for all items in the queue both processed and expected . We can now express our output function as the derivative of U ( t ) the output function for the integral of the error . Now we are calculating how many shares should be added or removed to the last output and not how much the shares should be . It also eliminates any dependency on time when calculating the error which increases resiliency .
fn shares(&self) -> usize {
        if let ControllerStatus::Disabled(shares) = self.state.get() {
            return shares;
        }
        let queue = self.queue.borrow();
        let mut expected = 0.0;
        let mut processed = 0.0;
        let now = Instant::now();
        for (exp, source) in queue.iter() {
            let remaining_time = exp.saturating_duration_since(now);
            trace!(
                "Remaining time for this source: {:#?}, total_units {}",
                remaining_time,
                source.total_units()
            );
            let time_fraction =
                1.0 - (remaining_time.as_secs_f64() / source.expected_duration().as_secs_f64());
            if remaining_time.as_nanos() == 0 && now.saturating_duration_since(*exp).as_secs() > 5 {
                // already too late, bump it up hard
                self.last_shares.set(1000);
                return 1000;
            }
            expected += source.total_units() as f64 * time_fraction;
            processed += source.processed_units() as f64;
        }
        // so little time has passed we can't really make any useful prediction
        if expected < 0.01 {
            return self.last_shares.get();
        }
        let error = 1.0 - processed / expected;
        let accumulated_error = self.accumulated_error.get();
        let acc = accumulated_error + error;
        self.accumulated_error.set(acc);
        let delta_error = error - self.last_error.get();
        self.last_error.set(error);
        // How did we pick our constants:
        //  * As with any stable PI controller we want the bulk of our gain to come from
        //    P.
        //  * As we normalize the maximum error to 1 we know that the gain should be at
        //    most 1000
        //  * physically, Ki can be expressed as Kp / Tau where Tau is a time constant,
        //    roughly equivalent to how many periods need to pass for the integral term
        //    to generate the same gain as the proportional term. We set that to 6 so
        //    the controller is not too sluggish, which is around 1.5 seconds on the
        //    default 250ms adjustment period.
        //
        //  We can then write X + X/6 = 1000, and solving for X we have the constants
        // below
        let kp = 850.0;
        let ki = kp / 6.0;
        let dshares = ki * error + kp * delta_error;
        let mut shares = (dshares + self.last_shares.get() as f64) as isize;
        shares = std::cmp::min(shares, 1000);
        shares = std::cmp::max(shares, self.min_shares.get() as isize);
        let shares = shares as usize;
        trace!(
            "processed: {}. expected: {} error: {}, delta_error {} , kp term {}, ki term {}, \
             shares: {}",
            processed,
            expected,
            error,
            delta_error,
            ki * error,
            kp * delta_error,
            shares
        );
        self.last_shares.set(shares);
        shares
    }
fn shares ( & self ) -> usize { if let ControllerStatus :: Disabled ( shares ) = self . state . get ( ) { return shares ; } let queue = self . queue . borrow ( ) ; let mut expected = 0.0 ; let mut processed = 0.0 ; let now = Instant :: now ( ) ; for ( exp , source ) in queue . iter ( ) { let remaining_time = exp . saturating_duration_since ( now ) ; trace ! ( " " remaining_time source total_units ( ) ) ; let time_fraction = 1.0 - ( remaining_time . as_secs_f64 ( ) / source . expected_duration ( ) . as_secs_f64 ( ) ) ; if remaining_time . as_nanos ( ) == 0 && now . saturating_duration_since ( * exp ) . as_secs ( ) > 5 { // already too late, bump it up hard self . last_shares . set ( 1000 ) ; return 1000 ; } expected += source . total_units ( ) as f64 * time_fraction ; processed += source . processed_units ( ) as f64 ; } // so little time has passed we can't really make any useful prediction if expected < 0.01 { return self . last_shares . get ( ) ; } let error = 1.0 - processed / expected ; let accumulated_error = self . accumulated_error . get ( ) ; let acc = accumulated_error + error ; self . accumulated_error . set ( acc ) ; let delta_error = error - self . last_error . get ( ) ; self . last_error . set ( error ) ; // How did we pick our constants: //  * As with any stable PI controller we want the bulk of our gain to come from //    P. //  * As we normalize the maximum error to 1 we know that the gain should be at //    most 1000 //  * physically, Ki can be expressed as Kp / Tau where Tau is a time constant, //    roughly equivalent to how many periods need to pass for the integral term //    to generate the same gain as the proportional term. We set that to 6 so //    the controller is not too sluggish, which is around 1.5 seconds on the //    default 250ms adjustment period. // //  We can then write X + X/6 = 1000, and solving for X we have the constants // below let kp = 850.0 ; let ki = kp / 6.0 ; let dshares = ki * error + kp * delta_error ; let mut shares = ( dshares + self . last_shares . get ( ) as f64 ) as isize ; shares = std :: cmp :: min ( shares , 1000 ) ; shares = std :: cmp :: max ( shares , self . min_shares . get ( ) as isize ) ; let shares = shares as usize ; trace ! ( " \
 " processed expected error delta_error ki error kp delta_error shares ) ; self . last_shares . set ( shares ) ; shares }
PI controller for shares.
PI controller for shares .
&self
&self