Applied IPC with Semaphore

ในตัวอย่างข้างล่างนี้จะเพิ่มเทคนิคของ semaphore เข้าไปในตัวอย่าง Multi-Remote Commanders โดยก่อนที่ thread จะเริ่มประมวลผลคำขอของ client ที่เชื่อมต่อผ่าน socket เข้ามา มันจะพยายามลดค่าของ semaphore โดยใช้ sem_wait() หากค่าของ semaphore เป็นศูนย์ (สล็อตทั้งหมดสำหรับการประมวลผลพร้อมกันถูกจองแล้ว) thread นั้นก็จะถูกบล็อก และรอจนกว่า thread อื่นจะประมวลผลเสร็จสิ้นและเรียก sem_post() (ซึ่งจะเพิ่มค่าของ semaphore)

วิธีนี้ semaphore จะจำกัดจำนวน thread ที่ทำงานพร้อมกันบนส่วนของโค้ดที่ใช้ทรัพยากรมาก ทำให้เซิร์ฟเวอร์ไม่ต้องเผชิญกับคำขอมากเกินไปในเวลาเดียวกัน วิธีการนี้เป็นหนึ่งในรูปแบบของการจำกัดอัตรา

ในบริบทของ client-server semaphores โดยทั่วไปจะถูกใช้ในฝั่งเซิร์ฟเวอร์ เนื่องจากเกิดการแข่งขันทรัพยากรเพเพราะมีการเข้าถึงพร้อมกันของ clients หลายตัว ดังนั้นรูปแบบการควบคุมนี้เป็นสิ่งที่พื้นฐานในสถานการณ์ที่คำขอของลูกค้าอาจต้องการทรัพยากรมาก และเซิร์ฟเวอร์มีความเสี่ยงที่จะขาดทรัพยากรหากมีการประมวลผลคำขอมากเกินไปในเวลาเดียวกัน 

// new_server_with_semaphore.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <pthread.h>

#include <semaphore.h>

#define PORT 8080
#define BUFFER_SIZE 1024

#define MAX_CONCURRENT_CLIENTS 3  // Max number of clients to handle concurrently

sem_t sem;

// Struct for arguments to pass to the thread handler
typedef struct {
    int socket;
} client_thread_args;

// Function to execute the command and retrieve the output
void execute_command(char* command, char* result, size_t result_size) {
    // Clear the result buffer
    memset(result, 0, result_size);

    FILE* pipe = popen(command, "r");
    if (!pipe) {
        snprintf(result, result_size, "Error: failed to open pipe.\n");
        return;
    }

    // Read command result into the buffer
    char buffer[128];
    while (!feof(pipe)) {
        if (fgets(buffer, sizeof(buffer), pipe) != NULL) {
            strncat(result, buffer, result_size - strlen(result) - 1);
        }
    }

    pclose(pipe);
}

// Thread function to handle communication with a client
void *client_handler(void *arg) {
    client_thread_args *args = (client_thread_args *)arg;

    int client_socket = args->socket;
    free(arg); // Free memory for arguments

    char buffer[BUFFER_SIZE] = {0};
    char result[BUFFER_SIZE] = {0};

    // Wait for the semaphore to become available
    sem_wait(&sem);

    // Communication loop with the connected client
    while (1) {
        memset(buffer, 0, BUFFER_SIZE); // Clear the buffer
        int read_size = read(client_socket, buffer, BUFFER_SIZE);
        if (read_size > 0) {
            printf("Received command: %s\n", buffer);

            // Special case: close connection if client sends "exit"
            if (strcmp(buffer, "exit") == 0) {
                printf("Closing connection...\n");
                close(client_socket);
                break; // Break out of the loop to end this thread
            }

            // Execute the command received from the client
            execute_command(buffer, result, BUFFER_SIZE);

            // Send back the result
            send(client_socket, result, strlen(result), 0);
            memset(result, 0, BUFFER_SIZE); // Clear the result buffer
            
            // After done processing the client's request
            sem_post(&sem); // Signal that this thread is done with the resource

        } else {
            printf("Client disconnected or read error...\n");
            close(client_socket);
            break; // Break out of the loop to end this thread
        }
    }

    return NULL; // Thread ends here
}

int main(int argc, char const *argv[]) {
    int server_fd, new_socket;
    struct sockaddr_in address;
    int opt = 1;
    int addrlen = sizeof(address);

    // Creating socket file descriptor
    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }

    // Forcefully attaching socket to the PORT 8080
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
        perror("setsockopt");
        exit(EXIT_FAILURE);
    }

    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(PORT);

    // Binding socket to the PORT address defined above
    if (bind(server_fd, (struct sockaddr *)&address, sizeof(address)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }

    // Display the server's IP and listening port
    char host[256];
    char *IP;
    struct hostent *host_entry;
    gethostname(host, sizeof(host));
    host_entry = gethostbyname(host);
    IP = inet_ntoa(*((struct in_addr*) host_entry->h_addr_list[0]));

    if (listen(server_fd, 3) < 0) {
        perror("listen");
        exit(EXIT_FAILURE);
    }

    // Initialize the semaphore to limit the number of concurrent client processing
    sem_init(&sem, 0, MAX_CONCURRENT_CLIENTS);

    printf("Server is running at IP: %s on Port: %d\n", IP, PORT);

    // Main loop to keep server running and accepting new connections
    while (1) {
        if ((new_socket = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen)) < 0) {
            perror("accept");
            exit(EXIT_FAILURE);
        }

        printf("Connection established...\n");

        // Create a thread for each client
        pthread_t client_thread;
        client_thread_args *args = (client_thread_args *)malloc(sizeof(*args));
        args->socket = new_socket;
        if (pthread_create(&client_thread, NULL, client_handler, args) != 0) {
            perror("Could not create thread");
            free(args); // Don't forget to free memory on failure
            close(new_socket);
        } else {
            pthread_detach(client_thread); // Automatically reclaim thread resources on exit
        }
    }

    // Destroy semaphore in any cleanup or exit code, if you ever break out of the infinite loop
    sem_destroy(&sem);

    return 0; // The server should never reach this part
}

ทำการคอมไพล์โปรแกรมทั้งสอง แล้วสร้างหน้าต่าง terminal จำนวน 4 หน้าต่าง ดังนี้

$ gcc -o new_server_with_semaphore new_server_with_semaphore.c -lpthread -Wall
$ ./new_server_with_semaphore
Server is running at IP: 10.211.55.3 on Port: 8080
Connection established...
Connection established...
Received command: ls
Result sent back to client. 
Received command: date
Result sent back to client. 
Received command: ifconfig
Result sent back to client. 
Received command: uname -a
Result sent back to client.

PreviousSemaphore ProgrammingNextPSoC™ IoT Development Kit

Last updated

Assoc. Prof. Wiroon Sriborrirux, Founder of Advance Innovation Center (AIC) and Bangsaen Design House (BDH), Electrical Engineering Department, Faculty of Engineering, Burapha University