//
// 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.
//

#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_CLIENT_CHANNEL_H
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_CLIENT_CHANNEL_H

#include <grpc/support/port_platform.h>

#include <map>
#include <memory>
#include <set>
#include <string>

#include "absl/status/status.h"
#include "absl/types/optional.h"

#include <grpc/support/log.h>

#include "src/core/ext/filters/client_channel/client_channel_factory.h"
#include "src/core/ext/filters/client_channel/config_selector.h"
#include "src/core/ext/filters/client_channel/dynamic_filters.h"
#include "src/core/ext/filters/client_channel/lb_policy.h"
#include "src/core/ext/filters/client_channel/resolver.h"
#include "src/core/ext/filters/client_channel/resolver_result_parsing.h"
#include "src/core/ext/filters/client_channel/retry_throttle.h"
#include "src/core/ext/filters/client_channel/service_config.h"
#include "src/core/ext/filters/client_channel/subchannel.h"
#include "src/core/ext/filters/client_channel/subchannel_pool_interface.h"
#include "src/core/lib/gprpp/sync.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/polling_entity.h"
#include "src/core/lib/iomgr/work_serializer.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/transport/connectivity_state.h"

//
// Client channel filter
//

// A client channel is a channel that begins disconnected, and can connect
// to some endpoint on demand. If that endpoint disconnects, it will be
// connected to again later.
//
// Calls on a disconnected client channel are queued until a connection is
// established.

// Channel arg key for server URI string.
#define GRPC_ARG_SERVER_URI "grpc.server_uri"

// Channel arg containing a pointer to the ClientChannel object.
#define GRPC_ARG_CLIENT_CHANNEL "grpc.internal.client_channel"

// Channel arg containing a pointer to the ServiceConfig object.
#define GRPC_ARG_SERVICE_CONFIG_OBJ "grpc.internal.service_config_obj"

// Max number of batches that can be pending on a call at any given
// time.  This includes one batch for each of the following ops:
//   recv_initial_metadata
//   send_initial_metadata
//   recv_message
//   send_message
//   recv_trailing_metadata
//   send_trailing_metadata
#define MAX_PENDING_BATCHES 6

namespace grpc_core {

class ClientChannel {
 public:
  static const grpc_channel_filter kFilterVtable;

  class LoadBalancedCall;

  // Returns the ClientChannel object from channel, or null if channel
  // is not a client channel.
  static ClientChannel* GetFromChannel(grpc_channel* channel);

  grpc_connectivity_state CheckConnectivityState(bool try_to_connect);

  // Starts a one-time connectivity state watch.  When the channel's state
  // becomes different from *state, sets *state to the new state and
  // schedules on_complete.  The watcher_timer_init callback is invoked as
  // soon as the watch is actually started (i.e., after hopping into the
  // client channel combiner).  I/O will be serviced via pollent.
  //
  // This is intended to be used when starting a watch from outside of C-core
  // via grpc_channel_watch_connectivity_state().  It should not be used
  // by other callers.
  void AddExternalConnectivityWatcher(grpc_polling_entity pollent,
                                      grpc_connectivity_state* state,
                                      grpc_closure* on_complete,
                                      grpc_closure* watcher_timer_init) {
    new ExternalConnectivityWatcher(this, pollent, state, on_complete,
                                    watcher_timer_init);
  }

  // Cancels a pending external watcher previously added by
  // AddExternalConnectivityWatcher().
  void CancelExternalConnectivityWatcher(grpc_closure* on_complete) {
    ExternalConnectivityWatcher::RemoveWatcherFromExternalWatchersMap(
        this, on_complete, /*cancel=*/true);
  }

  int NumExternalConnectivityWatchers() const {
    MutexLock lock(&external_watchers_mu_);
    return static_cast<int>(external_watchers_.size());
  }

  // Starts and stops a connectivity watch.  The watcher will be initially
  // notified as soon as the state changes from initial_state and then on
  // every subsequent state change until either the watch is stopped or
  // it is notified that the state has changed to SHUTDOWN.
  //
  // This is intended to be used when starting watches from code inside of
  // C-core (e.g., for a nested control plane channel for things like xds).
  void AddConnectivityWatcher(
      grpc_connectivity_state initial_state,
      OrphanablePtr<AsyncConnectivityStateWatcherInterface> watcher);
  void RemoveConnectivityWatcher(
      AsyncConnectivityStateWatcherInterface* watcher);

