server.c

#include "phenom/defs.h"
#include "phenom/configuration.h"
#include "phenom/job.h"
#include "phenom/log.h"
#include "phenom/sysutil.h"
#include "phenom/printf.h"
#include "phenom/listener.h"
#include "phenom/socket.h"
#include "phenom/stream.h"
#include "phenom/string.h"
#include "http_parser.h"
#include "server.h"
#include <fcntl.h>
#include "sys/stat.h"
#include <sysexits.h>
#include <unistd.h>
#include <stdlib.h>

const char *HEADER_END = "\r\n";

static ph_thread_pool_t *file_send_pool;

static ph_memtype_t mt_reqstate;
static ph_memtype_t mt_parser;
static ph_memtype_t mt_string;

static struct ph_memtype_def mt_reqstate_def = { "xx", "request_state", sizeof(request_state), PH_MEM_FLAGS_ZERO };
static struct ph_memtype_def mt_parser_def = { "xx,", "http_parser", sizeof(http_parser), PH_MEM_FLAGS_ZERO };
static struct ph_memtype_def mt_string_def = { "xx", "string", 0, PH_MEM_FLAGS_ZERO };

static struct ph_job_def file_send_job_def = {
  xx_send_file, PH_MEMTYPE_FIRST, NULL
};

const http_parser_settings parser_settings = {
  NULL, xx_handle_url, NULL, NULL, NULL, xx_handle_request_parsed, NULL, NULL
};

static void xx_tear_down_socket(ph_sock_t *sock)
{
  ph_sock_enable(sock, false);
  ph_sock_shutdown(sock, PH_SOCK_SHUT_RDWR);
  ph_sock_free(sock);
}

static request_state *xx_get_request_state(ph_sock_t *sock)
{
  http_parser *parser = calloc(1, sizeof(http_parser));
  http_parser_init(parser, HTTP_REQUEST);
  parser->data = sock;

  request_state *reqstate = ph_mem_alloc(mt_reqstate);
  reqstate->state = XX_PARSING_REQUEST;
  reqstate->parser = parser;
  reqstate->requested_file = ph_string_make_empty(mt_string, 16);

  return reqstate;
}

static void xx_clean_request_state(request_state *reqstate)
{
  if (reqstate->parser) {
    free(reqstate->parser);
  }

  if (reqstate->requested_file) {
    ph_string_delref(reqstate->requested_file);
  }

  ph_mem_free(mt_reqstate, reqstate);
}

static void xx_read_request(ph_sock_t *sock, request_state *reqstate)
{
  ph_buf_t *buf;
  size_t nparsed;

  while (reqstate->state == XX_PARSING_REQUEST) {
    buf = ph_sock_read_line(sock);
    if (!buf) {
      return;
    }

    nparsed = http_parser_execute(reqstate->parser, &parser_settings, ph_buf_mem(buf), ph_buf_len(buf));
    if ((uint64_t)nparsed != ph_buf_len(buf)) {
      if (strncmp(ph_buf_mem(buf), HEADER_END, ph_buf_len(buf)) == 0) {
        // http_parser doesn't handle the header end line entirely, so it's indistinguishable
        // from a real error.  catch it here, and move on.
        ph_buf_delref(buf);
        return;
      }

      ph_log(PH_LOG_ERR, "Failed parsing request line: parsed %u out of %u bytes", nparsed, ph_buf_len(buf));
    }

    ph_buf_delref(buf);
  }
}

static int xx_handle_url(http_parser *parser, const char *c, size_t len)
{
  ph_sock_t *sock = parser->data;
  request_state *reqstate = sock->job.data;

  ph_string_append_buf(reqstate->requested_file, c, (uint32_t)len);
}

static int xx_handle_request_parsed(http_parser *parser)
{
  ph_sock_t *sock = parser->data;
  request_state *reqstate = sock->job.data;

  reqstate->state = XX_REQUEST_PARSED;
}

static void xx_send_file(ph_job_t *job, ph_iomask_t why, void *data)
{
  ph_job_free(job);

  ph_sock_t *sock = job->data;
  request_state *reqstate = sock->job.data;

  ph_stream_t *f = ph_stm_file_open("foobar", O_RDONLY, 0);
  if (!f) {
    ph_log(PH_LOG_ERR, "Failed to open 'foobar'!");
    return;
  }

  struct stat st;
  stat("foobar", &st);

  ph_stm_printf(sock->stream, "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: text/plain\r\nContent-Length: %u\r\n\r\n", st.st_size);

  uint64_t nread, nwrote;
  if (!ph_stm_copy(f, sock->stream, PH_STREAM_READ_ALL, &nread, &nwrote)) {
    ph_log(PH_LOG_ERR, "Failed to copy file to socket!");

    ph_stm_close(f);
    return;
  }

  ph_stm_flush(sock->stream);
  ph_stm_close(f);

  reqstate->state = XX_RESPONSE_QUEUED;

  ph_sock_wakeup(sock);
}

