-
Notifications
You must be signed in to change notification settings - Fork 128
/
compilation_session.py
112 lines (81 loc) · 3.65 KB
/
compilation_session.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# Copyright (c) Facebook, Inc. and its affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.
from pathlib import Path
from typing import List, Optional, Tuple
from compiler_gym.service.proto import (
Action,
ActionSpace,
Benchmark,
Observation,
ObservationSpace,
)
class CompilationSession:
"""Base class for encapsulating an incremental compilation session.
To add support for a new compiler, subclass from this base and provide
implementations of the abstract methods, then call
:func:`create_and_run_compiler_service
<compiler_gym.service.runtime.create_and_run_compiler_service>` and pass in
your class type:
.. code-block:: python
from compiler_gym.service import CompilationSession
from compiler_gym.service import runtime
class MyCompilationSession(CompilationSession):
...
if __name__ == "__main__":
runtime.create_and_run_compiler_service(MyCompilationSession)
"""
compiler_version: str = ""
"""The compiler version."""
action_spaces: List[ActionSpace] = []
"""A list of action spaces describing the capabilities of the compiler."""
observation_spaces: List[ObservationSpace] = []
"""A list of feature vectors that this compiler provides."""
def __init__(
self, working_dir: Path, action_space: ActionSpace, benchmark: Benchmark
):
"""Start a CompilationSession.
Subclasses should initialize the parent class first.
:param working_dir: A directory on the local filesystem that can be used
to store temporary files such as build artifacts.
:param action_space: The action space to use.
:param benchmark: The benchmark to use.
"""
del action_space # Subclasses must use this.
del benchmark # Subclasses must use this.
self.working_dir = working_dir
def apply_action(self, action: Action) -> Tuple[bool, Optional[ActionSpace], bool]:
"""Apply an action.
:param action: The action to apply.
:return: A tuple: :code:`(end_of_session, new_action_space,
action_had_no_effect)`.
"""
raise NotImplementedError
def get_observation(self, observation_space: ObservationSpace) -> Observation:
"""Compute an observation.
:param observation_space: The observation space.
:return: An observation.
"""
raise NotImplementedError
def fork(self) -> "CompilationSession":
"""Create a copy of current session state.
Implementing this method is optional.
:return: A new CompilationSession with the same state.
"""
# No need to override this if you are not adding support to fork().
raise NotImplementedError("CompilationSession.fork() not supported")
def handle_session_parameter(self, key: str, value: str) -> Optional[str]:
"""Handle a session parameter send by the frontend.
Session parameters provide a method to send ad-hoc key-value messages to
a compilation session through the :meth:`env.send_session_parameter()
<compiler_gym.envs.CompilerEnv.send_session_parameter>` method. It us up
to the client/service to agree on a common schema for encoding and
decoding these parameters.
Implementing this method is optional.
:param key: The parameter key.
:param value: The parameter value.
:return: A string response message if the parameter was understood. Else
:code:`None` to indicate that the message could not be interpretted.
"""
pass