Ошибка в многопоточном серверном приложении...помогите разобраться!

[Robinson]

Здравствуйте!
Скачал недавно пример многопоточного клиент серверного приложения, пытаюсь разобраться что к чему и немного модифицировать

Ошибка возникает когда подключается больше 2х клиентов: один из клиентов не обменивается данными с сервером, хотя другие работают нормально....

Вот сервер

Код:

#include "stdafx.h"

#define MY_PORT 666 // Порт, который слушает сервер

// прототип функции, обслуживающий
// подключившихся пользователей
DWORD WINAPI SexToClient(LPVOID thread_info);

// глобальная переменная – количество
// активных пользователей 
int nclients = 0;

struct ThreadInfo
{
	SOCKET client_socket;
	HANDLE mutex;
};

int _tmain(int argc, _TCHAR* argv[])
{
	using namespace std;
	setlocale(LC_ALL, "Russian");

	WSADATA wsadata;
	cout << "TCP SERVER DEMO" << endl;

	if (WSAStartup(0x0202, &wsadata))
	{
		// Ошибка!
		cout << "Error WSAStartup " << WSAGetLastError() << endl;
		return -1;
	}

	SOCKET mysocket;

	if ((mysocket = socket(AF_INET, SOCK_STREAM, 0)) < 0)
	{
		// Ошибка!
		cout << "Error socket" << WSAGetLastError() << endl;
		WSACleanup();
		// Деиницилизация библиотеки Winsock
		return -1;
	}

	sockaddr_in local_addr;
	local_addr.sin_family = AF_INET;
	local_addr.sin_port = htons(MY_PORT);
	local_addr.sin_addr.s_addr = 0;

	if (bind(mysocket, (sockaddr *) &local_addr, sizeof(local_addr)))
	{
		// Ошибка
		cout << "Error bind " << WSAGetLastError() << endl;
		closesocket(mysocket);  // закрываем сокет!
		WSACleanup();
		return -1;
	}

	if (listen(mysocket, 0x100))
	{
		// Ошибка
		cout << "Error listen " << WSAGetLastError();
		closesocket(mysocket);
		WSACleanup();
		return -1;
	}

	cout << "Ожидание подключений" << endl;

	//SOCKET client_socket;  
	sockaddr_in client_addr;

	int client_addr_size = sizeof(client_addr);

            //ввожу i для задержки, чтобы успеть запустить несколько клиентов
	int i = 0;
	cin >> i;
	cout << i << endl;

	ThreadInfo thd_info;
	thd_info.mutex = CreateMutex(NULL, false, NULL);

	while(thd_info.client_socket = accept(mysocket, (sockaddr*) &client_addr, &client_addr_size))
	{
		nclients++;// увеличиваем счетчик подключившихся клиентов

		// пытаемся получить имя хоста
		HOSTENT* hst = gethostbyaddr((char*) &client_addr.sin_addr.s_addr, 4, AF_INET);

		// вывод сведений о клиенте
		cout << hst->h_name << inet_ntoa(client_addr.sin_addr) << " new connect!" << endl;
			
		DWORD thID;
		HANDLE my_thread = CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);
		CloseHandle(my_thread);
    }

	return 0;
}

DWORD WINAPI SexToClient(LPVOID thread_info)
{
	SOCKET my_sock = ((ThreadInfo*) thread_info)->client_socket;
	char buff[64];

	// цикл эхо-сервера: прием строки от клиента и возвращение ее клиенту
	int bytes_recv = 0;
	while( (bytes_recv = recv(my_sock, buff, sizeof(buff), 0)) && bytes_recv != SOCKET_ERROR)
		send(my_sock, buff, bytes_recv, 0);

	// если мы здесь, то произошел выход из цикла по причине возращения функцией recv ошибки – соединение клиентом разорвано
	if(WaitForSingleObject(((ThreadInfo*) thread_info)->mutex, INFINITE) == WAIT_OBJECT_0)
	{
		--nclients; // уменьшаем счетчик активных клиентов
		std::cout << "-disconnect, clients: " << nclients << std::endl;
	}
	else
	{
		std::cout << "wait" << std::endl;
	}

	ReleaseMutex(((ThreadInfo*) thread_info)->mutex);

	// закрываем сокет
	closesocket(my_sock);
	return 0;
}

