Dotaz: Jak osetrit komunikaci POSIX serveru s (a mezi) vice klienty?

[tomes.io]

Ahoj, pisu jako skolni projekt neco jako dropbox v C. Synchronizacni server a klient. Hapruje mi komunikace mezi nimi. Kdyz bezi server, tak se po pripojeni prvniho klienta zesychnronizuje s jeho "dropbox" adresare. Dalsi klienti se sychnronizuji uz podle domovskeho adresare serveru.

Klienti sve adresare monitoruji pomoci inotify a pri zachyceni udalosti (vytvoreni, prejmenovani, smazani souboru/adresare) poslou zpravu serveru spolu s cestou k afektovane polozce. A tady nastava problem.

Kdyz je k serveru pripojen jeden klient, vse vicemene funguje (vicemene protoze jeste nemam osetrene vsechny mallocy a napriklad pri zkopirovani velkeho mnozstvi polozek do klienta se abortne server). Klient se synchronizuje s adresarem serveru. Kdyz ale pripojim dalsiho klienta, tak se stane, ze server vypise, ze obdrzel prazdnou zpravu a hodi error:

Reading from socket: Bad file descriptor
Received message: 
ERROR writing to socket: Bad file descriptor

Jeden z klientu se tvari ze vse probehlo v poradku - obdrzel zpravu ze servru ze synchronizace probehla:

The directory New Folder was created.
Synced with server.

Druhy z klientu ovsem tuto zpravu neobdrzi a co je hlavni, jeho inotify prestane reagovat na udalosti v monitorovanem adresari. Nemuze tedy pokracovat v praci. Pritom se do tohoto "poskozeneho" adresare klienta zkopiruje vsechno jak ma. Dokonce se z nej u odstrani soubory, pokud si je jiny klient ve svem adresari smazal. Ale poskozeny klient uz ve svem adresari nic nemonitoruje a neposila tedy zpravy serveru.

Zajimalo by me, co zapricinuje, ze na socket dorazi prazdna, nebo necitelna zprava a jak to souvisi s tim, ze klient prestane monitorovat adresar. Zatim jsem se nedopatral, kde se zasekl a proc. Velmi bych ocenil, kdyby se na kod nekdo podival a poradil, protoze sam s timto asi nepohnu.

Kod pro server je nasledujici:

#define _GNU_SOURCE
#include <stdio.h>        
#include <stdlib.h>         
#include <string.h>        
#include <fcntl.h>          
#include <sys/stat.h>       
#include <errno.h>
#include <strings.h>
#include <sys/types.h> 
#include <sys/wait.h>
#include <unistd.h>
#include <netinet/in.h>      
#include <sys/socket.h>      
#include <arpa/inet.h>      
#include <sched.h>  
#include <pthread.h>          
#include <signal.h>         
#include <semaphore.h>      
#include <fcntl.h>
#include <sys/ioctl.h> 
#include <sys/inotify.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/sendfile.h>
#include <sys/stat.h>
#include <libgen.h>
#include <assert.h>
#include <ctype.h>

#define BUF_SIZE            1024     // buffer size in bytes
#define PORT_NUM            7011    // Port number for a Web server (TCP 5080)
#define PEND_CONNECTIONS     250     // pending connections to hold
#define TRUE                   1
#define FALSE                  0

/* store paths to clients folders */
char storage[100][1024];
pthread_mutex_t lock;
int clients;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;


int send_file(const char *source,const char *destination);

/**
 * @brief concatenates strings to create a path when copying a directory
 * 
 * @param s1
 * @param s2
 * @return 
 */

char* concat(const char *s1, const char *s2) {
    char *result = malloc(strlen(s1) + strlen(s2) + 1); //+1 for the zero-terminator

    if (!result)
        perror("ERROR when concatening");

    strcpy(result, s1);
    strcat(result, "/");
    strcat(result, s2);
    return result;
}
/**
 * @brief Tests if an item of the same name already exists in directory
 * 
 * @param parent
 * @param name
 * @return 
 */

int testdir(char *parent, char *name) {
    DIR *dir;
    struct dirent *entry;
    int ret = 1;

    if (!(dir = opendir(parent))) {
        printf("Creating a new directory. ");
        /* prevent trying to copy files to directories, which are not in
         destination folder yet */
        mkdir(parent, S_IRWXU | S_IRWXG | S_IRWXO);

        if (!(dir = opendir(parent)))
            perror("Cannot open the new dir.");
    }

    if (!(entry = readdir(dir)))
        perror("Cannot read dir. ");

    do {
        if (entry->d_type == DT_DIR) {
            if (strcmp(entry->d_name, ".") == 0 ||
                    strcmp(entry->d_name, "..") == 0)
                continue;

            if (strcmp(entry->d_name, name) == 0)
                ret = 0;

        } else {
            if (strcmp(entry->d_name, name) == 0)
                ret = 0;
        }

    } while ((entry = readdir(dir)));

    closedir(dir);
    return ret;
}

