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;
}