One to One network file transfer code C# socket based

Question

I have been developing a piece of software which has a central computer talk to many smaller nodes with a star network configuration. Each smaller node sends a file around 5MB in size once the central computer requests it This all happens at the same time so there will be no traffic on the network then there will suddenly be lots from multiple sources.

My implementation works fine it is just really slow when running on the hardware, One Raspberry Pi 3 running as a hub connected to Raspberry Pi Zeros via usb (they are acting as Ethernet adapters). At the moment I have it setup with just a smaller node and it is struggling to send data.

When connecting the Pi Zero to my computer and running the central computer code it takes ~1 second to complete the request (which is good) however when I run the code on the raspberry Pi 3 it takes ~15 minutes which is crazy to me. How do I make my code faster (or at least closer to what I see when connected to PC)?

I did some testing using the Pi3 and the Pi0 using iPerf I found that the connection theoretically supported 91.2Mbits/sec which should transfer my files in ~0.5 sec. To get more real world number I ran a python HTTP server and transferred an 88MB video file which was able to transfer at around 8.5MB/sec, still around 0.7 sec per file. My conclusion from this is that it must be the code that I have made that has a bottleneck in it or some sort of linux configuration issue (but that should affect these results too right?).

The code works by sending the data in chunks to keep everything reliable as i found that sending the as one large item resulted in limited reliability. The code essentially sends a request from the chunk requester to the chunk responser which passes on the data to external code which generates the data. Once the request data is generated the responder sends the full size of the data to the requester which sends an acknowledgement both then build structures to hold data (the requester made a large byte array, the responder breaks the data into chunks) once the responder is ready to send chunks it sends a message to the requester which starts requesting chunks. The requester requests all the chunks and then sends a message to the responder which completes the transaction.

I made a little diagram which shows this if that sounds confusing:

A message sent between the requester and responder has to follow this format:

<Data size, int> break message <data (with length of data size from the front)>

Chunk Requester:

    public byte[] Request(byte[] requestData)
    {
        if(!socket.Connected) throw new SocketNotConnectedException();

        //format the request data and send it to the camera
        {
            byte[] correct = InterconnectHelper.FormatSendData(requestData);
            socket.Send(correct);
        }

        //recieve data and return the result
        try
        {
            byte[] buffer = new byte[Constants.CameraBufferSize]; // this constant is 128 and only used for commands
            int recieved = socket.Receive(buffer);
            byte[] recievedData = InterconnectHelper.RecieveData(buffer, recieved, socket);

            int dataSize = int.Parse(Encoding.ASCII.GetString(recievedData));

            //send back the chunk size for the rest of the data
            recievedData = Encoding.ASCII.GetBytes(chunkSize.ToString()); // chunkSize is 20,000
            socket.Send(InterconnectHelper.FormatSendData(recievedData));

            //assemble the chunk array
            int chunkAmount = dataSize / chunkSize; //chunkSize is 20,000
            if(dataSize % chunkSize != 0) ++chunkAmount;
            byte[] returnData = new byte[dataSize];

            recieved = socket.Receive(buffer);
            recievedData = InterconnectHelper.RecieveData(buffer, recieved, socket);
            if(!recievedData.SequenceEqual(Constants.ReadyTransferBytes))
                throw new SocketUnexpectedDataException("Transfer ready message expected");

            buffer = new byte[Constants.HubBufferSize]; // buffer size is 400,000

            for(int i = 0; i < chunkAmount; i++)
            {
                socket.Send(InterconnectHelper.FormatSendData(Encoding.ASCII.GetBytes(i.ToString())));

                recieved = socket.Receive(buffer);
                recievedData = InterconnectHelper.RecieveData(buffer, recieved, socket);

                Array.Copy(recievedData, 0, returnData, i * chunkSize, recievedData.Length);
            }

            return returnData;
        }
        finally
        {
            socket.Send(InterconnectHelper.FormatSendData(Constants.EndTransferBytes));
        }
    }