/**
 * @brief Creates a path to server's or client's main directory
 * 
 * @param recpath - path of the destination source
 * @param ssfold - path to the server folder
 * @param len - length of characters to be cut from destination source
 * @return a path where a directory or file is handled.
 */

char* create_path(char* recpath, char* ssfold, int len) {
    char tempath [1024];
    char *buffer;

    strcpy(tempath, recpath + len);
    buffer = malloc(strlen(ssfold) + strlen(tempath) + 1);

    if (!buffer)
        perror("ERROR when creating path");

    strcpy(buffer, ssfold);
    strcat(buffer, tempath);
    return buffer;
}

/**
 * @brief Copies one directory to another
 * 
 * @param source
 * @param destination
 * @param level
 */

void copydir(const char *source, const char *destination, int level) {
    DIR *dir;
    struct dirent *entry;
    char path[1024];
    char dest[1024];
    char *sendpath;
    
    strcpy(dest, destination);

    if (!(dir = opendir(source)))
        return;

    if (!(entry = readdir(dir)))
        return;

    do {
        if (entry->d_type == DT_DIR) {

            int len = snprintf(path, sizeof (path) - 1, "%s/%s",
                    source, entry->d_name);
            path[len] = 0;

            if (strcmp(entry->d_name, ".") == 0 ||
                    strcmp(entry->d_name, "..") == 0)
                continue;

            mkdir(concat(dest, entry->d_name), S_IRWXU | S_IRWXG | S_IRWXO);
            copydir(path, concat(dest, entry->d_name), level + 1);
            
        } else {
            /* check that source file exists and can be opened */
            int len = snprintf(path, sizeof (path) - 1, "%s/%s",
                    source, entry->d_name);
            path[len] = 0;
            sendpath = concat(dest, entry->d_name);
            send_file(path, sendpath);
        }

    } while ((entry = readdir(dir)));

    closedir(dir);
}

/**
 * @brief Sends a file
 * 
 * @param source
 * @param destination
 */

int send_file(const char *source, const char *destination) {
    FILE * filer, * filew;
    int numr;
    const int SIZE_OF_BUFFER=1024;
    char buffer[SIZE_OF_BUFFER];
    char dest[1024];
    
    strcpy(dest,destination);

    if ((filer = fopen(source, "rb")) == NULL) {
        perror("send_file: open read file error.\n");
        printf("debug source: %s\n", source);
        //exit(1);
        return 0;
    }

    if ((filew = fopen(dest, "wb")) == NULL) {
        perror("send_file: open write file error.\n");
        printf("debug destination: %s\n", dest);
        fclose(filer);
        exit(1);
        return 0;
    }

    while (feof(filer) == 0) {

        if ((numr = fread(buffer, 1, SIZE_OF_BUFFER, filer)) != SIZE_OF_BUFFER) {

            if (ferror(filer) != 0) {
                perror("read file error.\n");
                fclose(filer);
                fclose(filew);
                exit(1);
            }
        }

        if (fwrite(buffer, 1, numr, filew) != numr) {
            perror("write file error.\n");
            fclose(filer);
            fclose(filew);
            exit(1);
        }
    }

    if(fclose(filer) != 0)
      printf("read file - close error.\n");//has read-file been deleted ... ???

    if(fclose(filew) != 0)
      printf("write file - close error.\n");//flush C buffer fails

    return 1;
    //fsync ????

}

/**
 * @biref Delete a directory
 * 
 * @param pathname
 */

void deldir(char *pathname) {
    DIR *dir;
    struct dirent *entry;
    char path[1024];

    if (!(dir = opendir(pathname)))
        return;

    if (!(entry = readdir(dir)))
        return;

    do {
        if (entry->d_type == DT_DIR) {

            int len = snprintf(path, sizeof (path) - 1, "%s/%s",
                    pathname, entry->d_name);
            path[len] = 0;

            if (strcmp(entry->d_name, ".") == 0 ||
                    strcmp(entry->d_name, "..") == 0)
                continue;

            deldir(path);

        } else {

            int len = snprintf(path, sizeof (path) - 1, "%s/%s",
                    pathname, entry->d_name);
            path[len] = 0;
            unlink(path);
        }

    } while ((entry = readdir(dir)));

    closedir(dir);
    rmdir(pathname);
}

/**
 * @brief Get a name of an affected item from a path
 * 
 * @param source
 * @return 
 */