  RefCountedPtr<LoadBalancedCall> CreateLoadBalancedCall(
      const grpc_call_element_args& args, grpc_polling_entity* pollent,
      grpc_closure* on_call_destruction_complete);

 private:
  class CallData;
  class ResolverResultHandler;
  class SubchannelWrapper;
  class ClientChannelControlHelper;
  class ConnectivityWatcherAdder;
  class ConnectivityWatcherRemover;

  // Represents a pending connectivity callback from an external caller
  // via grpc_client_channel_watch_connectivity_state().
  class ExternalConnectivityWatcher : public ConnectivityStateWatcherInterface {
   public:
    ExternalConnectivityWatcher(ClientChannel* chand,
                                grpc_polling_entity pollent,
                                grpc_connectivity_state* state,
                                grpc_closure* on_complete,
                                grpc_closure* watcher_timer_init);

    ~ExternalConnectivityWatcher() override;

    // Removes the watcher from the external_watchers_ map.
    static void RemoveWatcherFromExternalWatchersMap(ClientChannel* chand,
                                                     grpc_closure* on_complete,
                                                     bool cancel);

    void Notify(grpc_connectivity_state state,
                const absl::Status& /* status */) override;

    void Cancel();

   private:
    // Adds the watcher to state_tracker_. Consumes the ref that is passed to it
    // from Start().
    void AddWatcherLocked()
        ABSL_EXCLUSIVE_LOCKS_REQUIRED(chand_->work_serializer_);
    void RemoveWatcherLocked()
        ABSL_EXCLUSIVE_LOCKS_REQUIRED(chand_->work_serializer_);

    ClientChannel* chand_;
    grpc_polling_entity pollent_;
    grpc_connectivity_state initial_state_;
    grpc_connectivity_state* state_;
    grpc_closure* on_complete_;
    grpc_closure* watcher_timer_init_;
    Atomic<bool> done_{false};
  };

  struct ResolverQueuedCall {
    grpc_call_element* elem;
    ResolverQueuedCall* next = nullptr;
  };
  struct LbQueuedCall {
    LoadBalancedCall* lb_call;
    LbQueuedCall* next = nullptr;
  };

  ClientChannel(grpc_channel_element_args* args, grpc_error_handle* error);
  ~ClientChannel();

  // Filter vtable functions.
  static grpc_error_handle Init(grpc_channel_element* elem,
                                grpc_channel_element_args* args);
  static void Destroy(grpc_channel_element* elem);
  static void StartTransportOp(grpc_channel_element* elem,
                               grpc_transport_op* op);
  static void GetChannelInfo(grpc_channel_element* elem,
                             const grpc_channel_info* info);

  // Note: Does NOT return a new ref.
  grpc_error_handle disconnect_error() const {
    return disconnect_error_.Load(MemoryOrder::ACQUIRE);
  }

  // Note: All methods with "Locked" suffix must be invoked from within
  // work_serializer_.

