VPF::Асинхронный сервер. - Форум программистов

iddqd

Дата 1.3.2007, 21:50 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 491
Регистрация: 16.11.2005

Репутация: нет
Всего: 2

Код


using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Collections;
using System.Threading;

public class StateObject
{
    public Socket workSocket = null;
    public const int BufferSize = 1024;
    public byte[] buffer = new byte[BufferSize];
    public StringBuilder sb = new StringBuilder();
}

public class AsyncServer
{
    public static ManualResetEvent allDone = new ManualResetEvent(false);

    public AsyncServer()
    {
    }

    public static void StartListening()
    {
        byte[] bytes = new Byte[1024];

        IPHostEntry ipHostInfo = Dns.GetHostEntry(Dns.GetHostName());
        int it = 1;
        Console.WriteLine("Выберите IP адрес:");
        foreach (IPAddress ipa in ipHostInfo.AddressList)
        {
            Console.WriteLine("{0}. {1}", it++, ipa);
        }
        int c = Convert.ToInt32(Console.ReadLine());
        IPAddress ipAddress = ipHostInfo.AddressList[c-1];
        IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 8000);

        Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

        try
        {
            listener.Bind(localEndPoint);
            listener.Listen(100);

            while (true)
            {
                allDone.Reset();

                Console.WriteLine("Ожидание подключения...");
                listener.BeginAccept(new AsyncCallback(AcceptCallback), listener);
                allDone.WaitOne();
            }

        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }

        Console.WriteLine("\nНажмите ENTER для продолжения...");
        Console.Read();

    }

    public static void AcceptCallback(IAsyncResult ar)
    {
        allDone.Set();
        Socket listener = (Socket)ar.AsyncState;
        Socket handler = listener.EndAccept(ar);
        StateObject state = new StateObject();
        state.workSocket = handler;
        Console.WriteLine("Подключился {0}", handler.RemoteEndPoint.ToString());
        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
            new AsyncCallback(ReadCallback), state);
    }

    public static string ParseCmd(string cmd)
    {
        if (cmd.IndexOf("POS") > 0) return "POS";
        if (cmd.IndexOf("DWNSER") > 0) return "DWNSER";
        return "";
    }

    public static string ValueFromToken(string cmd, string token)
    {
        string[] tmp = cmd.Split(' ');
        Console.WriteLine("{0}\n{1}\n{2}\n", tmp[1], tmp[2], tmp[3]);
        //return tmp[12].Substring(3);
        foreach (string s in tmp)
        {
            if (s.IndexOf("=")<0) continue;
            if (s.Remove(s.IndexOf("=")) == token)
            {
                return s.Substring(s.IndexOf("=")+1);
            }
        }
        return "";
    }

    public static void ReadCallback(IAsyncResult ar)
    {
        String content = String.Empty;

        StateObject state = (StateObject)ar.AsyncState;
        Socket handler = state.workSocket;

        int bytesRead = handler.EndReceive(ar);
        if (bytesRead > 0)
        {
            //state.sb.Append(Encoding.ASCII.GetString(state.buffer, 0, bytesRead));
            q.Enqueue(Encoding.ASCII.GetString(state.buffer, 0, bytesRead));
            state.sb = new StringBuilder("");
            //content = state.sb.ToString();
            content = q.Dequeue().ToString();
            Console.WriteLine("Прочитано {0} байт из сокета. \n Данные : {1}", content.Length, content);
            if (content.IndexOf(">") > -1)
            {                
                switch (ParseCmd(content))
                {
                    case "POS":
                        Send(handler, "<ACK.SERVER=" + ValueFromToken(content, "AC") + ">");
                        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
                        break;
                    case "DWNSER":
                        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
                        break;
                }
            }
            else
            {
                handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
            }
        }
    }

    private static void Send(Socket handler, String data)
    {
        byte[] byteData = Encoding.ASCII.GetBytes(data);

        handler.BeginSend(byteData, 0, byteData.Length, 0,
            new AsyncCallback(SendCallback), handler);
    }

    private static void SendCallback(IAsyncResult ar)
    {
        try
        {
            Socket handler = (Socket)ar.AsyncState;

            int bytesSent = handler.EndSend(ar);
            Console.WriteLine("Отправлено {0} байт клиенту.", bytesSent);

        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    public static Queue q = new Queue();

    public static int Main(String[] args)
    {
        StartListening();
        return 0;
    }
}

Этот сервер отлично работает. Но в один момент клиент начинает ему посылать команду "POS" очень быстро. Более 5-ти команд в секунду и сервер не успевает их обрабатывать, не помогло даже создание очереди, пакеты смешиваются. Пожалуйста, скажите, как можно синхронизировать их действия?

Добавлено @ 21:51
Конкретней: процедура ReadCallback.

--------------------

Plamenk

Дата 2.3.2007, 13:59 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 258
Регистрация: 18.2.2003

Репутация: 1
Всего: 3

Попробуй сделать отдельный поток для функционала выполняемого в ReadCallback.

Например:

1. Твой ReadCallback просто добавляет StateObject в очередь и уведомляет поток обработчик, что пришло но вое сообщение;
2. Поток обработчик последовательно выбирает StateObject из очереди и выполняет обработку;

И ещё мне кажется что данный фрагмент кода может является источником проблем:

Код


case "POS":
      Send(handler, "<ACK.SERVER=" + ValueFromToken(content, "AC") + ">");
      handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
      break;

Ты начинаешь снова получать данные (handler.BeginReceive) не дождавшись окончание отправки (Send). Лучше либо снова получать данные после завершения операции Send (SendCallback), либо использовать для отправки новый Socket.

iddqd

Дата 2.3.2007, 15:52 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 491
Регистрация: 16.11.2005

Репутация: нет
Всего: 2

Сейчас вся эта красота выглядит вот так, но ошибка осталась. Очередь раньше вобще не имела смысла.

Код


using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Collections;
using System.Threading;

public class StateObject
{
    public Socket workSocket = null;
    public const int BufferSize = 1024;
    public byte[] buffer = new byte[BufferSize];
    public StringBuilder sb = new StringBuilder();
}

public class AsyncServer
{
    public static ManualResetEvent allDone = new ManualResetEvent(false);

    public AsyncServer()
    {
    }

    public static void StartListening()
    {
        IPHostEntry ipHostInfo = Dns.GetHostEntry(Dns.GetHostName());
        int it = 1;
        Console.WriteLine("Выберите IP адрес:");
        foreach (IPAddress ipa in ipHostInfo.AddressList)
        {
            Console.WriteLine("{0}. {1}", it++, ipa);
        }
        int c = Convert.ToInt32(Console.ReadLine());
        IPAddress ipAddress = ipHostInfo.AddressList[c-1];
        IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 8000);

        Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

        try
        {
            listener.Bind(localEndPoint);
            listener.Listen(100);

            while (true)
            {
                allDone.Reset();

                Console.WriteLine("Ожидание подключения...");
                listener.BeginAccept(new AsyncCallback(AcceptCallback), listener);
                allDone.WaitOne();
            }

        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }

        Console.WriteLine("\nНажмите ENTER для продолжения...");
        Console.Read();

    }

    public static void AcceptCallback(IAsyncResult ar)
    {
        allDone.Set();
        Socket listener = (Socket)ar.AsyncState;
        Socket handler = listener.EndAccept(ar);
        StateObject state = new StateObject();
        state.workSocket = handler;
        Console.WriteLine("Подключился {0}", handler.RemoteEndPoint.ToString());
        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
            new AsyncCallback(ReadCallback), state);
    }

    public static string ParseCmd(string cmd)
    {
        if (cmd.IndexOf("POS") > 0) return "POS";
        if (cmd.IndexOf("DWNSER") > 0) return "DWNSER";
        return "";
    }

    public static string ValueFromToken(string cmd, string token)
    {
        string[] tmp = cmd.Split(' ');
        foreach (string s in tmp)
        {
            if (s.IndexOf("=")<0) continue;
            if (s.Remove(s.IndexOf("=")) == token)
            {
                return s.Substring(s.IndexOf("=")+1);
            }
        }
        return "";
    }

    public static void ReadCallback(IAsyncResult ar)
    {
        String content = String.Empty;

        StateObject state = (StateObject)ar.AsyncState;
        Socket handler = state.workSocket;

        int bytesRead = handler.EndReceive(ar);
        if (bytesRead > 0)
        {
            state.sb.Append(Encoding.ASCII.GetString(state.buffer, 0, bytesRead));
            content = state.sb.ToString();
            if (content.IndexOf(">") > -1)
            {
                Console.WriteLine("Прочитано {0} байт из сокета. \n Данные : {1}", content.Length, content);
                switch (ParseCmd(content))
                {
                    case "POS":
                        Send(handler, "<ACK.SERVER=" + ValueFromToken(content, "AC") + ">");
                        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
                        break;
                    case "DWNSER":
                        handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
                        break;
                }
            }
            else
            {
                handler.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ReadCallback), state);
            }
        }
    }

    private static void Send(Socket handler, String data)
    {
        byte[] byteData = Encoding.ASCII.GetBytes(data);

        handler.BeginSend(byteData, 0, byteData.Length, 0,
            new AsyncCallback(SendCallback), handler);
    }

    private static void SendCallback(IAsyncResult ar)
    {
        try
        {
            Socket handler = (Socket)ar.AsyncState;

            int bytesSent = handler.EndSend(ar);
            Console.WriteLine("Отправлено {0} байт клиенту.", bytesSent);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }
    public static int Main(String[] args)
    {
        StartListening();
        return 0;
    }
}

