adamc@116: #include adamc@116: adamc@116: #include adamc@116: #include adamc@116: #include adamc@116: #include adamc@472: #include adamc@116: adamc@138: #include adamc@138: adamc@244: #include "urweb.h" adamc@117: adamc@311: int uw_backlog = 10; adamc@311: int uw_bufsize = 1024; adamc@116: adamc@138: typedef struct node { adamc@138: int fd; adamc@138: struct node *next; adamc@138: } *node; adamc@138: adamc@138: static node front = NULL, back = NULL; adamc@138: adamc@138: static int empty() { adamc@138: return front == NULL; adamc@138: } adamc@138: adamc@138: static void enqueue(int fd) { adamc@138: node n = malloc(sizeof(struct node)); adamc@138: adamc@138: n->fd = fd; adamc@138: n->next = NULL; adamc@138: if (back) adamc@138: back->next = n; adamc@138: else adamc@138: front = n; adamc@138: back = n; adamc@138: } adamc@138: adamc@138: static int dequeue() { adamc@138: int ret = front->fd; adamc@138: adamc@138: front = front->next; adamc@138: if (!front) adamc@138: back = NULL; adamc@138: adamc@138: return ret; adamc@138: } adamc@138: adamc@138: static pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER; adamc@138: static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; adamc@138: adamc@167: #define MAX_RETRIES 5 adamc@167: adamc@424: int uw_db_begin(uw_context); adamc@424: int uw_db_commit(uw_context); adamc@424: int uw_db_rollback(uw_context); adamc@424: adamc@424: static int try_rollback(uw_context ctx) { adamc@424: int r = uw_db_rollback(ctx); adamc@424: adamc@424: if (r) { adamc@424: printf("Error running SQL ROLLBACK\n"); adamc@424: uw_reset(ctx); adamc@424: uw_write(ctx, "HTTP/1.1 500 Internal Server Error\n\r"); adamc@424: uw_write(ctx, "Content-type: text/plain\r\n\r\n"); adamc@424: uw_write(ctx, "Error running SQL ROLLBACK\n"); adamc@424: } adamc@424: adamc@424: return r; adamc@424: } adamc@424: adamc@138: static void *worker(void *data) { adamc@324: int me = *(int *)data, retries_left = MAX_RETRIES; adamc@466: uw_context ctx = uw_init(0, 1024, 0); adamc@272: adamc@272: while (1) { adamc@311: failure_kind fk = uw_begin_init(ctx); adamc@272: adamc@272: if (fk == SUCCESS) { adamc@311: uw_db_init(ctx); adamc@272: printf("Database connection initialized.\n"); adamc@272: break; adamc@272: } else if (fk == BOUNDED_RETRY) { adamc@272: if (retries_left) { adamc@311: printf("Initialization error triggers bounded retry: %s\n", uw_error_message(ctx)); adamc@272: --retries_left; adamc@272: } else { adamc@311: printf("Fatal initialization error (out of retries): %s\n", uw_error_message(ctx)); adamc@311: uw_free(ctx); adamc@272: return NULL; adamc@272: } adamc@272: } else if (fk == UNLIMITED_RETRY) adamc@311: printf("Initialization error triggers unlimited retry: %s\n", uw_error_message(ctx)); adamc@272: else if (fk == FATAL) { adamc@311: printf("Fatal initialization error: %s\n", uw_error_message(ctx)); adamc@311: uw_free(ctx); adamc@272: return NULL; adamc@272: } else { adamc@311: printf("Unknown uw_handle return code!\n"); adamc@311: uw_free(ctx); adamc@272: return NULL; adamc@272: } adamc@272: } adamc@116: adamc@116: while (1) { adamc@311: char buf[uw_bufsize+1], *back = buf, *s; adamc@138: int sock; adamc@116: adamc@138: pthread_mutex_lock(&queue_mutex); adamc@138: while (empty()) adamc@138: pthread_cond_wait(&queue_cond, &queue_mutex); adamc@138: sock = dequeue(); adamc@138: pthread_mutex_unlock(&queue_mutex); adamc@138: adamc@138: printf("Handling connection with thread #%d.\n", me); adamc@138: adamc@138: while (1) { adamc@167: unsigned retries_left = MAX_RETRIES; adamc@311: int r = recv(sock, back, uw_bufsize - (back - buf), 0); adamc@138: adamc@138: if (r < 0) { adamc@138: fprintf(stderr, "Recv failed\n"); adamc@138: break; adamc@138: } adamc@138: adamc@138: if (r == 0) { adamc@138: printf("Connection closed.\n"); adamc@138: break; adamc@138: } adamc@138: adamc@138: printf("Received %d bytes.\n", r); adamc@138: adamc@138: back += r; adamc@138: *back = 0; adamc@138: adamc@138: if (s = strstr(buf, "\r\n\r\n")) { adamc@424: failure_kind fk; adamc@457: char *cmd, *path, *headers, path_copy[uw_bufsize+1], *inputs; adamc@138: adamc@475: s[2] = 0; adamc@144: adamc@138: if (!(s = strstr(buf, "\r\n"))) { adamc@138: fprintf(stderr, "No newline in buf\n"); adamc@138: break; adamc@138: } adamc@138: adamc@138: *s = 0; adamc@457: headers = s + 2; adamc@138: cmd = s = buf; adamc@401: adamc@401: printf("Read: %s\n", buf); adamc@138: adamc@138: if (!strsep(&s, " ")) { adamc@138: fprintf(stderr, "No first space in HTTP command\n"); adamc@138: break; adamc@138: } adamc@138: adamc@138: if (strcmp(cmd, "GET")) { adamc@138: fprintf(stderr, "Not ready for non-get command: %s\n", cmd); adamc@138: break; adamc@138: } adamc@138: adamc@138: path = s; adamc@138: if (!strsep(&s, " ")) { adamc@138: fprintf(stderr, "No second space in HTTP command\n"); adamc@138: break; adamc@138: } adamc@138: adamc@144: if (inputs = strchr(path, '?')) { adamc@144: char *name, *value; adamc@144: *inputs++ = 0; adamc@144: adamc@144: while (*inputs) { adamc@144: name = inputs; adamc@145: if (inputs = strchr(inputs, '&')) adamc@145: *inputs++ = 0; adamc@145: else adamc@145: inputs = strchr(name, 0); adamc@145: adamc@145: if (value = strchr(name, '=')) { adamc@144: *value++ = 0; adamc@311: uw_set_input(ctx, name, value); adamc@144: } adamc@145: else adamc@311: uw_set_input(ctx, name, ""); adamc@144: } adamc@144: } adamc@144: adamc@138: printf("Serving URI %s....\n", path); adamc@138: adamc@458: uw_set_headers(ctx, headers); adamc@458: adamc@167: while (1) { adamc@500: uw_write(ctx, ""); adamc@500: adamc@424: if (uw_db_begin(ctx)) { adamc@424: printf("Error running SQL BEGIN\n"); adamc@424: if (retries_left) adamc@424: --retries_left; adamc@424: else { adamc@424: fk = FATAL; adamc@424: uw_reset(ctx); adamc@464: uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\n\r"); adamc@464: uw_write_header(ctx, "Content-type: text/plain\r\n\r\n"); adamc@424: uw_write(ctx, "Error running SQL BEGIN\n"); adamc@424: adamc@424: break; adamc@424: } adamc@424: } adamc@167: adamc@462: uw_write_header(ctx, "HTTP/1.1 200 OK\r\n"); adamc@462: uw_write_header(ctx, "Content-type: text/html\r\n"); adamc@167: adamc@400: strcpy(path_copy, path); adamc@458: fk = uw_begin(ctx, path_copy); adamc@167: if (fk == SUCCESS) { adamc@464: uw_write(ctx, ""); adamc@464: adamc@424: if (uw_db_commit(ctx)) { adamc@424: fk = FATAL; adamc@424: adamc@424: printf("Error running SQL COMMIT\n"); adamc@424: uw_reset(ctx); adamc@464: uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\n\r"); adamc@464: uw_write_header(ctx, "Content-type: text/plain\r\n"); adamc@424: uw_write(ctx, "Error running SQL COMMIT\n"); adamc@424: } adamc@424: adamc@167: break; adamc@167: } else if (fk == BOUNDED_RETRY) { adamc@167: if (retries_left) { adamc@311: printf("Error triggers bounded retry: %s\n", uw_error_message(ctx)); adamc@167: --retries_left; adamc@167: } adamc@167: else { adamc@311: printf("Fatal error (out of retries): %s\n", uw_error_message(ctx)); adamc@167: adamc@311: uw_reset_keep_error_message(ctx); adamc@464: uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\n\r"); adamc@464: uw_write_header(ctx, "Content-type: text/plain\r\n"); adamc@311: uw_write(ctx, "Fatal error (out of retries): "); adamc@311: uw_write(ctx, uw_error_message(ctx)); adamc@311: uw_write(ctx, "\n"); adamc@424: adamc@424: try_rollback(ctx); adamc@424: adamc@424: break; adamc@167: } adamc@167: } else if (fk == UNLIMITED_RETRY) adamc@311: printf("Error triggers unlimited retry: %s\n", uw_error_message(ctx)); adamc@167: else if (fk == FATAL) { adamc@311: printf("Fatal error: %s\n", uw_error_message(ctx)); adamc@167: adamc@311: uw_reset_keep_error_message(ctx); adamc@464: uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\r\n"); adamc@464: uw_write_header(ctx, "Content-type: text/plain\r\n"); adamc@311: uw_write(ctx, "Fatal error: "); adamc@311: uw_write(ctx, uw_error_message(ctx)); adamc@311: uw_write(ctx, "\n"); adamc@167: adamc@424: try_rollback(ctx); adamc@424: adamc@167: break; adamc@167: } else { adamc@311: printf("Unknown uw_handle return code!\n"); adamc@167: adamc@311: uw_reset_keep_request(ctx); adamc@464: uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\n\r"); adamc@464: uw_write_header(ctx, "Content-type: text/plain\r\n"); adamc@311: uw_write(ctx, "Unknown uw_handle return code!\n"); adamc@167: adamc@424: try_rollback(ctx); adamc@424: adamc@167: break; adamc@167: } adamc@167: adamc@311: uw_reset_keep_request(ctx); adamc@424: adamc@424: if (try_rollback(ctx)) adamc@424: break; adamc@167: } adamc@138: adamc@311: uw_send(ctx, sock); adamc@138: adamc@138: printf("Done with client.\n\n"); adamc@324: uw_memstats(ctx); adamc@138: break; adamc@138: } adamc@116: } adamc@116: adamc@138: close(sock); adamc@311: uw_reset(ctx); adamc@116: } adamc@116: } adamc@102: adamc@477: static void help(char *cmd) { adamc@477: printf("Usage: %s [-p ] [-t ]\n", cmd); adamc@477: } adamc@477: adamc@138: int main(int argc, char *argv[]) { adamc@116: // The skeleton for this function comes from Beej's sockets tutorial. adamc@138: int sockfd; // listen on sock_fd adamc@116: struct sockaddr_in my_addr; adamc@116: struct sockaddr_in their_addr; // connector's address information adamc@116: int sin_size, yes = 1; adamc@472: int uw_port = 8080, nthreads = 1, i, *names, opt; adamc@472: adamc@477: while ((opt = getopt(argc, argv, "hp:t:")) != -1) { adamc@472: switch (opt) { adamc@472: case '?': adamc@472: fprintf(stderr, "Unknown command-line option"); adamc@477: help(argv[0]); adamc@472: return 1; adamc@138: adamc@477: case 'h': adamc@477: help(argv[0]); adamc@477: return 0; adamc@477: adamc@472: case 'p': adamc@472: uw_port = atoi(optarg); adamc@472: if (uw_port <= 0) { adamc@472: fprintf(stderr, "Invalid port number\n"); adamc@477: help(argv[0]); adamc@472: return 1; adamc@472: } adamc@472: break; adamc@472: adamc@472: case 't': adamc@472: nthreads = atoi(optarg); adamc@472: if (nthreads <= 0) { adamc@472: fprintf(stderr, "Invalid thread count\n"); adamc@477: help(argv[0]); adamc@472: return 1; adamc@472: } adamc@472: break; adamc@472: adamc@472: default: adamc@472: fprintf(stderr, "Unexpected getopt() behavior\n"); adamc@472: return 1; adamc@472: } adamc@138: } adamc@138: adamc@138: names = calloc(nthreads, sizeof(int)); adamc@116: adamc@116: sockfd = socket(PF_INET, SOCK_STREAM, 0); // do some error checking! adamc@116: adamc@116: if (sockfd < 0) { adamc@116: fprintf(stderr, "Listener socket creation failed\n"); adamc@116: return 1; adamc@116: } adamc@116: adamc@116: if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) < 0) { adamc@116: fprintf(stderr, "Listener socket option setting failed\n"); adamc@116: return 1; adamc@116: } adamc@116: adamc@116: my_addr.sin_family = AF_INET; // host byte order adamc@311: my_addr.sin_port = htons(uw_port); // short, network byte order adamc@116: my_addr.sin_addr.s_addr = INADDR_ANY; // auto-fill with my IP adamc@116: memset(my_addr.sin_zero, '\0', sizeof my_addr.sin_zero); adamc@116: adamc@116: if (bind(sockfd, (struct sockaddr *)&my_addr, sizeof my_addr) < 0) { adamc@116: fprintf(stderr, "Listener socket bind failed\n"); adamc@116: return 1; adamc@116: } adamc@116: adamc@311: if (listen(sockfd, uw_backlog) < 0) { adamc@116: fprintf(stderr, "Socket listen failed\n"); adamc@116: return 1; adamc@116: } adamc@116: adamc@116: sin_size = sizeof their_addr; adamc@116: adamc@311: printf("Listening on port %d....\n", uw_port); adamc@116: adamc@138: for (i = 0; i < nthreads; ++i) { adamc@138: pthread_t thread; adamc@138: names[i] = i; adamc@138: if (pthread_create(&thread, NULL, worker, &names[i])) { adamc@138: fprintf(stderr, "Error creating worker thread #%d\n", i); adamc@138: return 1; adamc@138: } adamc@138: } adamc@138: adamc@116: while (1) { adamc@138: int new_fd = accept(sockfd, (struct sockaddr *)&their_addr, &sin_size); adamc@116: adamc@116: if (new_fd < 0) { adamc@116: fprintf(stderr, "Socket accept failed\n"); adamc@116: return 1; adamc@116: } adamc@116: adamc@116: printf("Accepted connection.\n"); adamc@138: adamc@138: pthread_mutex_lock(&queue_mutex); adamc@138: enqueue(new_fd); adamc@139: pthread_cond_broadcast(&queue_cond); adamc@138: pthread_mutex_unlock(&queue_mutex); adamc@116: } adamc@102: }