3131
3232import java .io .IOException ;
3333import java .io .UncheckedIOException ;
34- import java .util .List ;
3534import java .util .Map ;
36- import java .util .concurrent .CopyOnWriteArrayList ;
3735import java .util .concurrent .locks .Lock ;
3836import java .util .concurrent .locks .ReentrantLock ;
3937import java .util .function .Consumer ;
4038import java .util .function .Function ;
4139
4240import org .apposed .appose .BuildException ;
43- import org .apposed .appose .Builder ;
4441import org .apposed .appose .Environment ;
4542import org .apposed .appose .Service ;
4643
4744/**
48- * An Appose worker that stays resident across tasks: its environment is
49- * built on first use, its worker process is started once and reused, and
50- * both are released only on request.
45+ * An Appose worker process that stays resident across tasks: it is started
46+ * on first use, reused afterward, and ended only on request.
5147 * <p>
5248 * Keeping the worker alive is what makes repeated runs fast: modules the
5349 * worker has imported stay imported, and objects a task hands to
5753 * free it.
5854 * </p>
5955 * <p>
60- * If the worker dies (e.g. it crashes, or {@link #kill()} is called), the
61- * next task transparently starts a new one.
56+ * A worker runs in a {@link LazyEnvironment}, which it builds if needed; any
57+ * number of workers may share one environment. If the worker dies (e.g. it
58+ * crashes, or {@link #kill()} is called), the next task transparently starts
59+ * a new process.
6260 * </p>
6361 */
6462public class ResidentWorker implements AutoCloseable {
6563
66- private final String name ;
67- private final Builder <?> builder ;
64+ private final LazyEnvironment environment ;
6865 private final Function <Environment , Service > launcher ;
69- private final List <BuildListener > listeners = new CopyOnWriteArrayList <>();
7066 private final Lock exclusive = new ReentrantLock ();
7167
7268 private volatile Consumer <String > debug ;
73- private Environment env ;
7469 private Service service ;
7570
7671 /**
77- * @param name Name of the environment, as reported to listeners.
78- * @param builder Builder for the environment; built on first use.
72+ * @param environment Environment the worker runs in.
7973 * @param launcher Creates the worker's service from the built environment,
8074 * e.g. {@code env -> env.python().init("import numpy")}.
8175 */
82- public ResidentWorker (final String name , final Builder <?> builder ,
76+ public ResidentWorker (final LazyEnvironment environment ,
8377 final Function <Environment , Service > launcher )
8478 {
85- this .name = name ;
79+ this .environment = environment ;
8680 this .launcher = launcher ;
87- // Note: Subscribe once, here, and dispatch to the current listeners,
88- // because builders accumulate subscribers with every call.
89- this .builder = builder //
90- .subscribeProgress ((title , current , maximum ) -> listeners .forEach (
91- l -> l .buildProgress (name , title , current , maximum ))) //
92- .subscribeOutput (text -> listeners .forEach (l -> l .buildOutput (name ,
93- text ))) //
94- .subscribeError (text -> listeners .forEach (l -> l .buildError (name ,
95- text )));
9681 }
9782
98- public String name () {
99- return name ;
83+ public LazyEnvironment environment () {
84+ return environment ;
10085 }
10186
10287 /**
@@ -108,11 +93,6 @@ public Lock exclusive() {
10893 return exclusive ;
10994 }
11095
111- public ResidentWorker addBuildListener (final BuildListener listener ) {
112- listeners .add (listener );
113- return this ;
114- }
115-
11696 /**
11797 * Sets where the worker's debug output goes: stderr lines, non-protocol
11898 * stdout lines and protocol traffic, as described by
@@ -123,21 +103,6 @@ public void debug(final Consumer<String> listener) {
123103 debug = listener ;
124104 }
125105
126- /** Gets the environment, building it if this is its first use. */
127- public synchronized Environment environment () throws BuildException {
128- if (env != null ) return env ;
129- listeners .forEach (l -> l .buildStarted (name ));
130- try {
131- env = builder .build ();
132- }
133- catch (final BuildException | RuntimeException exc ) {
134- listeners .forEach (l -> l .buildFinished (name , exc ));
135- throw exc ;
136- }
137- listeners .forEach (l -> l .buildFinished (name , null ));
138- return env ;
139- }
140-
141106 /**
142107 * Gets the worker's service, building the environment and starting the
143108 * worker process first if needed.
@@ -146,7 +111,7 @@ public synchronized Environment environment() throws BuildException {
146111 */
147112 public synchronized Service service () throws BuildException {
148113 if (service != null && service .isAlive ()) return service ;
149- final Service s = launcher .apply (environment ());
114+ final Service s = launcher .apply (environment . get ());
150115 s .debug (msg -> {
151116 final Consumer <String > listener = debug ;
152117 if (listener != null ) listener .accept (msg );
@@ -161,7 +126,7 @@ public synchronized Service service() throws BuildException {
161126 return service ;
162127 }
163128
164- /** Creates a task to run on the resident worker. */
129+ /** Creates a task to run on this worker. */
165130 public Service .Task task (final String script ,
166131 final Map <String , Object > inputs ) throws BuildException
167132 {
0 commit comments