.NET Framework

Απ: Ασύγχρονα Sockets

Παναγιώτης Καναβός — Thu, 06 Oct 2011 02:25:38 GMT

Τα 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()

Απ: Ασύγχρονα Sockets

George J. Capnias — Thu, 06 Oct 2011 02:10:53 GMT

Παναγιώτης Καναβός:
Επιπλέον, η χρήση του 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.

Απ: Ασύγχρονα Sockets

Παναγιώτης Καναβός — Sun, 02 Oct 2011 01:28:56 GMT

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

while (!_worker.CancellationPending)
{
    _listener.BeginAcceptTcpClient(asyncAccept =>
    {
                        ...

    }, null);
}

Έτσι ξεκινάς άπειρα BeginAcceptTcpClient μέχρι που σκάει η εφαρμογή.

Ο κώδικας που έχεις και δουλεύει δεν κάνει ασύγχρονο BeginAcceptTcpClient γιατί αμέσως μετά την Begin κάνεις Wait. Δεν έχει διαφορά ουσιαστικά σε σχέση με την κλήση της AcceptTcpClient μέσα σε loop. Θα μπορούσες να κάνεις ακριβώς το ίδιο και με lambdas:

while (!_worker.CancellationPending)
{
   IAsyncResult a= _listener.BeginAcceptTcpClient(asyncAccept =>
    {
                        ...

    }, null);
    a.AsyncWaitHandle.WaitOne();
}

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

Απ: Ασύγχρονα Sockets

George J. Capnias — Tue, 27 Sep 2011 03:31:16 GMT

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

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.

Ασύγχρονα Sockets

Παναγιώτης Καναβός — Sat, 24 Sep 2011 01:16:07 GMT

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

Καταρχήν, ο παρακάτω κώδικας είναι η σύγχρονη μορφή του server. Ο TcpListener δέχεται συνδέσεις σε ένα loop και ό,τι παραλάβει το αποθηκεύει σε ένα αρχείο. Το αξιοσημείωτο είναι ότι αυτή η μορφή είναι πολύ γρήγορη έτσι και τρέχει στο δικό της thread, καθώς δεν υπάρχει καθόλου κόστος από context switching ή την δημιουργία tasks. Χρειάστηκε να στείλω αρκετά μεγάλα μηνύματα (αρκετά KB) πριν αρχίσω να βλέπω διαφορά ανάμεσα σε αυτή τη μορφή και τις ασύγχρονες. Σημειωτέον, χρησιμοποίησα μέχρι 64 client threads τη φορά. Αν ο αριθμός client είναι πολύ μεγάλος, κάποιοι από αυτούς θα αρχίσουν να βλέπουν timeouts.

public class TcpServer 
{        
	public void Start(int port)
	{
		var listener = new TcpListener(IPAddress.Any, port);
		listener.Start();
		Console.WriteLine("Server Started");

		Accept(listener);
	}

	private void Accept(TcpListener listener)
	{
		while (true)
		{
			var client = listener.AcceptTcpClient();  //BLOCKING
			Process(client);
		}
	}
	

	private static void Process(TcpClient client)
	{
		using (NetworkStream stream = client.GetStream())
		{
			var fileName = Path.GetTempFileName();
			
			using (var fileStream = File.Open(fileName,FileMode.Create))
			{
				stream.CopyTo(fileStream);	//BLOCKING                    
			}
			File.Delete(fileName);
		}
		client.Close();
	}
}

O server ξεκινάει φυσικά με ένα Task.Factory.StartNew(()=>myServer.Start(12345)).

Μία πρώτη βελτίωση είναι η επεξεργασία να γίνει ασύγχρονα. Αυτό γίνεται απλά τρέχοντας την Process στο δικό της task.

private void Accept(TcpListener listener)
{
    while (true)
    {
        var client = listener.AcceptTcpClient();
        Task.Factory.StartNew(() => Process(client));
    }
}

Επόμενο βήμα είναι να γίνεται ασύγχρονα και η αποδοχή των συνδέσεων, και εδώ είναι το μανίκι. Με την TaskFactory.FromAsync() μπορώ να φτιάξω ένα task από τις BeginAcceptClient, EndAcceptClient. Μόλις τελειώσει αυτό το task όμως θα πρέπει να δημιουργήσω άλλο ένα παρόμοιο για να παραλάβει την επόμενη σύνδεση. Δεν μπορώ να βάλω την FromAsync στο loop γιατί θα δημιουργηθούν άπειρα task. Πρέπει αναγκαστικά να δημιουργήσω το νέο task αφού ολοκληρωθεί το προηγούμενο. Και φυσικά ένα task που έχει ολοκληρωθεί δεν ξαναξεκινάει, για να πεις ότι θα έβαζα ένα restart στο Continue.