char *get_suffix(char *source) {
    char *pch;
    char temp[1024];
    strcpy(temp, source);
    char *suffix = malloc(strlen(temp));
    /* find suffix */
    pch = strtok(temp, "/");
    while (pch != NULL) {
        strcpy(suffix, pch);
        pch = strtok(NULL, "/");
    }

    return suffix;
}

/**
 * @brief Get a parent directory from path
 * 
 * @param source
 * @return 
 */

char *get_parentdir(char *source) {
    char *suffix;
    char *result = malloc(strlen(source) + 3);
    /* backup path*/
    strcpy(result, source);
    /* find suffix */
    suffix = get_suffix(source);
    /* cut path */
    result[strlen(result) - strlen(suffix)] = 0;

    return result;
}

/**
 * @brief Handles received message
 * 
 * @param buffer
 * @param ssfold
 * @param count
 */

void handle_recv_msg(char *buffer, char *ssfold) {
    int i, j, flag, len;
    char *p_str;
    char recpath [1024];
    struct stat isdir;

    /* check if a new client wants to be connected */
    if (buffer[0] == '/') {

        if (clients == 0) {
            strcpy(storage[0], buffer);
            copydir(storage[0], ssfold, 0);
            
        } else {
            
            for (i = 0; i < clients + 1; i++) {
                
                if (strlen(storage[i]) == 0) {
                    strcpy(storage[i], buffer);
                    copydir(ssfold, storage[i], 0); /* synchronize the client with the server. */
                }
            }
        }

        clients++;
        flag = 0;
    } else {
        /* split the message into tokens*/
        j = 0;
        p_str = strtok(buffer, ":");

        while (p_str != NULL) {

            if (j == 0)
                flag = atoi(p_str);
            else if (j == 1)
                len = atoi(p_str);
            else
                strcpy(recpath, p_str);

            p_str = strtok(NULL, ":");
            j++;
        }

    }

    if ((flag == IN_CREATE)) {

        p_str = create_path(recpath, ssfold, len);  //create path to server's folder
        char *name = get_suffix(recpath);
        char *parent;
        
        
        if (stat(recpath, &isdir) == 0 && S_ISDIR(isdir.st_mode)) {
                                    
            if (testdir(get_parentdir(p_str), name)) {
                mkdir(p_str, S_IRWXU | S_IRWXG | S_IRWXO);
                copydir(recpath, p_str, 0);
            }

            for (i = 0; i < clients; i++) {

                p_str = create_path(recpath, storage[i], len);
                parent = get_parentdir(p_str);

                if (testdir(parent, name)) {
                    mkdir(p_str, S_IRWXU | S_IRWXG | S_IRWXO);
                    copydir(recpath, p_str, 0);
                    
                }
            }

        } else {

            if (testdir(get_parentdir(p_str), name))
                send_file(recpath, p_str);

            for (i = 0; i < clients; i++) {

                parent = get_parentdir(create_path(recpath, storage[i], len));

                if (testdir(parent, name))
                    send_file(recpath, create_path(recpath, storage[i], len));
            }
        }
    }

    if ((flag == IN_MOVED_TO)) {
        
        p_str = create_path(recpath, ssfold, len);
        char *name = get_suffix(recpath);
        char *parent;
        
        if (stat(recpath, &isdir) == 0 && S_ISDIR(isdir.st_mode)) {
                                 
            if (testdir(get_parentdir(p_str), name)) {               
                mkdir(p_str, S_IRWXU | S_IRWXG | S_IRWXO);
                copydir(recpath, p_str, 0);
            }
                
            for (i = 0; i < clients; i++) {

                p_str = create_path(recpath, storage[i], len);
                parent = get_parentdir(p_str);

                if (testdir(parent, name)) {
                    mkdir(create_path(recpath, storage[i], len), S_IRWXU | S_IRWXG | S_IRWXO);
                    copydir(recpath, p_str, 0);
                }                    
            }

        } else {

            if (testdir(get_parentdir(p_str), name))
                send_file(recpath, p_str);

            for (i = 0; i < clients; i++) {

                parent = get_parentdir(create_path(recpath, storage[i], len));
                
                if (testdir(parent, name))
                    send_file(recpath, create_path(recpath, storage[i], len));
            }
        }
    }

    if ((flag == IN_MOVED_FROM) || (flag == IN_DELETE)) {

        p_str = create_path(recpath, ssfold, len);
        char *name = get_suffix(recpath);
        char *parent;

        if (stat(create_path(recpath, ssfold, len), &isdir) == 0 && 
                S_ISDIR(isdir.st_mode)) {
            
            parent = get_parentdir(p_str);
            
            if (testdir(parent, name) == 0) {
                 
                deldir(p_str);
            }
            
            for (i = 0; i < clients; i++) {

                p_str = create_path(recpath, storage[i], len);
                parent = get_parentdir(p_str);

                if (testdir(parent, name) == 0) {
                    
                    deldir(p_str);
                }                    
            }

        } else {
            
            if (testdir(get_parentdir(p_str), name) == 0)
                unlink(p_str);
                                          
            for (i = 0; i < clients; i++) {
                
                p_str = create_path(recpath, storage[i], len);
                
                if (testdir(get_parentdir(p_str), name) == 0)              
                    unlink(p_str);
            }
        }
    }
}

