Subversion Repositories fpga-cf

/**

 *

 */

package edu.byu.cc.plieber.fpgaenet.fcp;

import java.io.*;

import java.net.*;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.LinkedBlockingQueue;

import javax.swing.text.html.MinimalHTMLWriter;

/**

 * The main class to instantiate for communication of an FCP/UDP/IP connection.

 * Provides methods to send data, request data, and received data. All data is

 * sent and received as byte arrays.

 *

 * @author Peter Lieber

 *

 */

public class FCPProtocol {

        /**

         * Creates new FCPProtocol object with default ports: 3001(remote), 3000(local)

         *

         * @throws IOException

         */

        public FCPProtocol() throws IOException {

                this.init(0x3000, 0x3001);

        }

        /**

         * Creates new FCPProtocol object with default local port, 3000

         * @param port Local UDP Port

         * @throws IOException

         */

        public FCPProtocol(int port) throws IOException {

                this.init(port, 0x3001);

        }

        /**

         * Creates new FCPProtocol object

         * @param port Local UDP Port

         * @param destport Remote UDP Port

         * @throws IOException

         */

        public FCPProtocol(int port, int destport) throws IOException {

                this.init(port, destport);

        }

        /**

         * Initializes protocol parameters and threads

         * @param port

         * @param destport

         * @throws IOException

         */

        private void init(int port, int destport) throws IOException {

                rec_cur = 1;

                rec_last_rcv = 0;

                snd_cur = 0;

                snd_last_ack = 0;

                timeout = 1000;

                socket = new DatagramSocket(port);

                socket.setSoTimeout(50);

                connectedPort = destport;

                connected = false;

                this.receivedQueue = new LinkedBlockingQueue<FCPPacket>();

                this.packetOutbox = new ConcurrentHashMap<Integer, FCPPacket>();

                recThread = new FCPReceiveThread(this);

                recThread.start();

                sendThread = new FCPSendThread(this);

                sendThread.start();

        }

        protected int recWindow = 1;

        protected int sendWindow = 20;

        protected volatile int rec_cur;

        protected volatile int rec_last_rcv;

        protected volatile int snd_cur;

        protected volatile int snd_last_ack;

        protected DatagramSocket socket;

        private FCPReceiveThread recThread;

        private FCPSendThread sendThread;

        protected boolean connected;

        protected InetAddress connectedAddress;

        protected int connectedPort;

        LinkedBlockingQueue<FCPPacket> receivedQueue;

        ConcurrentHashMap<Integer, FCPPacket> packetOutbox;

        protected long timeout = 500;

        public boolean packetsPending() {

                return !packetOutbox.isEmpty() || !sendThread.sendQueue.isEmpty();

        }

        /**

         * Sends acknowledgement packet (internal)

         * @param address

         * @param seq

         */

        protected void sendAck(InetAddress address, int seq) {

                // TODO Auto-generated method stub

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 1;

                fp.port = 0;

                fp.seq = seq;

                fp.len = 0;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

        }

        /**

         * Sends connection acknowledgement packet (internal)

         */

        public void sendConAck() {

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 3;

                fp.port = 0;

                fp.seq = 0;

                fp.len = 0;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

        }

        /**

         * Sends a request for some number of bytes.

         * @param port FCP port number

         * @param numBytes Number of bytes expected

         * @return

         * @throws FCPException

         */

        public boolean sendDataRequest(int port, int numBytes) throws FCPException {

                if (!this.connected) throw new FCPException("Not connected to FPGA!");

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 4;

                fp.port = port;

                fp.len = numBytes;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

                return true;

        }

        /**

         * Sends data through the specified FCP port. The data must be less than 1024 bytes long.

         * @param port FCP port number

         * @param data data to be sent

         * @param count number of bytes to send

         * @return

         * @throws IOException

         * @throws FCPException

         */

        public boolean send(int port, byte[] data, int count) throws FCPException {

                if (!this.connected) throw new FCPException("Not connected to FPGA!");

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 0;

                fp.port = port;

                fp.len = count;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                fp.data = data.clone();

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        throw new FCPException("Interrupted Send Operation");

                }

                checkHealth();