Μία λύση είναι το recursion που πρότεινε ο Νίκος ο Παλλαδινός στη συζήτηση για το Ασύγχρονο Retry. Η Accept ξαναγράφεται ως εξής:

private Task Accept(TcpListener listener)
{
    return Task.Factory
        .FromAsync<TcpClient>(listener.BeginAcceptTcpClient, listener.EndAcceptTcpClient, listener)
        .ContinueWith(tc =>{
            Accept((TcpListener)tc.AsyncState);
            Process(tc.Result);
        });
}

Μόλις έρθει μία σύνδεση, καλείται ξανά η Accept για να παραλάβει την επόμενη και το Process γίνεται και αυτό στο δικό του ξεχωριστό task.

Σειρά έχει να ξεφορτωθούμε και τα μπλοκαρίσματα κατά την αποθήκευση στα αρχεία. Και εδώ μπορούμε να δουλέψουμε ασύγχρονα όπως παραπάνω, χρησιμοποιώντας τις BeginRead/EndRead κλπ μαζί με την TaskFactory.FromAsync. Ή μπορούμε να κλέψουμε και να χρησιμοποιήσουμε τα stream extensions από τα ParallelExtensionsExtras, τα οποία έχουν έτοιμες ασύγχρονες μορφές για τις Stream.CopyTo, Stream.Read κλπ.

private static Task Process(TcpClient client)
{
	NetworkStream stream = client.GetStream();
	var fileName = Path.GetTempFileName();

	return stream
		.CopyStreamToFileAsync(fileName)
	.ContinueWith(t =>{
		var e = t.Exception;                                                                
		stream.Close();
		File.Delete(fileName);
		if (e != null) throw e;
	}, TaskContinuationOptions.ExecuteSynchronously) 
	.ContinueWith(t2=>
		client.Close());
}

Όλος μαζί ο server γίνεται πλέον

public class TcpServer 
{        
	public void Start(int port)
	{
		var listener = new TcpListener(IPAddress.Any, port);
		listener.Start();
		Console.WriteLine("Server Started");

		Accept(listener);
	}

	private Task Accept(TcpListener listener)
	{
		return Task.Factory
			.FromAsync<TcpClient>(listener.BeginAcceptTcpClient, listener.EndAcceptTcpClient, listener)
			.ContinueWith(tc =>{
				Accept((TcpListener)tc.AsyncState);
				return ProcessAsync(tc.Result);
			}).Unwrap();
	}
	

	private static Task Process(TcpClient client)
	{
		NetworkStream stream = client.GetStream();
		var fileName = Path.GetTempFileName();

		return stream
			.CopyStreamToFileAsync(fileName)
		.ContinueWith(t =>{
			var e = t.Exception;                                                                
			stream.Close();
			File.Delete(fileName);
			if (e != null) throw e;
		}, TaskContinuationOptions.ExecuteSynchronously) 
		.ContinueWith(t3=>
			client.Close());
	}       
}

Τί γνώμη έχετε και πως θα μπορούσε να βελτιωθεί ο κώδικας? Πως θα μπορούσε να γίνει πιο καθαρός και πιο μαζεμένος?

Σκέφτηκα καταρχήν να χρησιμοποιήσω Iterator όπως στο retry αλλά δεν θα βόλευε καθώς θέλω τα δύο βήματα της διαδικασίας (accept, process) να μπορούν να εκτελούνται ανεξάρτητα. Με τον iterator θα έπρεπε να τελειώσει πρώτα το Process για να ξεκινήσει το επόμενο accept. Το παρακάτω δηλαδή δεν παίζει όπως θέλουμε:

private IEnumerable<Task> AsyncIterator(TcpListener listener)
{
	while (true)
	{
		var acceptTask = Task.Factory
			.FromAsync<TcpClient>(listener.BeginAcceptTcpClient, listener.EndAcceptTcpClient, listener);
		yield return acceptTask;
		var client = acceptTask.Result;
		yield return Process(client);
	}
}

Καμμία καλύτερη ιδέα?