C#/.NET multithreaded application design

Question

The application to be designed serves as a bridge between two different systems.
One natively speaks TCP (RS232 actually, but there's a COM->ETH server in the line of communication) - the other one is an ERP system (able to talk through web services).

So in a picture it would look like that:

||device|| <----TCP----> ||my application|| <----WCF----> ||ERP||

Now as a web developer I never had much to do with threads (changed a bit with MVC and Tasks but still...).

Now I am challenged with a proper "server" design.

1. I already realized while(true) with a Sleep inside is probably not the smartest choice. Been there, done that..
2. From what I have read so far I need something that blocks. Like Console.ReadKey() only that I don't want to run this as a console application. It should run as a service in the background (Windows Service for example).

I had a look at the consumer/producer pattern using a BlockingCollection which is pretty neat. I now have my producer (WCF server) which triggers creation of consumers.

My question now is... how do I - from my WCF self host (=producer) - access my multiple consumer instances? I need to query their internal state and also I need to be able to destroy single consumer instances if they become stuck - it happens.

OperationsManager:

using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using WindowsService2.Producer;

namespace WindowsService2
{
    internal class OperationsManager
    {
        private CancellationTokenSource _cts;
        private List<Task> _tasks;
        private ILogger _logger;

        internal OperationsManager()
        {
            _cts = new CancellationTokenSource();
            _tasks = new List<Task>();
            _logger = new Logger("OperationsManager");
        }

        //internal int ConsumerCount { get; set; }

        internal void Start()
        {
            _logger.Log("Start() called.");

            StartProducer();
            StartConsumers();
        }

        private void StartConsumers()
        {
            // NOP.
        }

        private void StartProducer()
        {
            BlockingCollection<ProducerMessage> blockingCollection = new BlockingCollection<ProducerMessage>();

            Task producerTask = Task.Factory.StartNew(() =>
            {
                ManagementService managementService = new ManagementService(blockingCollection, _cts.Token);
                managementService.Produce();
            }, _cts.Token);

            _tasks.Add(producerTask);
        }

        internal void Stop()
        {
            _logger.Log("Stop() called.");

            _cts.Cancel();
            Task.WaitAll(_tasks.ToArray(), Timeout.Infinite);

            _cts.Dispose();
        }
    }
}

Producer

using System;
using System.Collections.Concurrent;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.Threading;
using WindowsService2.Producer.WCF;

namespace WindowsService2.Producer
{
    // this is my "producer"
    internal class ManagementService : Producer
    {
        private Logger _logger;
        private ServiceHost _serviceHost;

        internal ManagementService(BlockingCollection<ProducerMessage> blockingCollection, CancellationToken cancellationToken) 
            : base(blockingCollection, cancellationToken)
        {
            _logger = new Logger("ManagementService");
            _cancellationToken.Register(ShutDownWcfEndpoint);
        }

        public override void Produce()
        {
            _logger.Log("Produce() called.");

            StartWcfEndpoint();
        }

        private void StartWcfEndpoint()
        {
            // see https://msdn.microsoft.com/en-us/library/ms731758%28v=vs.110%29.aspx
            Uri baseAddress = new Uri("http://localhost:8080/hello");

            // Create the ServiceHost.
            _serviceHost = new ServiceHost(typeof(HelloWorldService), baseAddress);

            // Enable metadata publishing.
            ServiceMetadataBehavior smb = new ServiceMetadataBehavior();
            smb.HttpGetEnabled = true;
            smb.MetadataExporter.PolicyVersion = PolicyVersion.Policy15;
            _serviceHost.Description.Behaviors.Add(smb);

            // Open the ServiceHost to start listening for messages. Since
            // no endpoints are explicitly configured, the runtime will create
            // one endpoint per base address for each service contract implemented
            // by the service.
            _serviceHost.Open();
        }

        private void ShutDownWcfEndpoint()
        {
            _logger.Log("Shutting down WCF endpoint...");
            _serviceHost.Close();

            _logger.Log("Shutting down WCF endpoint... completed.");
        }
    }
}

Consumer

public class Consumer
{
    // No code yet.
    // It has an "internal state" and uses TcpListener to communicate with the device.
    // It should be able to take commands from the management service (producer) 
    // (like: destroy yourself, what's your status?, ...)
}

As you can see there are no consumers created yet. This is because I figured the design might not work after all.

• The management service (producer) needs to trigger new consumers. Therefore the management service needs to write into the BlockingCollection - but how? public static BlockingCollection?
• The management service should provide a method which shows the "internal state" of each consumer => how many consumers even are there?

I think the producer/consumer pattern might be the wrong choice here. What I need is "Hey I'm the management service and I have x number of "worker threads" and their state is collection[0].State.

How would you do that?

Before this question is being closed as too broad - it usually happens with this kind of question - at least on SO - please give me any advice on how to break the requirements into smaller more manageable pieces.

Answer 1

I think you have a pretty good idea going here.

I had a look at the consumer/producer pattern using a BlockingCollection which is pretty neat. I now have my producer (WCF server) which triggers creation of consumers.

The backend Server equivalent of consumer/producer pattern is a Async communication model. That doesnt mean it needs to be slow or infrequent. But rather than sleep, block threads etc. when condition "Full" is met, you wait for the client to say, send again now.

My question now is... how do I - from my WCF self host (=producer) - access my multiple consumer instances? I need to query their internal state and also I need to be able to destroy single consumer instances if they become stuck - it happens.

The async patterns is used in various models now. One interesting one that has some aspects you could use is registration.

If the producer is spawning clients, it should know all clients ids. I could then theoretically call , are you awake ? No response, kill it (using call to OS ?) and then respawn. At the risk of some data loss.

An example of how this kind of pattern works is SignalR (minus the Kill part) SignalR overview with Further reading

But I think it will provide enough example to model something similar to match your scenario.