Цитата

И ещё мне кажется что данный фрагмент кода может является источником проблем:

Он им и является smile

Никак не могу вылечить это. Что уже только не делал. Последняя попытка была сделать глобальный флаг и после обработки калбэка на Send менять его.

Цитата

Попробуй сделать отдельный поток для функционала выполняемого в ReadCallback.

Даже не представляю как это реализовать smile

--------------------

Plamenk

Дата 2.3.2007, 17:22 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 258
Регистрация: 18.2.2003

Репутация: 1
Всего: 3

Самое простое - попробуй для отправки (Send) использовать новый Socket.

Цитата

Даже не представляю как это реализовать

Если в самых общих чертатах, то:

1. Создаешь глобальную очередь;
2. Создаешь поток, который в цикле опрашивает очередь и если там есть сообщения обрабатывает их;
3. Модифицируешь ReadCallback, таким образом, чтобы она заполняла глобальную очередь и уведомляла поток о том, что в очереди появилось сообщение;

ЗЫ

Доступ к очереди необходимо синхронизовать (самое простое с помощью критических секций).

Это сообщение отредактировал(а) Plamenk - 2.3.2007, 17:23

iddqd

Дата 2.3.2007, 20:00 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 491
Регистрация: 16.11.2005

Репутация: нет
Всего: 2

Не хватает моих сил и знаний для этой реализации. Может кто-нибудь поможет парой-тройкой строчек кода?

--------------------

iddqd

Дата 9.3.2007, 10:05 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 491
Регистрация: 16.11.2005

Репутация: нет
Всего: 2

Код


Доступ к очереди необходимо синхронизовать (самое простое с помощью критических секций).

Вроде всё сделал. Осталась последняя проблема. Вот с этим (цитата) можно пояснить?

--------------------

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Распределённые приложения и сеть \| Следующая тема »