  void OnResolverResultChangedLocked(Resolver::Result result)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);
  void OnResolverErrorLocked(grpc_error_handle error)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void CreateOrUpdateLbPolicyLocked(
      RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config,
      Resolver::Result result) ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);
  OrphanablePtr<LoadBalancingPolicy> CreateLbPolicyLocked(
      const grpc_channel_args& args)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void UpdateStateAndPickerLocked(
      grpc_connectivity_state state, const absl::Status& status,
      const char* reason,
      std::unique_ptr<LoadBalancingPolicy::SubchannelPicker> picker)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void UpdateServiceConfigInControlPlaneLocked(
      RefCountedPtr<ServiceConfig> service_config,
      RefCountedPtr<ConfigSelector> config_selector,
      const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
      const char* lb_policy_name)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void UpdateServiceConfigInDataPlaneLocked()
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void CreateResolverLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);
  void DestroyResolverAndLbPolicyLocked()
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  grpc_error_handle DoPingLocked(grpc_transport_op* op)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void StartTransportOpLocked(grpc_transport_op* op)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  void TryToConnectLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(work_serializer_);

  // These methods all require holding resolution_mu_.
  void AddResolverQueuedCall(ResolverQueuedCall* call,
                             grpc_polling_entity* pollent)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(resolution_mu_);
  void RemoveResolverQueuedCall(ResolverQueuedCall* to_remove,
                                grpc_polling_entity* pollent)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(resolution_mu_);

  // These methods all require holding data_plane_mu_.
  void AddLbQueuedCall(LbQueuedCall* call, grpc_polling_entity* pollent)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(data_plane_mu_);
  void RemoveLbQueuedCall(LbQueuedCall* to_remove, grpc_polling_entity* pollent)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(data_plane_mu_);
  RefCountedPtr<ConnectedSubchannel> GetConnectedSubchannelInDataPlane(
      SubchannelInterface* subchannel) const
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(data_plane_mu_);

  //
  // Fields set at construction and never modified.
  //
  const bool deadline_checking_enabled_;
  const bool enable_retries_;
  grpc_channel_stack* owning_stack_;
  ClientChannelFactory* client_channel_factory_;
  const grpc_channel_args* channel_args_;
  RefCountedPtr<ServiceConfig> default_service_config_;
  std::string server_name_;
  UniquePtr<char> target_uri_;
  channelz::ChannelNode* channelz_node_;
  grpc_pollset_set* interested_parties_;

  //
  // Fields related to name resolution.  Guarded by resolution_mu_.
  //
  mutable Mutex resolution_mu_;
  // Linked list of calls queued waiting for resolver result.
  ResolverQueuedCall* resolver_queued_calls_ ABSL_GUARDED_BY(resolution_mu_) =
      nullptr;
  // Data from service config.
  grpc_error_handle resolver_transient_failure_error_
      ABSL_GUARDED_BY(resolution_mu_) = GRPC_ERROR_NONE;
  bool received_service_config_data_ ABSL_GUARDED_BY(resolution_mu_) = false;
  RefCountedPtr<ServiceConfig> service_config_ ABSL_GUARDED_BY(resolution_mu_);
  RefCountedPtr<ConfigSelector> config_selector_
      ABSL_GUARDED_BY(resolution_mu_);
  RefCountedPtr<DynamicFilters> dynamic_filters_
      ABSL_GUARDED_BY(resolution_mu_);

  //
  // Fields used in the data plane.  Guarded by data_plane_mu_.
  //
  mutable Mutex data_plane_mu_;
  std::unique_ptr<LoadBalancingPolicy::SubchannelPicker> picker_
      ABSL_GUARDED_BY(data_plane_mu_);
  // Linked list of calls queued waiting for LB pick.
  LbQueuedCall* lb_queued_calls_ ABSL_GUARDED_BY(data_plane_mu_) = nullptr;

  //
  // Fields used in the control plane.  Guarded by work_serializer.
  //
  std::shared_ptr<WorkSerializer> work_serializer_;
  ConnectivityStateTracker state_tracker_ ABSL_GUARDED_BY(work_serializer_);
  OrphanablePtr<Resolver> resolver_ ABSL_GUARDED_BY(work_serializer_);
  bool previous_resolution_contained_addresses_
      ABSL_GUARDED_BY(work_serializer_) = false;
  RefCountedPtr<ServiceConfig> saved_service_config_
      ABSL_GUARDED_BY(work_serializer_);
  RefCountedPtr<ConfigSelector> saved_config_selector_
      ABSL_GUARDED_BY(work_serializer_);
  absl::optional<std::string> health_check_service_name_
      ABSL_GUARDED_BY(work_serializer_);
  OrphanablePtr<LoadBalancingPolicy> lb_policy_
      ABSL_GUARDED_BY(work_serializer_);
  RefCountedPtr<SubchannelPoolInterface> subchannel_pool_
      ABSL_GUARDED_BY(work_serializer_);
  // The number of SubchannelWrapper instances referencing a given Subchannel.
  std::map<Subchannel*, int> subchannel_refcount_map_
      ABSL_GUARDED_BY(work_serializer_);
  // The set of SubchannelWrappers that currently exist.
  // No need to hold a ref, since the map is updated in the control-plane
  // work_serializer when the SubchannelWrappers are created and destroyed.
  std::set<SubchannelWrapper*> subchannel_wrappers_
      ABSL_GUARDED_BY(work_serializer_);
  // Pending ConnectedSubchannel updates for each SubchannelWrapper.
  // Updates are queued here in the control plane work_serializer and then
  // applied in the data plane mutex when the picker is updated.
  std::map<RefCountedPtr<SubchannelWrapper>, RefCountedPtr<ConnectedSubchannel>>
      pending_subchannel_updates_ ABSL_GUARDED_BY(work_serializer_);
  int keepalive_time_ ABSL_GUARDED_BY(work_serializer_) = -1;

  //
  // Fields accessed from both data plane mutex and control plane
  // work_serializer.
  //
  Atomic<grpc_error_handle> disconnect_error_;

  //
  // Fields guarded by a mutex, since they need to be accessed
  // synchronously via get_channel_info().
  //
  Mutex info_mu_;
  UniquePtr<char> info_lb_policy_name_ ABSL_GUARDED_BY(info_mu_);
  UniquePtr<char> info_service_config_json_ ABSL_GUARDED_BY(info_mu_);

  //
  // Fields guarded by a mutex, since they need to be accessed
  // synchronously via grpc_channel_num_external_connectivity_watchers().
  //
  mutable Mutex external_watchers_mu_;
  std::map<grpc_closure*, RefCountedPtr<ExternalConnectivityWatcher>>
      external_watchers_ ABSL_GUARDED_BY(external_watchers_mu_);
};

