path: root/snowcast_server_concurrent.c

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

#include "protocol.h"

#define LINE_MAX 1024
#define MAX_USERS 1000
#define MAX_PATH 50

typedef struct station {
    int seekIndex;
    char* filePath;
} station_t;

typedef struct user {
    int udpPort;
    int stationNum;
    int sockfd;
    pthread_t streamThread;
} user_t;


/* For safe condition variable usage, must use a boolean predicate and  */
/* a mutex with the condition.                                          */
int                 count = 0;
pthread_cond_t      cond  = PTHREAD_COND_INITIALIZER;
pthread_mutex_t     mutex = PTHREAD_MUTEX_INITIALIZER;

pthread_mutex_t     station_mutex = PTHREAD_MUTEX_INITIALIZER;

const char *port;
int num_stations;

int start_threads = 0;
int max_active_users = 0;

pthread_mutex_t     mutex_user_data = PTHREAD_MUTEX_INITIALIZER;
// array from index to user_data
user_t  *user_data;
int sockfd_to_user[MAX_USERS];

// stations array pointer
station_t *station_data;

void *send_udp_packet_routine(void* arg);
void *select_thread(void* arg);
void *synchronization_thread(void* arg);

void *get_in_addr(struct sockaddr *sa);

void *init_user(int sockfd);
void *update_user_udpPort(int sockfd, int udpPort);
void *update_user_station(int sockfd, int stationNum);
void *print_user_data(int sockfd);
void destroy_user(int sockfd);

void send_announce_message(int fd, int station_num);


// void *load_file(void* arg);

int main(int argc, char *argv[])
{
    // threads to control reading files at chunks while the other threads sleep
    // station_data = malloc(sizeof(station_t) * NUM_STATIONS);
    // check and assign arguments
    if (argc < 3) {
        fprintf(stderr,"usage: ./snowcast_server <listen port> <file0> [file 1] [file 2] ... \n");
        exit(1);
    }

    port = argv[1];
    num_stations = argc - 2;

    printf("port: %s\n", port);
    printf("num_stations: %d\n", num_stations);

    // init stations
    size_t totalSize = 0;
    // get size to malloc
    for (int i = 2; i < argc; i++)
    {
        printf("file: %s\n", argv[i]);
        totalSize += sizeof(int) + strlen(argv[i]);
    }
    station_data = malloc(totalSize);
    // assign the stations
    for (int i = 2; i < argc; i++)
    {
        station_data[i - 2] = (station_t) { 0, argv[i]};
    }

    // print all indexes in station data
    for (int i = 0; i < num_stations; i++)
    {
        printf("station %d: %s\n", i, station_data[i].filePath);
    }

    // make array of user data
    user_data = malloc(sizeof(user_t) * max_active_users);
    if (!user_data) {  perror("malloc"); return 1; }

    // make and start "select" thread that manages:
    // 1) new connections, 2) requests from current connections, 3)cloing connections
    pthread_t s_thread;
    pthread_create(&s_thread, NULL, select_thread, NULL);

    // start syncchronization thread to broadcast stations
    pthread_t sync_thread;
pthread_create(&sync_thread, NULL, synchronization_thread, NULL);

    // command line interface
    char input[LINE_MAX];
    while (1) {
	    char *line = fgets(input, LINE_MAX, stdin);

        if (line == NULL) {
            continue;
        } else if (strncmp("q\n", input, LINE_MAX) == 0) {
            // end code if type in q
            printf("Exiting.\n");
            break;
        } else if (strncmp("p\n", input, LINE_MAX) == 0) {
            // print all user data
            for (int i = 0; i < max_active_users; i++) {
                print_user_data(i);
            }
        } else if (strncmp("s\n", input, LINE_MAX) == 0) {
        }
    }

    return 0;
}

int sendall(int udp_sockfd, char *buf, int *len, struct addrinfo *thread_res)
{
    int MAX_PACKET_SIZE = 512;
    int total = 0;        // how many bytes we've sent
    int bytesleft = *len; // how many we have left to send
    int n;

    while(total < *len) {
        n = sendto(udp_sockfd, buf+total, MAX_PACKET_SIZE, 0, thread_res->ai_addr, thread_res->ai_addrlen);
        //         thread_res->ai_addr, thread_res->ai_addrlen)) == -1;
        if (n == -1) { break; }
        total += n;
        bytesleft -= n;
    }

    *len = total; // return number actually sent here

    return n==-1?-1:0; // return -1 on failure, 0 on success
} 


