#include #include #include #include #include #include #include #include #include #include #include #include #include "protocol.c" #define LINE_MAX 1024 #define MAX_USERS 1000 #define MAX_PATH 50 #define MAX_RATE_PER_SECOND 16*1024 / 2 // typedef struct station { // int streamFd; // char* filePath; // int fileBufferSize; // char fileBuffer[MAX_STREAM_RATE]; // } station_t; typedef struct station { pthread_t streamThread; int readfd; char *filePath; } station_t; int num_stations; station_t *stations; int setup_stations(int argc, char *argv[]); void *stream_routine(void *arg); typedef struct user { int udpPort; int stationNum; int sockfd; } 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; int max_sockfd = 0; pthread_mutex_t mutex_user_data = PTHREAD_MUTEX_INITIALIZER; // array from index to user_data user_t *user_data; int *sockfd_to_user; // stations array pointer // station_t *station_data; struct udp_packet_routine_args { int user_index; int buffer_size; char *file_buffer; }; void *send_udp_packet_routine(void* arg); void *select_routine(void* arg); void *sync_routine(void* arg); void *send_announce_routine(void* arg); void init_station(int station_num, const char *station_name); int parse(char buffer[LINE_MAX], char *tokens[LINE_MAX / 2]); void *print_info_routine(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_reply(int fd, int station_num); void send_invalid_command_reply(int fd, size_t message_size, char* message); // void *load_file(void* arg); main(int argc, char *argv[]) { // check and assign arguments if (argc < 3) { fprintf(stderr,"usage: ./snowcast_server [file 1] [file 2] ... \n"); exit(1); } // initizlize the port port = argv[1]; // initialize the stations & their threads if (setup_stations(argc, argv) == -1) { perror("setup_stations"); exit(1); } // make array of user data printf("max active users: %d\n", sizeof(user_t) * max_active_users); user_data = malloc(sizeof(user_t) * max_active_users); if (!user_data) { perror("malloc userdata"); return 1; } sockfd_to_user = malloc(sizeof(int) * max_active_users); if (!sockfd_to_user) { perror("malloc sockfd to user"); return 1; } // make and start "select" thread that manages: // 1) new connections, 2) requests from current connections, 3) closing connections pthread_t select_thread; pthread_create(&select_thread, NULL, select_routine, NULL); // start syncchronization thread to broadcast stations // pthread_t sync_thread; // pthread_create(&sync_thread, NULL, sync_routine, NULL); // command line interface char input[LINE_MAX]; memset(input, 0, LINE_MAX); char *tokens[LINE_MAX / 2]; while (read(STDIN_FILENO, input, LINE_MAX) > 0) { // init tokens memset(tokens, 0, (LINE_MAX / 2) * sizeof(char *)); // if 0, all whitespace if (!parse(input, tokens)) continue; char *command = tokens[0]; // if q, shutdown! if (!strcmp(command, "q")) { printf("Exiting.\n"); // TODO: exit better than break break; } // if p, print info else if (!strcmp(command, "p")) { // get the file descriptor int print_fd = 0; // see if there is a file path char *output_file_path = tokens[1]; if (output_file_path != NULL) { if ((print_fd = open(output_file_path, O_CREAT | O_WRONLY | O_TRUNC, S_IRWXU)) == -1) { perror("open"); continue; } } else { print_fd = STDOUT_FILENO; } // printf("print_fd: %d\n", print_fd); pthread_t print_info_thread; pthread_create(&print_info_thread, NULL, print_info_routine, print_fd); // note - this file descriptor is closed in the thread } else if (!strcmp(command, "u")) { // print all user data for (int i = 0; i < max_active_users; i++) { print_user_data(i); } } } return 0; } int read_file(int fd, char buffer[MAX_RATE_PER_SECOND], int station_num) { int bytes_read = read(fd, buffer, MAX_RATE_PER_SECOND); if (bytes_read < 0) { perror("read (from station file)"); return -1; } // printf("bytes read: %d\n", bytes_read); if (bytes_read == 0) { // printf("end of file, restarting\n"); pthread_t send_announce_thread; pthread_create(&send_announce_thread, NULL, send_announce_routine, station_num); if (lseek(fd, 0, SEEK_SET) == -1) { perror("lseek (in resarting file)"); return -1; } bytes_read = read(fd, buffer, MAX_RATE_PER_SECOND); if (bytes_read < 0) { perror("read (from station file, after restart)"); return -1; } } return bytes_read; } void *stream_routine_cleanup(void *arg) { int read_fd = (int) arg; close(read_fd); } void *stream_routine(void *arg) { int station_num = (int) arg; printf("stream routine %d\n", station_num); int read_fd = stations[station_num].readfd; pthread_cleanup_push(stream_routine_cleanup, read_fd); // make buffer which will be used to stream to children char buffer[MAX_RATE_PER_SECOND]; memset(buffer, 0, MAX_RATE_PER_SECOND); // if (!buffer) { perror("malloc (buffer in station thread)"); exit(1); } for (;;) { // load bytes into buffer int bytes_read = read_file(read_fd, buffer, station_num); if (bytes_read == -1) { exit(1); } // TODO: send buffer to children char *send_buffer = malloc(2 + bytes_read); for (int i = 0; i < max_active_users; i++) { if (!user_data[i].sockfd || user_data[i].sockfd == -1) continue; if (user_data[i].stationNum == station_num) { // send the udp packet int *send_buffer = malloc(2 + bytes_read); memset(send_buffer, 0, 2 + bytes_read); send_buffer[0] = i; send_buffer[1] = bytes_read; memcpy(send_buffer+2, buffer, bytes_read); // printf("sending udp packet to user %d\n", i); pthread_t t; pthread_create(&t, NULL, send_udp_packet_routine, send_buffer); } } free(send_buffer); usleep(1000000 / 2 - 5000); start_threads = 1; pthread_cond_broadcast(&cond); usleep(5000); start_threads = 0; memset(buffer, 0, MAX_RATE_PER_SECOND); } return (NULL); pthread_cleanup_pop(1); } int setup_stations(int argc, char *argv[]) { num_stations = argc - 2; // get the size to malloc int totalSize = 0; for(int i = 2; i < argc; i++) { totalSize += sizeof(pthread_t) + sizeof(int) + strlen(argv[i]); } // malloc the stations array stations = malloc(totalSize); if (!stations) { perror("malloc (stations pointer)"); return -1; } // assign the stations, and start the threads for (int i = 0; i < num_stations; i++) { stations[i].filePath = argv[i+2]; stations[i].readfd = open(argv[i+2], O_RDONLY); if (stations[i].readfd < 0) { perror("read (from station file)"); return -1; } pthread_create(&stations[i].streamThread, NULL, stream_routine, i); } printf("successfully created %d stations\n", num_stations); return 1; } void write_int_to_fd(int fd, int n) { int l = snprintf(NULL, 0, "%d", n); char *num = malloc(l + 1); if (!num) { perror("malloc write to fd"); return; } snprintf(num, l + 1, "%d", n); if (write(fd, num, strlen(num)) == -1) { perror("write"); } free(num); } void *print_info_routine(void *arg) { int print_fd = (int) arg; // printf("thread print_fd: %d\n", print_fd); // printf("num_stations: %d\n", num_stations); for (int i = 0; i < num_stations; i++) { write_int_to_fd(print_fd, i); char *comma = ","; write(print_fd, comma, strlen(comma)); // write file path char* file_path = stations[i].filePath; write(print_fd, file_path, strlen(file_path)); for (int j = 0; j < max_active_users; j++) { if (!user_data[j].sockfd || user_data[j].sockfd == -1) continue; if (user_data[j].stationNum == i) { char *localhost_ip = ",127.0.0.1:"; write(print_fd, localhost_ip, strlen(localhost_ip)); // write udpPort write_int_to_fd(print_fd, user_data[j].udpPort); } } // wrtie new line char *newline = "\n"; write(print_fd, newline, strlen(newline)); } if (print_fd != STDOUT_FILENO) close(print_fd); return (NULL); } int send_all_udp(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 } void udp_port_cleanup_handler(void *arg) { int sockfd = (int) arg; close(sockfd); } /* Make the manager routine */ void *send_udp_packet_routine(void *arg) { // printf("send udp packet routine\n"); int *buf = arg; // unpack args int user_index = buf[0]; int buffer_size = buf[1]; char *file_buffer = malloc(buffer_size); memcpy(file_buffer, buf+2, buffer_size); // printf("udp packet routine, user:%d\n size: %d\n", user_index, buffer_size); // 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; // 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 int int_port = user_data[user_index].udpPort; int length = snprintf( NULL, 0, "%d", int_port ); char* port = malloc( length + 1 ); if (!port) { perror("malloc on port"); return (NULL); } 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); } pthread_cleanup_push(udp_port_cleanup_handler, (void *)udp_sockfd); // 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; } // potential error here! // printf("station info - bytes read: %d, station_fd: %d, filePath: %s, buffersize: %d\n", bytes_read, station_info->streamFd, station_info->filePath, station_info->fileBufferSize); if (send_all_udp(udp_sockfd, file_buffer, &buffer_size, thread_res) == -1) { perror("send_all_udp"); printf("We only sent %d bytes because of the error!\n", buffer_size); } free(file_buffer); pthread_mutex_unlock(&m); pthread_cleanup_pop(1); return (NULL); } void *send_announce_routine(void *arg) { // unpack args int station_num = (int) arg; // send the announce messages for (int i = 0; i < max_active_users; i++) { if (user_data[i].sockfd == 0 || user_data[i].sockfd == -1) { continue; } // update the station of each user if (user_data[i].stationNum == station_num) { send_announce_reply(user_data[i].sockfd, station_num); } } } // void *sync_routine(void *arg) { // int c = 0; // while (1) // { // start_threads = 1; // pthread_cond_broadcast(&cond); // usleep(2000); // start_threads = 0; // usleep(1000000-2000); // } // } void *select_routine(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 { struct timeval tv; tv.tv_sec = 0; tv.tv_usec = TIMEOUT; if (setsockopt(newfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) { perror("setsockopt"); } uint8_t command_type = -1; if ((nbytes = recv(newfd, &command_type, 1, 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"); } // remove user from data structures close(newfd); // destroy_user(i); continue; } if (command_type != 0) { char * message = "must send a Hello message first"; send_invalid_command_reply(newfd, strlen(message), message); close(newfd); continue; } // get the udp port uint16_t udp_port = -1; int bytes_to_read = sizeof(uint16_t); if (recv_all(newfd, &udp_port, &bytes_to_read) == -1) { perror("recv_all"); close(newfd); // destroy_user(i); continue; } udp_port = ntohs(udp_port); FD_SET(newfd, &master); // add to master set if (newfd > fdmax) { // keep track of the max fdmax = newfd; } init_user(newfd); update_user_udpPort(newfd, udp_port); // send the welcome message to client struct Welcome welcome; welcome.replyType = 2; welcome.numStations = htons(num_stations); int bytes_to_send = sizeof(struct Welcome); if (send_all(newfd, &welcome, &bytes_to_send) == -1) perror("send_all"); } } else { // handle data from a client uint8_t command_type = -1; if ((nbytes = recv(i, &command_type, 1, 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_type == 0) { // we got a Hello commmand // get the udp port uint16_t udp_port = -1; int bytes_to_read = sizeof(uint16_t); if (recv_all(i, &udp_port, &bytes_to_read) == -1) { perror("recv_all"); close(i); FD_CLR(i, &master); destroy_user(i); continue; } if (user_data[sockfd_to_user[i]].udpPort != -1) { // send back in invalid command char * message = "must not sent more than one Hello message"; send_invalid_command_reply(i, strlen(message), message); // drop connection upon invalid command close(i); FD_CLR(i, &master); destroy_user(i); continue; } } else if (command_type == 1) { // we got a SetStation command // get the station number uint16_t station_number = -1; int bytes_to_read = sizeof(uint16_t); if (recv_all(i, &station_number, &bytes_to_read) == -1) { perror("recv_all"); close(i); FD_CLR(i, &master); destroy_user(i); continue; } station_number = ntohs(station_number); // check if user has a udpPort to stream to if (user_data[sockfd_to_user[i]].udpPort == -1) { // send back in invalid command char * message = "must send Hello message first"; send_invalid_command_reply(i, strlen(message), message); // drop connection upon invalid command close(i); FD_CLR(i, &master); destroy_user(i); continue; } // check if station num is in range if (station_number >= num_stations || station_number < 0) { // send back in invalid command char * message = "station number out of range"; send_invalid_command_reply(i, strlen(message), message); // drop connection upon invalid command close(i); FD_CLR(i, &master); destroy_user(i); continue; } // printf("setting station to %d\n", ntohs(command.number)); // update station of user update_user_station(i, station_number); send_announce_reply(i, station_number); } else { // send back in invalid command char * message = "invalid command"; send_invalid_command_reply(i, strlen(message), message); // drop connection upon invalid command close(i); FD_CLR(i, &master); destroy_user(i); } } } // END handle data from client } // END got new incoming connection } // END looping through file descriptors } // 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); if(sockfd > max_sockfd) { max_sockfd = sockfd; int * more_sockfd_to_user = realloc(sockfd_to_user, sizeof(int) * (max_sockfd + 1)); if (!more_sockfd_to_user) { perror("realloc"); exit(1); } sockfd_to_user = more_sockfd_to_user; } int running_index = 0; while(running_index < max_active_users) { if (user_data[running_index].sockfd == -1) { break; } running_index++; } if (running_index == max_active_users) { // printf("reached max active users\n"); // printf("making new memory\n"); max_active_users++; 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[running_index] = (user_t){-1, -1, sockfd}; sockfd_to_user[sockfd] = running_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\n", user_data[index].udpPort, user_data[index].stationNum, user_data[index].sockfd); } void destroy_user(int sockfd) { pthread_mutex_lock(&mutex_user_data); // stop the thread streaming to the user // pthread_t thread = user_data[sockfd_to_user[sockfd]].streamThread; // if (thread != -1) { // pthread_cancel(thread); // } // "remove" the user from the list of user data user_data[sockfd_to_user[sockfd]] = (user_t) {-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_reply(int fd, int station_num) { char* file_path = stations[station_num].filePath; int len_file_path = strlen(file_path); char *send_buffer = malloc(len_file_path+2); if (!send_buffer) { perror("malloc in send announce"); return; } send_buffer[0] = 3; send_buffer[1] = len_file_path; memcpy(send_buffer + 2, file_path, len_file_path); size_t bytes_to_send = len_file_path + 2; if (send_all(fd, send_buffer, &bytes_to_send) == -1) perror("send_all"); free(send_buffer); } void send_invalid_command_reply(int fd, size_t message_size, char* message) { char *send_buffer = malloc(message_size+2); if (!send_buffer) { perror("malloc in send invalid command"); return; } // type and payload size send_buffer[0] = 4; send_buffer[1] = message_size; memcpy(send_buffer + 2, message, message_size); int bytes_to_send = message_size + 2; if (send_all(fd, send_buffer, &bytes_to_send) == -1) perror("send"); free(send_buffer); } // void init_station(int station_num, const char* station_name) { // station_t *station = &station_data[station_num]; // // open the file // int stream_fd = open(station_name, O_RDONLY); // if (stream_fd == -1) { // perror("open"); // return; // } // station->streamFd = stream_fd; // station->filePath = station_name; // // setup file buffer // char stream_buffer[MAX_STREAM_RATE]; // memset(stream_buffer, 0, MAX_STREAM_RATE); // station->fileBufferSize = MAX_STREAM_RATE; // memcpy(&station->fileBufferSize, stream_buffer, MAX_STREAM_RATE); // // load the first buffer into the stations // seek_stations(station_num); // } // void seek_stations(int station_num) { // station_t *station_info = &station_data[station_num]; // memset(&station_info->fileBuffer, 0, MAX_STREAM_RATE); // int bytes_read = read(station_info->streamFd, &station_info->fileBuffer, MAX_STREAM_RATE); // lseek(station_info->streamFd, -16, SEEK_SET); // // printf("station info - bytes read: %d, station_fd: %d, filePath: %s, buffersize: %d\n", bytes_read, station_info->streamFd, station_info->filePath, station_info->fileBufferSize); // // time to restart the file // if (bytes_read == 0) { // if (lseek(station_info->streamFd, 0, SEEK_SET) == -1) { // perror("fseek"); // } // pthread_t send_announce_thread; // pthread_create(&send_announce_thread, NULL, send_announce_routine, (void *)station_num); // // load first chunk // bytes_read = read(station_info->streamFd, &station_info->fileBuffer, MAX_STREAM_RATE); // } // station_info->fileBufferSize = bytes_read; // } // Parses a buffer into tokens, from cs33 :) int parse(char buffer[LINE_MAX], char *tokens[LINE_MAX / 2]) { const char *regex = " \n\t\f\r"; char *current_token = strtok(buffer, regex); if (current_token == NULL) return 0; for (int i = 0; current_token != NULL; i++) { tokens[i] = current_token; current_token = strtok(NULL, regex); } return 1; }