Quelle  funnel-requestqueue.c   Sprache: C

// SPDX-License-Identifier: GPL-2.0-only
/*
 * Copyright 2023 Red Hat
 */

#include "funnel-requestqueue.h"

#include <linux/atomic.h>
#include <linux/compiler.h>
#include <linux/wait.h>

#include "funnel-queue.h"
#include "logger.h"
#include "memory-alloc.h"
#include "thread-utils.h"

/*
 * This queue will attempt to handle requests in reasonably sized batches instead of reacting
 * immediately to each new request. The wait time between batches is dynamically adjusted up or
 * down to try to balance responsiveness against wasted thread run time.
 *
 * If the wait time becomes long enough, the queue will become dormant and must be explicitly
 * awoken when a new request is enqueued. The enqueue operation updates "newest" in the funnel
 * queue via xchg (which is a memory barrier), and later checks "dormant" to decide whether to do a
 * wakeup of the worker thread.
 *
 * When deciding to go to sleep, the worker thread sets "dormant" and then examines "newest" to
 * decide if the funnel queue is idle. In dormant mode, the last examination of "newest" before
 * going to sleep is done inside the wait_event_interruptible() macro, after a point where one or
 * more memory barriers have been issued. (Preparing to sleep uses spin locks.) Even if the funnel
 * queue's "next" field update isn't visible yet to make the entry accessible, its existence will
 * kick the worker thread out of dormant mode and back into timer-based mode.
 *
 * Unbatched requests are used to communicate between different zone threads and will also cause
 * the queue to awaken immediately.
 */

enum {
 NANOSECOND = 1,
 MICROSECOND = 1000 * NANOSECOND,
 MILLISECOND = 1000 * MICROSECOND,
 DEFAULT_WAIT_TIME = 20 * MICROSECOND,
 MINIMUM_WAIT_TIME = DEFAULT_WAIT_TIME / 2,
 MAXIMUM_WAIT_TIME = MILLISECOND,
 MINIMUM_BATCH = 32,
 MAXIMUM_BATCH = 64,
};

struct uds_request_queue {
 /* Wait queue for synchronizing producers and consumer */
 struct wait_queue_head wait_head;
 /* Function to process a request */
 uds_request_queue_processor_fn processor;
 /* Queue of new incoming requests */
 struct funnel_queue *main_queue;
 /* Queue of old requests to retry */
 struct funnel_queue *retry_queue;
 /* The thread id of the worker thread */
 struct thread *thread;
 /* True if the worker was started */
 bool started;
 /* When true, requests can be enqueued */
 bool running;
 /* A flag set when the worker is waiting without a timeout */
 atomic_t dormant;
};

static inline struct uds_request *poll_queues(struct uds_request_queue *queue)
{
 struct funnel_queue_entry *entry;

 entry = vdo_funnel_queue_poll(queue->retry_queue);
 if (entry != NULL)
  return container_of(entry, struct uds_request, queue_link);

 entry = vdo_funnel_queue_poll(queue->main_queue);
 if (entry != NULL)
  return container_of(entry, struct uds_request, queue_link);

 return NULL;
}

static inline bool are_queues_idle(struct uds_request_queue *queue)
{
 return vdo_is_funnel_queue_idle(queue->retry_queue) &&
        vdo_is_funnel_queue_idle(queue->main_queue);
}

/*
 * Determine if there is a next request to process, and return it if there is. Also return flags
 * indicating whether the worker thread can sleep (for the use of wait_event() macros) and whether
 * the thread did sleep before returning a new request.
 */
static inline bool dequeue_request(struct uds_request_queue *queue,
       struct uds_request **request_ptr, bool *waited_ptr)
{
 struct uds_request *request = poll_queues(queue);

 if (request != NULL) {
  *request_ptr = request;
  return true;
 }

 if (!READ_ONCE(queue->running)) {
  /* Wake the worker thread so it can exit. */
  *request_ptr = NULL;
  return true;
 }

 *request_ptr = NULL;
 *waited_ptr = true;
 return false;
}

static void wait_for_request(struct uds_request_queue *queue, bool dormant,
        unsigned long timeout, struct uds_request **request,
        bool *waited)
{
 if (dormant) {
  wait_event_interruptible(queue->wait_head,
      (dequeue_request(queue, request, waited) ||
       !are_queues_idle(queue)));
  return;
 }

 wait_event_interruptible_hrtimeout(queue->wait_head,
        dequeue_request(queue, request, waited),
        ns_to_ktime(timeout));
}