void *my_thread(void * arg) {
    int my_clisock; //copy socket
    char buffer[BUF_SIZE];
    int n;
    //path to the server's main directory
    char ssfold[] = "/home/tomesh/Dropbox/C/sync"; 

    my_clisock = *(int *) arg; // copy the socket
    bzero(buffer, 1024);

    n = read(my_clisock, buffer, 1023);

    if (n < 0)
        perror("Reading from socket");

    printf("Received message: %s\n", buffer);
    
    if (strlen(buffer) > 6 ) {
        
        pthread_mutex_lock (&lock);
        handle_recv_msg(buffer, ssfold);  
        pthread_mutex_unlock (&lock);
        n = write(my_clisock, "Synced with server.", 19);
    }

    if (n < 0)
        perror("ERROR writing to socket");

    close(my_clisock); // close the client connection
    pthread_exit(NULL);
}

int main(void) {
    int server_s;
    int clisock;
    struct sockaddr_in server_addr;
    struct sockaddr_in client_addr;
    socklen_t addr_len;
    //unsigned int args;
    pthread_attr_t attr;
    pthread_t threads;
    int i;

    server_s = socket(AF_INET, SOCK_STREAM, 0);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(PORT_NUM);
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    bind(server_s, (struct sockaddr *) &server_addr, sizeof (server_addr));
    listen(server_s, PEND_CONNECTIONS);
    pthread_attr_init(&attr);
    pthread_attr_setschedpolicy(&attr, SCHED_FIFO); // FIFO scheduling for threads 
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); // Don't want threads (particualrly main)
                                                               // waiting on each other
    /* init storage */
    for (i = 0; i < 100; i++) {
        storage[i][0] = 0;
    }

    printf("server is ready ...\n");
    clients = 0;
    //sem_init(&sem, 0, 0);

    while (TRUE) {
        addr_len = sizeof (client_addr);
        clisock = accept(server_s, (struct sockaddr *) &client_addr, &addr_len);

        if (clisock < 0) {
            printf("ERROR - Unable to create socket \n");
            exit(FALSE);
            
        } else {
            
            pthread_create(&threads, &attr, my_thread, &clisock);
            sleep(0); // Giving threads some CPU time
        }
    }
    //close(server_s); // close the primary socket
    return 0;
}

a pro klienta:

#define _GNU_SOURCE
#include <string.h> 
#include <time.h>
#include <unistd.h>
#include <stdlib.h>    
#include <sys/types.h> 
#include <sys/wait.h>
#include <dirent.h>
#include <unistd.h>
#include <sys/inotify.h>
#include <stdio.h>  
#include <sys/stat.h>
#include <fcntl.h>   
#include <sys/ioctl.h>>
#include <string.h>
#include <errno.h>
#include <sys/sendfile.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <ctype.h>


int inotify;
char files[100][1024];
struct inotify_event *event;
struct sockaddr_in serv_addr;
int sockfd;


/**
 * @brief Test if only a correct message is about to be sent
 * 
 * @param s
 * @return 
 */

int is_empty(const char *s) {

    char buff[1024];
    strcpy(buff,s);;
    
    buff[strlen(buff)] = '\0';

    if (buff[0] == '/' || isdigit(buff[0])){      
        return 1;
    }
   
    return 0;
}

/**
 * @brief Sends a message to the server.
 * 
 * @param message
 * @param portno
 */

void send_msg(char *message, int portno) {
    
    int n;
    char recvbuff[255];
    
    serv_addr.sin_port = htons(portno);
    sockfd = socket(AF_INET, SOCK_STREAM, 0);

    if (sockfd < 0)
        perror("send_msg: ERROR opening socket");

    if (connect(sockfd, (struct sockaddr *) &serv_addr,
            sizeof (serv_addr)) < 0)
        perror("send msg: ERROR connecting");

    n = write(sockfd, message, strlen(message) +1);

    if (n < 0)
        perror("send msg: ERROR writing to socket");

    n = read(sockfd, recvbuff, 255);

    if (n < 0)
        perror(" send_msg: ERROR reading from socket");

    printf("%s\n", recvbuff);
    free(message);
    close(sockfd);
}

