Κάποιος ρώτησε πρόσφατα πως μπορεί να χρησιμοποιήσει το Task Parallel Library για να φτιάξει ένα ασύγχρονο socket server ο οποίος θα μπορεί να εξυπηρετεί πολλούς clients. Προφανώς θα πρέπει να χρησιμοποιηθούν οι ασύγχρονες μορφές των συναρτήσεων του TcpListener (BeginAccept αντί για Accept) κλπ. Το πρόβλημα είναι ότι ο κώδικας που δημιουργείται είναι λίγο μακαρόνι. Έκανα μία προσπάθεια που περιγράφω παρακάτω αλλά θα ήθελα σχόλια και βελτιώσεις.

---

Παναγιώτης Καναβός, Freelancer
Twitter: http://www.twitter.com/pkanavos

Εδώ ο ένοχος, που είχε την ανησυχία! Παραθέτω τον κώδικα που μου δουλεύει..

public class CommunicationService
    {
        const string MessageSeparator = "\r\n";

        private static CommunicationService _instance;
        public static CommunicationService Instance
        {
            get
            {
                if (_instance == null)
                    _instance = new CommunicationService();

                return _instance;
            }
        }

        BackgroundWorker _worker;

        private CommunicationService()
        {
            _worker = new BackgroundWorker();
            _worker.DoWork += new DoWorkEventHandler(workerOnDoWork);
            _worker.RunWorkerCompleted += new RunWorkerCompletedEventHandler(workerOnRunWorkerCompleted);
            _worker.WorkerSupportsCancellation = true;
        }

        ~CommunicationService()
        {
            _worker.Dispose();
            _worker = null;
        }

        public void RunWorker(int port)
        {
            RunWorker(new CommunicationServiceStartEventArgs() { ServerEndPoint = new IPEndPoint(0, port) });
        }

        public void RunWorker(CommunicationServiceStartEventArgs e)
        {
            if (_worker.IsBusy)
                throw new InvalidOperationException();

            _worker.RunWorkerAsync(e);
        }

        public void CancelWorker()
        {
            _worker.CancelAsync();
        }

        void AfterAcceptTcpClient(IAsyncResult resultAcceptTcpClient)
        {
            TcpListener _listener = resultAcceptTcpClient.AsyncState as TcpListener;
            TcpClient _client = _listener.EndAcceptTcpClient(resultAcceptTcpClient);
            NetworkStream _stream = _client.GetStream();

            TcpClientState _state = new TcpClientState(_client);
            _stream.BeginRead(_state.IncomingBuffer, 0, _state.IncomingBuffer.Length, AfterRead, _state);
        }

        void AfterRead(IAsyncResult resultRead)
        {
            TcpClientState _state = resultRead.AsyncState as TcpClientState;
            NetworkStream _stream = _state.Client.GetStream();
            int bytesread = _stream.EndRead(resultRead);

            if (bytesread > 0)
            {
                _state.IncomingData += Encoding.ASCII.GetString(_state.IncomingBuffer, 0, bytesread);

                string[] sentences = _state.IncomingData.Split(new string[] { MessageSeparator }, StringSplitOptions.None);
                if (sentences.Length > 1)
                {
                    for (int position = 0; position < sentences.Length - 1; position++)
                    {
                        this.OnMessageReceived(new CommunicationServiceReceiveEventArgs(sentences[position]) { ClientEndPoint = _state.Client.Client.RemoteEndPoint as IPEndPoint });
                    }
                    _state.IncomingData = sentences[sentences.Length - 1];
                }

                _stream.BeginRead(_state.IncomingBuffer, 0, _state.IncomingBuffer.Length, AfterRead, _state);
            }

        }

        void workerOnDoWork(object sender, DoWorkEventArgs e)
        {
            CommunicationServiceStartEventArgs args = e.Argument as CommunicationServiceStartEventArgs;
            if (args != null)
            {

                TcpListener _listener = new TcpListener(args.ServerEndPoint);
                _listener.Start();
                Console.WriteLine(">>> Listener started...");

                while (!_worker.CancellationPending)
                {
                    IAsyncResult a = _listener.BeginAcceptTcpClient(AfterAcceptTcpClient, _listener);
                    a.AsyncWaitHandle.WaitOne();
                }

                _listener.Stop();
                Console.WriteLine(">>> Listener stopped!");
                _listener = null;
            }
        }

        void workerOnRunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
        {
            Console.WriteLine(">>> Listener finished!");
        }


        public event EventHandler<CommunicationServiceReceiveEventArgs> MessageReceived;

        protected virtual void OnMessageReceived(CommunicationServiceReceiveEventArgs e)
        {
            EventHandler<CommunicationServiceReceiveEventArgs> handler = MessageReceived;
            if (handler != null)
                handler(this, e);
        }

    }

    public class CommunicationServiceStartEventArgs : EventArgs
    {
        public IPEndPoint ServerEndPoint { get; set; }
    }

    public class CommunicationServiceReceiveEventArgs : EventArgs
    {
        public string MessageText { get; set; }
        public string MessageBytes { get; set; }
        public IPEndPoint ClientEndPoint { get; set; }

        public CommunicationServiceReceiveEventArgs()
        { }

        public CommunicationServiceReceiveEventArgs(string message)
        {
            this.MessageText = message;
        }


    }

    public class TcpClientState
    {
        const int BufferSize = 512;

        public TcpClient Client { get; set; }
        public byte[] IncomingBuffer { get; set; }
        public string IncomingData { get; set; }

        public TcpClientState(TcpClient Client)
        {
            this.Client = Client;
            this.IncomingBuffer = new byte[BufferSize];
        }
    }