Сначала запускаю сервер, затем несколько клиентов, после чего ввожу i для начала работы сервера.
Только 2 клиента принимают и передают инфу, остальные не делают ничего....

Вот клиент

Код:

#include "stdafx.h"

#define PORT		666
#define SERVERADDR	"127.0.0.1"
#define BUFF_SIZE	64

int main(int argc, char* argv[])
  {
	using namespace std;

	char buff[BUFF_SIZE];

	cout << "TCP DEMO CLIENT" << endl;

	WSADATA wsadata;

	if (WSAStartup(0x202, &wsadata))
	{
		cout << "WSAStart error " << WSAGetLastError() << endl;
		return -1;
	}

	SOCKET my_sock;

	my_sock = socket(AF_INET,SOCK_STREAM,0);

	if (my_sock < 0)
	{
		cout << "Socket() error " << WSAGetLastError() << endl;
		return -1;
	}

	sockaddr_in dest_addr;
	dest_addr.sin_family = AF_INET;
	dest_addr.sin_port=htons(PORT);
	HOSTENT *hst;


	if (inet_addr(SERVERADDR) != INADDR_NONE) 
		dest_addr.sin_addr.s_addr = inet_addr(SERVERADDR);
	else

	if (hst = gethostbyname(SERVERADDR))// hst->h_addr_list содержит не массив адресов, а массив указателей на адреса
		((unsigned long *)&dest_addr.sin_addr)[0] = ((unsigned long **)hst->h_addr_list)[0][0];
	else 
	{
		cout << "Invalid address " << SERVERADDR << endl;
		closesocket(my_sock);
		WSACleanup();
		return -1;
	}

	if (connect(my_sock, (sockaddr *)&dest_addr, sizeof(dest_addr)))
	{
		cout << "Connect error " << WSAGetLastError() << endl;
		return -1;
	}

	cout << "Connection with " << SERVERADDR << " successful complete" << endl;


	int nsize;

	for(int i = 0; i < 10; i++)
	{
		send(my_sock, "Hallow, server!", sizeof("Hallow, server!"), 0); //передаем строку клиента серверу
			
		if((nsize = recv(my_sock, buff, sizeof(buff) - 1, 0)) == SOCKET_ERROR)
		{
			//некорректный выход
			cout << "Recv error " << WSAGetLastError() << endl;	
			closesocket(my_sock);
			WSACleanup();
			return 0;
		}

		// выводим на экран 
		cout << i << ": " << buff << endl;

		Sleep(2000);
	}

	cout << "Exit..." << endl;
	closesocket(my_sock);
	WSACleanup();
	return -1;
}

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

[RXL]

Robinson, в многопоточном приложении при работе с общими объектами необходимо пользоваться средствами синхронизации.

[Robinson]

как я понимаю общий объект у меня это глобальная переменная nclients
доступ к ней в каждом потоке я синхронизирую так

Код:

if(WaitForSingleObject(((ThreadInfo*) thread_info)->mutex, INFINITE) == WAIT_OBJECT_0)
	{
		--nclients; // уменьшаем счетчик активных клиентов
		std::cout << "-disconnect, clients: " << nclients << std::endl;
	}
	else
	{
		std::cout << "wait" << std::endl;
	}

	ReleaseMutex(((ThreadInfo*) thread_info)->mutex);

только ошибка возникает не здесь, а , ИМХО, когда пытаются одновременно подключиться несколько клиентов...
что именно мне нужно синхронизировать?

[Алексей++]

во всё не всматривался , но в глаза бросилось вот что:

Код:

HANDLE my_thread = CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);
CloseHandle(my_thread);

не знаю как другие, а я так с потоками не поступал никогда, да и мсдн пишет

The thread object remains in the system until the thread has terminated and all handles to it have been closed through a call to CloseHandle.

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

Код:

DWORD WINAPI SexToClient(LPVOID thread_info)
{
...
return 0;
}

[Robinson]

убрал CloseHandle(my_thread); - ничего не изменилось, но все равно спасибо за замечание...

[sss]

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

Код:

  send(my_sock, "Hallow, server!", sizeof("Hallow, server!"), 0); //передаем строку клиента серверу

Что нибудь проявится. Может быть данные не были отправлены из-за ошибки, например, WSAENOBUFS  ...

P.S.: Интересное применение sizeof, не правда ли!

[Domenik]

Код:

HANDLE my_thread = CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);
CloseHandle(my_thread);

Вполне приемлемая строка. CloseHandle(my_thread); означает лишь то, что с указателем на данный поток вы работать не будете, а не то, что поток уничтожается, если правильно описан поток, то возможно управления им из родительского потока и не потребуется.

Сам, когда писал чат сервер, делал аналогично, прекрасно все работало.


P.S. возможно Алексей1153++ прав, я лишь говорю о собственном опыте.

[Алексей++]

sss, а лучше так

Код:

const char* p="Hallow, server!";
send(my_sock, p, ::strlen(p)+1, 0); //передаем строку клиента серверу

Domenik, дело в том, что я не проверял, что будет, если так закрыть хендл только что запущенного потока Более того, я никогда его, хендл, не сохранял, то есть тупо так:

::CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);

и всё. Ну и обеспечиваю естественный выход из процедуры потока всегда. Пока что всё работает нормально
А сеть я пишу в MFC на полюбившемся CAsyncSocket - он такой ручной, чертяко ))

---------------
Эй, знатоки, разъясните ))

[sss]

Алексей1153++, а лучше так 

Код:

#define HELLOW_TEXT_409      TEXT("Hallow, server!")
#define HELLOW_TEXT_419      TEXT("Прывет, сервер!")
#define HELLOW_TEXT          HELLOW_TEXT_409

  HRESULT          hr;
  ExceptionReport  eReport;
  size_t           stCbLen;
  try
  {
    if ( FAILED( hr = ::StringCbLength( HELLOW_TEXT, BUFF_SIZE, &stCbLen)))
      throw new __EComError( hr);

    if ( SOCKET_ERROR == ::send( my_sock, HELLOW_TEXT, stCbLen, 0))
      throw new __ESystemError( ::WSAGetLastError());
  }
  catch( EExcept* e)
  {
    std::cout << "client exception report: " << e->GetReport( eReport)  << std::endl;
    e->Release();
  }
}

Я только так работаю со строками, уже года 2....

[Алексей++]

дефайны (да и то они не всегда нужны) пригодятся, когда текст много где использовался, а тут достаточно и одном месте ) Незачем загромождать, когда не нужно.

[Domenik]

sss

используй char, так намного удобней, а в дефайнах и так много всякой нечисти, и без текстовых строк).

П.С. Прошу прощения за мой английский =)

[Алексей++]

используй char, так намного удобней

чем же ? Поясни на примере

[Domenik]

Код:

...
char* p="<<Здесь буфер большой длинны>>";
send(my_sock, p, ::strlen(p)+1, 0); //передаем строку клиента серверу
delete[] p;
...

Плюс в том, что можно не плодить в памяти множество буферов, а создавать их, когда они понадобились, а когда они больше не нужны, их можно удалить.

Во-вторых,  define служит как вы сами и говорили не для этого, а для того, чтобы определить там константы, которые впоследствии будут использоваться множество раз.

В-третьих, мне удобней использовать char, не знаю точно почему, исторически так сложилось наверно…

[sss]

Domenik, и чё после delete[] p? Исключений не бывает?

P.S.: И кстати, с чего ты взял что я использую char а не wchar_t?

P.P.S.: Я вообще хотел привлечь внимание не к define, а к StringCbLength!!!

[Алексей++]

Domenik, ты вообще то не про char* , а про char говорил - это же разное ))

а тут

char* p="<<Здесь буфер большой длинны>>";

во первых, строка КОНСТАНТНАЯ, а не буфер (даст ли компилятор так сделать ?)
во вторых - а 256 длину максимум не хочешь ? ) Тоже компилятор скажет.

в самых третьих , если где то используешь две одинаковые константные строки, то это не будут буферы, это будут указатели на одну и ту же строку.

const char* p1="одинаковый текст";
const char* p2="одинаковый текст";
const char* p3="одинаковый текст";

с delete - это вообще цирк...

sss. надо было явно указать, что подчёркивал, а то мы щас устроим )))

[Domenik]

Алексей1153++

Код:

char* ttt	=	"1234567 <всего 1680 чаров + ZT> 567890";

delete[] ttt;

Лень считать, сколько символов, но 100% больше 256.. специально проверил только что(ибо в моей памяти всплывает цифра 65536)

По поводу delete[]...

