/*
 *
 * Copyright 2015 gRPC authors.
 *
 * 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 "src/core/ext/filters/client_channel/backup_poller.h"

#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include "src/core/ext/filters/client_channel/client_channel.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/support/env.h"
#include "src/core/lib/support/string.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/completion_queue.h"

#define DEFAULT_POLL_INTERVAL_MS 5000

typedef struct backup_poller {
  grpc_timer polling_timer;
  grpc_closure run_poller_closure;
  grpc_closure shutdown_closure;
  gpr_mu* pollset_mu;
  grpc_pollset* pollset;  // guarded by pollset_mu
  bool shutting_down;     // guarded by pollset_mu
  gpr_refcount refs;
  gpr_refcount shutdown_refs;
} backup_poller;

static gpr_once g_once = GPR_ONCE_INIT;
static gpr_mu g_poller_mu;
static backup_poller* g_poller = nullptr;  // guarded by g_poller_mu
// g_poll_interval_ms is set only once at the first time
// grpc_client_channel_start_backup_polling() is called, after that it is
// treated as const.
static int g_poll_interval_ms = DEFAULT_POLL_INTERVAL_MS;

static void init_globals() {
  gpr_mu_init(&g_poller_mu);
  char* env = gpr_getenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS");
  if (env != nullptr) {
    int poll_interval_ms = gpr_parse_nonnegative_int(env);
    if (poll_interval_ms == -1) {
      gpr_log(GPR_ERROR,
              "Invalid GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS: %s, "
              "default value %d will be used.",
              env, g_poll_interval_ms);
    } else {
      g_poll_interval_ms = poll_interval_ms;
    }
  }
  gpr_free(env);
}

static void backup_poller_shutdown_unref(grpc_exec_ctx* exec_ctx,
                                         backup_poller* p) {
  if (gpr_unref(&p->shutdown_refs)) {
    grpc_pollset_destroy(exec_ctx, p->pollset);
    gpr_free(p->pollset);
    gpr_free(p);
  }
}

static void done_poller(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
  backup_poller_shutdown_unref(exec_ctx, (backup_poller*)arg);
}

static void g_poller_unref(grpc_exec_ctx* exec_ctx) {
  if (gpr_unref(&g_poller->refs)) {
    gpr_mu_lock(&g_poller_mu);
    backup_poller* p = g_poller;
    g_poller = nullptr;
    gpr_mu_unlock(&g_poller_mu);
    gpr_mu_lock(p->pollset_mu);
    p->shutting_down = true;
    grpc_pollset_shutdown(exec_ctx, p->pollset,
                          GRPC_CLOSURE_INIT(&p->shutdown_closure, done_poller,
                                            p, grpc_schedule_on_exec_ctx));
    gpr_mu_unlock(p->pollset_mu);
    grpc_timer_cancel(exec_ctx, &p->polling_timer);
  }
}

static void run_poller(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
  backup_poller* p = (backup_poller*)arg;
  if (error != GRPC_ERROR_NONE) {
    if (error != GRPC_ERROR_CANCELLED) {
      GRPC_LOG_IF_ERROR("run_poller", GRPC_ERROR_REF(error));
    }
    backup_poller_shutdown_unref(exec_ctx, p);
    return;
  }
  gpr_mu_lock(p->pollset_mu);
  if (p->shutting_down) {
    gpr_mu_unlock(p->pollset_mu);
    backup_poller_shutdown_unref(exec_ctx, p);
    return;
  }
  grpc_error* err = grpc_pollset_work(exec_ctx, p->pollset, nullptr,
                                      grpc_exec_ctx_now(exec_ctx));
  gpr_mu_unlock(p->pollset_mu);
  GRPC_LOG_IF_ERROR("Run client channel backup poller", err);
  grpc_timer_init(exec_ctx, &p->polling_timer,
                  grpc_exec_ctx_now(exec_ctx) + g_poll_interval_ms,
                  &p->run_poller_closure);
}

void grpc_client_channel_start_backup_polling(
    grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties) {
  gpr_once_init(&g_once, init_globals);
  if (g_poll_interval_ms == 0) {
    return;
  }
  gpr_mu_lock(&g_poller_mu);
  if (g_poller == nullptr) {
    g_poller = (backup_poller*)gpr_zalloc(sizeof(backup_poller));
    g_poller->pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size());
    g_poller->shutting_down = false;
    grpc_pollset_init(g_poller->pollset, &g_poller->pollset_mu);
    gpr_ref_init(&g_poller->refs, 0);
    // one for timer cancellation, one for pollset shutdown
    gpr_ref_init(&g_poller->shutdown_refs, 2);
    GRPC_CLOSURE_INIT(&g_poller->run_poller_closure, run_poller, g_poller,
                      grpc_schedule_on_exec_ctx);
    grpc_timer_init(exec_ctx, &g_poller->polling_timer,
                    grpc_exec_ctx_now(exec_ctx) + g_poll_interval_ms,
                    &g_poller->run_poller_closure);
  }
  gpr_ref(&g_poller->refs);
  gpr_mu_unlock(&g_poller_mu);
  grpc_pollset_set_add_pollset(exec_ctx, interested_parties, g_poller->pollset);
}

void grpc_client_channel_stop_backup_polling(
    grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties) {
  if (g_poll_interval_ms == 0) {
    return;
  }
  grpc_pollset_set_del_pollset(exec_ctx, interested_parties, g_poller->pollset);
  g_poller_unref(exec_ctx);
}