SOCKET и потоки (и ещё MFC::CAsyncSocket)

[tom4ik85]

Здравствуйте. Решил обратится за помощью, потому как не очень представляю как реализовать поставленную мне задачу.
Программка должна быть сервером. К ней по сети (Ethernet) может подключится до 272 устройств.
Для сервера я могу реализовать сокет и он будет слушать.
Чтоб главное окошко не тормозить нужно обработку приема-получения данных запихивать в отдельный поток.
Т.к. неизвестно когда одно из 272 возможных клиентов захочет что-нибудь послать, а реагировать нужно как можно быстрее.
Так вот, каким образом мне реализовать чтоб возвращаемый по accept() сокет (который я и буду пользовать для связи) был в отдельном потоке. Какие потоку лучше передавать параметры?
Общение между главным и всеми остальными потоками планирую через сообщения сделать. Еще вопрос, можно ли запихнуть в параметры сообщения указатель на массив данных (те данные, что приняты сокетом).
И что будет если я запущу 272+1 потока? и как мне их потом различить?
в общем буду рад любой помощи
Заранее спасибо

[McZim]

это под венду?

[tom4ik85]

Да

[Вад]

Ну, ты можешь в одном потоке делать select на несколько сокетов, насколько я понимаю. Тогда 272 потока не понадобится...
Если же хочешь использовать сокет в отдельном потоке - создай класс по типу Receiver, который управляет потоком, и передавай туда свой accept-нутый сокет, а дальше получай/передавай из другого потока через сокет.

[tom4ik85]

В смысле селект на несколько сокетов?
И как мне без такого кол-ва потоков оперативно реагировать на сообщения из сети?
А гда можно посмотреть что такое Reciever и как он выглядит, чтоб что-то похожее сделать?
Нельзя ли сделать что-то типа:
Сервер слушает. Приходит запрос с клиента. При этом я запускаю поток и передаю ему какую-то ифу для инициализации (HWND диалога, int номер сокета, указатель на сокет или его копию и т.д.).
Сокет, тот что после accept вернулся создавать внутри потока и работать там с ним.
Т.е. поток для обработки сокета одинаковый и я его запускаю н-ное кол-во раз?
Формат данных то устройств одинаковый, потому и хочу как-то универсализировать это дело.

[McZim]

tom4ik85, может лучше почитать про концепцию потоковых серверов?

[tom4ik85]

А где почитать?
Буду благодарен за ссылки.

[Вад]

tom4ik85, Receiver - это я его условно назвал, первое что в голову пришло - это когда сокет + поток оборачиваются в класс, в частном случае, получающий данные (более общий - приём-передача).

Как это случалось реализовывать:
Есть некий владелец слушающего сокета, он в своём потоке сидит себе и слушает. Как только дослушался до accept, стучит куда надо (хозяину), отдавая ему новый открытый сокет.
Хозяин конструирует в зависимости от потребностей обёртку для этого нового сокета, имеющую свой поток и там получающую/передающую данные (обёртка может обладать своей очередью для приёма-передачи, например, или отдавать полученные данные наружу через переданный ей интерфейс). Дальше с сокетом вся работа идёт через обёртку.

Не берусь судить, насколько хороша такая схема с точки зрения проектирования, но вполне себе реализуемая

Что до функции select, то читай тот же MSDN Да и остальное про сокеты можешь почитать там же

Ещё, в случае потребности работы с отдельным потоком я обычно делаю так: есть класс, у него среди свойств есть поток (обёртка или хэндл). при создании потока для объекта туда передаётся указатель на сам объект. И дальше уже из функции потока вызывается нестатическая функция объекта. Главное - грамотно потоки завершать, чтобы до вызова деструктора (или во время него) поток был завершён, чтобы без эксцессов

[tom4ik85]

Примерно так:

Код:

class CMySock : CAsyncSocket
{
public:
     CMySock(CAsyncSock* sock);
     ~CMySock();
     ...........
protected:
     CAsyncSocket m_socket;
     HANDLE hThread;
     ...........
}

DWORD WINAPI Thread(void* lParam)
{
     Param* m_param = (Param*)lParam;
     CWnd* pWnd = m_param->pWnd; // указатель на мое окошко, чтоб сообщения посылать
     CAsyncSocket* s = m_param->sock;
     s->Recieve(...);
     ...........
     pWnd->SendMessage(WM_MY);
     ...........
     return 0;
}

CMySock::CMySock(CAsyncSock* sock)
{
     if(hThread == NULL)
     {
          sock->Accept(m_socket);
          ...........
          hthread = CreateThread(NULL, 0, Thread, &MyParam, CREATE_SUSPENDED, NULL);
     }
}

