1. Mã Nguồn Server Socket
Server được thiết kế để chạy trên một luồng duy nhất. Nó lắng nghe các kết nối từ client và quản lý danh sách các socket đã kết nối.
// Biến toàn cục cho server
int server_fd = -1;
pthread_t server_thread_id;
struct sockaddr_un server_address;
struct sockaddr_un client_address;
std::vector<int> connected_clients;
// Hàm luồng chính của server
void* server_thread_function(void* arg) {
LOG_INFO("Server bắt đầu lắng nghe.");
// Lắng nghe trên socket
int ret = listen(server_fd, 1);
if (ret == -1) {
LOG_ERROR("Không thể lắng nghe yêu cầu kết nối từ client: %d - %s", errno, strerror(errno));
close(server_fd);
unlink(SOCKET_PATH);
return nullptr;
}
const char* ready_magic = "QNX_CLIENT";
while (true) {
// Chấp nhận kết nối từ client
socklen_t addr_len = sizeof(client_address);
int client_socket = accept(server_fd, (struct sockaddr*)&client_address, &addr_len);
if (client_socket < 0) {
LOG_ERROR("Không thể chấp nhận kết nối từ client: %d - %s", errno, strerror(errno));
close(server_fd);
unlink(SOCKET_PATH);
return nullptr;
}
char buffer[64] = {0};
int recv_size = 0;
// Đọc dữ liệu từ client để xác thực
recv_size = read(client_socket, buffer, sizeof(buffer));
LOG_INFO("Đã đọc từ socket [%d]: %s", client_socket, buffer);
// Nếu client gửi đúng "magic" string, thêm vào danh sách kết nối
if (strncmp(buffer, ready_magic, strlen(ready_magic)) == 0) {
connected_clients.push_back(client_socket);
LOG_INFO("Client [%d] đã được xác thực và thêm vào danh sách.", client_socket);
} else {
LOG_WARN("Client [%d] gửi dữ liệu không hợp lệ, đóng kết nối.", client_socket);
close(client_socket);
}
}
return nullptr;
}
// Hàm khởi tạo server
void initialize_server() {
int ret;
LOG_INFO("Khởi tạo server socket tại [%s].", SOCKET_PATH);
// Tạo socket
server_fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (server_fd < 0) {
LOG_ERROR("Không thể tạo socket giao tiếp: %d - %s", errno, strerror(errno));
return;
}
// Cấu hình địa chỉ server
server_address.sun_family = AF_UNIX;
strncpy(server_address.sun_path, SOCKET_PATH, sizeof(server_address.sun_path) - 1);
unlink(SOCKET_PATH); // Xóa socket file cũ nếu tồn tại
// Gán socket với địa chỉ
ret = bind(server_fd, (struct sockaddr*)&server_address, sizeof(server_address));
if (ret == -1) {
LOG_ERROR("Không thể gán socket server [%s]: %d - %s", SOCKET_PATH, errno, strerror(errno));
close(server_fd);
unlink(SOCKET_PATH);
return;
}
// Tạo luồng để xử lý kết nối
if (pthread_create(&server_thread_id, nullptr, server_thread_function, nullptr) != 0) {
LOG_ERROR("Tạo luồng server thất bại: %d - %s", errno, strerror(errno));
} else {
LOG_INFO("Server đã khởi chạy thành công.");
}
}
2. Mã Nguồn Client Socket
Client được thiết kế để chạy trên nhiều luồng (ví dụ: 30 luồng). Mỗi luồng sẽ thực hiện một kết nối đến server, gửi tín hiệu sẵn sàng và sau đó chờ nhận dữ liệu.
// Cấu trúc để truyền dữ liệu cho luồng nhận
struct ClientInfo {
int client_socket;
struct sockaddr_un* server_addr;
};
// Gửi tín hiệu sẵn sàng đến server
int send_ready_signal(int fd) {
const char* ready_msg = "QNX_CLIENT";
LOG_INFO("Gửi tín hiệu sẵn sàng đến server.");
return write(fd, ready_msg, strlen(ready_msg));
}
// Luồng nhận dữ liệu từ server
void* client_receiver_thread(void* arg) {
if (arg == nullptr) {
LOG_ERROR("[client_receiver_thread] Tham số không được phép là NULL.");
return nullptr;
}
ClientInfo* client_data = static_cast(arg);
LOG_INFO("[client_receiver_thread] Luồng nhận bắt đầu (tid=%lu)...", pthread_self());
unsigned char buffer[1024] = {0};
ssize_t bytes_received;
// Gửi tín hiệu sẵn sàng
send_ready_signal(client_data->client_socket);
while (true) {
bytes_received = recv(client_data->client_socket, buffer, sizeof(buffer), 0);
if (bytes_received == 0) {
LOG_ERROR("[client_receiver_thread] Server đã đóng kết nối.");
break;
} else if (bytes_received == -1) {
LOG_ERROR("[client_receiver_thread] Lỗi nhận dữ liệu: %s", strerror(errno));
break;
}
LOG_INFO("Nhận dữ liệu (kích thước: %zd): %s", bytes_received, buffer);
memset(buffer, 0, sizeof(buffer));
}
close(client_data->client_socket);
delete client_data; // Giải phóng bộ nhớ
return nullptr;
}
// Luồng chính để kết nối đến server
void* client_connection_thread(void* arg) {
int* client_id = static_cast(arg);
LOG_INFO("[client_connection_thread] Bắt đầu kết nối client [%d]...", *client_id);
int client_socket;
int ret;
int connection_attempts = 0;
struct sockaddr_un server_address;
// Tạo socket Unix
client_socket = socket(AF_UNIX, SOCK_STREAM, 0);
if (client_socket < 0) {
LOG_ERROR("Không thể tạo socket giao tiếp");
return nullptr;
}
server_address.sun_family = AF_UNIX;
strncpy(server_address.sun_path, SOCKET_PATH, sizeof(server_address.sun_path) - 1);
// Thử kết nối đến server
ret = connect(client_socket, (struct sockaddr*)&server_address, sizeof(server_address));
while (ret == -1) {
connection_attempts++;
LOG_WARN("Kết nối đến server thất bại, thử lại lần %d", connection_attempts);
sleep(1);
ret = connect(client_socket, (struct sockaddr*)&server_address, sizeof(server_address));
}
// Chuẩn bị dữ liệu cho luồng nhận
ClientInfo* client_data = new ClientInfo;
client_data->client_socket = client_socket;
client_data->server_addr = &server_address;
// Tạo luồng nhận dữ liệu
pthread_t receiver_thread_id;
if (pthread_create(&receiver_thread_id, nullptr, client_receiver_thread, client_data) != 0) {
LOG_ERROR("Tạo luồng nhận thất bại");
delete client_data;
} else {
LOG_INFO("Luồng nhận đã được tạo.");
}
return nullptr;
}
// Hàm khởi tạo client
void initialize_client(int client_id) {
pthread_t connection_thread_id;
LOG_INFO("* * * * * * * * * * * * * * * * * * [initialize_client] Bắt đầu client [%d]...", client_id);
if (pthread_create(&connection_thread_id, nullptr, client_connection_thread, &client_id) != 0) {
LOG_ERROR("Tạo luồng kết nối thất bại");
}
// pthread_join(connection_thread_id, nullptr); // Không cần chờ nếu dùng thuộc tính PTHREAD_CREATE_DETACHED
}
3. Ví dụ Sử Dụng
Dưới đây là ví dụ về cách khởi chạy server và nhiều client trong cùng một ứng dụng.
#include <iostream>
#include <thread>
#include <vector>
// Định nghĩa hằng số đường dẫn socket
#define SOCKET_PATH "/tmp/qnx_socket_example"
// Macro logging đơn giản
#define LOG_INFO(msg, ...) std::cout << "[INFO] " << msg << std::endl
#define LOG_ERROR(msg, ...) std::cerr << "[ERROR] " << msg << std::endl
#define LOG_WARN(msg, ...) std::cout << "[WARN] " << msg << std::endl
int main(int argc, char* argv[]) {
// Khởi chạy server
initialize_server();
// Khởi chạy 5 client để minh họa
std::vector<std::thread> client_threads;
for (int i = 0; i < 5; ++i) {
client_threads.emplace_back(initialize_client, i);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
// Đợi các luồng client hoàn thành (trong trường hợp này, chúng sẽ chạy vô hạn)
for (auto& t : client_threads) {
t.join();
}
// Trong ứng dụng thực tế, cần có cơ chế dừng server một cách an toàn
// ...
return 0;
}