static void xx_handle_event(ph_sock_t *sock, ph_iomask_t why, void *data)
{
  request_state *reqstate = data;

  if (why & (PH_IOMASK_ERR | PH_IOMASK_TIME)) {
done:
    xx_clean_request_state(reqstate);
    xx_tear_down_socket(sock);
    return;
  }

  // try and parse any data right off the bat.
  xx_read_request(sock, reqstate);

  switch (reqstate->state) {
    case XX_ERROR:
      {
        goto done;
        break;
      }
    case XX_PARSING_REQUEST:
      {
        // nothing to do here.  we're still waiting on data.
        break;
      }
    case XX_REQUEST_PARSED:
      {
        ph_job_t *file_send_job = ph_job_alloc(&file_send_job_def);
        file_send_job->data = sock;

        // disable the socket while we wait for the file i/o to finish
        ph_sock_enable(sock, false);

        // queue our file i/o work
        ph_job_set_pool(file_send_job, file_send_pool);
        break;
      }
    case XX_RESPONSE_QUEUED:
      {
        // re-enable our socket now that we've done our writing
        ph_sock_enable(sock, true);
        reqstate->state = XX_RESPONDING;
        ph_sock_wakeup(sock);
      break;
      }
    case XX_RESPONDING:
      {
        if (ph_bufq_len(sock->wbuf) == 0 && !ph_job_has_pending_wakeup(&sock->job)) {
          goto done;
        }
        break;
      }
  }
}

static void xx_accept_connection(ph_listener_t *lstn, ph_sock_t *sock)
{
  ph_unused_parameter(lstn);

  request_state *reqstate = xx_get_request_state(sock);

  sock->job.data = reqstate;
  sock->callback = xx_handle_event;

  ph_sock_enable(sock, true);
}

int main(int argc, char **argv)
{
  uint16_t portno = 9999;
  char *addrstring = NULL;
  ph_sockaddr_t addr;
  ph_listener_t *listen;

  ph_library_init();
  ph_log_level_set(PH_LOG_INFO);

  // Set up the address that we're going to listen on
  if (ph_sockaddr_set_v6(&addr, addrstring, portno) != PH_OK) {
    ph_fdprintf(STDERR_FILENO, "Invalid address [%s]:%d", addrstring ? addrstring : "*", portno);
    exit(EX_USAGE);
  }

  mt_reqstate = ph_memtype_register(&mt_reqstate_def);
  mt_parser = ph_memtype_register(&mt_parser_def);
  mt_string = ph_memtype_register(&mt_string_def);

  file_send_pool = ph_thread_pool_define("file_send", 10240, 2);

  ph_nbio_init(2);
  ph_debug_console_start("/tmp/phenom-debug-console");

  listen = ph_listener_new("file-server", xx_accept_connection);
  ph_listener_bind(listen, &addr);
  ph_listener_enable(listen, true);

  ph_log(PH_LOG_INFO, "Listening on `P{sockaddr:%p}...", (void*)&addr);

  ph_sched_run();

  return 0;
}

/* vim:ts=2:sw=2:et:
 */

server.h

#include "phenom/job.h"
#include "phenom/socket.h"
#include "phenom/listener.h"
#include "http_parser.h"

#ifndef server_h
#define server_h

typedef uint64_t client_state;

#define XX_ERROR 0
#define XX_PARSING_REQUEST 1
#define XX_REQUEST_PARSED 2
#define XX_RESPONSE_QUEUED 4
#define XX_RESPONDING 86

typedef struct {
    client_state state;
    http_parser *parser;
    ph_string_t *requested_file;
} request_state;

static void xx_tear_down_socket(ph_sock_t *sock);
static request_state *xx_get_request_state(ph_sock_t *sock);
static void xx_clean_request_state(request_state *reqstate);
static int xx_handle_url(http_parser *parser, const char *c, size_t len);
static int xx_handle_request_parsed(http_parser *parser);
static void xx_send_file(ph_job_t *job, ph_iomask_t why, void *data);
static void xx_handle_event(ph_sock_t *sock, ph_iomask_t why, void *data);
static void xx_accept_connection(ph_listener_t *lstn, ph_sock_t *sock);

#endif