                return true;

        }

        /**

         * Sends data through the specified FCP port.  The data can be any length (within reason).

         * @param port

         * @param bytes

         * @param numBytes

         * @return

         * @throws FCPException

         */

        public boolean sendData(int port, List<Byte> bytes, int numBytes) throws FCPException {

                int offset = 0;

                int numRead = 0;

                while (offset < numBytes) {

                        numRead = Math.min(offset+1024, numBytes) - offset;

                        this.send(port, bytes.subList(offset, offset+numRead), numRead);

                        offset += 1024;

                }

                return true;

        }

        public void sendData(int port, ArrayList<Byte> bytes) throws FCPException {

                this.sendData(port, bytes, bytes.size());

        }

        public void sendData(int port, byte value) throws FCPException {

                this.send(port, value);

        }

        /**

         * Sends data through the specified FCP port. The data must be less than 1024 bytes long.

         * @param port FCP port number

         * @param bytes data to be sent

         * @param count number of bytes to send

         * @return

         * @throws FCPException

         */

        public boolean send(int port, List<Byte> bytes, int count) throws FCPException {

                if (!this.connected) throw new FCPException("Not connected to FPGA!");

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 0;

                fp.port = port;

                fp.len = count;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                fp.data = new byte[bytes.size()];

                for (int i=0; i<fp.data.length; i++) {

                        fp.data[i] = bytes.get(i);

                }

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

                return true;

        }

        private boolean send(int port, byte value) throws FCPException {

                if (!this.connected) throw new FCPException("Not connected to FPGA!");

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 0;

                fp.port = port;

                fp.len = 1;

                fp.dest = connectedAddress;

                fp.dstPort = connectedPort;

                fp.data = new byte[1];

                fp.data[0] = value;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

                return true;

        }

        private void checkHealth() throws FCPException {

                if (!this.sendThread.isAlive()) {

                        if (this.sendThread.sendTimeout) throw new FCPException("Send Error: timeout");

                        else if (this.sendThread.ioException) throw new FCPException("I/O Error");

                }

        }

        void processPacket(FCPPacket fcppacket) {

                if (fcppacket.command == 5) {

                        try {

                                this.receivedQueue.put(fcppacket);

                        } catch (InterruptedException e) {

                                // TODO Auto-generated catch block

                                e.printStackTrace();

                        }

                }

        }

        /**

         * Connects to an FPGA at the given address

         * @param address Address (usually derived from IP address)

         * @param port Remote UDP port (0x3001)

         */

        public void connect(InetAddress address, int port) {

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 2;

                fp.port = 0;

                fp.seq = 0;

                fp.len = 0;

                fp.dest = address;

                fp.dstPort = port;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

        }

        public void connect(InetAddress address) {

                FCPPacket fp = new FCPPacket();

                fp.version = 0;

                fp.command = 2;

                fp.port = 0;

                fp.seq = 0;

                fp.len = 0;

                fp.dest = address;

                fp.dstPort = 0x3001;

                try {

                        sendThread.sendQueue.put(fp);

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

        }

        /**

         * Disconnect from FPGA.  This method does not send anything to the FPGA,

         * but only ends the send and receiving threads of execution and closes

         * the UDP socket.

         */

        public void disconnect() {

                recThread.done = true;

                sendThread.done = true;

                try {

                        synchronized (recThread) {

                                recThread.join();

                        }

                        synchronized (sendThread) {

                                sendThread.join();

                        }

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

                socket.close();

        }

        void resetSW() {

                rec_cur = 1;

                rec_last_rcv = 0;

                snd_cur = 0;

                snd_last_ack = 0;

        }

        /**

         * Gets the response data that has been received due to a data request.

         * Blocks until data is available.

         *

         * @return

         * @throws InterruptedException

         */

        public byte[] getDataResponse() {

                FCPPacket packet;

                try {

                        packet = this.receivedQueue.take();

                } catch (InterruptedException e) {

                        // TODO Auto-generated catch block

                        packet = null;

                }

                if (packet != null)

                        return packet.data;

                else return null;

        }

        @Override

        public String toString() {

                return "FCP Protocol< FPGA: " + this.connectedAddress.toString() + " >";

        }

        void send(FCPPacket fcpPacket) {

                try {

                        this.socket.send(fcpPacket.wrapInDatagram());

                } catch (IOException e) {

                        // TODO Auto-generated catch block

                        e.printStackTrace();

                }

        }

        public boolean isConnected() {

                return this.connected;

        }

        public void setSendWindow(int i) {

                this.sendWindow = i;

        }

        public int getSourceUDPPort() {

                return this.socket.getLocalPort();

        }

        public int getDestUDPPort() {

                return this.connectedPort;

        }

        public InetAddress getDestIPAddress() {

                return this.connectedAddress;

        }

        public long getWhileCount() {

                return this.sendThread.whileCount;

        }

        public void resetWhileCount() {

                this.sendThread.whileCount = 0;

        }

}