content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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.
   */
  
  package org.apache.cassandra.transport;
  
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.TimeUnit;
  import java.util.function.Consumer;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.base.Predicate;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import io.netty.channel.Channel;
  import io.netty.channel.EventLoop;
  import io.netty.util.AttributeKey;
  import org.apache.cassandra.concurrent.DebuggableTask;
  import org.apache.cassandra.concurrent.LocalAwareExecutorPlus;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.exceptions.OverloadedException;
  import org.apache.cassandra.metrics.ClientMetrics;
  import org.apache.cassandra.net.FrameEncoder;
  import org.apache.cassandra.service.ClientWarn;
  import org.apache.cassandra.service.QueryState;
  import org.apache.cassandra.service.reads.thresholds.CoordinatorWarnings;
  import org.apache.cassandra.transport.ClientResourceLimits.Overload;
  import org.apache.cassandra.transport.Flusher.FlushItem;
  import org.apache.cassandra.transport.messages.ErrorMessage;
  import org.apache.cassandra.transport.messages.EventMessage;
  import org.apache.cassandra.utils.JVMStabilityInspector;
  import org.apache.cassandra.utils.MonotonicClock;
  import org.apache.cassandra.utils.NoSpamLogger;
  
  import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED;
  
  public class Dispatcher implements CQLMessageHandler.MessageConsumer<Message.Request>
  {
      private static final Logger logger = LoggerFactory.getLogger(Dispatcher.class);
  
      @VisibleForTesting
      static final LocalAwareExecutorPlus requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
                                                                               DatabaseDescriptor::setNativeTransportMaxThreads,
                                                                               "transport",
                                                                               "Native-Transport-Requests");
  
      /** CASSANDRA-17812: Rate-limit new client connection setup to avoid overwhelming during bcrypt
       *
       * authExecutor is a separate thread pool for handling requests on connections that need to be authenticated.
       * Calls to AUTHENTICATE can be expensive if the number of rounds for bcrypt is configured to a high value,
       * so during a connection storm checking the password hash would starve existing connected clients for CPU and
       * trigger timeouts if on the same thread pool as standard requests.
       *
       * Moving authentication requests to a small, separate pool prevents starvation handling all other
       * requests. If the authExecutor pool backs up, it may cause authentication timeouts but the clients should
       * back off and retry while the rest of the system continues to make progress.
       *
       * Setting less than 1 will service auth requests on the standard {@link Dispatcher#requestExecutor}
       */
      @VisibleForTesting
      static final LocalAwareExecutorPlus authExecutor = SHARED.newExecutor(Math.max(1, DatabaseDescriptor.getNativeTransportMaxAuthThreads()),
                                                                            DatabaseDescriptor::setNativeTransportMaxAuthThreads,
                                                                            "transport",
                                                                            "Native-Transport-Auth-Requests");
  
      private static final ConcurrentMap<EventLoop, Flusher> flusherLookup = new ConcurrentHashMap<>();
      private final boolean useLegacyFlusher;
  
      /**
       * Takes a Channel, Request and the Response produced by processRequest and outputs a FlushItem
       * appropriate for the pipeline, which is specific to the protocol version. V5 and above will
       * produce FlushItem.Framed instances whereas earlier versions require FlushItem.Unframed.
       * The instances of these FlushItem subclasses are specialized to release resources in the
       * right way for the specific pipeline that produced them.
       */
      // TODO parameterize with FlushItem subclass
      interface FlushItemConverter
      {
          FlushItem<?> toFlushItem(Channel channel, Message.Request request, Message.Response response);
      }
  
      public Dispatcher(boolean useLegacyFlusher)
      {
          this.useLegacyFlusher = useLegacyFlusher;
      }
  
      @Override
      public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure)
      {
          if (!request.connection().getTracker().isRunning())
          {
              // We can not respond with a custom, transport, or server exceptions since, given current implementation of clients,
              // they will defunct the connection. Without a protocol version bump that introduces an "I am going away message",
              // we have to stick to an existing error code.
              Message.Response response = ErrorMessage.fromException(new OverloadedException("Server is shutting down"));
              response.setStreamId(request.getStreamId());
              response.setWarnings(ClientWarn.instance.getWarnings());
              response.attach(request.connection);
              FlushItem<?> toFlush = forFlusher.toFlushItem(channel, request, response);
              flush(toFlush);
              return;
          }
  
          // if native_transport_max_auth_threads is < 1, don't delegate to new pool on auth messages
          boolean isAuthQuery = DatabaseDescriptor.getNativeTransportMaxAuthThreads() > 0 &&
                                (request.type == Message.Type.AUTH_RESPONSE || request.type == Message.Type.CREDENTIALS);
  
          // Importantly, the authExecutor will handle the AUTHENTICATE message which may be CPU intensive.
          LocalAwareExecutorPlus executor = isAuthQuery ? authExecutor : requestExecutor;
  
          executor.submit(new RequestProcessor(channel, request, forFlusher, backpressure));
          ClientMetrics.instance.markRequestDispatched();
      }
  
      public static class RequestTime
      {
          private final long enqueuedAtNanos;
          private final long startedAtNanos;
  
          public RequestTime(long createdAtNanos)
          {
              this(createdAtNanos, createdAtNanos);
          }
  
          public RequestTime(long enqueuedAtNanos, long startedAtNanos)
          {
              Preconditions.checkArgument(enqueuedAtNanos != -1);
              this.enqueuedAtNanos = enqueuedAtNanos;
              this.startedAtNanos = startedAtNanos;
          }
  
          public static RequestTime forImmediateExecution()
          {
              return new RequestTime(MonotonicClock.Global.preciseTime.now());
          }
  
          public RequestTime withStartedAt(long startedAtNanos)
          {
              return new RequestTime(enqueuedAtNanos, startedAtNanos);
          }
  
          public long startedAtNanos()
          {
              return startedAtNanos;
          }
  
          public long enqueuedAtNanos()
          {
              return enqueuedAtNanos;
          }
  
          /**
           * Base time is used by timeouts, and can be set to either when the request was added to the queue,
           * or when the processing has started, which is controlled by {@link DatabaseDescriptor#getCQLStartTime()}
           *
           * Since client read/write timeouts are usually aligned with server-side timeouts, it is desireable to use
           * enqueue time as a base. However, since client removes the handler `readTimeoutMillis` (which is 12 seconds
           * by default), the upper bound for any execution on the coordinator is 12 seconds (thanks to CASSANDRA-7392,
           * any replica-side query is capped by the verb timeout), if REQUEST option is used. But even simply allowing
           * such long timeouts also implicitly allows queues to grow large, since our queues are currently unbounded.
           *
           * Latency, however, is _always_ based on request processing time, since the amount of time that request spends
           * in the queue is not a representative metric of replica performance.
           */
          public long baseTimeNanos()
          {
              switch (DatabaseDescriptor.getCQLStartTime())
              {
                  case REQUEST:
                      return startedAtNanos();
                  case QUEUE:
                      return enqueuedAtNanos();
                  default:
                      throw new IllegalArgumentException("Unknown start time: " + DatabaseDescriptor.getCQLStartTime());
              }
          }
  
          /**
           * Given the current time and a base timeout for the verb return a request's expiration deadline,
           * the time at which it becomes eligible for load shedding.
           * The two factors to consider are the per-verb and client timeouts. Both are calculated by subtracting the
           * time already elapsed during the lifetime of the request from some base value.
           *
           * When deriving verb timeout, two alternative semantics are available. This timeout may represent either:
           *  * the total time available for a coordinator to process a client request and return its response
           *  * a time bound for a coordinator to send internode requests and gather responses from replicas
           *
           * The point from which elapsed time is measured here is configurable to accommodate these two different
           * options. For the former, the clock starts when a client request is received and enqueued by the coordinator.
           * For the latter, it starts when the request is dequeued by the coordinator and processing is started.
           * See {@link #baseTimeNanos()} for details.
           *
           * The client timeout represents how long the sender of a request is prepared to wait for a response. By
           * implication, after this window has passed any further work done on the server side is wasted effort. Ideally,
           * the base for this timeout would be set on a per-request basis but as this not currently supported in the
           * protocol, it is configured uniformly for all requests. See {@link DatabaseDescriptor#getNativeTransportTimeout}.
           * For this calculation, elapsed time is always measured from the point when a request is received and enqueued.
           *
           * Where verb timeout is based on queue admission, deadline computation is straightforward. The expiration
           * deadline is simply the current time plus the smaller of the verb and client timeouts.
           *
           * However, if verb timeout is based on when the request is dequeued, the implications are more nuanced.
           * In this scenario, while there may still be "headroom" available within the verb timeout, using it could
           * exceed the client timeout (which is always based on admission time).
           *
           * For example:
           *
           * * Client timeout base is 10 (cb), verb timeout base is 5 (vb)
           * * Request is enqueued at t1 (e)
           * * Request is dequeued at t8 (d)
           * * computeDeadline is called at t9 (n)
           *
           * If verb timeout is based on dequeuing, there would still some time remaining before a verb-based deadline.
           * elapsed  = (n - d)        ;  1
           * timeout  = (vb - elapsed) ;  4
           * deadline = (n + timeout)  ;  t13
           * ostensibly, the coordinator has until t13 to complete processing
           *
           * But as client timeout is measured from admission time, the request may exceeded the maximum wait period for
           * the client sooner.
           * elapsed  = (n - e)        ;  8
           * timeout  = (cb - elapsed) ;  2
           * deadline = (n + timeout)  ;  t11
           * So the coordinator actually only has until t11 to complete processing, beyond then the client will not accept
           * any response.
           *
           * @param verbExpiresAfterNanos the base timeout value for the verb being executed
           * @return the point in time after which no further processing should occur
           */
          public long computeDeadline(long verbExpiresAfterNanos)
          {
              long clientDeadline = clientDeadline();
  
              long verbDeadline = baseTimeNanos() + verbExpiresAfterNanos;
              // Whichever one is closer
              return Math.min(verbDeadline, clientDeadline);
          }
  
          public long computeTimeout(long now, long verbExpiresAfterNanos)
          {
              return computeDeadline(verbExpiresAfterNanos) - now;
          }
  
          /**
           * No request should survive native request deadline, but in order to err on the side of caution, we have this
           * swtich that allows hints to be submitted to mutation stage when cluster is potentially overloaded. Allowing
           * hints to be not bound by deadline can exacerbate overload, but since there are also correctness implications,
           * this seemed like a reasonable configuration option.
           */
          public boolean shouldSendHints()
          {
              if (!DatabaseDescriptor.getEnforceNativeDeadlineForHints())
                  return true;
  
              long now = MonotonicClock.Global.preciseTime.now();
              long clientDeadline = clientDeadline();
              return now < clientDeadline;
          }
  
          public long clientDeadline()
          {
              return enqueuedAtNanos() + DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS);
          }
  
          public long timeSpentInQueueNanos()
          {
              return startedAtNanos - enqueuedAtNanos;
          }
      }
  
      /**
       * It is important to keep this class an instance of {@link DebuggableTask}, either runnable or callable since this
       * is the only way we can keep it not wrapped into a callable on SEPExecutor submission path. And we need this
       * functionality for tracking time purposes.
       */
      public class RequestProcessor implements DebuggableTask.RunnableDebuggableTask
      {
          private final Channel channel;
          private final Message.Request request;
          private final FlushItemConverter forFlusher;
          private final Overload backpressure;
  
          private volatile long startTimeNanos;
  
          public RequestProcessor(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure)
          {
              this.channel = channel;
              this.request = request;
              this.forFlusher = forFlusher;
              this.backpressure = backpressure;
          }
  
          @Override
          public void run()
          {
              startTimeNanos = MonotonicClock.Global.preciseTime.now();
              processRequest(channel, request, forFlusher, backpressure, new RequestTime(request.createdAtNanos, startTimeNanos));
          }
  
          @Override
          public long creationTimeNanos()
          {
              return request.createdAtNanos;
          }
  
          @Override
          public long startTimeNanos()
          {
              return startTimeNanos;
          }
  
          @Override
          public String description()
          {
              return request.toString();
          }
  
          @Override
          public String toString()
          {
              return "RequestProcessor{" +
                     "request=" + request +
                     ", approxStartTimeNanos=" + startTimeNanos +
                     '}';
          }
      }
  
      /**
       * Checks if the item in the head of the queue has spent more than allowed time in the queue.
       */
      @Override
      public boolean hasQueueCapacity()
      {
          double threshold = DatabaseDescriptor.getNativeTransportQueueMaxItemAgeThreshold();
          if (threshold <= 0)
              return true;
  
          return requestExecutor.oldestTaskQueueTime() < (DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS) * threshold);
      }
  
      /**
       * Note: this method may be executed on the netty event loop, during initial protocol negotiation; the caller is
       * responsible for cleaning up any global or thread-local state. (ex. tracing, client warnings, etc.).
       */
      private static Message.Response processRequest(ServerConnection connection, Message.Request request, Overload backpressure, RequestTime requestTime)
      {
          long queueTime = requestTime.timeSpentInQueueNanos();
  
          // If we have already crossed the max timeout for all possible RPCs, we time out the query immediately.
          // We do not differentiate between query types here, since if we got into a situation when, say, we have a PREPARE
          // query that is stuck behind the EXECUTE query, we would rather time it out and catch up with a backlog, expecting
          // that the bursts are going to be short-lived.
          ClientMetrics.instance.queueTime(queueTime, TimeUnit.NANOSECONDS);
          if (queueTime > DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS))
          {
              ClientMetrics.instance.markTimedOutBeforeProcessing();
              return ErrorMessage.fromException(new OverloadedException("Query timed out before it could start"));
          }
  
          if (connection.getVersion().isGreaterOrEqualTo(ProtocolVersion.V4))
              ClientWarn.instance.captureWarnings();
  
          // even if ClientWarn is disabled, still setup CoordinatorTrackWarnings, as this will populate metrics and
          // emit logs on the server; the warnings will just be ignored and not sent to the client
          if (request.isTrackable())
              CoordinatorWarnings.init();
  
          switch (backpressure)
          {
              case NONE:
                  break;
              case REQUESTS:
              {
                  String message = String.format("Request breached global limit of %d requests/second and triggered backpressure.",
                                                 ClientResourceLimits.getNativeTransportMaxRequestsPerSecond());
  
                  NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, message);
                  ClientWarn.instance.warn(message);
                  break;
              }
              case BYTES_IN_FLIGHT:
              {
                  String message = String.format("Request breached limit(s) on bytes in flight (Endpoint: %d, Global: %d) and triggered backpressure.",
                                                 ClientResourceLimits.getEndpointLimit(), ClientResourceLimits.getGlobalLimit());
  
                  NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, message);
                  ClientWarn.instance.warn(message);
                  break;
              }
              case QUEUE_TIME:
              {
                  String message = String.format("Request has spent over %s time of the maximum timeout %dms in the queue",
                                                 DatabaseDescriptor.getNativeTransportQueueMaxItemAgeThreshold(),
                                                 DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.MILLISECONDS));
  
                  NoSpamLogger.log(logger, NoSpamLogger.Level.INFO, 1, TimeUnit.MINUTES, message);
                  ClientWarn.instance.warn(message);
                  break;
              }
          }
  
          QueryState qstate = connection.validateNewMessage(request.type, connection.getVersion());
  
          Message.logger.trace("Received: {}, v={}", request, connection.getVersion());
          connection.requests.inc();
          Message.Response response = request.execute(qstate, requestTime);
  
          if (request.isTrackable())
              CoordinatorWarnings.done();
  
          response.setStreamId(request.getStreamId());
          response.setWarnings(ClientWarn.instance.getWarnings());
          response.attach(connection);
          connection.applyStateTransition(request.type, response.type);
          return response;
      }
  
      /**
       * Note: this method may be executed on the netty event loop.
       */
      static Message.Response processRequest(Channel channel, Message.Request request, Overload backpressure, RequestTime requestTime)
      {
          try
          {
              return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime);
          }
          catch (Throwable t)
          {
              JVMStabilityInspector.inspectThrowable(t);
  
              if (request.isTrackable())
                  CoordinatorWarnings.done();
  
              Predicate<Throwable> handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true);
              ErrorMessage error = ErrorMessage.fromException(t, handler);
              error.setStreamId(request.getStreamId());
              error.setWarnings(ClientWarn.instance.getWarnings());
              return error;
          }
          finally
          {
              CoordinatorWarnings.reset();
              ClientWarn.instance.resetWarnings();
          }
      }
  
      /**
       * Note: this method is not expected to execute on the netty event loop.
       */
      void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure, RequestTime requestTime)
      {
          Message.Response response = processRequest(channel, request, backpressure, requestTime);
          FlushItem<?> toFlush = forFlusher.toFlushItem(channel, request, response);
          Message.logger.trace("Responding: {}, v={}", response, request.connection().getVersion());
          flush(toFlush);
      }
  
      private void flush(FlushItem<?> item)
      {
          EventLoop loop = item.channel.eventLoop();
          Flusher flusher = flusherLookup.get(loop);
          if (flusher == null)
          {
              Flusher created = useLegacyFlusher ? Flusher.legacy(loop) : Flusher.immediate(loop);
              Flusher alt = flusherLookup.putIfAbsent(loop, flusher = created);
              if (alt != null)
                  flusher = alt;
          }
  
          flusher.enqueue(item);
          flusher.start();
      }
  
      public boolean isDone()
      {
          return requestExecutor.getPendingTaskCount() == 0 && requestExecutor.getActiveTaskCount() == 0;
      }
  
      public static void shutdown()
      {
          requestExecutor.shutdown();
          authExecutor.shutdown();
      }
  
      /**
       * Dispatcher for EventMessages. In {@link Server.ConnectionTracker#send(Event)}, the strategy
       * for delivering events to registered clients is dependent on protocol version and the configuration
       * of the pipeline. For v5 and newer connections, the event message is encoded into an Envelope,
       * wrapped in a FlushItem and then delivered via the pipeline's flusher, in a similar way to
       * a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, Overload, RequestTime)}.
       * It's worth noting that events are not generally fired as a direct response to a client request,
       * so this flush item has a null request attribute. The dispatcher itself is created when the
       * pipeline is first configured during protocol negotiation and is attached to the channel for
       * later retrieval.
       *
       * Pre-v5 connections simply write the EventMessage directly to the pipeline.
       */
      static final AttributeKey<Consumer<EventMessage>> EVENT_DISPATCHER = AttributeKey.valueOf("EVTDISP");
      Consumer<EventMessage> eventDispatcher(final Channel channel,
                                             final ProtocolVersion version,
                                             final FrameEncoder.PayloadAllocator allocator)
      {
          return eventMessage -> flush(new FlushItem.Framed(channel,
                                                            eventMessage.encode(version),
                                                            null,
                                                            allocator,
                                                            f -> f.response.release()));
      }
  }
  