Chunk Responder:

    public byte[] RecieveData()
    {
        if (!Connected()) throw new SocketNotConnectedException();
        if (waitingForResponse) throw new ResponseNeededException("There is a pending request in progress, complete it first");

        int recieved = socket.Receive(buffer); // buffer size is 128

        waitingForResponse = true;

        return InterconnectHelper.RecieveData(buffer, recieved, socket);
    }

    public void SendResponse(byte[] data)
    {
        if (!Connected()) throw new SocketNotConnectedException();

        //Inform the how large the data will be
        socket.Send(GenerateInformationPackage(data));

        //check if request has been canceled
        int recieve = socket.Receive(buffer); // buffer size is 128
        byte[] command = InterconnectHelper.RecieveData(buffer, recieve, socket);
        if(command == Constants.EndTransferBytes)
        {
            waitingForResponse = false;
            return;
        }

        int chunkSize = int.Parse(Encoding.ASCII.GetString(command));
        int chunkAmount = data.Length / chunkSize;
        List<byte[]> chunks = new List<byte[]>(chunkAmount);

        //move the byte data into a chunk cache
        for (int i = 0; i < chunkAmount; i++)
        {
            byte[] sampleChunks = new byte[chunkSize];
            Array.Copy(data, i * chunkSize, sampleChunks, 0, chunkSize);

            chunks.Add(sampleChunks);
        }

        if (data.Length % chunkSize != 0)
        {
            byte[] sampleChunks = new byte[data.Length % chunkSize];
            Array.Copy(data, chunkAmount * chunkSize, sampleChunks, 0, data.Length % chunkSize);

            chunks.Add(sampleChunks);
        }

        socket.Send(InterconnectHelper.FormatSendData(Constants.ReadyTransferBytes));

        int sendChunk;
        recieve = socket.Receive(buffer);
        command = InterconnectHelper.RecieveData(buffer, recieve, socket);

        while (!command.SequenceEqual(Constants.EndTransferBytes))
        {
            sendChunk = int.Parse(Encoding.ASCII.GetString(command));
            socket.Send(InterconnectHelper.FormatSendData(chunks[sendChunk]));

            recieve = socket.Receive(buffer);
            command = InterconnectHelper.RecieveData(buffer, recieve, socket);
        }

        waitingForResponse = false;
    }

    private byte[] GenerateInformationPackage(byte[] data)
    {
        return InterconnectHelper.FormatSendData(Encoding.ASCII.GetBytes(data.Length.ToString()));
    }

Shared functionality (InterconnectHelper):

    public static byte[] FormatSendData(byte[] send)
    {
        byte[] formatted =
            new byte[send.Length + send.Length.ToString().Length + Constants.EndOfMessageBytes.Length];
        int position = 0;
        byte[] temp = Encoding.ASCII.GetBytes(send.Length.ToString());

        //add data length
        Array.Copy(temp, 0, formatted, position, temp.Length);
        position += temp.Length;

        //add message break
        temp = Constants.EndOfMessageBytes;
        Array.Copy(temp, 0, formatted, position, temp.Length);
        position += temp.Length;

        //add data
        Array.Copy(send, 0, formatted, position, send.Length);
        return formatted;
    }


    public static byte[] RecieveData(byte[] buffer, int recieved, ISocket socket)
    {
        //figure out if that was all the data
        int end = Helpers.ByteHelpers.SearchEOMStartIndex(buffer, recieved);// this returns the first index of the message break from the data 
        if (end < 0 || end == recieved) throw new InvalidDataException("Recieved data does not have a size specification");

        int length;

        bool success = int.TryParse(Encoding.ASCII.GetString(buffer, 0, end), out length);
        if (!success) throw new InvalidDataException("Could not convert data length to number");

        byte[] output = new byte[length];

        int throwAwayData = end + Constants.EndOfMessageBytes.Length;
        int filled = recieved - throwAwayData;

        if (length > buffer.Length - end - Constants.EndOfMessageBytes.Length || filled < length)
        {
            //the data must be collected over multiple recieves
            Array.Copy(buffer, throwAwayData, output, 0, filled);

            while (filled < length)
            {
                if (!Helpers.NetworkHelpers.Connected(socket)) throw new SocketDisconnectedException();

                recieved = socket.Receive(buffer);
                Array.Copy(buffer, 0, output, filled, recieved);
                filled += recieved;
            }

            return output;
        }

        Array.Copy(buffer, end + Constants.EndOfMessageBytes.Length, output, 0, length);
        return output;
    }

Answer 1

naming

Choosing correct as an identifier seems odd, when request would probably be more appropriate here.

buffer size

It would be helpful if you could discuss numeric details of latency, throughput, and buffer size, perhaps in comments. What bad effect would we see by reducing request chunkSize to 2k, or increasing to 200k?

Rather than incrementing upon zero remainder, the natural way to express it would be Math.Ceiling(dataSize / chunkSize).

Typo: you wanted received.

This seems like a very magic number:

        buffer = new byte[Constants.HubBufferSize]; // buffer size is 400,000

Why just twenty buffers? Is this somehow related to max image length coming off the camera? Is chunkAmount typically constrained to be less than twenty?

It would be very helpful to log detailed delta times for this portion of the code:

        for (int i = 0; i < chunkAmount; i++) {
            socket.Send(InterconnectHelper.FormatSendData(Encoding.ASCII.GetBytes(i.ToString())));
            recieved = socket.Receive(buffer);
            recievedData = InterconnectHelper.RecieveData(buffer, recieved, socket);

The if (data.Length % chunkSize != 0) test works, at the expense of copy-n-pasted statements. A more natural way to express it would be to keep incrementing (or decrementing) some bookkeeping variable by chunkSize or by size of final buffer, as appropriate.

memory size

Pre-calculating the large chunks object works fine on large memory servers, but perhaps less well on your Raspberry Pi. Monitor vmstat 1 to verify there's no paging thrashing.

latency vs. bandwidth

The smallish 20 KiB buffer size seems like trouble - it doesn't allow for the TCP congestion window to open up much. The fact that you idle the data channel while waiting a roundtrip to exchange command packets simply exacerbates that. Can you have multiple request commands outstanding? Can you use bigger buffers? Can you ditch this protocol in favor of rsync'ing files, perhaps files chopped into conveniently small pieces?