Datagram Socket Starving other threads in my UDP program

Question

As part of a course assignment, we have been tasked with adding an extra layer of reliability on top of the UDP layer java offers in order to send a large picture file. This is to be done using the Go-Back-N protocol: http://en.wikipedia.org/wiki/Go_back_N

From what i understand, the crux of this problem relies on being able to send packets while simultaneously checking if any Acknowledgements have come in for old packets which would allow you to move your window along.

I am currently doing this by having two threads: One which sends the next packets if there is space in the window; and one which continually just listens for any incoming acknowledgements and reacts appropriately.

My problem is that the program should be threaded such that it is as if these two threads are acting simulatneously, but in fact, it seems as if the ACKReceiver thread is getting a hugely disproportionate amount of time. From the thread dump it appears to "starve" the sending thread for a little while when it reaches the DataSocket.receive() line, blocking execution here and not giving the other thread an opportunity to run in the meantime.

I have had a look at the following question which seems to hint that the problem is something to do with the fact that DatagramSocket.receive is synchronized...but offers no usable solution to the problem:

Java Thread won't pause on I/O operation

Here is the code to the sender part of my code, i am relatively sure the receiver on the other side is perfectly fine (for one thing, i didn't have to use any threads to get that to work!):

import java.io.*;
import java.net.*;
import java.nio.ByteBuffer;

public class Sender3 {
    short base = 0;
    short nextSeqNum = 0;
    DatagramPacket[] packets;
    ByteBuffer bb;
    String endSys;
    int portNum;
    String fileName;
    int retryTime;
    int windowSize;
    DatagramSocket clientSocket;
    InetAddress IPAddress;
    boolean timedOut = false;

    public Sender3(String endSys, int portNum, String fileName, int retryTime, int windowSize){
        this.endSys = endSys;
        this.portNum = portNum;
        this.fileName = fileName;
        this.retryTime = retryTime;
        this.windowSize = windowSize;
    }

    public static void main(String args[]) throws Exception{
        //Check for current arguments and assign them
        if(args.length != 5){
            System.out.println("Invalid number of arguments. Please specify: <endSystem> <portNumber> <fileName> <retryTimeout><windowSize>");
            System.exit(1);
        }

        Sender3 sendy = new Sender3(args[0], Integer.parseInt(args[1]), args[2], Integer.parseInt(args[3]), Integer.parseInt(args[4]));

        sendy.go();
    }

    private void go() throws Exception{

        clientSocket = new DatagramSocket();



        bb = ByteBuffer.allocate(2);
        byte[] picData = new byte[1021];
        byte[] sendData = new byte[1024];

        Thread.yield()
        short seqNum = 0; 
        byte[] seqBytes = new byte[2];
        byte EOFFlag = 0;
        boolean acknowledged = false;
        int lastPacketRetrys = 0;
        int resends = 0;
        IPAddress = InetAddress.getByName(endSys);

        FileInputStream imReader = new FileInputStream(new File(fileName));
        double fileSizeKb = imReader.available() / 1021.0; //We add 3 bytes to every packet, so dividing by 1021 will give us total kb sent. 
        int packetsNeeded = (int) Math.ceil(fileSizeKb);
        packets = new DatagramPacket[packetsNeeded];
        long startTime = System.currentTimeMillis();
        long endTime;
        double throughput;

        //Create array of packets to send
        for(int i = 0; i < packets.length; i++){
            if(i == packets.length - 1){
                EOFFlag = 1;
                picData = new byte[imReader.available()];
                sendData = new byte[picData.length + 3];
            }
            imReader.read(picData);
            bb.putShort((short)i);
            bb.flip();
            seqBytes = bb.array();
            bb.clear();
            System.arraycopy(seqBytes, 0, sendData, 0, seqBytes.length);
            sendData[2] = EOFFlag;
            System.arraycopy(picData, 0, sendData, 3, picData.length);
            packets[i] = new DatagramPacket((byte[])sendData.clone(), sendData.length, IPAddress, portNum);
        }

        ACKGetter ackGet = new ACKGetter();
        Thread ackThread = new Thread(ackGet);
        ackThread.start();

        //System.out.println("timeout is: " + timedOut + " base is: " + base + " packet length is: " + packets.length + " nextSeqNum: " + nextSeqNum);

        while(base != packets.length){
            if(timedOut){
                //System.out.println("Timed out waiting for acknowledgement, resending all unACKed packets in window");
                clientSocket.setSoTimeout(retryTime);
                resends++;
                if(nextSeqNum == packets.length)
                    lastPacketRetrys++;
                //Resend all packets in window
                for (int i = base; i < nextSeqNum; i++){
                //  System.out.println("Resending packets with number: " + i);
                    clientSocket.send(packets[i]);
                }
                timedOut = false;
            }

            if(nextSeqNum - base < windowSize && nextSeqNum < packets.length){
                //System.out.println("sending packet with seqNum: " + nextSeqNum);
                clientSocket.send(packets[nextSeqNum]);
                if(base == nextSeqNum){
                    clientSocket.setSoTimeout(retryTime); 
                }
                nextSeqNum++;
            }
            else{
                //Thread.yield();
            }

        }




        if(lastPacketRetrys > 10){
            System.out.println("Last packet ACK was lost (we think). So we just gave up, number of retransmissions will probably be higher");
        }
        endTime = System.currentTimeMillis();
        throughput = 1000 * fileSizeKb / (endTime - startTime);
        clientSocket.close();
        imReader.close();
        System.out.println("Number of retransmissions: " + resends);
        System.out.println("Average throughput is: " + throughput + "Kb/s");

    }


    private class ACKGetter implements Runnable {
        //Listen out for ACKs and update pointers accordingly
        DatagramPacket ackPacket;
        byte[] ackData = new byte[2];
        public void run() {
            while(base != packets.length){
                if(base != nextSeqNum){
                    try{
                        ackPacket = new DatagramPacket(ackData, ackData.length);
                        clientSocket.receive(ackPacket);
                        ackData = ackPacket.getData();
                        bb.put(ackData[0]);
                        bb.put(ackData[1]);
                        bb.flip();
                        short ack = bb.getShort();
                        bb.clear();
                        if(base <= ack){
                            //System.out.println("acknowledgement for base num: " + base + "ack num:" + ack);
                            base = (short) (ack + 1);
                            //If theres nothing left in window, stop timing, otherwise restart the timer
                            if(base == nextSeqNum){
                                clientSocket.setSoTimeout(0);
                            }
                            else{
                                clientSocket.setSoTimeout(retryTime);
                            }
                        }
                        else{
                            //System.out.println("ACK didnt change anything: " + ack);
                        }
                    }
                    catch(Exception ex){
                        timedOut = true;
                        //System.out.println("Packet timed out...resending..");
                    }
                }

                Thread.yield();


            }
        }
    }
}

Answer 1

I think you are having a deadlock here because the reader thread is in clientSocket.receive() while the sender makes a call to clientSocket.setSoTimeout(). See the following DatagramSocket method definitions:

public synchronized void setSoTimeout(int timeout) throws SocketException {
...
public synchronized void receive(DatagramPacket p) throws IOException {

If you are receiving with a socket timeout of 0 then receive hangs waiting for a packet. If you issue a SIGQUIT your JVM will dump the threads and should show you the deadlock and you can track the stack frames to see where the sender and receiver are stuck.

To fix this you should stop changing the setSoTimeout value which sounds like very bad practice to me. I would switch to using DatagramChannel, making the socket non-blocking, and use NIO receive to do the reads. See the NIO docs for more information on how to use a channel Selector.