/* Make the manager routine */
void *send_udp_packet_routine(void *arg) {
    // unpack args
    int user_index = (int) arg;
    printf("thread : user_index: %d\n", user_index);
    // print user data
    print_user_data(user_index);

    // declare vairables to be used
    int did_work = 1;
    pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
    int s;
    int udp_sockfd;
    struct addrinfo thread_hints, *thread_res, *thread_servinfo;
    int error_code;

    // TODO: add error checking on these calls***

    // setup hints
    memset(&thread_hints, 0, sizeof thread_hints);
    thread_hints.ai_family = AF_INET;  // use IPv4 only
    thread_hints.ai_socktype = SOCK_DGRAM;
    thread_hints.ai_flags = AI_PASSIVE; // fill in my IP for me

    // setup the socket for client listener DATAGRAM (udp)
    // cover the port integer to a string
    int int_port = user_data[user_index].udpPort;
    int length = snprintf( NULL, 0, "%d", int_port );
    char* port = malloc( length + 1 );
    snprintf( port, length + 1, "%d", int_port );
    sprintf(port, "%d", int_port);

    if (error_code = getaddrinfo(NULL, port, &thread_hints, &thread_servinfo) != 0)
    {
        fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(error_code));
        return (NULL);
    }
    free(port);

    // loop through all the results and make a socket
    for(thread_res = thread_servinfo; thread_res != NULL; thread_res = thread_res->ai_next) {
        if ((udp_sockfd = socket(thread_res->ai_family, thread_res->ai_socktype,
                thread_res->ai_protocol)) == -1) {
            perror("talker: socket");
            continue;
        }
        break;
    }
    if (udp_sockfd == NULL) {
        fprintf(stderr, "talker: failed to create socket\n");
        return (NULL);
    }

    // bind(udp_sockfd, thread_res->ai_addr, thread_res->ai_addrlen);


    // freeaddrinfo(thread_servinfo);

    while (1) {
        // wait for 
        pthread_mutex_lock(&m);
        did_work = 0;
        while (!start_threads)
        {
            pthread_cond_wait(&cond, &m);
        }

        int station_num = user_data[user_index].stationNum;
        if (station_num == -1) {
            did_work = 1;
        }

        if (!did_work) {
            // sendto a random string of data to the user
            int station_num = user_data[user_index].stationNum;
            char *data = station_data[station_num].filePath;
            // printf("load data: thread %d \n", user_index);

            // get file path
            char* file_path = station_data[station_num].filePath;
            // get current seek chunk
            FILE* file_stream = fopen(file_path, "r");
            int current_chunk = station_data[station_num].seekIndex;
            if (fseek(file_stream, current_chunk, SEEK_SET) == -1) {
                perror("fseek");
                return (NULL);
            }
            size_t BYTES_PER_SECOND = 16*1024;
            // read 1000 bytes of the file
            char file_buffer[BYTES_PER_SECOND];
            if (fread(file_buffer, BYTES_PER_SECOND, 1, file_stream) == -1) {
                perror("fread");
                return (NULL);
            }
            fclose(file_stream);
            // printf("send data: thread %d \n", user_index);
            // int numbytes;
            // if ((numbytes = sendto(udp_sockfd, data, strlen(data), 0,
            //         thread_res->ai_addr, thread_res->ai_addrlen)) == -1) {
            //     perror("talker: sendto");
            //     return (NULL);
            // }
            // print the size of the file_buffer
            // printf("size of file_buffer: %lu\n", sizeof(file_buffer));

            int bytes_sent = sizeof(file_buffer);
            if (sendall(udp_sockfd, file_buffer, &bytes_sent, thread_res) == -1)
            {
                perror("sendall");
                printf("We only sent %d bytes because of the error!\n", bytes_sent);
            }
            // printf("We sent all %d bytes!\n", bytes_sent);

            did_work = 1;


            usleep(400000);
        }

        pthread_mutex_unlock(&m);

        usleep(100000);
    }
    return NULL;
}

void *synchronization_thread(void *arg) {
    int c = 0;
    while (1)
    {
        start_threads = 1;
        // printf("\nbroadcast %d\n", c++);
        pthread_cond_broadcast(&cond);
        usleep(2000);
        start_threads = 0;
        // printf("before loop");
        // update file seek index for each station
        size_t BYTES_PER_SECOND = 16*1024;
        // print num_stations
        // printf("num_stations: %d\n", num_stations);
        for (int i = 0; i < num_stations; i++)
        {
            // printf("checking station %d\n", i);
            // get size of file
            FILE* fp = fopen(station_data[i].filePath, "r");
            fseek(fp, 0L, SEEK_END);
            size_t size = ftell(fp);
            if (size == -1) {
                perror("ftell");
                return (NULL);
            }
            station_data[i].seekIndex += BYTES_PER_SECOND;
            // if the seek index is greater than the size of the file, reset it
            if (station_data[i].seekIndex >= size)
            {
                // printf("resetting seek index for station %d\n", i);
                station_data[i].seekIndex = 0;
            }
            fclose(fp);
        }


        usleep(2000);
        usleep(1000000-4000);
    }
}

