windows.jai

/* -farzher September 13, 2021
  http server using raw sockets

  i'm not sure exactly how tcp sockets are working here.
  it seems to reuse the same socket for some calls, but
  other times it'll close the socket and accept a new one

  known issues:
  1. the server requests a thread for every connection
     and if you have more connections than threads the new
     connections will get completely stuck while waiting
  2. todo using async polling apis instead of blocking api
*/

CONCURRENCY_ENUM :: enum u8 {THREAD_PER_REQUEST; THREAD_POOL; ASYNC;}
CONCURRENCY : CONCURRENCY_ENUM : .THREAD_PER_REQUEST;
#assert(CONCURRENCY != .ASYNC);




ListenThreadData   :: struct {ip:string; port:u16;};
listen_thread_data : ListenThreadData;
listen_thread      : Thread;
listen_called      : bool;
http_listen :: (ip: string = "0.0.0.0", port: u16 = 80) {
  listen_called = true;
  listen_thread_data.ip = ip;
  listen_thread_data.port = port;
  thread_init(*listen_thread, listen_thread_proc);
  thread_start(*listen_thread);
}

https_listen :: (port: u16 = 443, ip: string = "0.0.0.0", certificate: string = "", privatekey: string = "") {
  die("https_listen is not implement on windows");
}

#if CONCURRENCY == .THREAD_POOL {
  BACKLOG               :: 12;
  THREAD_POOL_COUNT     :: BACKLOG;
  connection_thread_pool: [THREAD_POOL_COUNT] Thread;
  connection_queue      : [..] MySocket;
  connection_queue_sema : Semaphore;
  connection_queue_lock : Mutex;
  work_add :: (work: MySocket) {
    lock(*connection_queue_lock);
    array_add(*connection_queue, work);
    unlock(*connection_queue_lock);
  }
  work_get :: () -> MySocket {
    lock(*connection_queue_lock);
    socket: MySocket = connection_queue[0];
    array_unordered_remove_by_index(*connection_queue, 0);
    unlock(*connection_queue_lock);
    return socket;
  }
} else {
  BACKLOG :: 1024;
}

listen_thread_proc :: (listen_thread: *Thread) -> s64 {
  ts: Temporary_Storage; context.temporary_storage = *ts;

  socket_init(); defer socket_uninit();

  listen_socket := socket_new();
  // listen_socket := socket_new(overlaped=true);
  defer closesocket(listen_socket);


  thread_data := listen_thread_data;

  bind(listen_socket, thread_data.ip, thread_data.port);
  listen(listen_socket, BACKLOG);
  ip_to_display := ifx thread_data.ip == "0.0.0.0" then "localhost" else thread_data.ip;
  log("HTTP Server listening on http://%:%", ip_to_display, thread_data.port);


  #if CONCURRENCY == .THREAD_POOL {
    array_reserve(*connection_queue, BACKLOG);
    init(*connection_queue_sema);
    init(*connection_queue_lock);
    for * thread: connection_thread_pool {
      thread_init(thread, thread_pool_proc);
      thread_start(thread);
    }
  }

  // iocp := CreateIoCompletionPort(INVALID_HANDLE_VALUE, null, 0, 0);

  while 1 {
    client_socket := accept(listen_socket);

    #if CONCURRENCY == .THREAD_POOL {
      work_add(client_socket);
      release(connection_queue_sema);
    } else if CONCURRENCY == .THREAD_PER_REQUEST {
      threads: [..]Thread; // leak
      thread: *Thread = array_add(*threads);
      thread.data = xx client_socket.socket;
      thread_init(thread, thread_request_proc);
      thread_start(thread);
    }

    tfree();
  }

  return 0;
}

thread_pool_proc :: (thread: *Thread) -> s64 {
  ts: Temporary_Storage; context.temporary_storage = *ts;

  while 1 {
    wait_for(*connection_queue_sema);
    client_socket := work_get();
    defer closesocket(client_socket);

    if client_socket.socket == 0  mylog("wut? socket 0"); // i don't think this has ever happened

    while 1 {
      request := recv_http_request(client_socket);
      if request.err then break;

      path_normal := trim(request.path, "/");

      // let the user's middleware handle the request!
      for middleware: middlewares {
        if middleware.method != .ANY && middleware.method != request.method then continue;
        if middleware.path != "*" && middleware.path != path_normal then continue;

        middleware.cb(*request);
        if request.response_sent then break; // once a middleware handles the request, we stop
      }

      // if none of the middleware responded to this request, 404
      if !request.response_sent then send_html(*request, tprint("<pre>Cannot % %</pre>", request.method, request.path), status_code=404);

      tfree();
    }

    tfree();
  }

  return 0;
}
thread_request_proc :: (thread: *Thread) -> s64 {
  ts: Temporary_Storage; context.temporary_storage = *ts;
  client_socket: MySocket;
  client_socket.socket = xx thread.data;
  defer closesocket(client_socket);

  while 1 {
    auto_release_temp();

    request := recv_http_request(client_socket);
    if request.err then break;

    handle_request(*request);
  }

  return 0;
}













recv_http_request :: (socket: MySocket) -> Request {
  request: Request;
  request.params.allocator = tallocator;

  request.socket         = socket;

  request.headers.allocator = tallocator;
  init(*request.headers);

  BUFFER_LEN :: 1024*10;
  request.raw.data = talloc(BUFFER_LEN);
  request.raw.count = 0;


  buffercursor := 0;
  startofline := 0;
  while 1 {
    bytes_received := recv(request.socket, request.raw.data + request.raw.count, xx(BUFFER_LEN - request.raw.count));

    if bytes_received == 0 { request.err = true; break; } // this maybe isn't an error? on connection that aren't keep-alive?
    if bytes_received < 0  { request.err = true; break; }

    // chunk: string;
    // chunk.data = request.raw.data + request.raw.count;
    // chunk.count = bytes_received;
    // mylog("chunk", chunk);

    if request.body.data then request.body.count += bytes_received;

    request.raw.count += bytes_received;
    bytes_read_this_chunk := 0;
    while request.body.data == null && buffercursor < request.raw.count - 1 { defer buffercursor+=1; defer bytes_read_this_chunk+=1;
      if request.raw[buffercursor] == #char "\r" && request.raw[buffercursor + 1] == #char "\n" {
        defer startofline = buffercursor + 2;

        line: string = ---;
        line.data = request.raw.data + startofline;
        line.count = buffercursor - startofline;

        if startofline == 0 {
          http_request_info := split(line, " ");
          request.method = to_METHOD(http_request_info[0]);
          request.path = http_request_info[1];
          // request.http_version = http_request_info[2];
          continue;
        }

        kv := split(line, ": ");
        table_add(*request.headers, kv[0], kv[1]);

        if kv[0] == {
          case "Content-Length";    request.content_length = xx to_integer(kv[1]);
          case "Sec-WebSocket-Key"; request.websocket_key  = kv[1];
        }

        if request.raw[buffercursor+2] == #char "\r" && request.raw[buffercursor+2 + 1] == #char "\n" {
          // double newline, start of body
          request.body.data = request.raw.data + buffercursor+4;
          request.body.count = bytes_received - bytes_read_this_chunk - 4;
          break;
        }
      }
    }

    if request.body.data && request.body.count == request.content_length then break;
  }

  return request;
}




#import,dir "modules/mybasic";
#import,dir "modules/mysocket";

#import "Windows";
#import "File";
#import "Thread";
release :: (sema: Semaphore) {ReleaseSemaphore(sema.event, 1, null); }