Έχω ένα singleton, που χρησιμοποιεί ένα BackgroundWorker object για να μην είναι blocking, και στην συνέχεια, περιμένει να γίνει μια σύνδεση (blocking). Με το που γίνει η σύνδεση κάνει async accept, και διαδοχικά async reads μέχρι να κλείσει το port, ή να μην υπάρχουν άλλα δεδομένα να διαβάσει. Επειδή τα threads διαδέχονται το ένα το άλλο, για την φυσική συνέχεια μεταξύ τους, υπάρχει ένα state object που περιέχει το TCPClient που ελέγχει την σύνδεση, και το buffer που γεμίζει. Ο τρόπος που επικοινωνεί με τον "έξω κόσμο" είναι ένα event.

Ο κώδικας που ενεργοποιεί το server και παίρνει τα event:

class Program
    {
        static SubscriptionServiceClient  subscriptionService;

        static void Main(string[] args)
        {
            CommunicationService.Instance.MessageReceived += new EventHandler<CommunicationServiceReceiveEventArgs>(InstanceOnMessageReceived);
            CommunicationService.Instance.RunWorker(4070);

            //Thread.Sleep(60000);
            Console.Read();

            CommunicationService.Instance.CancelWorker();
        }

        static void InstanceOnMessageReceived(object sender, CommunicationServiceReceiveEventArgs e)
        {
            try
            {
                Console.WriteLine("{0}:{1} {2}", e.ClientEndPoint.Address, e.ClientEndPoint.Port, e.MessageText);
                subscriptionService.SendMessage(e.MessageText);
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }

Ξεκινάω το service και κάνω block το thread με ένα Console.Read()...

Η αλήθεια είναι ότι πριν καταλήξω σε αυτή την λύση είχα δοκιμάσει μια άλλη, που με απογοήτευσε... Έπαιζε για περίπου 60" και μετά κατέβαζε το μηχάνημα μιας και έτρωγε όλη την μνήμη μου - δεν κατάφερα να βρω τι έφταιγε, αν ενδιαφέρεσαι, να το δεις...

    public class CommunicationService
    {
        private static CommunicationService _instance;
        public static CommunicationService Instance
        {
            get
            {
                if (_instance == null)
                    _instance = new CommunicationService();

                return _instance;
            }
        }

        BackgroundWorker _worker;

        private CommunicationService()
        {
            _worker = new BackgroundWorker();
            _worker.DoWork += new DoWorkEventHandler(workerOnDoWork);
            _worker.RunWorkerCompleted += new RunWorkerCompletedEventHandler(workerOnRunWorkerCompleted);
            _worker.WorkerSupportsCancellation = true;
        }

        ~CommunicationService()
        {
            _worker.Dispose();
            _worker = null;
        }

        public void RunWorker(int port)
        {
            RunWorker(new CommunicationServiceStartEventArgs() { ServerEndPoint = new IPEndPoint(0, port) });
        }

        public void RunWorker(CommunicationServiceStartEventArgs e)
        {
            if (_worker.IsBusy)
                throw new InvalidOperationException();

            _worker.RunWorkerAsync(e);
        }

        public void CancelWorker()
        {
            _worker.CancelAsync();
        }

        void workerOnDoWork(object sender, DoWorkEventArgs e)
        {
            CommunicationServiceStartEventArgs args = e.Argument as CommunicationServiceStartEventArgs;
            if (args != null)
            {
                List<TcpClient> clients = new List<TcpClient>();

                TcpListener _listener = new TcpListener(args.ServerEndPoint);
                _listener.Start();
                Console.WriteLine(">>> Listener started...");

                while (!_worker.CancellationPending)
                {
                    _listener.BeginAcceptTcpClient(asyncAccept =>
                    {
                        byte[] incomingBuffer = new byte[512];
                        TcpClient _client = _listener.EndAcceptTcpClient(asyncAccept);
                        clients.Add(_client);

                        NetworkStream _stream = _client.GetStream();
                        IAsyncResult _readResult = _stream.BeginRead(incomingBuffer, 0, incomingBuffer.Length, null, null);

                        while (!_worker.CancellationPending)
                        {
                            // wait for the read operation to complete 
                            _readResult.AsyncWaitHandle.WaitOne();
                         
                            int bytesRead = _stream.EndRead(_readResult);

                            // if zero bytes read, the connection is closed? 
                            if (bytesRead == 0)
                            {
                                break;
                            }

                            // raise event with data 
                            string message = Encoding.ASCII.GetString(incomingBuffer, 0, bytesRead);
                            this.OnMessageReceived(new CommunicationServiceReceiveEventArgs(message) { ClientEndPoint = _client.Client.RemoteEndPoint as IPEndPoint });

                            // start the read operation again 
                            _readResult = _stream.BeginRead(incomingBuffer, 0, incomingBuffer.Length, null, null);
                        }

                        _client.Close();
                        clients.Remove(_client);

                    }, null);
                }

                foreach (TcpClient _client in clients)
                {
                    _client.Close();
                }

                _listener.Stop();
                _listener = null;
            }
        }

        void workerOnRunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
        {
            Console.WriteLine(">>> Listener finished!");
        }


        public event EventHandler<CommunicationServiceReceiveEventArgs> MessageReceived;

        protected virtual void OnMessageReceived(CommunicationServiceReceiveEventArgs e)
        {
            EventHandler<CommunicationServiceReceiveEventArgs> handler = MessageReceived;
            if (handler != null)
                handler(this, e);
        }

    }

    public class CommunicationServiceStartEventArgs : EventArgs
    {
        public IPEndPoint ServerEndPoint { get; set; }
    }

    public class CommunicationServiceReceiveEventArgs : EventArgs
    {
        public string MessageText { get; set; }
        public string MessageBytes { get; set; }
        public IPEndPoint ClientEndPoint { get; set; }

        public CommunicationServiceReceiveEventArgs()
        { }

        public CommunicationServiceReceiveEventArgs(string message)
        {
            this.MessageText = message;
        }


    }

Πάνω κάτω το ίδιο μοντέλο - υπάρχει το BackgroundWorker αλλά αντί να υπάρχουν τα async callbacks, προσπαθεί να γίνει υλοποίηση με lamdas μέσα στο workerOnDoWork() για να κάνουν το async. Αν μπορείς να εντοπίσεις που έχω κάνει το λάθος...

 

George J.

---

George J. Capnias: Χειροπρακτικός Υπολογιστών, Ύψιστος Γκουράρχης της Κουμπουτερολογίας w: capnias.org, t: @gcapnias, l: gr.linkedin.com/in/gcapnias

Το πρόβλημα δεν είναι στα lambdas αλλά στο ότι ξεκινάς συνέχεια ασύγχρονα operations μέσα σε ένα loop:

---

Παναγιώτης Καναβός, Freelancer
Twitter: http://www.twitter.com/pkanavos

Παναγιώτης Καναβός:

Επιπλέον, η χρήση του BackgroundWorker δεν προσφέρει κανένα πλεονέκτημα σε σχέση με την χρήση ενός απλού TaskFactory.StartNew καθώς δεν χρησιμοποιείς ούτε καν το ProgressEvent. Ουσιαστικά σηκώνεις ένα και μοναδικό thread και όταν τελειώσει γράφεις κάτι στην κονσόλα. Ίσα-ίσα πρέπει να γράψεις παραπάνω διάσπαρτο κώδικα για να χειριστείς τα events που σηκώνει ο Background worker.

Εδώ θέλω να σταθούμε λίγο - ωραία η όλη ασύγχρονη προσέγγιση, αλλά χωρίς την αναφορά progress και την δυνατότητα canceling της διαδικασίας, δεν νομίζω ότι είναι εύκολο να έχεις κάτι "fire & forget" μέσα σε μια εφαρμογή. Η δική μου προσέγγιση της υλοποίησης "progress report & canceling" είναι ο BackgroundWorker, και όχι υλοποίηση του asynchronous.

Μην ξεχνάς ότι ο παραπάνω κώδικας με τα lamdas δεν μου έπαιξε ποτέ σωστά, και ο άλλος κώδικας είναι work in progress ενός draft.

Πως θα πρότεινες να υλοποιηθεί το progress reporting και το canceling μιας ασύγχρονης ακολουθίας;

 

George J.

---

George J. Capnias: Χειροπρακτικός Υπολογιστών, Ύψιστος Γκουράρχης της Κουμπουτερολογίας w: capnias.org, t: @gcapnias, l: gr.linkedin.com/in/gcapnias

Τα lambdas δεν έχουν να κάνουν με το ότι δεν έπαιξε σωστά ο κώδικας. Αν δεν είχες βάλει το WaitOne και ο κώδικας με τα callbacks θα έσκαγε με τον ίδιο τρόπο. Και οι δύο μορφές του κώδικα είναι ουσιαστικά οι ίδιες.

 Όσον αφορά το progress reporting, δεν χρησιμοποιείς πουθενά το ProgressChanged event οπότε τί εννοείς reporting?

Το cancellation γίνεται μέσω του CancellationSource και του CancellationToken που μπορείς να περάσεις σαν παράμετρο σε όλα τα tasks. Καλώντας έτσι την CancellationSource.Cancel() θα ακυρωθούν όλα τα tasks τα οποία ακούνε στο ίδιο token. Αυτό σημαίνει ότι μπορείς να συνδέσεις parent και child tasks έτσι ώστε να ακυρωθούν όλα όσο γίνεται πιο γρήγορα όταν κληθεί η Cancel, χωρίς να χρειαστεί να κάνεις εσύ κάποιο έλεγχο. Αντίθετα με τον BackgroundWorker θα πρέπει να ελέγχεις πρίν από κάθε iteration και κάθε BeginXXX ότι δεν έχει σηκωθεί το CancellationPending.

Ο κώδικας για τη χρήση του CancellationSource είναι πολύ απλός:

var source = new CancellationTokenSource(); 
Task.Factory.StartNew(() => 
    { 

......
    },source.Token);

.....

source.Cancel()

---

Παναγιώτης Καναβός, Freelancer
Twitter: http://www.twitter.com/pkanavos