void *select_thread(void *arg) {
    fd_set master;    // master file descriptor list
    fd_set read_fds;  // temp file descriptor list for select()
    int fdmax;        // maximum file descriptor number

    int listener;     // listening socket descriptor
    int newfd;        // newly accept()ed socket descriptor
    struct sockaddr_storage remoteaddr; // client address
    socklen_t addrlen;

    char buf[256];    // buffer for client data
    int nbytes;


    char remoteIP[INET6_ADDRSTRLEN];

    int yes=1;        // for setsockopt() SO_REUSEADDR, below
    int i, j, rv;

    struct addrinfo hints, *ai, *p;

    // const char* port = argv[1];

    FD_ZERO(&master);    // clear the master and temp sets
    FD_ZERO(&read_fds);

    // LISTENER: get us a socket and bind it
    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE;
    if ((rv = getaddrinfo(NULL, port, &hints, &ai)) != 0) {
        fprintf(stderr, "snowcast_server: %s\n", gai_strerror(rv));
        exit(1);
    }

    for(p = ai; p != NULL; p = p->ai_next) {
        listener = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
        if (listener < 0) { 
            continue;
        }
        
        // lose the pesky "address already in use" error message
        setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));

        if (bind(listener, p->ai_addr, p->ai_addrlen) < 0) {
            close(listener);
            continue;
        }

        break;
    }

    // if we got here, it means we didn't get bound
    if (p == NULL) {
        fprintf(stderr, "snowcast_server: failed to bind\n");
        exit(2);
    }

    freeaddrinfo(ai); // all done with this

     // listen
    if (listen(listener, 10) == -1) {
        perror("listen");
        exit(3);
    }

    // add the listener to the master set
    FD_SET(listener, &master);

    // keep track of the biggest file descriptor
    fdmax = listener; // so far, it's this one

    while(1) {
        read_fds = master; // copy it
        if (select(fdmax+1, &read_fds, NULL, NULL, NULL) == -1) {
            perror("select");
            exit(4);
        }

        // run through the existing connections looking for data to read
        for(i = 0; i <= fdmax; i++) {
            if (FD_ISSET(i, &read_fds)) { // we got one!!
                if (i == listener) {
                    // handle new connections
                    addrlen = sizeof remoteaddr;
                    newfd = accept(listener,
                        (struct sockaddr *)&remoteaddr,
                        &addrlen);

                    if (newfd == -1) {
                        perror("accept");
                    } else {
                        FD_SET(newfd, &master); // add to master set
                        if (newfd > fdmax) {    // keep track of the max
                            fdmax = newfd;
                        }
                        printf("selectserver: new connection from %s on "
                            "socket %d\n.",
                            inet_ntop(remoteaddr.ss_family,
                                get_in_addr((struct sockaddr*)&remoteaddr),
                                remoteIP, INET6_ADDRSTRLEN),
                            newfd);
                        // init user with this newfd
                        init_user(newfd);

                        // send the welcome message to client
                        struct Welcome welcome;
                        welcome.replyType = 2;
                        welcome.numStations = htons(num_stations);
                        int numbytes;
                        if ((numbytes=send(newfd, &welcome, sizeof(struct Welcome), 0)) == -1)
                            perror("send");

                        //print the num bytes
                        // print the size of the struct welcome
                        printf("size of welcome struct: %lu\n", sizeof(struct Welcome));
                        printf("sent %d bytes\n", numbytes);
                    }
                } else {
                    // handle data from a client
                    struct Command command;
                    if ((nbytes = recv(i, (char*)&command, sizeof(struct Command), 0)) <= 0) {
                        // got error or connection closed by client
                        if (nbytes == 0) {
                            // connection closed
                            printf("selectserver: socket %d hung up\n", i);
                        } else {
                            perror("recv");
                        }
                        close(i); // bye!
                        FD_CLR(i, &master); // remove from master set
                        // remove user from data structures
                        destroy_user(i);
                    } else {
                        // we got some data from a client
                        if (command.commandType == 0) {
                            // hello message with udpPort
                            printf("udpPort (from Hello) for new connection is %d.\n", ntohs(command.number));
                            // update udp port of user
                            update_user_udpPort(i, ntohs(command.number));
                        }
                        else if (command.commandType == 1) {
                            int station_num = ntohs(command.number);
                            printf("setting station to %d\n", ntohs(command.number));
                            // update station of user
                            update_user_station(i, ntohs(command.number));

                            send_announce_message(i, station_num);
                        }
                        else {
                            // send back in invalid command
                            struct InvalidCommand invalid;
                            invalid.replyType = 4;
                            invalid.replyStringSize = 21;
                            // make a string with the command.commmandType type in it
                            invalid.replyString = "Invalid command type";
                            if ((send(i, &invalid, sizeof(struct InvalidCommand), 0)) == -1)
                                perror("send");
                            
                            // drop connection upon invalid command
                            close(i);
                            FD_CLR(i, &master);
                        }
                    }
                } // END handle data from client
            } // END got new incoming connection
        } // END looping through file descriptors

        // broadcast the new files over the udp socket list for each use

    } // END for(;;)--and you thought it would never end!
}