/**
 * @brief Recursively watches a directory.
 * 
 * @param file
 */

void watch_dir(char *file) {
    int ret;
    struct stat buf;
    struct dirent *data;

    if (lstat(file, &buf))
        return;

    if (!S_ISDIR(buf.st_mode))
        return;

    //Now recursive part
    DIR *dir = opendir(file);

    if (dir == NULL)
        return;

    while ((data = readdir(dir)) != NULL) {
        if (strcmp(data->d_name, ".") == 0 ||
                strcmp(data->d_name, "..") == 0)
            continue;

        if (strstr(data->d_name, ".") == data->d_name)
            continue;

        char *path2 = (char *) malloc(1024 + sizeof (path2));
        asprintf(&path2, "%s/%s", file, data->d_name);

        if (lstat(path2, &buf))
            continue;
        //If it's a dir then launch watch_dir himself on it
        if (S_ISDIR(buf.st_mode))
            watch_dir(path2);

        free(path2);
    }

    closedir(dir);
    // now set watch for the argument of watch_dir
    ret = inotify_add_watch(inotify, file, IN_ALL_EVENTS);

    if (ret == -1) {
        printf("%s\n", file);
        perror("Error in watcher setting\n");
        exit(0);
    }

    strcpy(files[ret], file);
    return;
}

/**
 * @brief Handles notified events.
 * 
 * @param path
 * @param dest
 */

void handle_event(char *path, char *dest, int portno) {
    
    char *message = (char *) malloc((strlen(path) * sizeof (message)) + 10);   

    if (event->mask & IN_CREATE) {
        if (event->mask & IN_ISDIR) {
            
            printf("The directory %s was created.\n", event->name);
            watch_dir(dest);
            
        } else {
            
            printf("The file %s was created.\n", event->name);
        }

        asprintf(&message, "%d:%d:%s", IN_CREATE,
                (int) strlen(dest), path);
        
        int j = is_empty(message);

        if (j == 1)
            send_msg(message, portno);
        
        //printf("")
    }

    if (event->mask & IN_DELETE) {
        if (event->mask & IN_ISDIR) {
            
            printf("The directory %s was deleted.\n", event->name);
            
        } else {
            
            printf("The file %s was deleted.\n", event->name);
        }

        asprintf(&message, "%d:%d:%s", IN_DELETE,
                (int) strlen(dest), path);

        int j = is_empty(message);

        if (j == 1)
            send_msg(message, portno);
    }
    
    if (event->mask & IN_MODIFY) {
        if (event->mask & IN_ISDIR) {
            
            printf("The directory %s was modified.\n", event->name);
            
        } else {

            printf("The file %s was modified.\n", event->name);
        }

        asprintf(&message, "%d:%d:%s", IN_MODIFY,
                (int) strlen(dest), path);

        int j = is_empty(message);

        if (j == 1)
            send_msg(message, portno);
    }

    if (event->mask & IN_MOVED_FROM) {
        if (event->mask & IN_ISDIR) {

            printf("Directory %s with cookie %d has been "
                    "moved from the sync directory.\n",
                    event->name, event->cookie);
            
        } else {
            
            printf("Directory %s with cookie %d has been "
                    "moved from the sync directory.\n",
                    event->name, event->cookie);
        }

        asprintf(&message, "%d:%d:%s", IN_MOVED_FROM,
                (int) strlen(dest), path);
        
        int j = is_empty(message);

        if (j == 1)
            send_msg(message, portno);
    }
    if (event->mask & IN_MOVED_TO) {
        if (event->mask & IN_ISDIR) {
            
            printf("Directory %s with cookie %d has been "
                    "moved to the sync directory.\n",
                    event->name, event->cookie);
            
        } else {
            
            printf("Directory %s with cookie %d has been "
                    "moved to the sync directory.\n",
                    event->name, event->cookie);
        }

        asprintf(&message, "%d:%d:%s", IN_MOVED_TO,
                (int) strlen(dest), path);
        
        int j = is_empty(message);

        if (j == 1)
            send_msg(message, portno);
    }
}

