network/servers/mod_pool.c - master - bvs22/draft

network/servers/mod_pool.c
248 строк4 KB
Batomunkuev Vladimir
17 фев 2025, 08:48
17 фев 2025, 08:487478e9f
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <pthread.h>
#include <sys/signal.h>

#define NUM_THREADS 5
#define PORT_SERVER 7777
#define MAX_THREADS 20

struct Task
{
	int client_socket;
	struct sockaddr_in client_add;
	struct Task *next;
};

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex_task = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_t *thread;
int fd_socket = -1;
struct Task *task = NULL;
unsigned int cur_tasks = 0;
unsigned int cur_treads = 0;

void clean(int signal)
{
	if (fd_socket != -1)
	{
		close(fd_socket);
	}
	printf("\n");
	exit(0);
}

void handle_client(int new_sock, long id)
{
	char buffer[256] = {0};
	char *msg = "Hello client!";
	ssize_t byte_in;

	if((byte_in = recv(new_sock, buffer, sizeof(buffer) - 1, 0)) == -1)
	{
		perror("recv");
		close(new_sock);
		return;
	}

	buffer[byte_in] = '\0';

	printf("server %ld: %s\n", id, buffer);

	if (send(new_sock, msg, strlen(msg), 0) == -1)
	{
		perror("send");
		close(new_sock);
		return;
	}

	close(new_sock);
}

void *thread_func(void *arg)
{
	struct Task *cur_task;
	long id = *(long *)arg;

	while(1)
	{
		pthread_mutex_lock(&mutex_task);
		if (cur_tasks < 10 && cur_treads > NUM_THREADS)
		{
			cur_treads--;
			free(arg);
			pthread_mutex_unlock(&mutex_task);
			pthread_exit(NULL);
		}
		pthread_mutex_unlock(&mutex_task);

		pthread_mutex_lock(&mutex);

		while(task == NULL)
		{
			pthread_cond_wait(&cond, &mutex);
		}

		cur_task = task;
		task = task->next;
		pthread_mutex_unlock(&mutex);

		handle_client(cur_task->client_socket, id);
		free(cur_task);

		pthread_mutex_lock(&mutex_task);
		cur_tasks--;
		pthread_mutex_unlock(&mutex_task);
	}

	return NULL;
}

void manag_thread()
{
	pthread_mutex_lock(&mutex_task);

	long *id; 

	if (cur_tasks > cur_treads && cur_treads < MAX_THREADS)
	{
		thread = realloc(thread, sizeof(pthread_t) * (cur_treads + 5));
		for (int i = cur_treads; i < cur_treads + 5; i++)
		{
			id = malloc(sizeof(long));
			*id = i;
			pthread_create(&thread[i], NULL, thread_func, id);
		}
		cur_treads = cur_treads + 5;
	}
	pthread_mutex_unlock(&mutex_task);
}

void queue_task(int new_socket, struct sockaddr_in client_add)
{
	struct Task *new_task;
	struct Task *cur;

	if ((new_task = malloc(sizeof(struct Task))) == NULL)
	{
		perror("malloc");
		return;
	}


	pthread_mutex_lock(&mutex_task);
	cur_tasks++;
	pthread_mutex_unlock(&mutex_task);

	pthread_mutex_lock(&mutex);

	manag_thread();

	new_task->client_socket = new_socket;
	new_task->client_add = client_add;
	new_task->next = NULL;

	if (task == NULL)
	{
		task = new_task;
	}
	else
	{
		cur = task;
		while(cur->next != NULL)
		{
			cur = cur->next;
		}
		cur->next = new_task;
	}

	pthread_cond_signal(&cond);
	pthread_mutex_unlock(&mutex);
}

int main()
{
	int new_socket;
	long *thread_id;
	struct sockaddr_in server_add, client_add;
	socklen_t len_add = sizeof(client_add);

	signal(SIGINT, clean);

	if ((fd_socket = socket(AF_INET, SOCK_STREAM, 0)) == -1)
	{
		perror("socket");
		return 1;
	}

	memset(&server_add, 0, sizeof(server_add));
	server_add.sin_family = AF_INET;
	server_add.sin_addr.s_addr = INADDR_ANY;
	server_add.sin_port = htons(PORT_SERVER);

	if (bind(fd_socket, (struct sockaddr *)&server_add, sizeof(server_add)) == -1)
	{
		perror("bind");
		close(fd_socket);
		return 1;
	}

	if (listen(fd_socket, 10) == -1)
	{
		perror("listen");
		close(fd_socket);
		return 1;
	}

	thread = malloc(sizeof(pthread_t) * NUM_THREADS);
	if (thread == NULL)
	{
		perror("malloc");
		close(fd_socket);
		return 1;
	}

	for (int i = 0; i < NUM_THREADS; i++)
	{
		thread_id = malloc(sizeof(long));
		if (thread_id == NULL)
		{
			perror("malloc");
			close(fd_socket);
			free(thread);
			return 1;
		}
		*thread_id = i;
		if(pthread_create(&thread[i], NULL, thread_func, thread_id) != 0)
		{
			perror("pthread_create");
			close(fd_socket);
			return 1;
		}
	}
	cur_treads = NUM_THREADS;

	while(1)
	{
		if ((new_socket = accept(fd_socket, (struct sockaddr *)&client_add, &len_add)) == -1)
		{
			perror("accept");
			close(fd_socket);
			return 1;
		}

		queue_task(new_socket, client_add);
	}

	for (int i = 0; i < cur_treads; i++)
	{
		pthread_join(thread[i], NULL);
	}

	return 0;
}