//
// ClientChannel::LoadBalancedCall
//

// This object is ref-counted, but it cannot inherit from RefCounted<>,
// because it is allocated on the arena and can't free its memory when
// its refcount goes to zero.  So instead, it manually implements the
// same API as RefCounted<>, so that it can be used with RefCountedPtr<>.
class ClientChannel::LoadBalancedCall
    : public RefCounted<LoadBalancedCall, PolymorphicRefCount, kUnrefCallDtor> {
 public:
  // If on_call_destruction_complete is non-null, then it will be
  // invoked once the LoadBalancedCall is completely destroyed.
  // If it is null, then the caller is responsible for checking whether
  // the LB call has a subchannel call and ensuring that the
  // on_call_destruction_complete closure passed down from the surface
  // is not invoked until after the subchannel call stack is destroyed.
  LoadBalancedCall(ClientChannel* chand, const grpc_call_element_args& args,
                   grpc_polling_entity* pollent,
                   grpc_closure* on_call_destruction_complete);
  ~LoadBalancedCall() override;

  void StartTransportStreamOpBatch(grpc_transport_stream_op_batch* batch);

  // Invoked by channel for queued LB picks when the picker is updated.
  static void PickSubchannel(void* arg, grpc_error_handle error);
  // Helper function for performing an LB pick while holding the data plane
  // mutex.  Returns true if the pick is complete, in which case the caller
  // must invoke PickDone() or AsyncPickDone() with the returned error.
  bool PickSubchannelLocked(grpc_error_handle* error)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(&ClientChannel::data_plane_mu_);
  // Schedules a callback to process the completed pick.  The callback
  // will not run until after this method returns.
  void AsyncPickDone(grpc_error_handle error);

  RefCountedPtr<SubchannelCall> subchannel_call() const {
    return subchannel_call_;
  }

 private:
  class LbQueuedCallCanceller;
  class Metadata;
  class LbCallState;

  // Returns the index into pending_batches_ to be used for batch.
  static size_t GetBatchIndex(grpc_transport_stream_op_batch* batch);
  void PendingBatchesAdd(grpc_transport_stream_op_batch* batch);
  static void FailPendingBatchInCallCombiner(void* arg,
                                             grpc_error_handle error);
  // A predicate type and some useful implementations for PendingBatchesFail().
  typedef bool (*YieldCallCombinerPredicate)(
      const CallCombinerClosureList& closures);
  static bool YieldCallCombiner(const CallCombinerClosureList& /*closures*/) {
    return true;
  }
  static bool NoYieldCallCombiner(const CallCombinerClosureList& /*closures*/) {
    return false;
  }
  static bool YieldCallCombinerIfPendingBatchesFound(
      const CallCombinerClosureList& closures) {
    return closures.size() > 0;
  }
  // Fails all pending batches.
  // If yield_call_combiner_predicate returns true, assumes responsibility for
  // yielding the call combiner.
  void PendingBatchesFail(
      grpc_error_handle error,
      YieldCallCombinerPredicate yield_call_combiner_predicate);
  static void ResumePendingBatchInCallCombiner(void* arg,
                                               grpc_error_handle ignored);
  // Resumes all pending batches on subchannel_call_.
  void PendingBatchesResume();

  static void RecvTrailingMetadataReadyForLoadBalancingPolicy(
      void* arg, grpc_error_handle error);
  void InjectRecvTrailingMetadataReadyForLoadBalancingPolicy(
      grpc_transport_stream_op_batch* batch);

  void CreateSubchannelCall();
  // Invoked when a pick is completed, on both success or failure.
  static void PickDone(void* arg, grpc_error_handle error);
  // Removes the call from the channel's list of queued picks if present.
  void MaybeRemoveCallFromLbQueuedCallsLocked()
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(&ClientChannel::data_plane_mu_);
  // Adds the call to the channel's list of queued picks if not already present.
  void MaybeAddCallToLbQueuedCallsLocked()
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(&ClientChannel::data_plane_mu_);

  ClientChannel* chand_;

  // TODO(roth): Instead of duplicating these fields in every filter
  // that uses any one of them, we should store them in the call
  // context.  This will save per-call memory overhead.
  grpc_slice path_;  // Request path.
  gpr_cycle_counter call_start_time_;
  grpc_millis deadline_;
  Arena* arena_;
  grpc_call_stack* owning_call_;
  CallCombiner* call_combiner_;
  grpc_call_context_element* call_context_;
  grpc_polling_entity* pollent_;
  grpc_closure* on_call_destruction_complete_;

  // Set when we get a cancel_stream op.
  grpc_error_handle cancel_error_ = GRPC_ERROR_NONE;

  // Set when we fail inside the LB call.
  grpc_error_handle failure_error_ = GRPC_ERROR_NONE;

  grpc_closure pick_closure_;

  // Accessed while holding ClientChannel::data_plane_mu_.
  ClientChannel::LbQueuedCall queued_call_
      ABSL_GUARDED_BY(&ClientChannel::data_plane_mu_);
  bool queued_pending_lb_pick_ ABSL_GUARDED_BY(&ClientChannel::data_plane_mu_) =
      false;
  LbQueuedCallCanceller* lb_call_canceller_
      ABSL_GUARDED_BY(&ClientChannel::data_plane_mu_) = nullptr;

  RefCountedPtr<ConnectedSubchannel> connected_subchannel_;
  const LoadBalancingPolicy::BackendMetricData* backend_metric_data_ = nullptr;
  std::function<void(grpc_error_handle, LoadBalancingPolicy::MetadataInterface*,
                     LoadBalancingPolicy::CallState*)>
      lb_recv_trailing_metadata_ready_;

  RefCountedPtr<SubchannelCall> subchannel_call_;

  // For intercepting recv_trailing_metadata_ready for the LB policy.
  grpc_metadata_batch* recv_trailing_metadata_ = nullptr;
  grpc_closure recv_trailing_metadata_ready_;
  grpc_closure* original_recv_trailing_metadata_ready_ = nullptr;

  // Batches are added to this list when received from above.
  // They are removed when we are done handling the batch (i.e., when
  // either we have invoked all of the batch's callbacks or we have
  // passed the batch down to the subchannel call and are not
  // intercepting any of its callbacks).
  grpc_transport_stream_op_batch* pending_batches_[MAX_PENDING_BATCHES] = {};
};

}  // namespace grpc_core

#endif  // GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_CLIENT_CHANNEL_H