int main(int agrc, char *argv[]) {

    int portno, n;
    struct hostent *server;
    char buffer[4096];
    char *dest = malloc(strlen(argv[1]) + 1);

    strcpy(dest, argv[1]);
    inotify = inotify_init();

    if (!inotify) {
        perror("Open inotify device");
        fprintf(stderr, "Are you sure your kernel supports inotify?\n");
    }
    //launch recursive watching
    watch_dir(dest);
    printf("Start monitoring!\n");
    portno = atoi(argv[2]);
    server = gethostbyname("localhost");

    if (server == NULL) {
        fprintf(stderr, "ERROR, no such host\n");
        exit(0);
    }

    bzero((char *) &serv_addr, sizeof (serv_addr));
    serv_addr.sin_family = AF_INET;
    
    bcopy((char *) server->h_addr,
            (char *) &serv_addr.sin_addr.s_addr,
            server->h_length);

    /*First connection to the server. 
     * Client sends to a server a path 
     * to client's "dropbox" folder*/
    serv_addr.sin_port = htons(portno);
    sockfd = socket(AF_INET, SOCK_STREAM, 0);

    if (connect(sockfd, (struct sockaddr *) &serv_addr,
            sizeof (serv_addr)) < 0)
        perror("ERROR connecting");

    n = write(sockfd, dest, strlen(dest) + 1);

    if (n < 0)
        perror("ERROR writing to socket");

    close(sockfd);

    while (1) {
        int ret;
        int i = 0;
        ret = read(inotify, buffer, 4096);

        if (ret < 0)
            perror("read");

        if (ret == 0)
            sleep(1);

        while (i < ret) {
            char *name = NULL;
            event = (struct inotify_event *) &buffer[i];

            if (event->len)
                name = &buffer[i] + sizeof (struct inotify_event);
           
            char *path = (char *) malloc(1024 * sizeof (path));

            asprintf(&path, "%s/%s", files[event->wd], name);                     
            handle_event(path, dest, portno);           
            i += sizeof (struct inotify_event) +event->len;
        }
    }

    close(sockfd);
    return 0;
}

---

---

Odpovědi

[tomes.io]

Po dalsim patrani jsem si zkusil pridat vypis hodnoty socketu v mainu a thread fci:

pro main:

} else {
            
            printf("clisock: %d\n", clisock);
            pthread_create(&threads, &attr, my_thread, &clisock);
            sleep(0); // Giving threads some CPU time
        }

a v thread fci:

my_clisock = *(int *) arg; // copy the socket
    bzero(buffer, 1024);

    n = read(my_clisock, buffer, 1023);
    printf("thread my_clisock: %d\n", my_clisock);

    if (n < 0)
        perror("Reading from socket");

Vysledek je nasledujici:

server is ready ...
clisock: 4
thread my_clisock: 4
Received message: /home/tomesh/Dropbox/C/dropbox
clisock: 4
thread my_clisock: 4
Received message: /home/tomesh/Dropbox/C/dropbox2
clisock: 4
thread my_clisock: 4
Received message: 256:30:/home/tomesh/Dropbox/C/dropbox/Cviceni 01
clisock: 4
clisock: 5
thread my_clisock: 5
Received message: 256:30:/home/tomesh/Dropbox/C/dropbox/Cviceni 01/Final
thread my_clisock: 5
Reading from socket: Bad file descriptor
Received message: 
ERROR writing to socket: Bad file descriptor
clisock: 5
thread my_clisock: 5
Received message: 256:30:/home/tomesh/Dropbox/C/dropbox/Cviceni 02
clisock: 5
thread my_clisock: 5
Received message: 256:30:/home/tomesh/Dropbox/C/dropbox/Cviceni 02/02
clisock: 6
thread my_clisock: 6
Received message: 256:30:/home/tomesh/Dropbox/C/dropbox/Cviceni 02/final

Zda se, ze bude problem s vlakny. Vypada to, ze se acceptuje nejaka hodnota socketu (clisock: 4), spusti se vlakno, ale system ho zastavi, mezitim se provede dalsi loop, acceptne se nova hodnota socketu (clisock: 5) a spusti se nove vlakno, ktere vypise odpovidajici hodnotu noveho socketu (thread my_clisock: 5).

Pak se znovu rozjede predtim pozastavene vlakno, ktere ale ma (z jakeho duvodu?) jako argument hodnotu uz noveho socketu a tu si zapise do sve promenne (thread my_clisock: 5).

Jenze uz nema, co by cetl, tak se vyhodi error a buffer je prazdny. Otazka, zni, jak je mozne, aby se ve vlakne takto zmenila hodnota promenne (pokud mam pravdu) a jak tento problem vyresit?

[Sten]

Problém je jednoduchý. Do pthread_create se předává ukazatel. Ten se předá threadu. Jenže ten ukazatel mezitím změnil hodnotu, protože se acceptnul další požadavek.

[Michal Kubeček]

Jenže ten ukazatel mezitím změnil hodnotu,

Přesněji: ne ukazatel, ale proměnná, na kterou ukazuje.

[tomes.io]

Chapu, ale nenapada me zpusob, jak to osetrit.

[Šangala]

