This is a C# program that is a reference implementation, it will be used to make a C version after it is stable and bug free. The programs are supposed to simulate sensor data collection and processing nodes. One node gathers data from a sensor, then distributes the data to all the connected clients. The data gather node should only send the latest data point, and it should only send the data when the required number of clients are available. Also it should only send the next data point when all the clients have finished processing the data.
This is the server code. This continually reads data from a sensor. If there are the right number of clients connected then it sends the latest data to each of the clients. It got really long because I kept finding new ways that client disconnects and stuff would break it, so making it simpler without breaking it would be great. The client count is controlled by the ClientCount variable, set to some low number for testing.
using System;
namespace ServerDistribute
{
class Program
{
const int ClientCount = 3;
const int BufferSize = 1024 * 1024 * 3;
public class ThreadData
{
public System.Threading.Thread Thread;
public System.Threading.AutoResetEvent DataTransferSignal;
public System.Threading.AutoResetEvent DataProcessCompleteSignal;
public byte[] DataBuffer;
public System.Net.Sockets.Socket Socket;
}
//not sure if stack or queue is better
static System.Collections.Generic.Stack<System.Threading.AutoResetEvent> AvailableDataTransferSignals = new System.Collections.Generic.Stack<System.Threading.AutoResetEvent>(ClientCount);
static System.Threading.AutoResetEvent[] DataTransferSignals = new System.Threading.AutoResetEvent[ClientCount];
static System.Collections.Generic.List<ThreadData> ClientThreadData = new System.Collections.Generic.List<ThreadData>();
static object ClientThreadDataLock = new object();
static System.Threading.AutoResetEvent ThreadEventMutex = new System.Threading.AutoResetEvent(false);
static void ProcessSocket(object state)
{
var threadData = (ThreadData)state;
using var client = threadData.Socket;
byte[] recBuffer = new byte[1];
threadData.DataProcessCompleteSignal.Set();
for (; ; )
{
try
{
Console.WriteLine("Client: Wait for data transfer signal");
threadData.DataTransferSignal.WaitOne();
Console.WriteLine("Client: Send data to client");
client.Send(threadData.DataBuffer);
Console.WriteLine("Client: Get process complete from client");
client.Receive(recBuffer);
threadData.DataProcessCompleteSignal.Set();
// System.Threading.Thread.Sleep(1000);
}
catch (Exception ex)
{
Console.WriteLine("Client: Got ex: " + ex.Message);
AvailableDataTransferSignals.Push(threadData.DataProcessCompleteSignal);
lock (ClientThreadDataLock)
ClientThreadData.Remove(threadData);
ThreadEventMutex.Set();
return;
}
}
}
static void ListenForClients()
{
string socketLocation = "C:\\Users\\Public\\Documents\\temp\\distribute.sock";
if (System.IO.File.Exists(socketLocation))
System.IO.File.Delete(socketLocation);
Console.WriteLine("Server: Creating socket");
using var socket = new System.Net.Sockets.Socket(System.Net.Sockets.AddressFamily.Unix, System.Net.Sockets.SocketType.Stream, System.Net.Sockets.ProtocolType.Unspecified);
socket.SendBufferSize = 1;
Console.WriteLine("Server: Creating end point");
var socketEndPoint = new System.Net.Sockets.UnixDomainSocketEndPoint(socketLocation);
Console.WriteLine("Server: Binding");
socket.Bind(socketEndPoint);
Console.WriteLine("Server: Listening");
socket.Listen();
ThreadEventMutex.Set();
Console.WriteLine("Server: Accepting clients");
for (; ; )
{
Console.WriteLine("Server: Waiting for empty slot");
ThreadEventMutex.WaitOne();
Console.WriteLine("Server: Waiting for new client");
var client = socket.Accept();
Console.WriteLine("Server: Got a client");
var pts = new System.Threading.ParameterizedThreadStart(ProcessSocket);
var t = new System.Threading.Thread(pts);
var newTD = new ThreadData()
{
DataBuffer = null,
Thread = t,
Socket = client,
DataTransferSignal = new System.Threading.AutoResetEvent(false)
};
lock (ClientThreadDataLock)
{
newTD.DataProcessCompleteSignal = AvailableDataTransferSignals.Pop();
ClientThreadData.Add(newTD);
if (ClientThreadData.Count < ClientCount)
ThreadEventMutex.Set();
}
t.Start(newTD);
}
}
static byte[] ActiveCaptureBuffer = new byte[BufferSize];
static byte[] SendableCaptureBuffer = new byte[BufferSize];
static object CaptureBufferLock = new object();
static System.Threading.AutoResetEvent DataCapturedSignal = new System.Threading.AutoResetEvent(false);
static void CaptureData()
{
byte[] tempBuffer;
using var rng = System.Security.Cryptography.RandomNumberGenerator.Create();
Console.WriteLine("Capture: Start capturing data");
for (; ; )
{
//simulate read data from sensor
rng.GetBytes(ActiveCaptureBuffer);
// Console.WriteLine("Got data");
//simulate read delay
// System.Threading.Thread.Sleep(250);
//this might make a delay in the capture if the copy thread takes too long to copy data into the local buffer...
lock (CaptureBufferLock)
{
tempBuffer = ActiveCaptureBuffer;
ActiveCaptureBuffer = SendableCaptureBuffer;
SendableCaptureBuffer = tempBuffer;
DataCapturedSignal.Set();
}
}
}
static void CopyData()
{
for (; ; )
{
//I think this might get stuck if a client is removed
Console.WriteLine("Copy: Wait for all clients to be ready to process new data");
System.Threading.AutoResetEvent.WaitAll(DataTransferSignals);
Console.WriteLine("Copy: Wait for new data for clients");
DataCapturedSignal.WaitOne();
lock (ClientThreadDataLock)
if (ClientThreadData.Count < ClientCount)
continue;
//may cause a delay in the capture thread
Console.WriteLine("Copy: Copy data from capture buffer");
byte[] clientDataBuffer = new byte[BufferSize];
lock (CaptureBufferLock)
Array.Copy(SendableCaptureBuffer, clientDataBuffer, BufferSize);
lock (ClientThreadDataLock)
{
if (ClientThreadData.Count < ClientCount)
continue;
Console.WriteLine("Copy: Signal clients to use new data");
foreach (var td in ClientThreadData)
{
td.DataBuffer = clientDataBuffer;
td.DataTransferSignal.Set();
}
}
}
}
static void Main(string[] args)
{
for (int i = 0; i < DataTransferSignals.Length; ++i)
{
DataTransferSignals[i] = new System.Threading.AutoResetEvent(false);
AvailableDataTransferSignals.Push(DataTransferSignals[i]);
}
var serverThread = new System.Threading.Thread(ListenForClients);
serverThread.Start();
var dataCaptureThread = new System.Threading.Thread(CaptureData);
dataCaptureThread.Start();
var dataCopyThread = new System.Threading.Thread(CopyData);
dataCopyThread.Start();
Console.ReadLine();
}
}
}
This is the client code. This is supposed to connenct to the server and pull data from it, process the data, and then tell the server that it's ready for new data. Right now it uses the ProcessTime variable to simulate data processing with a thread sleep.
using System;
namespace ClientDistribute
{
class Program
{
const int BufferSize = 1024 * 1024 * 3;
const int ProcessTime = 1000;
static void Main(string[] args)
{
string socketLocation = "C:\\Users\\Public\\Documents\\temp\\distribute.sock";
Console.WriteLine("Creating socket");
using var socket = new System.Net.Sockets.Socket(System.Net.Sockets.AddressFamily.Unix, System.Net.Sockets.SocketType.Stream, System.Net.Sockets.ProtocolType.Unspecified);
Console.WriteLine("Creating end point");
var socketEndPoint = new System.Net.Sockets.UnixDomainSocketEndPoint(socketLocation);
Console.WriteLine("Connecting");
socket.Connect(socketEndPoint);
var recBuffer = new byte[BufferSize];
int recLen;
for (; ; )
{
recLen = socket.Receive(recBuffer);
Console.WriteLine($"Got {recLen} bytes, first byte: 0x{recBuffer[0]:X2}");
//simulate processing the data
System.Threading.Thread.Sleep(ProcessTime);
socket.Send(recBuffer, 0, 1, System.Net.Sockets.SocketFlags.None);
}
Console.WriteLine("Leaving main");
}
}
}
Mostly I'm wondering if the implementation is correct and won't break under clients disconnecting and reconnecting or similar. Also performance improvements if they exist would be great. I don't think security is too important since these will all be on a single network that does not have an external connection. I'm not too concerned about style, but if there's any clever ways to simplify the code or make it easier to read without breaking the functionality then that might be good too.
ProcessTimevariable to different values. \$\endgroup\$