Coverage for mlos_bench/mlos_bench/tests/optimizers/toy_optimization_loop_test.py: 100%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5"""Toy optimization loop to test the optimizers on mock benchmark environment.""" 

6 

7import logging 

8from typing import Tuple 

9 

10import pytest 

11 

12from mlos_bench.environments.base_environment import Environment 

13from mlos_bench.environments.mock_env import MockEnv 

14from mlos_bench.optimizers.base_optimizer import Optimizer 

15from mlos_bench.optimizers.convert_configspace import tunable_values_to_configuration 

16from mlos_bench.optimizers.mlos_core_optimizer import MlosCoreOptimizer 

17from mlos_bench.optimizers.mock_optimizer import MockOptimizer 

18from mlos_bench.tunables.tunable_groups import TunableGroups 

19from mlos_core.data_classes import Suggestion 

20from mlos_core.optimizers.bayesian_optimizers.smac_optimizer import SmacOptimizer 

21from mlos_core.util import config_to_series 

22 

23# For debugging purposes output some warnings which are captured with failed tests. 

24DEBUG = True 

25logger = logging.debug 

26if DEBUG: 

27 logger = logging.warning 

28 

29 

30def _optimize(env: Environment, opt: Optimizer) -> Tuple[float, TunableGroups]: 

31 """Toy optimization loop.""" 

32 assert opt.not_converged() 

33 

34 while opt.not_converged(): 

35 

36 with env as env_context: 

37 

38 tunables = opt.suggest() 

39 

40 logger("tunables: %s", str(tunables)) 

41 # pylint: disable=protected-access 

42 if isinstance(opt, MlosCoreOptimizer) and isinstance(opt._opt, SmacOptimizer): 

43 config = tunable_values_to_configuration(tunables) 

44 config_series = config_to_series(config) 

45 logger("config: %s", str(config)) 

46 try: 

47 logger( 

48 "prediction: %s", 

49 opt._opt.surrogate_predict(suggestion=Suggestion(config=config_series)), 

50 ) 

51 except RuntimeError: 

52 pass 

53 

54 assert env_context.setup(tunables) 

55 

56 (status, _ts, output) = env_context.run() 

57 assert status.is_succeeded() 

58 assert output is not None 

59 score = output["score"] 

60 assert isinstance(score, float) 

61 assert 60 <= score <= 120 

62 logger("score: %s", str(score)) 

63 

64 opt.register(tunables, status, output) 

65 

66 (best_score, best_tunables) = opt.get_best_observation() 

67 assert best_score is not None and len(best_score) == 1 

68 assert isinstance(best_tunables, TunableGroups) 

69 return (best_score["score"], best_tunables) 

70 

71 

72def test_mock_optimization_loop(mock_env_no_noise: MockEnv, mock_opt: MockOptimizer) -> None: 

73 """Toy optimization loop with mock environment and optimizer.""" 

74 (score, tunables) = _optimize(mock_env_no_noise, mock_opt) 

75 assert score == pytest.approx(64.9, 0.01) 

76 assert tunables.get_param_values() == { 

77 "vmSize": "Standard_B2ms", 

78 "idle": "halt", 

79 "kernel_sched_migration_cost_ns": 117026, 

80 "kernel_sched_latency_ns": 149827706, 

81 } 

82 

83 

84def test_mock_optimization_loop_no_defaults( 

85 mock_env_no_noise: MockEnv, 

86 mock_opt_no_defaults: MockOptimizer, 

87) -> None: 

88 """Toy optimization loop with mock environment and optimizer.""" 

89 (score, tunables) = _optimize(mock_env_no_noise, mock_opt_no_defaults) 

90 assert score == pytest.approx(60.97, 0.01) 

91 assert tunables.get_param_values() == { 

92 "vmSize": "Standard_B2s", 

93 "idle": "halt", 

94 "kernel_sched_migration_cost_ns": 49123, 

95 "kernel_sched_latency_ns": 234760738, 

96 } 

97 

98 

99def test_flaml_optimization_loop(mock_env_no_noise: MockEnv, flaml_opt: MlosCoreOptimizer) -> None: 

100 """Toy optimization loop with mock environment and FLAML optimizer.""" 

101 (score, tunables) = _optimize(mock_env_no_noise, flaml_opt) 

102 assert score == pytest.approx(60.15, 0.01) 

103 assert tunables.get_param_values() == { 

104 "vmSize": "Standard_B2s", 

105 "idle": "halt", 

106 "kernel_sched_migration_cost_ns": -1, 

107 "kernel_sched_latency_ns": 13718105, 

108 } 

109 

110 

111# @pytest.mark.skip(reason="SMAC is not deterministic") 

112def test_smac_optimization_loop(mock_env_no_noise: MockEnv, smac_opt: MlosCoreOptimizer) -> None: 

113 """Toy optimization loop with mock environment and SMAC optimizer.""" 

114 (score, tunables) = _optimize(mock_env_no_noise, smac_opt) 

115 expected_score = 70.33 

116 expected_tunable_values = { 

117 "vmSize": "Standard_B2s", 

118 "idle": "mwait", 

119 "kernel_sched_migration_cost_ns": 297669, 

120 "kernel_sched_latency_ns": 290365137, 

121 } 

122 assert score == pytest.approx(expected_score, 0.01) 

123 assert tunables.get_param_values() == expected_tunable_values