у меня просто такой сумбур в голове
я только недавно начал изучать потоки, и что-то медленно движется
Только сильно не ругайте, если глупостей много.
Лучше скажите где неправильно - буду разбираться.

[Алексей++]

tom4ik85, не нужны потоки -
один сокет пользуешь чисто как слушалку - когда клиент хочет подключиться, вызывается виртуальный метод OnAccept.
Слушающий сокет создаёт пару для подключаюзщегося сокета (через Accept) и забывает про них. Совсем
Если подключать клиент по какой то причине не захотелось,  надо вызвать Accept вхолостую - создать временный парный сокет, асептнуть и тут же убить. Иначе почему то удалённый сокет будет долбиться и долбиться

у клиентского сокета , когда его асептнули, будет вызван метод OnConnect

Ещё надо включить соответствующие события у сокетов - через AsyncSelect

[lapulya]

Не уверен, что подобный подход (я имеюю ввиду бомбардировку очереди сообщений окна сообщениями от 272 устройств, особеннопри их активности) не будет тормозить работу пользователя с GUI. Лучше, но и сложнее сделать как предлагали выше, а именно

1) есть GUI работающий в смоем потоке
2) сервер слушает в другом отдельном потоке запросы на соединение
3) по приходу запроса создаем отдельный поток в котором осуществляется работа с конкретным клиентом, данное соединение пинуется и при завершении общения сторон (короректно или не корректно) поток завершает

вот при таком раскладе все будет как часы (неоднократно проверено). Естессно надо все как следует обернуть ))) У меня все есть ))) полностью слолюшн дать не могу, но сложные момент могу мммм продемонстрировать...

Да, сразу предупреждаю, там чистый API ни какого MFC и т.п.

[Алексей++]

Да, сразу предупреждаю, там чистый API ни какого MFC и т.п.

MFC - легко Класс CAsyncSocket. И оборачивать не нужно, и потоки сами сделаются

[pups]

Прошу консультации опытных специалистов по CAsyncSocket.

Моя задача:
Написать приложение КЛИЕНТ, который будет  забирать у сервера миллион (или более) "единиц" информации в короткие сроки. При этом размер получаемой информации (одного запроса) колеблется от 0 до 100 (и более) мегабайт. Команды при обращению к серверу все различные, но в некоторых случаях имеют цикличность.

Пример работы моего клиента:
У нас есть события, которые  вызываются системой при работе клиента, это Receive. Это событие происходит когда серверу есть что отдать для клиента. Рассмотрим пример работы клиента:

... Receive()
{
// В этой функции мы должны получить данные от сервера (функция вызывается автоматически событиями, если сервер что то отдает)
// Создали буфер для получения сообщения
// Получили сообщение или часть сообщения от сервера т.к. размера буфера может не хватить
// и функция будет вызвана (событием) повторно для получения оставшейся части.
// т.е. функция будет вызвана столько раз, пока не получим всю информацию по запросу.
}

... Send()
{
while(...)
{
// Отправляем запрос на сервер
// Режим ожидания пока не получим полностью информацию на текущий запрос (при этом обрабатываем все системные события)
}
}

ВОПРОС: Как получить (создать) событие, которое сообщит, что данные текущего запроса от сервера уже получены полностью?
т.е. если мы не дождемся получения оставшейся части и отправим следующий запрос к серверу, то мы его перегрузим, а нам нужно забрать от сервера еще много информации с минимальными задержками по времени.  Есть какие то события, которые сообщат мне, что текущее сообщение получено полностью?

[Алексей++]

pups, нет такого события, которое покажет, что ВСЕ данные пришли. Происходит несколько иначе:

произведи свой класс от CAsyncSocket (лучше визардом - удобнее будет потом виртуальные функции переопределить)

Код:

class CMyS : public CAsyncSocket
{
};

для клиентского сокета при помощи AsyncSelect установи флаги

Код:

AsyncSelect(FD_READ|FD_CONNECT|FD_CLOSE);

таким образом,
1) виртуальная функция OnClose будет вызываться, когда сокет (или удалённая пара) разорвал соединение
2) виртуальная функция OnConnect будет вызываться, когда сервер подключил сокет

   и самое для тебя интересное:

