-# Copyright (c) 2022 Cisco and/or its affiliates.
+# Copyright (c) 2024 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
debug_list = list()
trace_list = list()
# Block until input object appears.
- dimension, dilled_function, param_focus_tracker, max_samples = (
- communication_pipe.recv()
- )
+ (
+ dimension,
+ dilled_function,
+ param_focus_tracker,
+ max_samples,
+ ) = communication_pipe.recv()
debug_list.append(
f"Called with param_focus_tracker {param_focus_tracker!r}"
)
if max_samples and samples >= max_samples:
break
sample_point = generate_sample(
- param_focus_tracker.averages, param_focus_tracker.covariance_matrix,
- dimension, scale_coeff
+ param_focus_tracker.averages,
+ param_focus_tracker.covariance_matrix,
+ dimension,
+ scale_coeff,
)
- trace(u"sample_point", sample_point)
+ trace("sample_point", sample_point)
samples += 1
- trace(u"samples", samples)
+ trace("samples", samples)
value, log_weight = value_logweight_function(trace, *sample_point)
- trace(u"value", value)
- trace(u"log_weight", log_weight)
- trace(u"focus tracker before adding", param_focus_tracker)
+ trace("value", value)
+ trace("log_weight", log_weight)
+ trace("focus tracker before adding", param_focus_tracker)
# Update focus related statistics.
param_distance = param_focus_tracker.add_without_dominance_get_distance(
sample_point, log_weight
# The code above looked at weight (not importance).
# The code below looks at importance (not weight).
log_rarity = param_distance / 2.0 / scale_coeff
- trace(u"log_rarity", log_rarity)
+ trace("log_rarity", log_rarity)
log_importance = log_weight + log_rarity
- trace(u"log_importance", log_importance)
+ trace("log_importance", log_importance)
value_tracker.add(value, log_importance)
# Update sampled statistics.
param_sampled_tracker.add_get_shift(sample_point, log_importance)
debug_list.append(f"integrator used {samples!s} samples")
debug_list.append(
- u" ".join([
- u"value_avg", str(value_tracker.average),
- u"param_sampled_avg", repr(param_sampled_tracker.averages),
- u"param_sampled_cov", repr(param_sampled_tracker.covariance_matrix),
- u"value_log_variance", str(value_tracker.log_variance),
- u"value_log_secondary_variance",
- str(value_tracker.secondary.log_variance)
- ])
+ " ".join(
+ [
+ "value_avg",
+ str(value_tracker.average),
+ "param_sampled_avg",
+ repr(param_sampled_tracker.averages),
+ "param_sampled_cov",
+ repr(param_sampled_tracker.covariance_matrix),
+ "value_log_variance",
+ str(value_tracker.log_variance),
+ "value_log_secondary_variance",
+ str(value_tracker.secondary.log_variance),
+ ]
+ )
)
communication_pipe.send(
(value_tracker, param_focus_tracker, debug_list, trace_list, samples)
-# Copyright (c) 2023 Cisco and/or its affiliates.
+# Copyright (c) 2024 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
log_xerfcx_10 = math.log(xerfcx_limit - math.exp(10) * erfcx(math.exp(10)))
def __init__(
- self, measurer, trial_duration_per_trial, packet_loss_ratio_target,
- trial_number_offset=0, timeout=7200.0, trace_enabled=False):
+ self,
+ measurer,
+ trial_duration_per_trial,
+ packet_loss_ratio_target,
+ trial_number_offset=0,
+ timeout=7200.0,
+ trace_enabled=False,
+ ):
"""Store rate measurer and additional parameters.
The measurer must never report negative loss count.
trial_number += 1
logging.info(f"Trial {trial_number!r}")
results = self.measure_and_compute(
- self.trial_duration_per_trial * trial_number, transmit_rate,
- trial_result_list, min_rate, max_rate, focus_trackers
+ self.trial_duration_per_trial * trial_number,
+ transmit_rate,
+ trial_result_list,
+ min_rate,
+ max_rate,
+ focus_trackers,
)
measurement, average, stdev, avg1, avg2, focus_trackers = results
zeros += 1
if (trial_number - self.trial_number_offset) <= 1:
next_load = max_rate
elif (trial_number - self.trial_number_offset) <= 3:
- next_load = (measurement.relative_forwarding_rate / (
- 1.0 - self.packet_loss_ratio_target))
+ next_load = measurement.relative_forwarding_rate / (
+ 1.0 - self.packet_loss_ratio_target
+ )
else:
next_load = (avg1 + avg2) / 2.0
if zeros > 0:
if lossy_loads[0] > next_load:
diminisher = math.pow(2.0, 1 - zeros)
next_load = lossy_loads[0] + diminisher * next_load
- next_load /= (1.0 + diminisher)
+ next_load /= 1.0 + diminisher
# On zero measurement, we need to drain obsoleted low losses
# even if we did not use them to increase next_load,
# in order to get to usable loses at higher loads.
# TODO: chi is from https://en.wikipedia.org/wiki/Nondimensionalization
chi = (load - mrr) / spread
chi0 = -mrr / spread
- trace(u"stretch: load", load)
- trace(u"mrr", mrr)
- trace(u"spread", spread)
- trace(u"chi", chi)
- trace(u"chi0", chi0)
+ trace("stretch: load", load)
+ trace("mrr", mrr)
+ trace("spread", spread)
+ trace("chi", chi)
+ trace("chi0", chi0)
if chi > 0:
log_lps = math.log(
load - mrr + (log_plus(0, -chi) - log_plus(0, chi0)) * spread
)
- trace(u"big loss direct log_lps", log_lps)
+ trace("big loss direct log_lps", log_lps)
else:
two_positive = log_plus(chi, 2 * chi0 - log_2)
two_negative = log_plus(chi0, 2 * chi - log_2)
if two_positive <= two_negative:
log_lps = log_minus(chi, chi0) + log_spread
- trace(u"small loss crude log_lps", log_lps)
+ trace("small loss crude log_lps", log_lps)
return log_lps
two = log_minus(two_positive, two_negative)
three_positive = log_plus(two_positive, 3 * chi - log_3)
three = log_minus(three_positive, three_negative)
if two == three:
log_lps = two + log_spread
- trace(u"small loss approx log_lps", log_lps)
+ trace("small loss approx log_lps", log_lps)
else:
log_lps = math.log(log_plus(0, chi) - log_plus(0, chi0))
log_lps += log_spread
- trace(u"small loss direct log_lps", log_lps)
+ trace("small loss direct log_lps", log_lps)
return log_lps
@staticmethod
# TODO: The stretch sign is just to have less minuses. Worth changing?
chi = (mrr - load) / spread
chi0 = mrr / spread
- trace(u"Erf: load", load)
- trace(u"mrr", mrr)
- trace(u"spread", spread)
- trace(u"chi", chi)
- trace(u"chi0", chi0)
+ trace("Erf: load", load)
+ trace("mrr", mrr)
+ trace("spread", spread)
+ trace("chi", chi)
+ trace("chi0", chi0)
if chi >= -1.0:
- trace(u"positive, b roughly bigger than m", None)
+ trace("positive, b roughly bigger than m", None)
if chi > math.exp(10):
first = PLRsearch.log_xerfcx_10 + 2 * (math.log(chi) - 10)
- trace(u"approximated first", first)
+ trace("approximated first", first)
else:
first = math.log(PLRsearch.xerfcx_limit - chi * erfcx(chi))
- trace(u"exact first", first)
+ trace("exact first", first)
first -= chi * chi
second = math.log(PLRsearch.xerfcx_limit - chi * erfcx(chi0))
second -= chi0 * chi0
intermediate = log_minus(first, second)
- trace(u"first", first)
+ trace("first", first)
else:
- trace(u"negative, b roughly smaller than m", None)
+ trace("negative, b roughly smaller than m", None)
exp_first = PLRsearch.xerfcx_limit + chi * erfcx(-chi)
exp_first *= math.exp(-chi * chi)
exp_first -= 2 * chi
second = math.log(PLRsearch.xerfcx_limit - chi * erfcx(chi0))
second -= chi0 * chi0
intermediate = math.log(exp_first - math.exp(second))
- trace(u"exp_first", exp_first)
- trace(u"second", second)
- trace(u"intermediate", intermediate)
+ trace("exp_first", exp_first)
+ trace("second", second)
+ trace("intermediate", intermediate)
result = intermediate + math.log(spread) - math.log(erfc(-chi0))
- trace(u"result", result)
+ trace("result", result)
return result
@staticmethod
def find_critical_rate(
- trace, lfit_func, min_rate, max_rate, loss_ratio_target,
- mrr, spread):
+ trace, lfit_func, min_rate, max_rate, loss_ratio_target, mrr, spread
+ ):
"""Given ratio target and parameters, return the achieving offered load.
This is basically an inverse function to lfit_func
loss_rate = math.exp(lfit_func(trace, rate, mrr, spread))
loss_ratio = loss_rate / rate
if loss_ratio > loss_ratio_target:
- trace(u"halving down", rate)
+ trace("halving down", rate)
rate_hi = rate
elif loss_ratio < loss_ratio_target:
- trace(u"halving up", rate)
+ trace("halving up", rate)
rate_lo = rate
- trace(u"found", rate)
+ trace("found", rate)
return rate
@staticmethod
:rtype: float
"""
log_likelihood = 0.0
- trace(u"log_weight for mrr", mrr)
- trace(u"spread", spread)
+ trace("log_weight for mrr", mrr)
+ trace("spread", spread)
for result in trial_result_list:
- trace(u"for tr", result.intended_load)
- trace(u"lc", result.loss_count)
- trace(u"d", result.intended_duration)
+ trace("for tr", result.intended_load)
+ trace("lc", result.loss_count)
+ trace("d", result.intended_duration)
# _rel_ values use units of intended_load (transactions per second).
log_avg_rel_loss_per_second = lfit_func(
trace, result.intended_load, mrr, spread
log_trial_likelihood *= -result.loss_count
log_trial_likelihood -= log_plus(0.0, +log_avg_abs_loss_per_trial)
log_likelihood += log_trial_likelihood
- trace(u"avg_loss_per_trial", math.exp(log_avg_abs_loss_per_trial))
- trace(u"log_trial_likelihood", log_trial_likelihood)
+ trace("avg_loss_per_trial", math.exp(log_avg_abs_loss_per_trial))
+ trace("log_trial_likelihood", log_trial_likelihood)
return log_likelihood
def measure_and_compute(
- self, trial_duration, transmit_rate, trial_result_list,
- min_rate, max_rate, focus_trackers=(None, None), max_samples=None):
+ self,
+ trial_duration,
+ transmit_rate,
+ trial_result_list,
+ min_rate,
+ max_rate,
+ focus_trackers=(None, None),
+ max_samples=None,
+ ):
"""Perform both measurement and computation at once.
High level steps: Prepare and launch computation worker processes,
# See https://stackoverflow.com/questions/15137292/large-objects-and-multiprocessing-pipes-and-send
worker = multiprocessing.Process(
target=Integrator.try_estimate_nd,
- args=(worker_pipe_end, 5.0, self.trace_enabled)
+ args=(worker_pipe_end, 5.0, self.trace_enabled),
)
worker.daemon = True
worker.start()
)
value = math.log(
self.find_critical_rate(
- trace, fitting_function, min_rate, max_rate,
- self.packet_loss_ratio_target, mrr, spread
+ trace,
+ fitting_function,
+ min_rate,
+ max_rate,
+ self.packet_loss_ratio_target,
+ mrr,
+ spread,
)
)
return value, logweight
raise RuntimeError(f"Worker {name} did not finish!")
result_or_traceback = pipe.recv()
try:
- value_tracker, focus_tracker, debug_list, trace_list, sampls = (
- result_or_traceback
- )
+ (
+ value_tracker,
+ focus_tracker,
+ debug_list,
+ trace_list,
+ sampls,
+ ) = result_or_traceback
except ValueError:
raise RuntimeError(
f"Worker {name} failed with the following traceback:\n"
)
return _PartialResult(value_tracker, focus_tracker, sampls)
- stretch_result = stop_computing(u"stretch", stretch_pipe)
- erf_result = stop_computing(u"erf", erf_pipe)
+ stretch_result = stop_computing("stretch", stretch_pipe)
+ erf_result = stop_computing("erf", erf_pipe)
result = PLRsearch._get_result(measurement, stretch_result, erf_result)
logging.info(
f"measure_and_compute finished with trial result "
# Named tuples, for multiple local variables to be passed as return value.
_PartialResult = namedtuple(
- u"_PartialResult", u"value_tracker focus_tracker samples"
+ "_PartialResult", "value_tracker focus_tracker samples"
)
"""Two stat trackers and sample counter.
"""
_ComputeResult = namedtuple(
- u"_ComputeResult",
- u"measurement avg stdev stretch_exp_avg erf_exp_avg trackers"
+ "_ComputeResult",
+ "measurement avg stdev stretch_exp_avg erf_exp_avg trackers",
)
"""Measurement, 4 computation result values, pair of trackers.
-# Copyright (c) 2021 Cisco and/or its affiliates.
+# Copyright (c) 2024 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
:returns: Expression constructing an equivalent instance.
:rtype: str
"""
- return f"ScalarStatTracker(log_sum_weight={self.log_sum_weight!r}," \
+ return (
+ f"ScalarStatTracker(log_sum_weight={self.log_sum_weight!r},"
f"average={self.average!r},log_variance={self.log_variance!r})"
+ )
def copy(self):
"""Return new ScalarStatTracker instance with the same state as self.
if absolute_shift > 0.0:
log_square_shift = 2 * math.log(absolute_shift)
log_variance = log_plus(
- log_variance, log_square_shift + log_sample_ratio)
+ log_variance, log_square_shift + log_sample_ratio
+ )
if log_variance is not None:
log_variance += old_log_sum_weight - new_log_sum_weight
self.log_sum_weight = new_log_sum_weight
One typical use is for Monte Carlo integrator to decide whether
the partial sums so far are reliable enough.
"""
+
def __init__(
- self, log_sum_weight=None, average=0.0, log_variance=None,
- log_sum_secondary_weight=None, secondary_average=0.0,
- log_secondary_variance=None, max_log_weight=None):
+ self,
+ log_sum_weight=None,
+ average=0.0,
+ log_variance=None,
+ log_sum_secondary_weight=None,
+ secondary_average=0.0,
+ log_secondary_variance=None,
+ max_log_weight=None,
+ ):
"""Initialize new tracker instance, empty by default.
:param log_sum_weight: Natural logarithm of sum of weights
:rtype: str
"""
sec = self.secondary
- return f"ScalarDualStatTracker(log_sum_weight={self.log_sum_weight!r},"\
- f"average={self.average!r},log_variance={self.log_variance!r}," \
- f"log_sum_secondary_weight={sec.log_sum_weight!r}," \
- f"secondary_average={sec.average!r}," \
- f"log_secondary_variance={sec.log_variance!r}," \
+ return (
+ f"ScalarDualStatTracker(log_sum_weight={self.log_sum_weight!r},"
+ f"average={self.average!r},log_variance={self.log_variance!r},"
+ f"log_sum_secondary_weight={sec.log_sum_weight!r},"
+ f"secondary_average={sec.average!r},"
+ f"log_secondary_variance={sec.log_variance!r},"
f"max_log_weight={self.max_log_weight!r})"
+ )
def add(self, scalar_value, log_weight=0.0):
"""Return updated both stats after addition of another sample.
"""
def __init__(
- self, dimension=2, log_sum_weight=None, averages=None,
- covariance_matrix=None):
+ self,
+ dimension=2,
+ log_sum_weight=None,
+ averages=None,
+ covariance_matrix=None,
+ ):
"""Initialize new tracker instance, two-dimensional empty by default.
If any of latter two arguments is None, it means
:returns: Expression constructing an equivalent instance.
:rtype: str
"""
- return f"VectorStatTracker(dimension={self.dimension!r}," \
- f"log_sum_weight={self.log_sum_weight!r}," \
- f"averages={self.averages!r}," \
+ return (
+ f"VectorStatTracker(dimension={self.dimension!r},"
+ f"log_sum_weight={self.log_sum_weight!r},"
+ f"averages={self.averages!r},"
f"covariance_matrix={self.covariance_matrix!r})"
+ )
def copy(self):
"""Return new instance with the same state as self.
:rtype: VectorStatTracker
"""
return VectorStatTracker(
- self.dimension, self.log_sum_weight, self.averages[:],
- copy.deepcopy(self.covariance_matrix)
+ self.dimension,
+ self.log_sum_weight,
+ self.averages[:],
+ copy.deepcopy(self.covariance_matrix),
)
def reset(self):
Code Review / csit.git / commitdiff