void *init_user(int sockfd) {
    // add the user to the list of user data
    pthread_mutex_lock(&mutex_user_data);

    // this is to save memory space. 
    // in general, the displacement of 4 is where a user "used to be"
    int user_index = max_active_users++;
    int running_index = 0;
    while(running_index++ < max_active_users)
    {
        if (user_data[running_index].sockfd == -1)
        {
            user_index = running_index;
            break;
        }
        // printf("reusing memory\n");
    }
    // have to make more memory
    if (user_index == max_active_users) {
        ///printf("making new memory\n");
        user_t *more_users = realloc(user_data, sizeof(user_t) * max_active_users);
        if (!more_users) { perror("realloc"); exit(1); }
        user_data = more_users;
    }

    // map TCP sockfd to this user index
    user_data[user_index] = (user_t){-1, -1, sockfd, -1};
    sockfd_to_user[sockfd] = user_index;
    // free(user_stream_threads);
    pthread_mutex_unlock(&mutex_user_data);
}
void *update_user_udpPort(int sockfd, int udpPort) {
    pthread_mutex_lock(&mutex_user_data);
    // get the user
    user_t *user = &user_data[sockfd_to_user[sockfd]];
    // set the udpPort
    user->udpPort = udpPort;
    // start the stream thread, now that we have the udpPort
    pthread_create(&user->streamThread, NULL, send_udp_packet_routine, (void *)sockfd_to_user[sockfd]);
    pthread_mutex_unlock(&mutex_user_data);
}
void *update_user_station(int sockfd, int stationNum) {
    pthread_mutex_lock(&mutex_user_data);
    user_data[sockfd_to_user[sockfd]].stationNum = stationNum;
    pthread_mutex_unlock(&mutex_user_data);
}
void *print_user_data(int index) {
    printf("udpPort: %d, stationNum: %d, sockfd: %d, threadId:%d\n", 
        user_data[index].udpPort, user_data[index].stationNum, user_data[index].sockfd, user_data[index].streamThread);
}
void destroy_user(int sockfd) {
    pthread_mutex_lock(&mutex_user_data);

    // stop the thread streaming to the user
    pthread_cancel(user_data[sockfd_to_user[sockfd]].streamThread);
    // pthread_kill(user_data[sockfd_to_user[sockfd]].streamThread, SIGINT);
    // "remove" the user from the list of user data
    user_data[sockfd_to_user[sockfd]] = (user_t) {-1, -1, -1, -1};
    // map sockfd to -1
    sockfd_to_user[sockfd] = -1;

    pthread_mutex_unlock(&mutex_user_data);
}


void *get_in_addr(struct sockaddr *sa)
{
    if (sa->sa_family == AF_INET) {
        return &(((struct sockaddr_in*)sa)->sin_addr);
    }

    return &(((struct sockaddr_in6*)sa)->sin6_addr);
}

void send_announce_message(int fd, int station_num) {
    char* file_path = station_data[station_num].filePath;
    int len_file_path = strlen(file_path);

    char *send_buffer = malloc(len_file_path+2);
    send_buffer[0] = 3;
    send_buffer[1] = len_file_path;

    memcpy(send_buffer + 2, file_path, len_file_path);

    printf("buffer: %s\n", send_buffer);

    int bytessent;
    if ((bytessent = send(fd, send_buffer, len_file_path + 2, 0)) == -1)
        perror("send");
    // print the number of bytes sent
    printf("sent %d bytes\n", bytessent);

    free(send_buffer);

}