3) виртуальная функция OnReceive будет вызываться, когда система имеет принятые от сервера данные, которые ещё не считаны твоим сокетом
Далее происходит следующее
  а) если в OnReceive не делать ничего, то эта функция так и будет постоянно вызываться, так как ещё есть данные, которые ты не считал
  б) если запретить обработку сообщений ( AsyncSelect(/*FD_READ*/0|FD_CONNECT|FD_CLOSE); ) то больше до разрешения эта функция не вызовется. Но так делать не нужно тебе
  в) можно вызвать CAsyncSocket::Receive() , чтоб принять часть данных в твой приёмный буфер. Функция - виртуальная, поэтому можно вызвать и просто Receive(), а можно вообще переопределить у себя в классе. Мне пока хватало просто вызвать родную функцию CAsyncSocket

Код:

void CMyS::OnReceive(int nErrorCode) 
{
	//имеются ещё не принятые данные

	//принять данные
	{
		//BYTE* m_ReadBuf		- твой приёмный буфер
		//DWORD m_dwdReadBufLen 	- длина этого буфера

		int nRead=0;

		nRead=CAsyncSocket::Receive(m_ReadBuf,m_dwdReadBufLen,0);

		if(nRead==SOCKET_ERROR || nRead<=0)
		{
			//НЕТ ДАННЫХ или какой то косяк
		}
		else
		{
			//в буфере m_ReadBuf находятся nRead байт новых данных
		}
	}

	CAsyncSocket::OnReceive(nErrorCode);
}

просто определить, что получены ВСЕ данные, не выйдет. К примеру, сервер послал тебе пакет 100 метров , затем пакет 1 метр. Ты примешь, возможно, сразу 101 метр. Как разделять пакеты - это уже твоя задача, ты читаешь данные небольшими кусками (скажем по 2 метра), отдаёшь их в накопительный буфер в своей основной программе. Программа после каждого принятого куска определяет, получен ли осмысленный пакет, если да  -она его обрабатывает и удаляет из накопительного буфера, смещая оставщийся конец ещё не докаченных данных в начало накопительного буфера.
Для чёткого разделения пакетов надо придумывать свой протокол

[pups]

pups, нет такого события, которое покажет, что ВСЕ данные пришли. Происходит несколько иначе:
... СКИП ...
просто определить, что получены ВСЕ данные, не выйдет. К примеру, сервер послал тебе пакет 100 метров , затем пакет 1 метр. Ты примешь, возможно, сразу 101 метр. Как разделять пакеты - это уже твоя задача, ты читаешь данные небольшими кусками (скажем по 2 метра), отдаёшь их в накопительный буфер в своей основной программе. Программа после каждого принятого куска определяет, получен ли осмысленный пакет, если да  -она его обрабатывает и удаляет из накопительного буфера, смещая оставщийся конец ещё не докаченных данных в начало накопительного буфера.
Для чёткого разделения пакетов надо придумывать свой протокол

в том то и вопрос! как разделить?! то что вы описали все работает как часы! я даже пробовал реализацию такую:
если размер моего буфера больше чем размер принятого пакета то это последняя передача.

_но_ сервера юникс (или ...) сами организуют (определяют) размер пакета в зависимости от своей загруженности (или на то есть другие причины) т.е. может получится так, что я определил размер буфера 1024, а сервер мне сначала отдает 900 байт, а потом шлет по 1024. это уже проверено.

примечание: сервер не мой. я не могу в качестве разделителя установить какую то последовательность символов, которая будет определять конец последней пачки.
да и в предыдущей реализации (не моей) было так организовано, но это создавало ошибки из за большой вероятности содержимых символов (конца сообщения) в середине передачи сообщения.

думаю, что должно быть какое то системное сообщение (событие). по крайней мере должно меняться состояние сокета т.к. событие рецейв вызывается от изменения состояния сокета. значит если состояние изменилось обратно, то можно это как-то поймать.

буду благодарен за любые (полезные) рекомендации.

зы. сейчас читаю мсдн chapter 7: winsock basics. там что то описывается про это, но так плохо описано, что есть предположение, что писал человек, который сам плавает в этом вопросе. при наличии знаний, там писать нужно три строчки, а писатель проявил свою "поэтическую вольность" и рассказал обо всем, но только не о том, что нужно.

[Алексей++]

разделить как - свой протокол придумать. Например, я вот это использую такой
https://forum.shelek.ru/index.php/topic,15504.0.html

но можешь что то своё придумать ) Например 4 байта - длина пакета, потом пакет, потом CRC 2 байта

[pups]

разделить как - свой протокол придумать. Например, я вот это использую такой
https://forum.shelek.ru/index.php/topic,15504.0.html
но можешь что то своё придумать ) Например 4 байта - длина пакета, потом пакет, потом CRC 2 байта

Во во! Майкрософт тоже это рекомендует:

