/*
 * Copyright (c) 1999-2001 Caucho Technology.  All rights reserved.
 *
 * Caucho Technology permits redistribution, modification and use
 * of this file in source and binary form ("the Software") under the
 * Caucho Developer Source License ("the License").  The following
 * conditions must be met:
 *
 * 1. Each copy or derived work of the Software must preserve the copyright
 *    notice and this notice unmodified.
 *
 * 2. Redistributions of the Software in source or binary form must include 
 *    an unmodified copy of the License, normally in a plain ASCII text
 *
 * 3. The names "Resin" or "Caucho" are trademarks of Caucho Technology and
 *    may not be used to endorse products derived from this software.
 *    "Resin" or "Caucho" may not appear in the names of products derived
 *    from this software.
 *
 * 4. Caucho Technology requests that attribution be given to Resin
 *    in any manner possible.  We suggest using the "Resin Powered"
 *    button or creating a "powered by Resin(tm)" link to
 *    http://www.caucho.com for each page served by Resin.
 *
 * This Software is provided "AS IS," without a warranty of any kind. 
 * ALL EXPRESS OR IMPLIED REPRESENTATIONS AND WARRANTIES, INCLUDING ANY
 * IMPLIED WARRANTY OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE
 * OR NON-INFRINGEMENT, ARE HEREBY EXCLUDED.
 *
 * CAUCHO TECHNOLOGY AND ITS LICENSORS SHALL NOT BE LIABLE FOR ANY DAMAGES
 * SUFFERED BY LICENSEE OR ANY THIRD PARTY AS A RESULT OF USING OR
 * DISTRIBUTING SOFTWARE. IN NO EVENT WILL Caucho OR ITS LICENSORS BE LIABLE
 * FOR ANY LOST REVENUE, PROFIT OR DATA, OR FOR DIRECT, INDIRECT, SPECIAL,
 * CONSEQUENTIAL, INCIDENTAL OR PUNITIVE DAMAGES, HOWEVER CAUSED AND
 * REGARDLESS OF THE THEORY OF LIABILITY, ARISING OUT OF THE USE OF OR
 * INABILITY TO USE SOFTWARE, EVEN IF HE HAS BEEN ADVISED OF THE POSSIBILITY
 * OF SUCH DAMAGES.      
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>
#include <dirent.h>
#include <time.h>
#include <signal.h>
/* probably system-dependent */
#include <dlfcn.h>
#include <jni.h>
#include <locale.h>

#include "cse.h"
#include "proxy.h"
#include "proxy_jni.h"

#define RING_SIZE 32
#define HEADER_SIZE (16 * 1024 - 256)

typedef struct t_java_req {
  jobject request_obj;
  jbyteArray uri;
  jcharArray header_buffer;
  jintArray header_offsets;
} t_java_req;

struct server_t {
  registry_t *node;
  int ss;

  JNIEnv *env;
  jobject server_obj;

  int is_dead;

  pthread_t accept_thread;

  pthread_mutex_t lock;
  pthread_cond_t accept_cond;

  pthread_mutex_t ssl_lock;

  /* Socket ring buffer from the accept thread to the connections */
  int accept_sockets[RING_SIZE];
  /* Address ring buffer from the accept thread to the connections */
  struct sockaddr_in accept_addresses[RING_SIZE];
  /* Head of the ring buffer */
  int accept_head;
  /* Tail of the ring buffer */
  int accept_tail;

  /* Maximum number of connections allowed waiting for an accept */
  int accept_max;
  /* Current number of connections waiting for an accept */
  int accept_count;
  /* Current number of connections being created */
  int create_count;

  /* Maximum number of connections */
  int connection_max;
  /* Current number of connections */
  int connection_count;

  /* Time of the last free */
  int free_time;

  /* Debug id */
  int connection_id;
  
  connection_t *conn_head;

  void *context;
  connection_ops_t *ops;
};

int g_now_seconds;

static int std_open(connection_t *conn, int fd);
static int std_read(connection_t *conn, char *buf, int len);
static int std_write(connection_t *conn, char *buf, int len);
static int std_close(connection_t *conn);
static void std_free(connection_t *conn);

struct connection_ops_t std_ops = {
  std_open,
  std_read,
  std_write,
  std_close,
  std_free,
};

void close_server(server_t *server);

static int
std_open(connection_t *conn, int fd)
{
  conn->fd = fd;
}

static int
std_read(connection_t *conn, char *buf, int len)
{
  return recv(conn->fd, buf, len, 0);
}

static int
std_write(connection_t *conn, char *buf, int len)
{
  return write(conn->fd, buf, len);
}

static int
std_close(connection_t *conn)
{
  int fd = conn->fd;
  conn->fd = -1;

  if (fd > 0)
    return close(fd);
  else
    return -1;
}

static void
std_free(connection_t *conn)
{
}

