RequestQueue.h

/*
 * Copyright (c) 2008-2014, Pedigree Developers
 *
 * Please see the CONTRIB file in the root of the source tree for a full
 * list of contributors.
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
 
#ifndef REQUEST_QUEUE_H
#define REQUEST_QUEUE_H
#include "pedigree/kernel/Atomic.h"
#include "pedigree/kernel/compiler.h"
#include "pedigree/kernel/process/Mutex.h"
#include "pedigree/kernel/process/SchedulerWorkerWake.h"
#include "pedigree/kernel/process/WaitQueue.h"
#include "pedigree/kernel/processor/state_forward.h"
#include "pedigree/kernel/processor/types.h"
#include "pedigree/kernel/utilities/IntrusiveMpscQueue.h"
#include "pedigree/kernel/utilities/StaticString.h"
#include "pedigree/kernel/utilities/String.h"
 
#include <config.h>
#if THREADS
#include "pedigree/kernel/machine/TimerHandler.h"
#endif
 
class Thread;
struct ThreadPlacement;
class Timer;
class PerProcessorScheduler;
 
#define REQUEST_QUEUE_NUM_PRIORITIES 4
 
class EXPORTED_PUBLIC RequestQueue {
 public:
  class PreallocatedRequest;
 
  enum class OverrunStatus {
    Clear,
    Armed,
    Stalled,
    Overloaded,
  };
 
#if THREADS && HOSTED && PEDIGREE_HOSTED_SMOKE_TESTS
  OverrunStatus sampleOverrunForTest();
 
  using HostedSmokeHook = void (*)(void*);
 
  void setAfterPreallocatedAdmissionHookForTest(HostedSmokeHook hook, void* context) {
    m_AfterPreallocatedAdmissionHook = hook;
    m_AfterPreallocatedAdmissionContext = context;
  }
 
  void setAfterIntakeExchangeHookForTest(HostedSmokeHook hook, void* context) {
    m_AfterIntakeExchangeHook = hook;
    m_AfterIntakeExchangeContext = context;
  }
 
  size_t workerTransientRetriesForTest() {
    return m_WorkerTransientRetries.value();
  }
 
  size_t guardedTransientRetriesForTest() {
    return m_GuardedTransientRetries.value();
  }
 
  size_t publisherDrainRetriesForTest() {
    return m_PublisherDrainRetries.value();
  }
#endif
 
#if THREADS && PEDIGREE_CONCURRENCY_SMOKE_TESTS
  using ConcurrencySmokeHook = void (*)(void*);
 
  void setAfterPreallocatedClaimHookForTest(ConcurrencySmokeHook hook, void* context) {
    m_AfterPreallocatedClaimHook = hook;
    m_AfterPreallocatedClaimContext = context;
  }
 
#endif
 
#if THREADS
 private:
  class RequestQueueOverrunChecker : public TimerHandler {
    friend class RequestQueue;
 
    RequestQueueOverrunChecker()
        : m_LastQueueSize(0),
          m_LastProgressGeneration(0),
          m_Tick(0),
          m_HasBacklogBaseline(false),
          queue(0) {}
 
   private:
    virtual void timer(uint64_t delta);
    OverrunStatus sample(size_t& lastSize, size_t& currentSize);
    void resetBaselineLocked();
 
    size_t m_LastQueueSize;
    size_t m_LastProgressGeneration;
    uint64_t m_Tick;
    bool m_HasBacklogBaseline;
 
    RequestQueue* queue;
  };
#endif
 
 protected:
  class Request;
 
  struct IntakeNode {
    explicit IntakeNode(Request* request = nullptr) : next(nullptr), owner(request) {}
 
    IntakeNode* next;
    Request* owner;
  };
 
  class Request {
   public:
    Request(size_t requestPriority, bool asynchronous, uint64_t requestP1, uint64_t requestP2,
            uint64_t requestP3, uint64_t requestP4, uint64_t requestP5, uint64_t requestP6,
            uint64_t requestP7, uint64_t requestP8,
            PreallocatedRequest* preallocatedOwner = nullptr)
        : p1(requestP1),
          p2(requestP2),
          p3(requestP3),
          p4(requestP4),
          p5(requestP5),
          p6(requestP6),
          p7(requestP7),
          p8(requestP8),
          m_ReturnValue(0),
#if THREADS
          m_Completion(),
          m_References(asynchronous ? 1 : 2),
#endif
          m_Next(nullptr),
          m_Intake(this),
          m_Priority(requestPriority),
          m_Asynchronous(asynchronous),
          m_Rejected(false),
          m_Completed(false),
          m_pPreallocatedOwner(preallocatedOwner) {
    }
 
    uint64_t p1, p2, p3, p4, p5, p6, p7, p8;
 
   private:
    friend class PreallocatedRequest;
    friend class RequestQueue;
 
    ~Request() = default;
 
    uint64_t m_ReturnValue;
#if THREADS
    WaitQueue m_Completion;
    Atomic<size_t> m_References;
#endif
    Request* m_Next;
    IntakeNode m_Intake;
    size_t m_Priority;
    bool m_Asynchronous;
    bool m_Rejected;
    bool m_Completed;
    PreallocatedRequest* m_pPreallocatedOwner;
 
    Request(const Request&);
    void operator=(const Request&);
  };
 
  class IntakeLane {
   public:
    IntakeLane() : m_Stub(), m_Queue(m_Stub) {}
 
    IntakeNode m_Stub;
    IntrusiveMpscQueue<IntakeNode, &IntakeNode::next> m_Queue;
 
   private:
    IntakeLane(const IntakeLane&) = delete;
    IntakeLane& operator=(const IntakeLane&) = delete;
  };
 
 public:
  class EXPORTED_PUBLIC PreallocatedRequest {
   public:
    using ReleaseCallback = void (*)(void*);
 
    PreallocatedRequest();
    PreallocatedRequest(ReleaseCallback releaseCallback, void* releaseContext);
    ~PreallocatedRequest();
 
    bool isAvailable() const;
 
   private:
    friend class RequestQueue;
    NOT_COPYABLE_OR_ASSIGNABLE(PreallocatedRequest);
 
    enum State {
      Idle,
      Claimed,
      Published,
      Releasing,
    };
 
    Request m_Request;
    Atomic<size_t> m_State;
    Atomic<size_t> m_ReleaseDepth;
    ReleaseCallback m_ReleaseCallback;
    void* m_ReleaseContext;
  };
 
  enum class PreallocatedPublishResult {
    Accepted,
    // Another accepted publication, or a claim that cannot roll back,
    // already owns this token.
    TokenBusy,
    QueueStopped,
    // Retained for source and module ABI compatibility. Preallocated
    // publications do not consume allocation admission.
    QueueFull,
    InvalidPriority,
  };
 
  RequestQueue(const String& name);
  virtual ~RequestQueue();
 
  // Action to perform when a duplicate request is found in the queue.
  enum ActionOnDuplicate {
    // Block waiting for it to complete, and return its return value.
    Block,
    // Ignore the duplicate and create a new request.
    NewRequest,
    // Return immediately, ignoring the result from the request.
    ReturnImmediately
  };
 
  enum class LifecycleState {
    Stopped,
    Accepting,
    Stopping,
    Destroyed,
  };
 
  virtual void initialise();
 
  virtual void destroy();
 
  MUST_USE_RESULT uint64_t addRequest(size_t priority, uint64_t p1 = 0, uint64_t p2 = 0,
                                      uint64_t p3 = 0, uint64_t p4 = 0, uint64_t p5 = 0,
                                      uint64_t p6 = 0, uint64_t p7 = 0, uint64_t p8 = 0);
 
  MUST_USE_RESULT uint64_t addRequest(size_t priority, ActionOnDuplicate action, uint64_t p1 = 0,
                                      uint64_t p2 = 0, uint64_t p3 = 0, uint64_t p4 = 0,
                                      uint64_t p5 = 0, uint64_t p6 = 0, uint64_t p7 = 0,
                                      uint64_t p8 = 0);
 
  uint64_t addAsyncRequest(size_t priority, uint64_t p1 = 0, uint64_t p2 = 0, uint64_t p3 = 0,
                           uint64_t p4 = 0, uint64_t p5 = 0, uint64_t p6 = 0, uint64_t p7 = 0,
                           uint64_t p8 = 0);
 
  MUST_USE_RESULT bool publishAsyncRequest(size_t priority, uint64_t p1 = 0, uint64_t p2 = 0,
                                           uint64_t p3 = 0, uint64_t p4 = 0, uint64_t p5 = 0,
                                           uint64_t p6 = 0, uint64_t p7 = 0, uint64_t p8 = 0);
 
  bool canWaitForCompletion();
 
  MUST_USE_RESULT bool waitForPreallocated(PreallocatedRequest& request);
 
  MUST_USE_RESULT PreallocatedPublishResult publishPreallocated(PreallocatedRequest& request,
                                                                size_t priority, uint64_t p1 = 0,
                                                                uint64_t p2 = 0, uint64_t p3 = 0,
                                                                uint64_t p4 = 0, uint64_t p5 = 0,
                                                                uint64_t p6 = 0, uint64_t p7 = 0,
                                                                uint64_t p8 = 0);
 
  MUST_USE_RESULT PreallocatedPublishResult republishPreallocatedWhileReleasing(
      PreallocatedRequest& request, size_t priority, uint64_t p1 = 0, uint64_t p2 = 0,
      uint64_t p3 = 0, uint64_t p4 = 0, uint64_t p5 = 0, uint64_t p6 = 0, uint64_t p7 = 0,
      uint64_t p8 = 0);
 
  MUST_USE_RESULT bool halt();
 
  MUST_USE_RESULT bool resume();
 
  LifecycleState getLifecycleState();
 
  bool drain();
 
 protected:
  virtual bool workerPlacement(ThreadPlacement&) const {
    return false;
  }
 
  virtual uint64_t executeRequest(uint64_t p1, uint64_t p2, uint64_t p3, uint64_t p4, uint64_t p5,
                                  uint64_t p6, uint64_t p7, uint64_t p8) = 0;
 
  RequestQueue(const RequestQueue&);
  void operator=(const RequestQueue&);
 
  virtual bool compareRequests(const Request& a, const Request& b) {
    return false;
  }
 
  virtual void cancelRequest(const Request& request) {}
 
  static int trampoline(void* p);
 
  int work();
 
  enum class NextRequestResult {
    Item,
    Empty,
    Retry,
  };
 
  NextRequestResult getNextRequest(Request*& request);
 
  bool drainIntakeLocked(size_t priority);
 
  void publishRequest(Request* request);
 
  Request* findDuplicate(const Request& request);
 
  static void completeRequest(Request* request, uint64_t returnValue, bool rejected);
 
  void discardRequest(Request* request);
 
  uint64_t addAsyncRequestInternal(size_t priority, uint64_t p1, uint64_t p2, uint64_t p3,
                                   uint64_t p4, uint64_t p5, uint64_t p6, uint64_t p7, uint64_t p8);
 
  PreallocatedPublishResult publishPreallocatedRequest(PreallocatedRequest& request,
                                                       PreallocatedRequest::State availableState,
                                                       size_t priority, uint64_t p1, uint64_t p2,
                                                       uint64_t p3, uint64_t p4, uint64_t p5,
                                                       uint64_t p6, uint64_t p7, uint64_t p8);
 
  void releasePreallocatedRequest(Request* request);
 
  void invokeCancelRequest(const Request& request);
 
  bool callbackActiveOnCurrentThread() const;
 
#if THREADS
  static void retainRequest(Request* request);
  void releaseRequest(Request* request);
  uint64_t waitForRequest(Request* request);
 
  bool startWorker();
  bool stopWorker();
 
  void closePreallocatedAdmission();
  void waitForPreallocatedPublishers();
#endif
 
  static constexpr size_t PublicationClosed = static_cast<size_t>(1) << ((sizeof(size_t) * 8) - 1);
  static constexpr size_t PublicationCountMask = ~PublicationClosed;
 
  IntakeLane m_IntakeLanes[REQUEST_QUEUE_NUM_PRIORITIES];
 
  Request* m_pRequestQueue[REQUEST_QUEUE_NUM_PRIORITIES];
  Request* m_pRequestQueueTail[REQUEST_QUEUE_NUM_PRIORITIES];
 
  Request* m_pActiveRequest;
 
  Atomic<size_t> m_State;
 
#if THREADS
  Mutex m_LifecycleMutex;
 
  WaitQueue m_RequestQueueWaiters;
  WaitQueue m_WorkerWaiters;
 
  Thread* m_pThread;
 
  Atomic<PerProcessorScheduler*> m_pWorkerScheduler;
  SchedulerWorkerWake m_WorkerWake;
 
  Atomic<size_t> m_bWorkerReady;
 
  Atomic<size_t> m_bWorkerActive;
 
  size_t m_WorkerProgressGeneration;
 
  RequestQueueOverrunChecker m_OverrunChecker;
  Timer* m_pOverrunTimer;
 
  Atomic<size_t> m_PublicationState;
 
#if HOSTED && PEDIGREE_HOSTED_SMOKE_TESTS
  HostedSmokeHook m_AfterPreallocatedAdmissionHook;
  void* m_AfterPreallocatedAdmissionContext;
  HostedSmokeHook m_AfterIntakeExchangeHook;
  void* m_AfterIntakeExchangeContext;
  Atomic<size_t> m_WorkerTransientRetries;
  Atomic<size_t> m_GuardedTransientRetries;
  Atomic<size_t> m_PublisherDrainRetries;
#endif
#if PEDIGREE_CONCURRENCY_SMOKE_TESTS
  ConcurrencySmokeHook m_AfterPreallocatedClaimHook;
  void* m_AfterPreallocatedClaimContext;
#endif
#endif
 
  size_t m_nMaxAsyncRequests;
  Atomic<size_t> m_nAsyncRequests;
 
  Atomic<size_t> m_nTotalRequests;
 
  Atomic<size_t> m_nActiveRequests;
 
  NormalStaticString m_Name;
};
 
#endif