7906

Committed 05 Jun 2026 01:51PM UTC coverage: 36.125% (-0.1%) from 36.24%

Build # 7906

Build Type

push

github

Committed by

rowleya

Commit Message

Make use of the new append feature when updating a log

Coverage Stats

1903 of 5902 branches covered (32.24%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

30 existing lines in 5 files now uncovered.

8925 of 24072 relevant lines covered (37.08%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

37.36

/SpiNNaker-allocserv/src/main/java/uk/ac/manchester/spinnaker/alloc/compat/V1CompatService.java

/*
 * Copyright (c) 2021 The University of Manchester
 *
 * 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
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package uk.ac.manchester.spinnaker.alloc.compat;

import static com.fasterxml.jackson.databind.PropertyNamingStrategies.SNAKE_CASE;
import static java.lang.Thread.interrupted;
import static java.util.Objects.nonNull;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.Executors.newCachedThreadPool;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.slf4j.LoggerFactory.getLogger;

import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;

import org.slf4j.Logger;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.google.errorprone.annotations.RestrictedApi;

import uk.ac.manchester.spinnaker.alloc.ForTestingOnly;
import uk.ac.manchester.spinnaker.alloc.SpallocProperties;
import uk.ac.manchester.spinnaker.alloc.SpallocProperties.CompatibilityProperties;
import uk.ac.manchester.spinnaker.alloc.model.Prototype;
import uk.ac.manchester.spinnaker.utils.UsedInJavadocOnly;
import uk.ac.manchester.spinnaker.utils.ValueHolder;

/**
 * Implementation of the old style Spalloc interface. Delegates handling a
 * single connection to a {@link V1CompatTask} instance (manufactured by Spring
 * as a {@linkplain Prototype prototype} bean).
 *
 * @author Donal Fellows
 */
@Service("spalloc-v1-compatibility-service")
@UsedInJavadocOnly(Prototype.class)
public class V1CompatService {
        private static final Logger log = getLogger(V1CompatService.class);

        /** The overall service properties. */
        @Autowired
        private SpallocProperties mainProps;

        /**
         * Factory for {@linkplain V1CompatTask tasks}. Only use via
         * {@link #getTask(Socket) getTask(...)} or the test API.
         */
        @Autowired
        private ObjectProvider<V1CompatTask> taskFactory;

        /** How we make threads. */
        private final ThreadFactory threadFactory;

        /** The service socket. */
        private ServerSocket serv;

        /** The main network service listener thread. */
        private Thread servThread;

        /**
         * How to serialize and deserialize JSON. Would be a bean except then we'd
         * be wiring by name.
         */
        private final ObjectMapper mapper;

        /** How the majority of threads are launched by the service. */
        private ExecutorService executor;

        private Duration shutdownTimeout;

        V1CompatService() {
                mapper = JsonMapper.builder().propertyNamingStrategy(SNAKE_CASE)
                                .build();
                var group = new ThreadGroup("spalloc-legacy-service");
                var counter = new ValueHolder<>(1);
                threadFactory = r -> {
                        var t = new Thread(group, r,
                                        "spalloc-legacy-" + counter.update(i -> i + 1));
                        t.setUncaughtExceptionHandler((thread, ex) -> log
                                        .error("uncaught exception in {}", thread, ex));
                        return t;
                };
        }

        /** A class that can reach into a compat service. */
        abstract static class Aware {
                private final V1CompatService srv;

                Aware(V1CompatService service) {
                        srv = requireNonNull(service);
                }

                /**
                 * @return The executor to use.
                 */
                protected final ExecutorService getExecutor() {
                        return requireNonNull(srv.executor);
                }

                /**
                 * @return The JSON mapper to use if necessary.
                 */
                protected final ObjectMapper getJsonMapper() {
                        return requireNonNull(srv.mapper);
                }

                /** @return The relevant properties. */
                protected final CompatibilityProperties getProperties() {
                        return srv.mainProps.getCompat();
                }
        }

        @PostConstruct
        private void open() throws IOException {
                var props = mainProps.getCompat();
                if (props.getThreadPoolSize() > 0) {
                        log.info("setting thread pool size to {}",
                                        props.getThreadPoolSize());
                        executor = newFixedThreadPool(props.getThreadPoolSize(),
                                        threadFactory);
                } else {
                        log.info("using unbounded thread pool");
                        executor = newCachedThreadPool(threadFactory);
                }

                if (props.isEnable()) {
                        var addr = new InetSocketAddress(props.getHost(), props.getPort());
                        serv = new ServerSocket();
                        serv.bind(addr);
                        servThread = threadFactory.newThread(this::acceptConnections);
                        servThread.setName("spalloc-legacy-service");
                        log.info("launching listener thread {} on address {}", servThread,
                                        addr);
                        servThread.start();
                }

                this.shutdownTimeout = props.getShutdownTimeout();
        }

        @PreDestroy
        private void close() throws IOException, InterruptedException {
                if (nonNull(serv)) {
                        log.info("shutting down listener thread {}", servThread);
                        // Shut down the server socket first; no new clients
                        servThread.interrupt();
                        serv.close();
                        servThread.join();
                }

                // Shut down the clients
                var remainingTasks = executor.shutdownNow();
                if (!remainingTasks.isEmpty()) {
                        log.warn("there are {} compat tasks outstanding",
                                        remainingTasks.size());
                }
                if (executor.awaitTermination(shutdownTimeout.toMillis(),
                                MILLISECONDS)) {
                        log.info("compat service stopped");
                } else {
                        log.warn("compat service executor ({}) still running!", executor);
                }
        }

        /**
         * Make a task.
         *
         * @param socket
         *            The connected socket that the task will be handling.
         * @return The task instance.
         */
        private V1CompatTask getTask(Socket socket) {
                return taskFactory.getObject(this, socket);
        }

        /**
         * Main service loop. Accepts connections and dispatches them to workers.
         */
        private void acceptConnections() {
                try {
                        while (acceptConnection()) {
                                continue;
                        }
                } finally {
                        try {
                                serv.close();
                        } catch (IOException e) {
                                log.warn("IO error", e);
                        }
                }
        }

        /**
         * Accept a single connection and dispatch it to a worker task in a thread.
         *
         * @return If {@code false}, we want to stop accepting connections.
         */
        private boolean acceptConnection() {
                try {
                        var service = getTask(serv.accept());
                        executor.execute(() -> service.handleConnection());
                } catch (SocketException e) {
                        // Check here; interrupt = shutting down = no errors, please
                        if (interrupted()) {
                                return false;
                        }
                        if (serv.isClosed()) {
                                return false;
                        }
                        log.warn("IO error", e);
                } catch (IOException e) {
                        log.warn("IO error", e);
                }
                // If we've been interrupted here, we want the main loop to stop
                return !interrupted();
        }

        /**
         * Not a public API! Operations for testing only.
         *
         * @hidden
         */
        @ForTestingOnly
        public interface TestAPI {
                /**
                 * Make an instance of {@link V1CompatTask} that we can talk to.
                 *
                 * @param in
                 *            How to send a message to the task. Should be
                 *            <em>unconnected</em>.
                 * @param out
                 *            How to receive a message from the task. Should be
                 *            <em>unconnected</em>.
                 * @return A future that can be cancelled to shut things down.
                 * @throws Exception
                 *             If various things go wrong.
                 */
                Future<?> launchInstance(PipedWriter in, PipedReader out)
                                throws Exception;
        }

        /**
         * Not a public API!
         *
         * @return Test interface.
         * @deprecated Only for testing.
         * @hidden
         */
        @ForTestingOnly
        @RestrictedApi(explanation = "just for testing", link = "index.html",
                        allowedOnPath = ".*/src/test/java/.*")
        @Deprecated
        public TestAPI getTestApi() {
                ForTestingOnly.Utils.checkForTestClassOnStack();
                return new TestAPI() {
                        @Override
                        public Future<?> launchInstance(PipedWriter in, PipedReader out)
                                        throws Exception {
                                var service = taskFactory.getObject(V1CompatService.this,
                                                new PipedReader(in), new PipedWriter(out));
                                return executor.submit(() -> service.handleConnection());
                        }
                };
        }
}

1	/*
2	* Copyright (c) 2021 The University of Manchester
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* https://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package uk.ac.manchester.spinnaker.alloc.compat;
17
18	import static com.fasterxml.jackson.databind.PropertyNamingStrategies.SNAKE_CASE;
19	import static java.lang.Thread.interrupted;
20	import static java.util.Objects.nonNull;
21	import static java.util.Objects.requireNonNull;
22	import static java.util.concurrent.Executors.newCachedThreadPool;
23	import static java.util.concurrent.Executors.newFixedThreadPool;
24	import static java.util.concurrent.TimeUnit.MILLISECONDS;
25	import static org.slf4j.LoggerFactory.getLogger;
26
27	import java.io.IOException;
28	import java.io.PipedReader;
29	import java.io.PipedWriter;
30	import java.net.InetSocketAddress;
31	import java.net.ServerSocket;
32	import java.net.Socket;
33	import java.net.SocketException;
34	import java.time.Duration;
35	import java.util.concurrent.ExecutorService;
36	import java.util.concurrent.Future;
37	import java.util.concurrent.ThreadFactory;
38
39	import jakarta.annotation.PostConstruct;
40	import jakarta.annotation.PreDestroy;
41
42	import org.slf4j.Logger;
43	import org.springframework.beans.factory.ObjectProvider;
44	import org.springframework.beans.factory.annotation.Autowired;
45	import org.springframework.stereotype.Service;
46
47	import com.fasterxml.jackson.databind.ObjectMapper;
48	import com.fasterxml.jackson.databind.json.JsonMapper;
49	import com.google.errorprone.annotations.RestrictedApi;
50
51	import uk.ac.manchester.spinnaker.alloc.ForTestingOnly;
52	import uk.ac.manchester.spinnaker.alloc.SpallocProperties;
53	import uk.ac.manchester.spinnaker.alloc.SpallocProperties.CompatibilityProperties;
54	import uk.ac.manchester.spinnaker.alloc.model.Prototype;
55	import uk.ac.manchester.spinnaker.utils.UsedInJavadocOnly;
56	import uk.ac.manchester.spinnaker.utils.ValueHolder;
57
58	/**
59	* Implementation of the old style Spalloc interface. Delegates handling a
60	* single connection to a {@link V1CompatTask} instance (manufactured by Spring
61	* as a {@linkplain Prototype prototype} bean).
62	*
63	* @author Donal Fellows
64	*/
65	@Service("spalloc-v1-compatibility-service")
66	@UsedInJavadocOnly(Prototype.class)
67	public class V1CompatService {
68	private static final Logger log = getLogger(V1CompatService.class);	2✔
69
70	/** The overall service properties. */
71	@Autowired
72	private SpallocProperties mainProps;
73
74	/**
75	* Factory for {@linkplain V1CompatTask tasks}. Only use via
76	* {@link #getTask(Socket) getTask(...)} or the test API.
77	*/
78	@Autowired
79	private ObjectProvider<V1CompatTask> taskFactory;
80
81	/** How we make threads. */
82	private final ThreadFactory threadFactory;
83
84	/** The service socket. */
85	private ServerSocket serv;
86
87	/** The main network service listener thread. */
88	private Thread servThread;
89
90	/**
91	* How to serialize and deserialize JSON. Would be a bean except then we'd
92	* be wiring by name.
93	*/
94	private final ObjectMapper mapper;
95
96	/** How the majority of threads are launched by the service. */
97	private ExecutorService executor;
98
99	private Duration shutdownTimeout;
100
101	V1CompatService() {	2✔
102	mapper = JsonMapper.builder().propertyNamingStrategy(SNAKE_CASE)	2✔
103	.build();	2✔
104	var group = new ThreadGroup("spalloc-legacy-service");	2✔
105	var counter = new ValueHolder<>(1);	2✔
106	threadFactory = r -> {	2✔
107	var t = new Thread(group, r,	2✔
108	"spalloc-legacy-" + counter.update(i -> i + 1));	2✔
109	t.setUncaughtExceptionHandler((thread, ex) -> log	2✔
110	.error("uncaught exception in {}", thread, ex));	×
111	return t;	2✔
112	};
113	}	2✔
114
115	/** A class that can reach into a compat service. */
116	abstract static class Aware {
117	private final V1CompatService srv;
118
119	Aware(V1CompatService service) {	2✔
120	srv = requireNonNull(service);	2✔
121	}	2✔
122
123	/**
124	* @return The executor to use.
125	*/
126	protected final ExecutorService getExecutor() {
127	return requireNonNull(srv.executor);	2✔
128	}
129
130	/**
131	* @return The JSON mapper to use if necessary.
132	*/
133	protected final ObjectMapper getJsonMapper() {
134	return requireNonNull(srv.mapper);	2✔
135	}
136
137	/** @return The relevant properties. */
138	protected final CompatibilityProperties getProperties() {
139	return srv.mainProps.getCompat();	×
140	}
141	}
142
143	@PostConstruct
144	private void open() throws IOException {
145	var props = mainProps.getCompat();	2✔
146	if (props.getThreadPoolSize() > 0) {	2✔
147	log.info("setting thread pool size to {}",	2✔
148	props.getThreadPoolSize());	2✔
149	executor = newFixedThreadPool(props.getThreadPoolSize(),	2✔
150	threadFactory);
151	} else {
152	log.info("using unbounded thread pool");	2✔
153	executor = newCachedThreadPool(threadFactory);	2✔
154	}
155
156	if (props.isEnable()) {	2!
157	var addr = new InetSocketAddress(props.getHost(), props.getPort());	×
158	serv = new ServerSocket();	×
159	serv.bind(addr);	×
160	servThread = threadFactory.newThread(this::acceptConnections);	×
161	servThread.setName("spalloc-legacy-service");	×
162	log.info("launching listener thread {} on address {}", servThread,	×
163	addr);
164	servThread.start();	×
165	}
166
167	this.shutdownTimeout = props.getShutdownTimeout();	2✔
168	}	2✔
169
170	@PreDestroy
171	private void close() throws IOException, InterruptedException {
UNCOV 172	if (nonNull(serv)) {	×
173	log.info("shutting down listener thread {}", servThread);	×
174	// Shut down the server socket first; no new clients
175	servThread.interrupt();	×
176	serv.close();	×
177	servThread.join();	×
178	}
179
180	// Shut down the clients
UNCOV 181	var remainingTasks = executor.shutdownNow();	×
UNCOV 182	if (!remainingTasks.isEmpty()) {	×
183	log.warn("there are {} compat tasks outstanding",	×
184	remainingTasks.size());	×
185	}
UNCOV 186	if (executor.awaitTermination(shutdownTimeout.toMillis(),	×
187	MILLISECONDS)) {
UNCOV 188	log.info("compat service stopped");	×
189	} else {
190	log.warn("compat service executor ({}) still running!", executor);	×
191	}
UNCOV 192	}	×
193
194	/**
195	* Make a task.
196	*
197	* @param socket
198	* The connected socket that the task will be handling.
199	* @return The task instance.
200	*/
201	private V1CompatTask getTask(Socket socket) {
202	return taskFactory.getObject(this, socket);	×
203	}
204
205	/**
206	* Main service loop. Accepts connections and dispatches them to workers.
207	*/
208	private void acceptConnections() {
209	try {
210	while (acceptConnection()) {	×
211	continue;	×
212	}
213	} finally {
214	try {
215	serv.close();	×
216	} catch (IOException e) {	×
217	log.warn("IO error", e);	×
218	}	×
219	}
220	}	×
221
222	/**
223	* Accept a single connection and dispatch it to a worker task in a thread.
224	*
225	* @return If {@code false}, we want to stop accepting connections.
226	*/
227	private boolean acceptConnection() {
228	try {
229	var service = getTask(serv.accept());	×
230	executor.execute(() -> service.handleConnection());	×
231	} catch (SocketException e) {	×
232	// Check here; interrupt = shutting down = no errors, please
233	if (interrupted()) {	×
234	return false;	×
235	}
236	if (serv.isClosed()) {	×
237	return false;	×
238	}
239	log.warn("IO error", e);	×
240	} catch (IOException e) {	×
241	log.warn("IO error", e);	×
242	}	×
243	// If we've been interrupted here, we want the main loop to stop
244	return !interrupted();	×
245	}
246
247	/**
248	* Not a public API! Operations for testing only.
249	*
250	* @hidden
251	*/
252	@ForTestingOnly
253	public interface TestAPI {
254	/**
255	* Make an instance of {@link V1CompatTask} that we can talk to.
256	*
257	* @param in
258	* How to send a message to the task. Should be
259	* <em>unconnected</em>.
260	* @param out
261	* How to receive a message from the task. Should be
262	* <em>unconnected</em>.
263	* @return A future that can be cancelled to shut things down.
264	* @throws Exception
265	* If various things go wrong.
266	*/
267	Future<?> launchInstance(PipedWriter in, PipedReader out)
268	throws Exception;
269	}
270
271	/**
272	* Not a public API!
273	*
274	* @return Test interface.
275	* @deprecated Only for testing.
276	* @hidden
277	*/
278	@ForTestingOnly
279	@RestrictedApi(explanation = "just for testing", link = "index.html",
280	allowedOnPath = "./src/test/java/.")
281	@Deprecated
282	public TestAPI getTestApi() {
283	ForTestingOnly.Utils.checkForTestClassOnStack();	2✔
284	return new TestAPI() {	2✔
285	@Override
286	public Future<?> launchInstance(PipedWriter in, PipedReader out)
287	throws Exception {
288	var service = taskFactory.getObject(V1CompatService.this,	2✔
289	new PipedReader(in), new PipedWriter(out));
290	return executor.submit(() -> service.handleConnection());	2✔
291	}
292	};
293	}
294	}

SpiNNakerManchester / JavaSpiNNaker / 7906

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous