13333362741

Committed 14 Feb 2025 04:31PM UTC coverage: 38.411% (-0.2%) from 38.564%

Build # 13333362741

Build Type

push

github

Committed by

rowleya

Commit Message

Put this in the right place and fill in the gaps

Run Details

12 of 374 new or added lines in 3 files covered. (3.21%)

227 existing lines in 8 files now uncovered.

9168 of 23868 relevant lines covered (38.41%)

1.15 hits per line

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

0.0

/SpiNNaker-comms/src/main/java/uk/ac/manchester/spinnaker/alloc/client/ProxiedTransceiver.java

/*
 * Copyright (c) 2022 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.client;

import static uk.ac.manchester.spinnaker.machine.ChipLocation.ZERO_ZERO;
import static uk.ac.manchester.spinnaker.utils.InetFactory.getByNameQuietly;

import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.google.errorprone.annotations.CheckReturnValue;
import com.google.errorprone.annotations.MustBeClosed;

import uk.ac.manchester.spinnaker.alloc.client.SpallocClient.Job;
import uk.ac.manchester.spinnaker.connections.EIEIOConnection;
import uk.ac.manchester.spinnaker.connections.SCPConnection;
import uk.ac.manchester.spinnaker.connections.model.Connection;
import uk.ac.manchester.spinnaker.machine.ChipLocation;
import uk.ac.manchester.spinnaker.machine.HasCoreLocation;
import uk.ac.manchester.spinnaker.machine.MachineVersion;
import uk.ac.manchester.spinnaker.machine.MemoryLocation;
import uk.ac.manchester.spinnaker.spalloc.exceptions.SpallocServerException;
import uk.ac.manchester.spinnaker.transceiver.ParallelSafe;
import uk.ac.manchester.spinnaker.transceiver.ProcessException;
import uk.ac.manchester.spinnaker.transceiver.SpinnmanException;
import uk.ac.manchester.spinnaker.transceiver.Transceiver;

/** A transceiver that routes messages across the proxy. */
final class ProxiedTransceiver extends Transceiver {

    private final Job job;

    private final ProxyProtocolClient websocket;

    private final Map<InetAddress, ChipLocation> hostToChip = new HashMap<>();

    /**
     * @param version
     *            The version of the machine connected to.
     * @param connections
     *            The proxied connections we will use.
     * @param hostToChip
     *            The mapping from addresses to chip locations, to enable
     *            manufacturing of proxied {@link EIEIOConnection}s.
     * @param websocket
     *            The proxy handle.
     * @throws SpinnmanException
     * @throws IOException
     *             If we couldn't finish setting up our networking.
     * @throws InterruptedException
     *             If communications are interrupted.
     * @throws SpinnmanExcception
     *             If SpiNNaker rejects a message.
     */
    @MustBeClosed
    ProxiedTransceiver(Job job,
            ProxyProtocolClient websocket)
            throws IOException, SpinnmanException, InterruptedException {
        super(getVersion(job), getConnections(job, websocket));
        this.job = job;
        this.websocket = websocket;

        for (Connection conn : getConnections()) {
            if (conn instanceof ProxiedSCPConnection) {
                ProxiedSCPConnection pConn = (ProxiedSCPConnection) conn;
                hostToChip.put(pConn.getRemoteIPAddress(), pConn.getChip());
            }
        }
    }

    private static MachineVersion getVersion(Job job) throws IOException {
        var machine = job.machine();
        return MachineVersion.bySize(machine.getWidth(), machine.getHeight());
    }

    private static List<Connection> getConnections(Job job,
            ProxyProtocolClient ws) throws IOException, InterruptedException {
        var conns = new ArrayList<Connection>();
        var machine = job.machine();
        InetAddress bootChipAddress = null;
        for (var bc : machine.getConnections()) {
            var chipAddr = getByNameQuietly(bc.getHostname());
            var chipLoc = bc.getChip().asChipLocation();
            conns.add(new ProxiedSCPConnection(chipLoc, ws, chipAddr));
            if (chipLoc.equals(ZERO_ZERO)) {
                bootChipAddress = chipAddr;
            }
        }
        if (bootChipAddress != null) {
            conns.add(new ProxiedBootConnection(ws, bootChipAddress));
        }
        return conns;
    }

    /** {@inheritDoc} */
    @Override
    public void close() throws IOException {
        super.close();
        websocket.close();
    }

    @Override
    public SCPConnection createScpConnection(ChipLocation chip,
            InetAddress addr) throws IOException {
        try {
            return new ProxiedSCPConnection(chip, websocket, addr);
        } catch (InterruptedException e) {
            throw new IOException("failed to proxy connection", e);
        }
    }

    @Override
    protected EIEIOConnection newEieioConnection(InetAddress localHost,
            Integer localPort) throws IOException {
        try {
            return new ProxiedEIEIOListenerConnection(hostToChip, websocket);
        } catch (InterruptedException e) {
            throw new IOException("failed to proxy connection", e);
        }
    }


    @Override
    @ParallelSafe
    public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
            InputStream dataStream, int numBytes)
            throws IOException, ProcessException, InterruptedException {
        ByteBuffer data = ByteBuffer.allocate(numBytes);
        byte[] buffer = new byte[1024];
        int remaining = numBytes;
        while (remaining > 0) {
            int toRead = Math.min(remaining, buffer.length);
            int read = dataStream.read(buffer, 0, toRead);
            if (read < 0) {
                throw new EOFException();
            }
            data.put(buffer, 0, read);
            remaining -= read;
        }
        try {
            this.job.writeMemory(core, baseAddress, data);
        } catch (SpallocServerException e) {
            throw new IOException(e);
        }
    }

    @Override
    @ParallelSafe
    public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
            File dataFile)
            throws IOException, ProcessException, InterruptedException {
        try (var stream = new FileInputStream(dataFile)) {
            writeMemory(core, baseAddress, stream, (int) dataFile.length());
        }
    }

    @Override
    @ParallelSafe
    public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
            ByteBuffer data)
            throws IOException, ProcessException, InterruptedException {
        try {
            this.job.writeMemory(core, baseAddress, data);
        } catch (SpallocServerException e) {
            throw new IOException(e);
        }
    }

    @Override
    @CheckReturnValue
    @ParallelSafe
    public ByteBuffer readMemory(HasCoreLocation core,
            MemoryLocation baseAddress, int length)
            throws IOException, ProcessException, InterruptedException {
        try {
            return job.readMemory(core, baseAddress, length);
        } catch (SpallocServerException e) {
            throw new IOException(e);
        }
    }

}

1	/*
2	* Copyright (c) 2022 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.client;
17
18	import static uk.ac.manchester.spinnaker.machine.ChipLocation.ZERO_ZERO;
19	import static uk.ac.manchester.spinnaker.utils.InetFactory.getByNameQuietly;
20
21	import java.io.EOFException;
22	import java.io.File;
23	import java.io.FileInputStream;
24	import java.io.IOException;
25	import java.io.InputStream;
26	import java.net.InetAddress;
27	import java.nio.ByteBuffer;
28	import java.util.ArrayList;
29	import java.util.HashMap;
30	import java.util.List;
31	import java.util.Map;
32
33	import com.google.errorprone.annotations.CheckReturnValue;
34	import com.google.errorprone.annotations.MustBeClosed;
35
36	import uk.ac.manchester.spinnaker.alloc.client.SpallocClient.Job;
37	import uk.ac.manchester.spinnaker.connections.EIEIOConnection;
38	import uk.ac.manchester.spinnaker.connections.SCPConnection;
39	import uk.ac.manchester.spinnaker.connections.model.Connection;
40	import uk.ac.manchester.spinnaker.machine.ChipLocation;
41	import uk.ac.manchester.spinnaker.machine.HasCoreLocation;
42	import uk.ac.manchester.spinnaker.machine.MachineVersion;
43	import uk.ac.manchester.spinnaker.machine.MemoryLocation;
44	import uk.ac.manchester.spinnaker.spalloc.exceptions.SpallocServerException;
45	import uk.ac.manchester.spinnaker.transceiver.ParallelSafe;
46	import uk.ac.manchester.spinnaker.transceiver.ProcessException;
47	import uk.ac.manchester.spinnaker.transceiver.SpinnmanException;
48	import uk.ac.manchester.spinnaker.transceiver.Transceiver;
49
50	/** A transceiver that routes messages across the proxy. */
51	final class ProxiedTransceiver extends Transceiver {
52
53	private final Job job;
54
55	private final ProxyProtocolClient websocket;
56
NEW 57	private final Map<InetAddress, ChipLocation> hostToChip = new HashMap<>();	×
58
59	/**
60	* @param version
61	* The version of the machine connected to.
62	* @param connections
63	* The proxied connections we will use.
64	* @param hostToChip
65	* The mapping from addresses to chip locations, to enable
66	* manufacturing of proxied {@link EIEIOConnection}s.
67	* @param websocket
68	* The proxy handle.
69	* @throws SpinnmanException
70	* @throws IOException
71	* If we couldn't finish setting up our networking.
72	* @throws InterruptedException
73	* If communications are interrupted.
74	* @throws SpinnmanExcception
75	* If SpiNNaker rejects a message.
76	*/
77	@MustBeClosed
78	ProxiedTransceiver(Job job,
79	ProxyProtocolClient websocket)
80	throws IOException, SpinnmanException, InterruptedException {
NEW 81	super(getVersion(job), getConnections(job, websocket));	×
NEW 82	this.job = job;	×
NEW 83	this.websocket = websocket;	×
84
NEW 85	for (Connection conn : getConnections()) {	×
NEW 86	if (conn instanceof ProxiedSCPConnection) {	×
NEW 87	ProxiedSCPConnection pConn = (ProxiedSCPConnection) conn;	×
NEW 88	hostToChip.put(pConn.getRemoteIPAddress(), pConn.getChip());	×
89	}
NEW 90	}	×
NEW 91	}	×
92
93	private static MachineVersion getVersion(Job job) throws IOException {
NEW 94	var machine = job.machine();	×
NEW 95	return MachineVersion.bySize(machine.getWidth(), machine.getHeight());	×
96	}
97
98	private static List<Connection> getConnections(Job job,
99	ProxyProtocolClient ws) throws IOException, InterruptedException {
NEW 100	var conns = new ArrayList<Connection>();	×
NEW 101	var machine = job.machine();	×
NEW 102	InetAddress bootChipAddress = null;	×
NEW 103	for (var bc : machine.getConnections()) {	×
NEW 104	var chipAddr = getByNameQuietly(bc.getHostname());	×
NEW 105	var chipLoc = bc.getChip().asChipLocation();	×
NEW 106	conns.add(new ProxiedSCPConnection(chipLoc, ws, chipAddr));	×
NEW 107	if (chipLoc.equals(ZERO_ZERO)) {	×
NEW 108	bootChipAddress = chipAddr;	×
109	}
NEW 110	}	×
NEW 111	if (bootChipAddress != null) {	×
NEW 112	conns.add(new ProxiedBootConnection(ws, bootChipAddress));	×
113	}
NEW 114	return conns;	×
115	}
116
117	/** {@inheritDoc} */
118	@Override
119	public void close() throws IOException {
NEW 120	super.close();	×
NEW 121	websocket.close();	×
NEW 122	}	×
123
124	@Override
125	public SCPConnection createScpConnection(ChipLocation chip,
126	InetAddress addr) throws IOException {
127	try {
NEW 128	return new ProxiedSCPConnection(chip, websocket, addr);	×
NEW 129	} catch (InterruptedException e) {	×
NEW 130	throw new IOException("failed to proxy connection", e);	×
131	}
132	}
133
134	@Override
135	protected EIEIOConnection newEieioConnection(InetAddress localHost,
136	Integer localPort) throws IOException {
137	try {
NEW 138	return new ProxiedEIEIOListenerConnection(hostToChip, websocket);	×
NEW 139	} catch (InterruptedException e) {	×
NEW 140	throw new IOException("failed to proxy connection", e);	×
141	}
142	}
143
144
145	@Override
146	@ParallelSafe
147	public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
148	InputStream dataStream, int numBytes)
149	throws IOException, ProcessException, InterruptedException {
NEW 150	ByteBuffer data = ByteBuffer.allocate(numBytes);	×
NEW 151	byte[] buffer = new byte[1024];	×
NEW 152	int remaining = numBytes;	×
NEW 153	while (remaining > 0) {	×
NEW 154	int toRead = Math.min(remaining, buffer.length);	×
NEW 155	int read = dataStream.read(buffer, 0, toRead);	×
NEW 156	if (read < 0) {	×
NEW 157	throw new EOFException();	×
158	}
NEW 159	data.put(buffer, 0, read);	×
NEW 160	remaining -= read;	×
NEW 161	}	×
162	try {
NEW 163	this.job.writeMemory(core, baseAddress, data);	×
NEW 164	} catch (SpallocServerException e) {	×
NEW 165	throw new IOException(e);	×
NEW 166	}	×
NEW 167	}	×
168
169	@Override
170	@ParallelSafe
171	public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
172	File dataFile)
173	throws IOException, ProcessException, InterruptedException {
NEW 174	try (var stream = new FileInputStream(dataFile)) {	×
NEW 175	writeMemory(core, baseAddress, stream, (int) dataFile.length());	×
176	}
NEW 177	}	×
178
179	@Override
180	@ParallelSafe
181	public void writeMemory(HasCoreLocation core, MemoryLocation baseAddress,
182	ByteBuffer data)
183	throws IOException, ProcessException, InterruptedException {
184	try {
NEW 185	this.job.writeMemory(core, baseAddress, data);	×
NEW 186	} catch (SpallocServerException e) {	×
NEW 187	throw new IOException(e);	×
NEW 188	}	×
NEW 189	}	×
190
191	@Override
192	@CheckReturnValue
193	@ParallelSafe
194	public ByteBuffer readMemory(HasCoreLocation core,
195	MemoryLocation baseAddress, int length)
196	throws IOException, ProcessException, InterruptedException {
197	try {
NEW 198	return job.readMemory(core, baseAddress, length);	×
NEW 199	} catch (SpallocServerException e) {	×
NEW 200	throw new IOException(e);	×
201	}
202	}
203
204	}

SpiNNakerManchester / JavaSpiNNaker / 13333362741

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