Когда создается массив, в памяти выделяется область, под этот массив, функция delete[] освобождает эту область и позволяет использовать ее для других нужд.

Предлогаю закончить обсуждение темы про char...

[Алексей++]

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

1681 символов. Мне тоже лень считать, за меня это студия делает

Код:

char* ttt ="123";
delete[] ttt;

По поводу delete[]...

Когда создается массив, в памяти выделяется область, под этот массив, функция delete[] освобождает эту область и позволяет использовать ее для других нужд.

1) где оператор new ?
2) писал ли ты таким образом прогрмаммы и со скольких школ тебя за это уволили ?

[Domenik]

1) оператор new в перспективе, в отдаленном светлом будущем...
2) так не пишу.

естественно память должна выделяться оператором new если вы об этом.


понял я уже что накосячил....

[Robinson]

ещё раз спасибо, но опять ничего не изменилось....может быть такой случай высосан из пальца и нет смысла так тестить?

P.S.

Код:

HANDLE my_thread = CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);

Sleep(1); // Наверно ламерское решение, но работает...

[sss]

Robinson, ошибок send не возникает? Моё мнение - дело не в создании потока.

Алексей1153++, злой ты...

[Алексей++]

sss, конечно злой ))) Невысыпаюсь. А так я добрый, чорный и пушыстый.

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

[zubr]

Алексей1153++, CloseHandle надо делать не для удаления потока (CloseHandle не удаляет поток), а для удаления хендла потока (некоего числа выделяемого системой), чтобы не мусорить систему. Представь, если какая нибудь программа (или множество программ) в системе каждую секунду создают n-ое колличество потоков, хендлы которых не удаляются, через какое то время работы этих программ системе может не хватать 32 бит под генерацию нового хендла.

[Алексей++]

zubr, хорошо, буду знать )

[Антон (LogRus)]

Robinson, добавь вывод в консоль:
1. сразу после получения нового соединения
2. внутри функции потока, при входе.
3. в цикле, можно выводить размер полученных данных и сами данные, особенно, если клиентов научить передавать ID процесса и что бы они его писали в консоль (пригодится при анализе)

Код:

	while( (bytes_recv = recv(my_sock, buff, sizeof(buff), 0)) && bytes_recv != SOCKET_ERROR)
		send(my_sock, buff, bytes_recv, 0);

в начале каждого сообщения в консоль пиши ThreadID

проведи испытания для случая со Sleep и без него.

Код:

HANDLE my_thread = CreateThread(NULL, NULL, SexToClient, &thd_info, NULL, &thID);

Sleep(1); // Наверно ламерское решение, но работает...

[Алексей++]

я понял, почему у менябез закрывания хендла всё работает - я же вызываю MFC-шную версию запуска
::AfxBeginThread

а там завёрнуто создание класса потока CWinThread, а по выходу из потока (return 0) вызывается AfxEndThread, которая всё автоматом освобождает. А я как-то не задумывался над этим раньше - я полагал, что эта функция тупо вызывает CreateThread, а там, оказывается, они наворотили огородище

Попробовал также и CreateThread + CloseHandle - всё нормально, поток остаётся, хендл убивается (счётчики в диспетчере тоже всё подтвердили )) )

[sss]

LogRus, sizeof(buff) не смущает?

Здесь случай ооочень тяжелый. Здесь не об работе потоков надо речь вести, а об знании языка и используемого API...

[Алексей++]

если так

Код:

BYTE buff[10];

то всё правильно,
а вот ежели динамический массив - то тогда косяк )

[sss]

Алексей1153++, что там recv ожидает на входе?

int recv( SOCKET s, char FAR* buf,  int len,  int flags);

Так что там всегда sizeof(char*)...

в общем пример всегда посылает 'Hall', не законченный нулем. Потом это дело с удовольствием выводится на cout.... Интересно, откуда он (поток) знает, что буфер закончился?

[Алексей++]

эээ...

Код:

The recv function receives data from a connected or bound socket.

int recv(
  SOCKET s,
  char* buf,
  int len,
  int flags
);

Parameters
s 
[in] Descriptor identifying a connected socket. 
buf 
[out] Buffer for the incoming data. 
len 
[in] Length of buf, in bytes 
flags 
[in] Flag specifying the way in which the call is made. 

len - Length of buf, in bytes

[sss]

Ну я про именно этот пример... Там однозначно buff имеет тип char*