|
@@ -1,5 +1,6 @@
|
1
|
1
|
# coding: utf-8
|
2
|
2
|
import collections
|
|
3
|
+import typing
|
3
|
4
|
from copy import copy
|
4
|
5
|
from multiprocessing import Queue
|
5
|
6
|
from multiprocessing import Process
|
|
@@ -7,12 +8,16 @@ from queue import Empty
|
7
|
8
|
import time
|
8
|
9
|
|
9
|
10
|
from synergine2.base import BaseObject
|
|
11
|
+from synergine2.exceptions import SynergineException
|
10
|
12
|
from synergine2.share import shared
|
11
|
13
|
from synergine2.config import Config
|
12
|
14
|
from synergine2.log import SynergineLogger
|
13
|
15
|
from synergine2.simulation import Subject
|
14
|
16
|
from synergine2.simulation import Event
|
15
|
17
|
|
|
18
|
+if typing.TYPE_CHECKING:
|
|
19
|
+ from synergine2.core import Core
|
|
20
|
+
|
16
|
21
|
STOP_SIGNAL = '__STOP_SIGNAL__'
|
17
|
22
|
|
18
|
23
|
|
|
@@ -60,6 +65,8 @@ class Terminal(BaseObject):
|
60
|
65
|
# List of subscribed Event classes. Terminal will not receive events
|
61
|
66
|
# who are not instance of listed classes
|
62
|
67
|
subscribed_events = [Event]
|
|
68
|
+ # Permit to execute terminal in main process, only one terminal can use this
|
|
69
|
+ main_process = False
|
63
|
70
|
|
64
|
71
|
def __init__(
|
65
|
72
|
self,
|
|
@@ -76,6 +83,7 @@ class Terminal(BaseObject):
|
76
|
83
|
self.cycle_events = []
|
77
|
84
|
self.event_handlers = collections.defaultdict(list)
|
78
|
85
|
self.asynchronous = asynchronous
|
|
86
|
+ self.core_process = None # type: Process
|
79
|
87
|
|
80
|
88
|
def accept_event(self, event: Event) -> bool:
|
81
|
89
|
for event_class in self.subscribed_events:
|
|
@@ -88,6 +96,26 @@ class Terminal(BaseObject):
|
88
|
96
|
self._output_queue = output_queue
|
89
|
97
|
self.run()
|
90
|
98
|
|
|
99
|
+ def execute_as_main_process(self, core: 'Core') -> None:
|
|
100
|
+ """
|
|
101
|
+ This method is called when the terminal have to be the main process. It will
|
|
102
|
+ create a process with the run of core and make it's job here.
|
|
103
|
+ """
|
|
104
|
+ output_queue = Queue()
|
|
105
|
+ input_queue = Queue()
|
|
106
|
+
|
|
107
|
+ self.logger.info('Start core in a process')
|
|
108
|
+ self.core_process = Process(target=core.run, kwargs=dict(
|
|
109
|
+ from_terminal=self,
|
|
110
|
+ from_terminal_input_queue=output_queue,
|
|
111
|
+ from_terminal_output_queue=input_queue,
|
|
112
|
+ ))
|
|
113
|
+ self.core_process.start()
|
|
114
|
+
|
|
115
|
+ # Core is started, continue this terminal job
|
|
116
|
+ self.logger.info('Core started, continue terminal job')
|
|
117
|
+ self.start(input_queue=input_queue, output_queue=output_queue)
|
|
118
|
+
|
91
|
119
|
def run(self):
|
92
|
120
|
"""
|
93
|
121
|
Override this method to create your daemon terminal
|
|
@@ -161,10 +189,19 @@ class TerminalManager(BaseObject):
|
161
|
189
|
self.outputs_queues = {}
|
162
|
190
|
self.inputs_queues = {}
|
163
|
191
|
|
|
192
|
+ def get_main_process_terminal(self) -> typing.Optional[Terminal]:
|
|
193
|
+ main_process_terminals = [t for t in self.terminals if t.main_process]
|
|
194
|
+ if main_process_terminals:
|
|
195
|
+ if len(main_process_terminals) > 1:
|
|
196
|
+ raise SynergineException('There is more one main process terminal !')
|
|
197
|
+ return main_process_terminals[0]
|
|
198
|
+ return None
|
|
199
|
+
|
164
|
200
|
def start(self) -> None:
|
165
|
201
|
self.logger.info('Start terminals')
|
166
|
|
- for terminal in self.terminals:
|
167
|
|
- # TODO: logs
|
|
202
|
+ # We exclude here terminal who is run from main process
|
|
203
|
+ terminals = [t for t in self.terminals if not t.main_process]
|
|
204
|
+ for terminal in terminals:
|
168
|
205
|
output_queue = Queue()
|
169
|
206
|
self.outputs_queues[terminal] = output_queue
|
170
|
207
|
|