void
handler(int signum)
{
  resin_close_all();
  fprintf(stderr, "Segmentation Violation (%d) pid:%d\n", signum, getpid());
  exit(11);
}

static void
fill_fields(JNIEnv *env, jobject server_obj, t_java_req *java_req)
{
  jobject request_obj;
  
  request_obj = (*env)->CallObjectMethod(env, server_obj,
                                         g_java.create_request);
  if ((*env)->ExceptionOccurred(env)) {
    (*env)->ExceptionDescribe(env);
    exit(1);
  }
  request_obj = (*env)->NewGlobalRef(env, request_obj);
  
  java_req->request_obj = request_obj;

  java_req->uri = (*env)->GetObjectField(env, request_obj, g_java.f_uri);
  
  java_req->header_buffer = (*env)->GetObjectField(env, request_obj,
                                                   g_java.f_header_buffer);
  java_req->header_offsets = (*env)->GetObjectField(env, request_obj,
                                                    g_java.f_header_offsets);
}

/**
 * Returns true if the server will allow a HTTP/1.1 keepalive
 */
int
server_allow_keepalive(server_t *server)
{
  return (server->connection_count < server->connection_max - 50 ||
          server->accept_count >= server->accept_max);
}

/**
 * Handle a connection from a user.
 *
 * @param sock the opened socket.
 * @param req the request object
 * @param res the response object
 * @param env the Java thread environment
 * @param java_req pointers to the java request.
 */
static void
do_request(connection_t *conn, request_t *req, response_t *res,
           JNIEnv *env, t_java_req *java_req)
{
  int len;
  char buf[4096];
  char *uri;
  char *ptr;
  short *headers;
  char *header_ptr;
  int *header_offsets;
  int header_start;
  jobject request_obj = java_req->request_obj;
  jint local_addr;
  jint local_port;
  jint remote_addr;
  jint remote_port;
  jint ssl_bits;
    
  int i, j, k;
  int result;

  local_addr = ntohl(req->server_addr.sin_addr.s_addr);
  local_port = ntohs(req->server_addr.sin_port);
  remote_addr = ntohl(req->remote_addr.sin_addr.s_addr);
  remote_port = ntohs(req->remote_addr.sin_port);
  ssl_bits = conn->ssl_bits;

  do {
    response_start(res);
    if (request_parse(req) < 0)
      break;

    uri = (*env)->GetPrimitiveArrayCritical(env, java_req->uri, 0);
    ptr = req->uri;
    for (len = 0; ptr[len]; len++)
      uri[len] = ptr[len];

    (*env)->ReleasePrimitiveArrayCritical(env, java_req->uri, uri, 0);
    (*env)->SetIntField(env, request_obj, g_java.f_uri_length, len);

    headers = (*env)->GetPrimitiveArrayCritical(env, java_req->header_buffer, 0);
    header_offsets = (*env)->GetPrimitiveArrayCritical(env,
                                                       java_req->header_offsets,
                                                       0);

    i = 0;
    j = 0;
  
    header_offsets[j++] = 0;
    for (k = 0; req->method[k] && i < HEADER_SIZE; k++)
      headers[i++] = req->method[k];
  
    header_offsets[j++] = i;
    for (k = 0; req->protocol[k] && i < HEADER_SIZE; k++)
      headers[i++] = req->protocol[k];

    header_offsets[j++] = i;
    for (k = 0; req->host && req->host[k] && i < HEADER_SIZE; k++)
      headers[i++] = req->host[k];

    header_offsets[j++] = i;
    for (k = 0;
         conn->ssl_cipher && conn->ssl_cipher[k] && i < HEADER_SIZE;
         k++)
      headers[i++] = conn->ssl_cipher[k];
    header_offsets[j] = i;

    header_start = i;
  
    len = req->header_length;
    header_ptr = req->headers;
    for (k = 0; k < len && i < HEADER_SIZE; k++)
      headers[i++] = header_ptr[k];
  
    len = req->header_size;
    for (k = 0; k <= len; k++) {
      header_offsets[j++] = header_start + req->header_keys[k];
      header_offsets[j++] = header_start + req->header_values[k];
    }
  
    (*env)->ReleasePrimitiveArrayCritical(env, java_req->header_buffer,
                                          headers, 0);
  
    (*env)->ReleasePrimitiveArrayCritical(env, java_req->header_offsets,
                                          header_offsets, 0);
    (*env)->SetIntField(env, request_obj, g_java.f_header_size, len);

    result = (*env)->CallBooleanMethod(env, request_obj, g_java.dispatch,
                                       req, local_addr, local_port,
                                       remote_addr, remote_port, ssl_bits);

    if (! result) {
      LOG(("dead\n"));
      close_server(req->server);
    }

    if ((*env)->ExceptionOccurred(env)) {
      (*env)->ExceptionDescribe(env);
    }

    cse_flush_request(res);
  } while (! req->server->is_dead && req->keepalive && conn->fd > 0);

  conn->ops->close(conn);
}