Moc jsem to nestudoval, ale minimálně to lze tak, že se vytvoří dynamicky v main()-u

int *clisock = new int();

nebo

int *clisock = new int(accept(server_s, (struct sockaddr *) &client_addr, &addr_len));

a uvolní se pomocí delete v my_thread().

[tomes.io]

To by bylo hezky, jenze tohle je C, ne C++ :) takze int() ani delete mi nepomuzou...

[Šangala]

To jsi mě dostal :), princip je stejný, jen funkce jiné, hádej jaké .

[tomes.io]

Jasne, pochopil jsem :) Jenze tohle mi hodi segfault hned, jak se pripoji prvni nebo druhy klient.

int *arg = malloc(sizeof(int));

while (TRUE) {
        addr_len = sizeof (client_addr);       
        clisock = accept(server_s, (struct sockaddr *) &client_addr, &addr_len);
        *arg = clisock;

        if (clisock < 0) {
            printf("ERROR - Unable to create socket \n");
            exit(FALSE);
            
        } else {
 
            pthread_create(&threads, &attr, my_thread, arg);
            free(arg);

        }
    }

Asi jsem se ztratil v tom pretypovani nebo ja nevim

[tomes.io]

Aha, mallocoval jsem mimo while loop, takto uz to vypada, ze to funguje jak ma:

while (TRUE) {
        addr_len = sizeof (client_addr);      
        clisock = accept(server_s, (struct sockaddr *) &client_addr, &addr_len);
        int *arg = malloc(sizeof(int));
        *arg = clisock;

        if (clisock < 0) {
            printf("ERROR - Unable to create socket \n");
            exit(FALSE);
            
        } else {

            pthread_create(&threads, &attr, my_thread, arg);
        }

    }

Jeste si dovolim chvili testovat, nez tohle tema uzavru (navic mam jeden dotaz jeste nize).

Kazdopadne dekuju :)

[Šangala]

Otázkou je jestli tam potřebuješ clisock (zbytečně se to znepřehledňuje)

int *arg = (int *) malloc(sizeof(int));
*arg = accept(server_s, (struct sockaddr *) &client_addr, &addr_len);

A nezapomenout na free() v tom thread-u ;)

[Šangala]

Vytvoření thread-u by bylo dobré testovat a při zklamání provést hned free(arg);. Jinak otázku níže, nevím kterou (, která by nebyla vyřešena).

[Šangala]

Přečti si.

Není to k problému, ale ke sleep(0) komentář nemusí být pravdivý, minimálně nic takového není zaručeno a nikdy nebylo…

[Michal Kubeček]

Přesně tak. Když chce někdo provést yield, má zavolat funkci, která je k tomu určena, ne něco, o čem je empiricky odpozorováno, že to za určitých okolností může udělat.

Tady je navíc špatně i ten. Stejně hned následuje blokující accept() a pokud náhodou blokovat nebude, tak proto, že čeká další spojení, které je žádoucí obsloužit co nejdřív, abychom minimalizovali jednak latence, jednak riziko, že se zaplní backlog. Koneckonců, proč se vzdávat procesoru, když není nikde psáno, že jich není dost pro všechny thready…

[Michal Kubeček]

Nevím, jestli je to přesně ten problém, na který jste narazil, ale způsob předávání parametru hlavní funkci threadu je špatně. Pokud v hlavním threadu proběhne druhý accept() dřív, než si první "child" stihne zkopírovat číslo deskriptoru, tak už ho mezitím stihnete přepsat, takže oba "child" thready budou používat stejnou hodnotu. Ta chyba by pak mohla být způsobena tím, že jeden z threadů zkusí z toho deskriptoru číst poté, co ho druhý zavřel.

Pokud opravdu potřebujete předávat jen int, tak je nejjednodušší ošidit tím, že se spolehnete na to, že na všech příčetných platformách je sizeof(int) <= sizeof(void*) a prostě číslo deskriptoru přetypujete na void* a zpátky. Jinak je potřeba naalokovat pro každý vytvářený thread samostatný prostor, aby nemohlo dojít ke kolizi.

Ještě dva praktické tipy takhle na první zběžné prolétnutí:

1. V multithreadových aplikacích nealokujte velké datové struktury na zásobníku. Resource limit pro velikost zásobníku (ulimit -s, např. v OpenSuSE defaultně 8 MB) se počítá pro všechny thready dohromady, takže je žádoucí mít zásobníky threadů co nejmenší.

2. V okamžiku, kdy v každém threadu voláte handle_recv_msg() pod mutexem, tak se v podstatě dobrovolně zbavujete výhody paralelizace, protože v té funkci bude vždy jen jeden thread a ostatní na něj budou muset čekat. Obecně je lepší mít kritické sekce co nejmenší.