Things get a little complicated if your message sizes vary. It is necessary to impose your own protocol to let the receiver know how big the forthcoming message will be. For example, the first 4 bytes written to the receiver will always be the integer size in bytes of the forthcoming message. The receiver will start every read by looking at the first 4 bytes, converting them to an integer, and determining how many additional bytes that message comprises.

Но я понятия не имею как создать свой протокол. Я его никогда их не делал. Тем более, что не известно как он себя поведет с большими обьемами данных и со всей прочей лабудой. Это мне придется неделю писать протокол, потом месяц отлавливать в нем ошибки ....

Может можно как-то получить (узнать) состояние сокета? Зная состояние сокета, можно создавать свои события.

[Алексей++]

Но я понятия не имею как создать свой протокол

взять лист бумаги и расписать структуру протокола. Ну вот, самое простое и не замороченное:

Код:

#pragma pack(push,1)

struct s_MyProtocolCRC
{
	friend struct s_MyProtocolHeader;
private:
	WORD wCRC;  //контрольная сумма всего пакета (без учёта байтов CRC, естественно)
	
public:
	void FillCRC(const void* pData, DWORD dwdLen)
	{
		wCRC=функция_подсчёта_контрольной_суммы(pData,dwdLen);
	}
	
	s_MyProtocolCRC():wCRC(0){}
	
	s_MyProtocolCRC(const void* pData, DWORD dwdLen)
	{
		FillCRC(pData,dwdLen);
	}
	
	bool CheckCRC(WORD wCRCforControl) const
	{
		return wCRCforControl==wCRC;
	}
};

struct s_MyProtocolHeader
{
	enum{e_sign=0x88};//сигнатура-константа

	BYTE bySign;			// сигнатура
	DWORD dwdLen_H_DATA_CRC;// длина всего пакета (с учётом заголовка и CRC)
	
	s_MyProtocolHeader(DWORD dwdDataLen):bySign(e_sign)
	{
		dwdLen_H_DATA_CRC=sizeof(*this)+dwdDataLen+sizeof(s_MyProtocolCRC);
	}

	bool CheckSign() const
	{
		return bySign==e_sign;
	}
};

#pragma pack(pop)

шлём данные

Код:

//BYTE* pBuf - буфер с данными для передачи
//DWORD dwdLen - длина данных для передачи

s_MyProtocolHeader hdr(dwdLen);
s_MyProtocolCRC crc(pBuf,dwdLen);

CAsyncSocket::Send(&hdr,sizeof(hdr),0);
CAsyncSocket::Send(pBuf,dwdLen,0);
CAsyncSocket::Send(&crc,sizeof(crc),0);

принимаем данные

Код:

BYTE* pBuf;// - буфер с некими накопленными принятыми данными
DWORD dwdLen;// - длина данных


DWORD dwdProcessed=0;
for(;;)
{
	const s_MyProtocolHeader* phdr=(s_MyProtocolHeader*)pBuf;
	//контроль на достаточную длину для заголовка
	if(dwdLen<sizeof(*phdr))break;
	//контроль сигнатуры
	if(!phdr->CheckSign())break;

	//контроль на достаточную длину
	if(dwdLen<phdr->dwdLen_H_DATA_CRC)break;

	const s_MyProtocolCRC* pCRC=0;

	const void* pData=phdr+sizeof(*phdr);
	const DWORD dwdDataLen=phdr->dwdLen_H_DATA_CRC-sizeof(*phdr)-sizeof(*pCRC);

	pCRC=(s_MyProtocolCRC*)(((BYTE*)pData)+dwdDataLen);

	WORD crc=функция_подсчёта_контрольной_суммы(pData,dwdDataLen);
	if(!pCRC->CheckCRC(crc))break;

	dwdProcessed=phdr->dwdLen_H_DATA_CRC;

	//pData - данные
	//dwdDataLen - их длина

	//обработка данных
	//...

	break;
}

//из принятого обработано dwdProcessed байт

[McZim]

offtop

Алексей1153++,

шлём данные

 

у меня шлём, прочиталось как шлЕм, сижу и думаю, какой такой шлем?

[Вад]

Подчеркну на всякий случай: нужен именно транспортный протокол над TCP. По сути, всего лишь способ сигнализации о начале пакета, его размере и типе (если нужно). Можно ещё контрольную сумму передавать, например.
Когда приходилось делать что-то подобное, обычно хватало простой заголовочной структуры для пакета, после которой уже передавались основные данные.

[Алексей++]

McZim, может сказывается, что ты букафку "ё" не уважаешь? )

[lapulya]