/**
 * Creates a new connection.  If there are already too many connections,
 * return null.
 */
static connection_t *
connection_create(server_t *server)
{
  connection_t *conn;
  int id;
  int count;

  pthread_mutex_lock(&server->lock);
  count = server->connection_count++;
  if (server->connection_count > server->connection_max) {
    pthread_mutex_unlock(&server->lock);
    return 0;
  }
  id = server->connection_id++;
  pthread_mutex_unlock(&server->lock);

  /*
  fprintf(stderr, "alloc conn id=%d count=%d\n", id, count);
  */
  
  conn = (connection_t *) cse_malloc(sizeof(connection_t));
  if (! conn)
    return 0;
  memset(conn, 0, sizeof(connection_t));
  
  conn->id = id;
  conn->server = server;
  conn->request = request_create();
  if (! conn->request) {
    cse_free(conn);
    return 0;
  }
  conn->response = response_create();
  if (! conn->response) {
    request_free(conn->request);
    cse_free(conn);
    return 0;
  }
  conn->request->server = server;

  conn->request->response = conn->response;
  conn->response->request = conn->request;
  
  conn->request->conn = conn;
  conn->response->conn = conn;

  conn->context = server->context;
  conn->ops = server->ops;
  conn->fd = -1;
  conn->ssl_lock = &server->ssl_lock;

  pthread_mutex_lock(&server->lock);
  
  conn->next = server->conn_head;
  if (server->conn_head)
    server->conn_head->prev = conn;
  server->conn_head = conn;
  
  pthread_mutex_unlock(&server->lock);

  LOG(("create id=%d count=%d accept-count=%d create-count=%d ring=%d\n",
       id, count,
       server->accept_count, server->create_count,
       (server->accept_head - server->accept_tail + RING_SIZE) % RING_SIZE));

  return conn;
}

/**
 * Frees the connection.
 *
 * @param conn the connection to be freed.
 */
static void
connection_free(connection_t *conn)
{
  int count;
  connection_t *next;
  connection_t *prev;

  if (! conn)
    return;
  
  pthread_mutex_lock(&conn->server->lock);
  count = conn->server->connection_count--;

  next = conn->next;
  prev = conn->prev;

  if (next)
    next->prev = prev;
  if (prev)
    prev->next = next;
  else
    conn->server->conn_head = next;
  
  pthread_mutex_unlock(&conn->server->lock);

  LOG(("free id=%d count=%d accept-count=%d\n",
       conn->id, count, conn->server->accept_count));

  if (conn->request)
    request_free(conn->request);
  conn->request = 0;
  if (conn->response)
    response_free(conn->response);
  conn->response = 0;

  if (conn->ops)
    conn->ops->free(conn);
  
  cse_free(conn);
}

/**
 * Main thread loop for the request
 *
 * @param v_server pointer to the server object
 */
static void *
request_run(void *v_connection)
{
  connection_t *conn = v_connection;
  server_t *server = conn->server;
  request_t *req = conn->request;
  response_t *res = conn->response;
  t_java_req java_req;
  JNIEnv *env;
  int result;
  struct timespec wait_time;

  setlocale(LC_TIME, "POSIX");
  setlocale(LC_ALL, "POSIX");

  LOG(("new connection id:%d conn-count:%d accept-count:%d create-count:%d ring:%d\n",
       conn->id, server->connection_count, server->accept_count, server->create_count,
       (server->accept_head - server->accept_tail + RING_SIZE) % RING_SIZE));
  
  pthread_mutex_lock(&server->lock);
  server->create_count--;
  pthread_mutex_unlock(&server->lock);

  result = (*g_java.vm)->AttachCurrentThread(g_java.vm, (void**)&env, 0);

  /* exit on any segv */
  /* signal(SIGSEGV, handler); */

  fill_fields(env, server->server_obj, &java_req);
  
  wait_time.tv_sec = 1;
  wait_time.tv_nsec = 0;

  while (! server->is_dead) {
    struct sockaddr_in *sin;
    int sock = -1;
    int tail;

    if (server->accept_count > server->accept_max) {
      int now = time(0);
      int is_dead = 0;
      
      if ((now >> 2) != (server->free_time >> 2)) {
        server->free_time = now;
        is_dead = 1;
      }

      if (is_dead)
        break;
    }

    pthread_mutex_lock(&server->lock);
    tail = server->accept_tail;

    if (server->accept_head != tail) {
      sock = server->accept_sockets[tail];
      req->remote_addr = server->accept_addresses[tail];
      server->accept_tail = (tail + 1) % RING_SIZE;
    }
    else {
      int now = time(0);
      struct timespec time;

      time.tv_sec = now + 5;
      time.tv_nsec = 0;
      
      server->accept_count++;
      pthread_cond_timedwait(&server->accept_cond, &server->lock, &time);
      server->accept_count--;
    }
    pthread_mutex_unlock(&server->lock);

    if (sock >= 0) {
      int len = sizeof(req->server_addr);
      getsockname(sock, (struct sockaddr *) &req->server_addr, &len);

      if (conn->ops && conn->ops->open(conn, sock) >= 0) {
        req->offset = 0;
        req->length = 0;
        res->length = 0;
        do_request(conn, req, res, env, &java_req);
      }
    }
  }

  if (java_req.request_obj)
    (*env)->DeleteGlobalRef(env, java_req.request_obj);

  (*g_java.vm)->DetachCurrentThread(g_java.vm);

  connection_free(conn);

  return 0;
}