static void request_queue_worker(void *arg)
{
 struct uds_request_queue *queue = arg;
 struct uds_request *request = NULL;
 unsigned long time_batch = DEFAULT_WAIT_TIME;
 bool dormant = atomic_read(&queue->dormant);
 bool waited = false;
 long current_batch = 0;

 for (;;) {
  wait_for_request(queue, dormant, time_batch, &request, &waited);
  if (likely(request != NULL)) {
   current_batch++;
   queue->processor(request);
  } else if (!READ_ONCE(queue->running)) {
   break;
  }

  if (dormant) {
   /*
 * The queue has been roused from dormancy. Clear the flag so enqueuers can
 * stop broadcasting. No fence is needed for this transition.
 */
   atomic_set(&queue->dormant, false);
   dormant = false;
   time_batch = DEFAULT_WAIT_TIME;
  } else if (waited) {
   /*
 * We waited for this request to show up. Adjust the wait time to smooth
 * out the batch size.
 */
   if (current_batch < MINIMUM_BATCH) {
    /*
 * If the last batch of requests was too small, increase the wait
 * time.
 */
    time_batch += time_batch / 4;
    if (time_batch >= MAXIMUM_WAIT_TIME) {
     atomic_set(&queue->dormant, true);
     dormant = true;
    }
   } else if (current_batch > MAXIMUM_BATCH) {
    /*
 * If the last batch of requests was too large, decrease the wait
 * time.
 */
    time_batch -= time_batch / 4;
    if (time_batch < MINIMUM_WAIT_TIME)
     time_batch = MINIMUM_WAIT_TIME;
   }
   current_batch = 0;
  }
 }

 /*
 * Ensure that we process any remaining requests that were enqueued before trying to shut
 * down. The corresponding write barrier is in uds_request_queue_finish().
 */
 smp_rmb();
 while ((request = poll_queues(queue)) != NULL)
  queue->processor(request);
}

int uds_make_request_queue(const char *queue_name,
      uds_request_queue_processor_fn processor,
      struct uds_request_queue **queue_ptr)
{
 int result;
 struct uds_request_queue *queue;

 result = vdo_allocate(1, struct uds_request_queue, __func__, &queue);
 if (result != VDO_SUCCESS)
  return result;

 queue->processor = processor;
 queue->running = true;
 atomic_set(&queue->dormant, false);
 init_waitqueue_head(&queue->wait_head);

 result = vdo_make_funnel_queue(&queue->main_queue);
 if (result != VDO_SUCCESS) {
  uds_request_queue_finish(queue);
  return result;
 }

 result = vdo_make_funnel_queue(&queue->retry_queue);
 if (result != VDO_SUCCESS) {
  uds_request_queue_finish(queue);
  return result;
 }

 result = vdo_create_thread(request_queue_worker, queue, queue_name,
       &queue->thread);
 if (result != VDO_SUCCESS) {
  uds_request_queue_finish(queue);
  return result;
 }

 queue->started = true;
 *queue_ptr = queue;
 return UDS_SUCCESS;
}

static inline void wake_up_worker(struct uds_request_queue *queue)
{
 if (wq_has_sleeper(&queue->wait_head))
  wake_up(&queue->wait_head);
}

void uds_request_queue_enqueue(struct uds_request_queue *queue,
          struct uds_request *request)
{
 struct funnel_queue *sub_queue;
 bool unbatched = request->unbatched;

 sub_queue = request->requeued ? queue->retry_queue : queue->main_queue;
 vdo_funnel_queue_put(sub_queue, &request->queue_link);

 /*
 * We must wake the worker thread when it is dormant. A read fence isn't needed here since
 * we know the queue operation acts as one.
 */
 if (atomic_read(&queue->dormant) || unbatched)
  wake_up_worker(queue);
}

void uds_request_queue_finish(struct uds_request_queue *queue)
{
 if (queue == NULL)
  return;

 /*
 * This memory barrier ensures that any requests we queued will be seen. The point is that
 * when dequeue_request() sees the following update to the running flag, it will also be
 * able to see any change we made to a next field in the funnel queue entry. The
 * corresponding read barrier is in request_queue_worker().
 */
 smp_wmb();
 WRITE_ONCE(queue->running, false);

 if (queue->started) {
  wake_up_worker(queue);
  vdo_join_threads(queue->thread);
 }

 vdo_free_funnel_queue(queue->main_queue);
 vdo_free_funnel_queue(queue->retry_queue);
 vdo_free(queue);
}