2
\$\begingroup\$

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.

\$\endgroup\$
5
  • \$\begingroup\$ For testing clients I put copies of the client program in multiple folders and changed the ProcessTime variable to different values. \$\endgroup\$
    – temp
    Commented Nov 13, 2021 at 3:29
  • \$\begingroup\$ Just a small thing I've noticed: AvailableDataTransferSignals is protected by ClientThreadDataLock when you create a new listening thread, but it's unprotected in the catch block. If multiple listeners break simultaneously, there might be problems. \$\endgroup\$ Commented Nov 13, 2021 at 21:10
  • \$\begingroup\$ @AndreySarafanov Oh right because it's not a concurrent stack. \$\endgroup\$
    – temp
    Commented Nov 14, 2021 at 4:30
  • \$\begingroup\$ @temp Is there any particular reason why did you implemented in a synchronous way instead of async? \$\endgroup\$ Commented Nov 15, 2021 at 10:01
  • \$\begingroup\$ @PeterCsala I guess the reason would be that I don't know how to use async. Is it a big difference for how it performs? \$\endgroup\$
    – temp
    Commented Nov 15, 2021 at 18:13

1 Answer 1

1
\$\begingroup\$

The code looks good as far as I can tell.

Just a few functional observations.

I obviously have no clue how the system will operate in real life, but I think the idea of hardcoding the number of clients is a bit odd. I assume that it will go into some sort of limping mode if say only two are connected. Perhaps it would be better to let the sensors identify themselves. As long as you got one fizz sensor, a buzz sensor and a bang sensor you are good to go. Having three buzz sensors will not do =)

What should happen if more than ClientCount sensors try to connect. Currently, it looks like the connection was successful, but it's not receiving any data.

Additionally, you may have to consider what should happen if a sensor goes offline and then come back to life. Perhaps you should devise some way to get it back up to speed with all the events it missed.

\$\endgroup\$
5
  • \$\begingroup\$ Old data is not important and needs to be discarded. The number of clients is based on the specific processing that needs to be done on them so that is going to be a fixed number and if I need to change it then it will be easy enough to change the code value. If there are less than the required number of clients then it should pause on sending data because the data is only useful with the required number of clients running their data processing things on it. Basically it's just one sensor and 3 or however many processing nodes for the data from the one sensor. \$\endgroup\$
    – temp
    Commented Nov 15, 2021 at 3:24
  • \$\begingroup\$ Oh also the sensor should not send data to the clients unless they are all done processing the previously sent data. \$\endgroup\$
    – temp
    Commented Nov 15, 2021 at 3:26
  • \$\begingroup\$ Cool. I would perhaps suggest that you split it up, on the design level, into something like a main controller, sensor(s) and processing units. Sensors are generally expected to only measure something. \$\endgroup\$
    – jdt
    Commented Nov 15, 2021 at 12:29
  • \$\begingroup\$ The sensor and processing is split up on account of the processing is done in the ClientDistribute program and the sensor data capture is done in the ServerDistribute program. I'm not sure if I can split it up more, but I guess I could have a program that captures the sensor data and sends it to another program that will then distribute it to clients, but I think the latency might be damaged from that. \$\endgroup\$
    – temp
    Commented Nov 15, 2021 at 22:36
  • \$\begingroup\$ @temp It’s your project so you can do whatever you want :-) My head is just currently in the embedded space with loop detectors and rear window washer reservoir level sensors. If your going the C/C++ route it might be worthwhile to have a look at using MPI \$\endgroup\$
    – jdt
    Commented Nov 16, 2021 at 14:35

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.