static int
start_connection(server_t *server)
{
  connection_t *conn;
  int result;

  pthread_mutex_lock(&server->lock);
  server->create_count++;
  pthread_mutex_unlock(&server->lock);
    
  conn = connection_create(server);

  if (conn) {
    result = pthread_create(&conn->thread, 0, request_run, conn);

    if (! result)
      return 1;

    connection_free(conn);
    
    pthread_mutex_lock(&server->lock);
    server->create_count--;
    pthread_mutex_unlock(&server->lock);
  }
  else {
    pthread_mutex_lock(&server->lock);
    server->connection_count--;
    server->create_count--;
    pthread_mutex_unlock(&server->lock);
  }
    
  LOG(("fail create\n"));

  return 0;
}

/**
 * Main loop for the accept server.
 *
 * @param server the server variable
 */
static void *
accept_run(void *v_server)
{
  server_t *server = v_server;
  while (! server->is_dead) {
    int head = server->accept_head;
    int next_head = (head + 1) % RING_SIZE;
    struct sockaddr_in sin;
    int len = sizeof(sin);
    int sock;
    int needs_new_connection = 0;

    memset(&sin, 0, sizeof(sin));
    sock = accept(server->ss, (struct sockaddr *) &sin, &len);

    if (sock < 0)
      continue;

    /* If overflow, kill it quickly */
    if (next_head == server->accept_tail) {
      LOG(("Overflow\n"));
      close(sock);
      continue;
    }

    server->accept_sockets[head] = sock;
    server->accept_addresses[head] = sin;

    pthread_mutex_lock(&server->lock);
    server->accept_head = next_head;
    
    if (server->accept_count + server->create_count > 0)
      pthread_cond_signal(&server->accept_cond);
    else
      needs_new_connection = 1;
    
    pthread_mutex_unlock(&server->lock);

    while (needs_new_connection) {
      struct timeval wait_time;

      wait_time.tv_sec = 0;
      wait_time.tv_usec = 10000;

      select(0, 0, 0, 0, &wait_time);

      needs_new_connection = 0;
      
      pthread_mutex_lock(&server->lock);
    
      if (server->accept_head == server->accept_tail) {
      }
      else if (server->accept_count + server->create_count > 0)
        pthread_cond_signal(&server->accept_cond);
      else
        needs_new_connection = 1;
    
      pthread_mutex_unlock(&server->lock);

      if (needs_new_connection && start_connection(server))
        needs_new_connection = 0;
    }
  }

  return 0;
}

/**
 * Starts the server
 *
 * @param ss the server socket file descriptor
 * @param env the JDK environment
 * @param server_obj the Java JniServer object
 *
 * @return the C-based server object.
 */
server_t *
start_server(registry_t *node, int ss, JNIEnv *env, jobject server_obj)
{
  server_t *server = cse_malloc(sizeof(server_t));
  int result;
  
  memset(server, 0, sizeof(server_t));
  server->node = node;
  server->ss = ss;
  server->env = env;
  server->accept_max = 5;
  server->connection_max = 100;
  server->server_obj = server_obj;
  server->ops = &std_ops;

  if (node && cse_next_link(node->first, "ssl")) {
    server->context = ssl_create(node);
    if (! server->context) {
      fprintf(stderr, "SSL isn't supported.\n");
      exit(1);
    }
    server->ops = ssl_get_ops();
  }

  pthread_mutex_init(&server->lock, 0);
  pthread_cond_init(&server->accept_cond, 0);

  g_now_seconds = time(0);
    
  result = pthread_create(&server->accept_thread, 0, accept_run, server);

  return server;
}

int
server_is_dead(server_t *server)
{
  return ! server || server->is_dead;
}

void
close_server(server_t *server)
{
  fprintf(stderr, "closing server\n");
  
  server->is_dead = 1;

  pthread_mutex_lock(&server->lock);
  pthread_cond_broadcast(&server->accept_cond);
  pthread_mutex_unlock(&server->lock);
}