spolu s cestou k afektovane polozce

Ale fuj…

[tomes.io]

Ano, zpusob predavani je spatne. Myslim, ze chapu. Jenze me to nijak nenakoplo k tomu to osetrit. Zavedeni pomocne promenne, ze ktere by se teprve predalo cislo deskriptoru fci

pthread_create

nic neresi:

 clisock = accept(server_s, (struct sockaddr *) &client_addr, &addr_len);
        arg = clisock;

Takze jak to osetrit, aby se do vytvoreni vlakna uchovala hodnota deskriptoru?

Ani to pretypovani nefunguje ciste. Pri:

pthread_create(&threads, &attr, my_thread, (void *)arg);

Prekladac mi vynada:

warning: cast to pointer from integer of different size [-Wint-to-pointer-cast]

[Michal Kubeček]

Však jsem také psal, že to není moc čisté. Mirek Dušín by se nepochybně červenal, ale pokud jste si jistý, že se tam ten int vejde, tak to bude fungovat. Ten warning by šel možná odstranit mezikrokem přes nějaký jiný celočíselný typ, který má stejnou velikost jako pointer, ale to samozřejmě nemění nic na tom, že to prostě čisté řešení není.

[luky]

Void pointer musi byt nejsirsi mozny na dane architekture a minimalni limit pro velikost objektu je 65535, takze void pointer nikdy nemuze byt mensi nez 2 byty. Staci si proto nastavit limit pro pocet otevrenych souboru na 65536 a musi to vzdy fungovat.

[Michal Kubeček]

Takze jak to osetrit, aby se do vytvoreni vlakna uchovala hodnota deskriptoru?

Pokud chcete psát multithreadové aplikace, je potřeba naučit se, že tohle bez spolupráce s tím vytvářeným threadem prostě nelze. Jakékoli předpoklady o tom, jak "běží čas" v různých threadech, jsou oprávněné pouze pokud použijete odpovídající synchronizační primitiva (mutexy, semafory, podmínkové proměnné, bariéry apod.). Jak už bylo řečeno výše, nejjednodušší je prostě prostor pro data, která chcete threadu předat, alokovat dynamicky (pro každý thread zvlášť) a nechat ho, ať ho sám uvolní, až je nebude potřebovat. Při opačném problému (tj. jak předat něco z threadu zpátky "rodiči") lze převzetí a úklid navázat třeba na pthread_join().

Pokud vám malloc()/free() připadají neefektivní, můžete si to alokovat sám v předalokovaném bufferu, ale opět je potřeba důsledně ošetřit souběžné přístupy ke společným informacím z různých threadů.

[Josef Kufner]

Další možností je vykašlat se na vlákna a nastudovat select. Výrazně si tím ulehčíš život. Vše bude probíhat v jednom vlákně, jen musíš ke všem úkolům přistupovat asynchroně, aby se select dostával ke slovu dostatečně často. Tímto přístupem také můžeš obsloužit nejvíce klientů najednou.

Pokud máš klientů málo současně, šel bych do forkování nových procesů. I když tam je nebezpečí, že si docela naběhneš se sdílením dat mezi procesy. Na druhou stranu, pokud ti stačí zamykat soubory, bude to nejjednodušší. Případné inotify eventy můžeš přeposílat po interním ovládacím socketu z hlavního procesu.

Pokud opravdu trváš na vláknech, osvědčilo se mi vytvořit si jednu strukturu, která obsahuje všechna data specifická pro konkrétního klienta (jeho socket, pthread vlákno a cokoliv dalšího). Když přijde klient, vyrobíš si pomocí malloc novou instanci téhle struktury, předáš ji čerstvě vytvořenému vláknu a zapomeneš, že existuje (případně si necháš pointer v nějakém globálním seznamu klientů). V podstatě je to takový objektový přístup, stejně jak to dělá stdio s FILE*. Všechno kouzlo je v jasném rozdělení, který kus paměti patří kterému vláknu, a jedna struktura se hlídá docela dobře.

[Michal Kubeček]

Tímto přístupem také můžeš obsloužit nejvíce klientů najednou.

Jen ve specifických aplikacích. Pokud bude mít víc procesorů a obsluha klienta bude znamenat netriviální zátěž procesoru, bude to naopak to nejméně efektivní řešení. Jinak bych spíš doporučil poll(); vždy, když vidím rozhraní pro select(), mám neodbytný pocit, že ten, kdo ho navrhoval, se chtěl vývojářům za něco pomstít.

[tomes.io]

Jo, ta struktura pro klienta je v planu :)

[tomes.io]

Diky vsem za pomoc, tipy a rady :)