text: /*
   * twemproxy - A fast and lightweight proxy for memcached protocol.
   * Copyright (C) 2011 Twitter, Inc.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  #include <stdlib.h>
  #include <unistd.h>
  #include <nc_core.h>
  #include <nc_conf.h>
  #include <nc_server.h>
  #include <nc_proxy.h>
  
  static uint32_t ctx_id; /* context generation */
  static uint32_t reserved_fds = 0;
  
  rstatus_t
  core_calc_connections(struct context *ctx)
  {
      int status;
      uint32_t threads_count;
      struct rlimit limit;
  
      ASSERT(ctx->role == NC_CONTEXT_ROLE_MASTER);
      ASSERT(ctx->workers != NULL);
  
      status = getrlimit(RLIMIT_NOFILE, &limit);
      if (status < 0) {
          log_error("getrlimit failed: %s", strerror(errno));
          return NC_ERROR;
      }
  
      threads_count = array_n(ctx->workers);
      reserved_fds = threads_count*(2 + 2) + RESERVED_FDS;
      ctx->max_nsconn = ctx->max_nsconn * threads_count - (threads_count - 1) * array_n(&ctx->pool) ;
  
      ctx->max_nfd = (uint32_t)limit.rlim_cur;
      ctx->max_ncconn = ctx->max_nfd - ctx->max_nsconn - reserved_fds;
      log_debug(LOG_NOTICE, "max fds %"PRIu32" max client conns %"PRIu32" "
                "max server conns %"PRIu32"", ctx->max_nfd, ctx->max_ncconn,
                ctx->max_nsconn);
  
      return NC_OK;
  }
  
  struct context *
  core_ctx_create(struct instance *nci, int role)
  {
      rstatus_t status;
      struct context *ctx;
  
      ctx = nc_alloc(sizeof(*ctx));
      if (ctx == NULL) {
          return NULL;
      }
      ctx->id = ++ctx_id;
      ctx->cf = NULL;
      ctx->stats = NULL;
      ctx->evb = NULL;
      array_null(&ctx->pool);
      ctx->max_timeout = nci->interval;
      ctx->timeout = ctx->max_timeout;
      ctx->max_nfd = 0;
      ctx->max_ncconn = 0;
      ctx->max_nsconn = 0;
      ctx->msgb = NULL;
      ctx->mbufb = NULL;
      ctx->cb = NULL;
      ctx->role = NC_CONTEXT_ROLE_NULL;
      ctx->workers = NULL;
      ctx->manager = NULL;
  
      /* parse and create configuration */
      ctx->cf = conf_create(nci->conf_filename);
      if (ctx->cf == NULL) {
          goto error;
      }
  
      /* initialize server pool from configuration */
      status = server_pool_init(&ctx->pool, &ctx->cf->pool, ctx);
      if (status != NC_OK) {
          goto error;
      }
  
      /* initialize event handling for client, proxy and server */
      ctx->evb = event_base_create(EVENT_SIZE, &core_core);
      if (ctx->evb == NULL) {
          goto error;
      }
  
      /* init the memory */
      status = nc_memory_init(ctx, nci->mbuf_chunk_size);
      if (status != NC_OK) {
          goto error;
      }
  
      /* create stats per server pool */
      ctx->stats = stats_create(ctx, nci->hostname);
      if (ctx->stats == NULL) {
          goto error;
      }
  
      /* init the connect base */
      status = nc_connect_init(ctx);
      if (status != NC_OK) {
          goto error;
      }
  
      if (role == NC_CONTEXT_ROLE_MASTER) {        
          /* initialize proxy per server pool */
          status = proxy_init(ctx);
          if (status != NC_OK) {
              goto error;
          }
  
          status = manage_init(ctx, nci->addr, nci->port);
          if (status != NC_OK) {
              goto error;
          }
      } else if (role == NC_CONTEXT_ROLE_WORKER) {    
          /* preconnect? servers in server pool */
          status = server_pool_preconnect(ctx);
          if (status != NC_OK) {
              goto error;
          }
      } else {
          NOT_REACHED();
      }
  
      ctx->role = role;
  
      log_debug(LOG_VVERB, "created ctx %p id %"PRIu32"", ctx, ctx->id);
  
      return ctx;
  
  error:
      
      core_ctx_destroy(ctx);
      return NULL;
  }
  
  void
  core_ctx_destroy(struct context *ctx)
  {
      log_debug(LOG_VVERB, "destroy ctx %p id %"PRIu32"", ctx, ctx->id);
      proxy_deinit(ctx);
      server_pool_disconnect(ctx);
      event_base_destroy(ctx->evb);
      stats_destroy(ctx->stats);
      server_pool_deinit(&ctx->pool);
      conf_destroy(ctx->cf);
      nc_memory_deinit(ctx);
      nc_connect_deinit(ctx);
      manage_deinit(ctx);
      nc_free(ctx);
  }
  
  static rstatus_t
  core_recv(struct context *ctx, struct conn *conn)
  {
      rstatus_t status;
  
      status = conn->recv(ctx, conn);
      if (status != NC_OK) {
          log_debug(LOG_INFO, "recv on %c %d failed: %s",
                    conn->client ? 'c' : (conn->proxy ? 'p' : 's'), conn->sd,
                    strerror(errno));
      }
  
      return status;
  }
  
  static rstatus_t
  core_send(struct context *ctx, struct conn *conn)
  {
      rstatus_t status;
  
      status = conn->send(ctx, conn);
      if (status != NC_OK) {
          log_debug(LOG_INFO, "send on %c %d failed: status: %d errno: %d %s",
                    conn->client ? 'c' : (conn->proxy ? 'p' : 's'), conn->sd,
                    status, errno, strerror(errno));
      }
  
      return status;
  }
  
  static void
  core_close(struct context *ctx, struct conn *conn)
  {
      rstatus_t status;
      char type, *addrstr;
  
      ASSERT(conn->sd > 0);
  
      if (conn->client) {
          type = 'c';
          addrstr = nc_unresolve_peer_desc(conn->sd);
      } else {
          type = conn->proxy ? 'p' : 's';
          addrstr = nc_unresolve_addr(conn->addr, conn->addrlen);
      }
      log_debug(LOG_DEBUG, "close %c %d '%s' on event %04"PRIX32" eof %d done "
                "%d rb %zu sb %zu%c %s", type, conn->sd, addrstr, conn->events,
                conn->eof, conn->done, conn->recv_bytes, conn->send_bytes,
                conn->err ? ':' : ' ', conn->err ? strerror(conn->err) : "");
  
      status = event_del_conn(ctx->evb, conn);
      if (status < 0) {
          log_warn("event del conn %c %d failed, ignored: %s",
                   type, conn->sd, strerror(errno));
      }
  
      conn->close(ctx, conn);
  }
  
  static void
  core_error(struct context *ctx, struct conn *conn)
  {
      rstatus_t status;
      char type = conn->client ? 'c' : (conn->proxy ? 'p' : 's');
  
      status = nc_get_soerror(conn->sd);
      if (status < 0) {
          log_warn("get soerr on %c %d failed, ignored: %s", type, conn->sd,
                    strerror(errno));
      }
      conn->err = errno;
  
      core_close(ctx, conn);
  }
  
  static void
  core_timeout(struct context *ctx)
  {
      for (;;) {
          struct msg *msg;
          struct conn *conn;
          int64_t now, then;
  
          msg = msg_tmo_min(ctx->msgb);
          if (msg == NULL) {
              ctx->timeout = ctx->max_timeout;
              return;
          }
  
          /* skip over req that are in-error or done */
  
          if (msg->error || msg->done) {
              msg_tmo_delete(msg);
              continue;
          }
  
          /*
           * timeout expired req and all the outstanding req on the timing
           * out server
           */
  
          conn = msg->tmo_rbe.data;
          then = msg->tmo_rbe.key;
  
          now = nc_msec_now();
          if (now < then) {
              int delta = (int)(then - now);
              ctx->timeout = MIN(delta, ctx->max_timeout);
              return;
          }
  
          log_debug(LOG_INFO, "req %"PRIu64" on s %d timedout", msg->id, conn->sd);
  
          msg_tmo_delete(msg);
          conn->err = ETIMEDOUT;
  
          core_close(ctx, conn);
      }
  }
  
  rstatus_t
  core_core(void *arg, uint32_t events)
  {
      rstatus_t status;
      struct conn *conn = arg;
      struct context *ctx;
  
      if (conn->owner == NULL) {
          log_warn("conn is already unrefed!");
          return NC_OK;
      }
  
      ctx = conn_to_ctx(conn);
  
      log_debug(LOG_VVERB, "event %04"PRIX32" on %c %d", events,
                conn->client ? 'c' : (conn->proxy ? 'p' : 's'), conn->sd);
  
      conn->events = events;
  
      /* error takes precedence over read | write */
      if (events & EVENT_ERR) {
          core_error(ctx, conn);
          return NC_ERROR;
      }
  
      /* read takes precedence over write */
      if (events & EVENT_READ) {
          status = core_recv(ctx, conn);
          if (status != NC_OK || conn->done || conn->err) {
              core_close(ctx, conn);
              return NC_ERROR;
          }
      }
  
      if (events & EVENT_WRITE) {
          status = core_send(ctx, conn);
          if (status != NC_OK || conn->done || conn->err) {
              core_close(ctx, conn);
              return NC_ERROR;
          }
      }
  
      return NC_OK;
  }
  
  rstatus_t
  core_loop(struct context *ctx)
  {
      int nsd;
  
      nsd = event_wait(ctx->evb, ctx->timeout);
      if (nsd < 0) {
          return nsd;
      }
  
      core_timeout(ctx);
  
      return NC_OK;
  }
  