Я может задачи не понял, но почему мы не посылать клиенту в самом начале размер пересылаемого "пакета" данных в байтах, а на клиенте при приеме просто считать принятый объем, и если пришло не все, значит надо ждать оставшиеся данные. Это самый что ни наесть стандартный прием. И MFC не нужен... да простит меня пушистый за такое ... пример как это реализовать только на моей памяти я указывал тут раза два...

[Вад]

lapulya, собственно, я примерно о том же

[Алексей++]

lapulya, да никто не обижается, кто чем привык, тем и пользуется ) Если реализовывать в железяке - то да, по байтикам проще. А в компутере озу много, можно сильно не изголяться )

и если пришло не все, значит надо ждать оставшиеся данные

а в примере у меня так и есть, просто тонкости там не прописаны, это ж отлаживать надо

ну а MFC тут лишь для удобства, если хочется и много времени свободного - сокеты в руки и вперёд ) Всё один к одному так же будет, только дольше писать

[lapulya]

Алексей1153++, ну у тебя там очень много букав написано))) а у меня весь код просто состоит в цикле с recv и выходом их цикла по считыванию необходимого количества информации (кстати не понял причем тут железка, я про программу, которая может работать где угодно, более того, товарищ pups писал о приеме миллионов запросов в короткие сроки причем каждый "пакет" может быть 100 Мб и более, тут с ОЗУ и надорваться можно, придется еще и на диск скидывать ну или уж я не знаю что там за конфигурация аппаратного обеспечения)

[Алексей++]

ну у тебя там очень много букав написано)))

вовсе не много, для интереса скомпилируй и посмотри ассеимблерный код

тут с ОЗУ и надорваться можно, придется еще и на диск скидывать

а всё зависит не от клиента вовсе, а от того, как сервер будет данные присылать. Согласен ? Например, не пакет с 100 сообщениями, а 100 сообщений друг за другом.

[lapulya]

Ну я не про ассемблерный код, а про прикладной на с/с++, вот мой, он короче и показывает суть

Код:

	int full_length; // длина пакета в байтах (передано с сервера или определена протоколом)
	char * buffer; // выделенный буфер данных размера full_length

	int rev_length = 0;
	nParam = sizeof( SOCKADDR_IN ); 
	while (rev_length < full_length)
	{
		int recv_now = recvfrom( sock, buffer + rev_length, full_length - rev_length, 0, &addr, &nParam );
		if (recv_now == SOCKET_ERROR ) 
		{
			TRACE( WSAGetLastError() ); 
			continue; 
		}

		rev_length += recv_now;
	}

[Алексей++]

так у меня сам протокол реализован практически )

[lapulya]

Я по этому и написал "много букав - не осилил", я же не говорю, что это не рабочий вариант, просто я так понял пупсу нужна концепция и я, а точнее Вад чуть раньше предложил наиболее простую. Это также протокол, только он много проще - всего надо передавать в начале пакета (первыми 4 байтами ну или как договоримся) размер данных пакета и все.

[pups]

так у меня сам протокол реализован практически )

Алексей. Можно ли мне обьяснить "крестьянину", что Вы подразумеваете под словом "протокол"? Прошу Вас мне "на пальцах" обьяснить. Я совершенно не понимаю о чем идет речь. В моем понимании протокол, это формат передаваемых данных. Если я создам свой формат передачи, то на сервере должен быть такой же формат, иначе сервер меня не поймет. А т.к. к серверу я не имею доступа, то наверное использование протокола не приемлемо?
Я не понимаю о каком протоколе идет речь и на каком уровне предется писать.
Как я увидел из Вашего примера, то .... ничего не понял. Вы используете базовый класс Асинхронного сокета, создаете какие то структуры которые как я понимаю должны содержать структуру протокола.

Если Вы меня понимаете, прошу Вас обьясните мне дураку "на пальцах" теорию того что Вы предлагаете, чтобы мне и окружающим было понятно.

Этот вопрос очень насущный и актуальный. Много людей пытаются реализовать данную задачу. Я этот вопрос ковыряю больше месяца, перерыл кучу классов на кодепрожекте, коде гуру и т.д.

P.S. Для всех: Очень прошу не создавать в теме флейм, чтобы обсуждение основного вопроса не переросло в обсуждение других вопросов, которые уже давно решены и не интересны.
Задача разделить  поступаемый поток данных от сервера, с целью контроля передачи очереди сообщений.
Буду очень благодарен за любые рабочие варианты (благодарности типа заметок внутри программы с благодарностью о помощи, пива и т.д. предусматриваются и даже не упоминаются).

ЗЫ2. Покрутил предлагаемый вариант lapulya, но не смог ему ума придать. Он сразу возвращает минус 1 и ничего не принимает. Последние два параметра не обязательные и я передавал туда NULL.