#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include /* Compile time constants */ #define FILED_SENDFILE_MAX 16777215 /* Default values */ #define MAX_FAILURE_COUNT 30 #define PORT 8080 #define THREAD_COUNT 10 #define BIND_ADDR "::" #define CACHE_SIZE 8192 /* Arguments for worker threads */ struct filed_worker_thread_args { int fd; }; /* File information */ struct filed_fileinfo { pthread_mutex_t mutex; char *path; int fd; size_t len; char *lastmod; char lastmod_b[64]; char *type; }; /* Request variables */ struct filed_http_request { /** Buffers **/ struct filed_fileinfo fileinfo; char path_b[1010]; char tmpbuf[1010]; /** HTTP Request information **/ char *path; /*** Path being requested ***/ struct { struct { int present; off_t offset; /*** Range start ***/ ssize_t length; /*** Range length ***/ } range; } headers; }; /* Global variables */ /** Open File cache **/ struct filed_fileinfo *filed_fileinfo_fdcache; unsigned int filed_fileinfo_fdcache_size = CACHE_SIZE; /* Initialize process */ static int filed_init(void) { unsigned int idx; int mutex_init_ret; mlockall(MCL_CURRENT | MCL_FUTURE); filed_fileinfo_fdcache = malloc(sizeof(*filed_fileinfo_fdcache) * filed_fileinfo_fdcache_size); if (filed_fileinfo_fdcache == NULL) { return(1); } for (idx = 0; idx < filed_fileinfo_fdcache_size; idx++) { mutex_init_ret = pthread_mutex_init(&filed_fileinfo_fdcache[idx].mutex, NULL); if (mutex_init_ret != 0) { return(1); } filed_fileinfo_fdcache[idx].path = strdup(""); filed_fileinfo_fdcache[idx].fd = -1; filed_fileinfo_fdcache[idx].lastmod = ""; filed_fileinfo_fdcache[idx].type = ""; } signal(SIGPIPE, SIG_IGN); return(0); } /* Listen on a particular address/port */ static int filed_listen(const char *address, unsigned int port) { struct sockaddr_in6 addr; int pton_ret, bind_ret, listen_ret; int fd; addr.sin6_family = AF_INET6; addr.sin6_flowinfo = 0; addr.sin6_scope_id = 0; addr.sin6_port = htons(port); pton_ret = inet_pton(AF_INET6, address, addr.sin6_addr.s6_addr); if (pton_ret != 1) { return(-1); } fd = socket(AF_INET6, SOCK_STREAM, 0); if (fd < 0) { return(fd); } bind_ret = bind(fd, (const struct sockaddr *) &addr, sizeof(addr)); if (bind_ret < 0) { close(fd); return(-1); } listen_ret = listen(fd, 128); if (listen_ret != 0) { close(fd); return(-1); } return(fd); } /* Log a message */ //#define FILED_DONT_LOG #ifdef FILED_DONT_LOG # define filed_logging_thread_init() 0 # define filed_log_msg_debug(x, ...) /**/ # define filed_log_msg(x) /**/ #else /* Initialize logging thread */ static int filed_logging_thread_init(void) { /* XXX:TODO: Unimplemented */ return(0); } /* XXX:TODO: Unimplemented */ #define filed_log_msg_debug(x, ...) { fprintf(stderr, x, __VA_ARGS__); fprintf(stderr, "\n"); fflush(stderr); } static void filed_log_msg(const char *buffer) { /* XXX:TODO: Unimplemented */ fprintf(stderr, "%s\n", buffer); return; } #endif /* Format time per RFC2616 */ static char *filed_format_time(char *buffer, size_t buffer_len, const time_t timeinfo) { struct tm timeinfo_tm, *timeinfo_tm_p; timeinfo_tm_p = gmtime_r(&timeinfo, &timeinfo_tm); if (timeinfo_tm_p == NULL) { return("unknown"); } buffer[buffer_len - 1] = '\0'; buffer_len = strftime(buffer, buffer_len - 1, "%a, %d %b %Y %H:%M:%S GMT", timeinfo_tm_p); return(buffer); } /* hash */ static unsigned int filed_hash(const unsigned char *value, unsigned int modulus) { unsigned char curr; unsigned int retval; retval = modulus - 1; while ((curr = *value)) { if (curr < 32) { curr = 255 - curr; } else { curr -= 32; } retval <<= 5; retval += curr; value++; } retval = retval % modulus; return(retval); } /* Open a file and return file information */ static struct filed_fileinfo *filed_open_file(const char *path, struct filed_fileinfo *buffer) { struct filed_fileinfo *cache; unsigned int cache_idx; off_t len; int fd; cache_idx = filed_hash((const unsigned char *) path, filed_fileinfo_fdcache_size); cache = &filed_fileinfo_fdcache[cache_idx]; filed_log_msg_debug("Locking mutex for idx: %lu", (unsigned long) cache_idx); pthread_mutex_lock(&cache->mutex); filed_log_msg_debug("Completed locking mutex for idx: %lu", (unsigned long) cache_idx); if (strcmp(path, cache->path) != 0) { filed_log_msg_debug("Cache miss for idx: %lu: OLD \"%s\", NEW \"%s\"", (unsigned long) cache_idx, cache->path, path); fd = open(path, O_RDONLY | O_LARGEFILE); if (fd < 0) { pthread_mutex_unlock(&cache->mutex); return(NULL); } free(cache->path); if (cache->fd >= 0) { close(cache->fd); } len = lseek(fd, 0, SEEK_END); lseek(fd, 0, SEEK_SET); cache->fd = fd; cache->len = len; cache->path = strdup(path); /* XXX:TODO: Determine */ cache->type = "video/mp4"; cache->lastmod = filed_format_time(cache->lastmod_b, sizeof(cache->lastmod_b), time(NULL) - 30); } else { filed_log_msg_debug("Cache hit for idx: %lu: PATH \"%s\"", (unsigned long) cache_idx, path); } /* * We have to make a duplicate FD, because once we release the cache * mutex, the file descriptor may be closed */ fd = dup(cache->fd); if (fd < 0) { pthread_mutex_unlock(&cache->mutex); return(NULL); } buffer->fd = fd; buffer->len = cache->len; buffer->type = cache->type; memcpy(buffer->lastmod_b, cache->lastmod_b, sizeof(buffer->lastmod_b)); buffer->lastmod = buffer->lastmod_b + (cache->lastmod - cache->lastmod_b); pthread_mutex_unlock(&cache->mutex); return(buffer); } /* Process an HTTP request and return the path requested */ static struct filed_http_request *filed_get_http_request(FILE *fp, struct filed_http_request *buffer_st) { char *method, *path; char *buffer, *tmpbuffer, *workbuffer, *workbuffer_next; size_t buffer_len, tmpbuffer_len; off_t range_start, range_end; ssize_t range_length; int range_request; int i; range_start = 0; range_end = 0; range_request = 0; range_length = -1; buffer = buffer_st->path_b; buffer_len = sizeof(buffer_st->path_b); tmpbuffer = buffer_st->tmpbuf; tmpbuffer_len = sizeof(buffer_st->tmpbuf); filed_log_msg("WAIT_FOR_REQUEST FD=..."); fgets(buffer, buffer_len, fp); method = buffer; buffer = strchr(buffer, ' '); if (buffer == NULL) { filed_log_msg("GOT_REQUEST FD=... ERROR=format"); return(NULL); } *buffer = '\0'; buffer++; path = buffer; buffer = strchr(buffer, ' '); if (buffer != NULL) { *buffer = '\0'; buffer++; } filed_log_msg("GOT_REQUEST FD=... PATH=..."); filed_log_msg("WAIT_FOR_HEADERS FD=..."); for (i = 0; i < 100; i++) { fgets(tmpbuffer, tmpbuffer_len, fp); if (strncasecmp(tmpbuffer, "Range: ", 7) == 0) { workbuffer = tmpbuffer + 7; if (strncasecmp(workbuffer, "bytes=", 6) == 0) { workbuffer += 6; range_request = 1; range_start = strtoull(workbuffer, &workbuffer_next, 10); workbuffer = workbuffer_next; if (*workbuffer == '-') { workbuffer++; if (*workbuffer != '\r' && *workbuffer != '\n') { range_end = strtoull(workbuffer, &workbuffer_next, 10); } } } } if (memcmp(tmpbuffer, "\r\n", 2) == 0) { break; } } filed_log_msg("GOT_HEADERS FD=..."); /* We only handle the "GET" method */ if (strcasecmp(method, "get") != 0) { return(NULL); } /* Determine range */ if (range_end != 0) { if (range_end <= range_start) { return(NULL); } range_length = range_end - range_start; filed_log_msg_debug("Computing length parameter: %llu = %llu - %llu", (unsigned long long) range_length, (unsigned long long) range_end, (unsigned long long) range_start ); } /* Fill up structure to return */ buffer_st->path = path; buffer_st->headers.range.present = range_request; buffer_st->headers.range.offset = range_start; buffer_st->headers.range.length = range_length; return(buffer_st); } /* Return an error page */ static void filed_error_page(FILE *fp, const char *date_current, int error_number) { char *error_string = "ERRORUnable to process request"; fprintf(fp, "HTTP/1.1 %i OK\r\nDate: %s\r\nServer: filed\r\nLast-Modified: %s\r\nContent-Length: %llu\r\nContent-Type: %s\r\nConnection: close\r\n\r\n%s", error_number, date_current, date_current, (unsigned long long) strlen(error_string), "text/html", error_string ); } /* Handle a single request from a client */ static void filed_handle_client(int fd, struct filed_http_request *request) { struct filed_fileinfo *fileinfo; ssize_t sendfile_ret; size_t sendfile_len, sendfile_sent, sendfile_size; off_t sendfile_offset; char *path; char *date_current, date_current_b[64]; int http_code; FILE *fp; /* Determine current time */ date_current = filed_format_time(date_current_b, sizeof(date_current_b), time(NULL)); /* Open socket as ANSI I/O for ease of use */ fp = fdopen(fd, "w+b"); if (fp == NULL) { close(fd); return; } request = filed_get_http_request(fp, request); filed_log_msg("PROCESS_REPLY_START FD=... PATH=... RANGE_START=... RANGE_LENGTH=..."); if (request == NULL || request->path == NULL) { filed_error_page(fp, date_current, 500); filed_log_msg("PROCESS_REPLY_COMPLETE FD=... ERROR=500"); fclose(fp); return; } path = request->path; http_code = -1; fileinfo = filed_open_file(path, &request->fileinfo); if (fileinfo == NULL) { filed_error_page(fp, date_current, 404); filed_log_msg("PROCESS_REPLY_COMPLETE FD=... ERROR=404"); } else { if (request->headers.range.offset != 0 || request->headers.range.length >= 0) { if ((size_t) request->headers.range.offset >= fileinfo->len) { filed_log_msg("PROCESS_REPLY_COMPLETE FD=... ERROR=416"); filed_error_page(fp, date_current, 416); } else { if (request->headers.range.length < 0) { filed_log_msg_debug("Computing length to fit in bounds: fileinfo->len = %llu, request->headers.range.offset = %llu", (unsigned long long) fileinfo->len, (unsigned long long) request->headers.range.offset ); request->headers.range.length = fileinfo->len - request->headers.range.offset; } filed_log_msg_debug("Partial request, starting at: %llu and running for %llu bytes", (unsigned long long) request->headers.range.offset, (unsigned long long) request->headers.range.length); http_code = 206; } } else { if (request->headers.range.present) { http_code = 206; } else { http_code = 200; } request->headers.range.offset = 0; request->headers.range.length = fileinfo->len; } if (http_code > 0) { fprintf(fp, "HTTP/1.1 %i OK\r\nDate: %s\r\nServer: filed\r\nLast-Modified: %s\r\nContent-Length: %llu\r\nAccept-Ranges: bytes\r\nContent-Type: %s\r\nConnection: close\r\n", http_code, date_current, fileinfo->lastmod, (unsigned long long) request->headers.range.length, fileinfo->type ); if (http_code == 206) { fprintf(fp, "Content-Range: bytes %llu-%llu/%llu\r\n", (unsigned long long) request->headers.range.offset, (unsigned long long) (request->headers.range.offset + request->headers.range.length - 1), (unsigned long long) fileinfo->len ); } fprintf(fp, "\r\n"); fflush(fp); filed_log_msg("PROCESS_REPLY_COMPLETE FD=... STATUS=20X"); #ifdef FILED_NONBLOCK_HTTP int socket_flags; fd_set rfd, wfd; char sinkbuf[8192]; ssize_t read_ret; FD_ZERO(&rfd); FD_ZERO(&wfd); FD_SET(fd, &rfd); FD_SET(fd, &wfd); socket_flags = fcntl(fd, F_GETFL); fcntl(fd, F_SETFL, socket_flags | O_NONBLOCK); #endif filed_log_msg("SEND_START IFD=... OFD=... BYTES=..."); sendfile_offset = request->headers.range.offset; sendfile_len = request->headers.range.length; sendfile_sent = 0; while (1) { if (sendfile_len > FILED_SENDFILE_MAX) { sendfile_size = FILED_SENDFILE_MAX; } else { sendfile_size = sendfile_len; } sendfile_ret = sendfile(fd, fileinfo->fd, &sendfile_offset, sendfile_size); if (sendfile_ret <= 0) { #ifdef FILED_NONBLOCK_HTTP if (errno == EAGAIN) { sendfile_ret = 0; while (1) { select(fd + 1, &rfd, &wfd, NULL, NULL); if (FD_ISSET(fd, &rfd)) { read_ret = read(fd, sinkbuf, sizeof(sinkbuf)); if (read_ret <= 0) { break; } } if (FD_ISSET(fd, &wfd)) { read_ret = 1; break; } } if (read_ret <= 0) { break; } } else { break; } #else break; #endif } sendfile_len -= sendfile_ret; sendfile_sent += sendfile_ret; if (sendfile_len == 0) { break; } } filed_log_msg("SEND_COMPLETE STATUS=... IFD=... OFD=... BYTES=... BYTES_SENT=..."); } close(fileinfo->fd); filed_log_msg("CLOSE_FILE FD=..."); } filed_log_msg("CLOSE_CONNECTION FD=..."); fclose(fp); return; } /* Handle incoming connections */ static void *filed_worker_thread(void *arg_v) { struct filed_worker_thread_args *arg; struct filed_http_request request; struct sockaddr_in6 addr; socklen_t addrlen; int failure_count = 0, max_failure_count = MAX_FAILURE_COUNT; int master_fd, fd; /* Read arguments */ arg = arg_v; master_fd = arg->fd; while (1) { /* Failure loop prevention */ if (failure_count > max_failure_count) { break; } /* Accept a new client */ addrlen = sizeof(addr); fd = accept(master_fd, (struct sockaddr *) &addr, &addrlen); /* * If we fail, make a note of it so we don't go into a loop of * accept() failing */ if (fd < 0) { /* Log the new connection */ filed_log_msg("ACCEPT_FAILED"); failure_count++; continue; } /* Log the new connection */ filed_log_msg("NEW_CONNECTION SRC_ADDR=... SRC_PORT=... FD=..."); /* Reset failure count*/ failure_count = 0; /* Handle socket */ filed_handle_client(fd, &request); } /* Report error */ filed_log_msg("THREAD_DIED ABNORMAL"); return(NULL); } /* Create worker threads */ static int filed_worker_threads_init(int fd, int thread_count) { struct filed_worker_thread_args *arg; pthread_t threadid; int pthread_ret; int i; for (i = 0; i < thread_count; i++) { arg = malloc(sizeof(*arg)); arg->fd = fd; pthread_ret = pthread_create(&threadid, NULL, filed_worker_thread, arg); if (pthread_ret != 0) { return(-1); } } return(0); } /* Run process */ int main(int argc, char **argv) { int port = PORT, thread_count = THREAD_COUNT; const char *bind_addr = BIND_ADDR; int init_ret; int fd; /* XXX: TODO: Process arguments */ argc = argc; argv = argv; /* Create listening socket */ fd = filed_listen(bind_addr, port); if (fd < 0) { perror("filed_listen"); return(1); } /* Become a daemon */ /* XXX:TODO: Become a daemon */ /* Initialize */ init_ret = filed_init(); if (init_ret != 0) { perror("filed_init"); return(3); } /* Create logging thread */ init_ret = filed_logging_thread_init(); if (init_ret != 0) { perror("filed_logging_thread_init"); return(4); } /* Create worker threads */ init_ret = filed_worker_threads_init(fd, thread_count); if (init_ret != 0) { perror("filed_worker_threads_init"); return(4); } /* Wait for threads to exit */ /* XXX:TODO: Monitor thread usage */ while (1) { sleep(60); } /* Return in failure */ return(2); }