Line data    Source code

       1             : /**
       2             : Copyright (c) 2019 Roman Katuntsev <sbkarr@stappler.org>
       3             : Copyright (c) 2023 Stappler LLC <admin@stappler.dev>
       4             : 
       5             : Permission is hereby granted, free of charge, to any person obtaining a copy
       6             : of this software and associated documentation files (the "Software"), to deal
       7             : in the Software without restriction, including without limitation the rights
       8             : to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
       9             : copies of the Software, and to permit persons to whom the Software is
      10             : furnished to do so, subject to the following conditions:
      11             : 
      12             : The above copyright notice and this permission notice shall be included in
      13             : all copies or substantial portions of the Software.
      14             : 
      15             : THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
      16             : IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
      17             : FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
      18             : AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
      19             : LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
      20             : OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
      21             : THE SOFTWARE.
      22             : **/
      23             : 
      24             : #ifndef STAPPLER_THREADS_SPTHREADTASKQUEUE_H_
      25             : #define STAPPLER_THREADS_SPTHREADTASKQUEUE_H_
      26             : 
      27             : #include "SPThreadTask.h"
      28             : 
      29             : namespace STAPPLER_VERSIONIZED stappler::thread {
      30             : 
      31             : class Worker;
      32             : 
      33             : class TaskQueue : public RefBase<memory::StandartInterface> {
      34             : public:
      35             :         using Ref = RefBase<memory::StandartInterface>;
      36             :         using TaskMap = std::map<uint32_t, std::vector<Rc<Task>>, std::less<void>>;
      37             : 
      38             :         enum class Flags {
      39             :                 None = 0,
      40             : 
      41             :                 // allow to submit task for specific thread via `perform(Map<uint32_t, Vector<Rc<Task>>> &&)`
      42             :                 // it requires additional space for internal local queues and less optimal thread scheduling
      43             :                 // with `condition_variable_any` instead of `condition_variable`
      44             :                 LocalQueue = 1,
      45             : 
      46             :                 // allow queue to be externally cancelled with `performAll` and `waitForAll`
      47             :                 // it requires extra condition to track count of tasks executing
      48             :                 Cancelable = 2,
      49             : 
      50             :                 // allow to wait for event on queue's main thread via 'wait'
      51             :                 Waitable = 4,
      52             :         };
      53             : 
      54             :         struct WorkerContext;
      55             : 
      56             :         static const TaskQueue *getOwner();
      57             : 
      58             :         TaskQueue(StringView name = StringView(), std::function<void()> &&wakeup = std::function<void()>());
      59             :         ~TaskQueue();
      60             : 
      61             :         void finalize();
      62             : 
      63             :         void performAsync(Rc<Task> &&task);
      64             : 
      65             :         void perform(Rc<Task> &&task, bool first = false);
      66             :         void perform(std::function<void()> &&, Ref * = nullptr, bool first = false);
      67             : 
      68             :         bool perform(TaskMap &&tasks);
      69             : 
      70             :         void update(uint32_t *count = nullptr);
      71             : 
      72             :         void onMainThread(Rc<Task> &&task);
      73             :         void onMainThread(std::function<void()> &&func, Ref *target);
      74             : 
      75             :         bool spawnWorkers(Flags);
      76             : 
      77             :         // maxOf<uint32_t> - set id to next available
      78             :         bool spawnWorkers(Flags, uint32_t threadId, uint16_t threadCount, StringView name = StringView());
      79             :         bool cancelWorkers();
      80             : 
      81             :         void performAll(Flags flags);
      82             :         bool waitForAll(TimeInterval = TimeInterval::seconds(1));
      83             : 
      84             :         bool wait(uint32_t *count = nullptr);
      85             :         bool wait(TimeInterval, uint32_t *count = nullptr);
      86             : 
      87             :         void lock();
      88             :         void unlock();
      89             : 
      90         100 :         StringView getName() const { return _name; }
      91             : 
      92             :         std::vector<std::thread::id> getThreadIds() const;
      93             : 
      94     3419997 :         size_t getOutputCounter() const { return _outputCounter.load(); }
      95             : 
      96             :         uint16_t getThreadCount() const;
      97             : 
      98             : protected:
      99             :         friend class Worker;
     100             : 
     101             :         Rc<Task> popTask(uint32_t idx);
     102             :         void onMainThreadWorker(Rc<Task> &&task);
     103             : 
     104             :         WorkerContext *_context = nullptr;
     105             : 
     106             :         std::mutex _inputMutexQueue;
     107             :         std::mutex _inputMutexFree;
     108             :         memory::PriorityQueue<Rc<Task>> _inputQueue;
     109             : 
     110             :         std::mutex _outputMutex;
     111             :         std::vector<Rc<Task>> _outputQueue;
     112             :         std::vector<Pair<std::function<void()>, Rc<Ref>>> _outputCallbacks;
     113             : 
     114             :         std::atomic<size_t> _outputCounter = 0;
     115             :         std::atomic<size_t> _tasksCounter = 0;
     116             : 
     117             :         StringView _name = StringView("TaskQueue");
     118             : 
     119             :         std::function<void()> _wakeup;
     120             : };
     121             : 
     122             : SP_DEFINE_ENUM_AS_MASK(TaskQueue::Flags)
     123             : 
     124             : }
     125             : 
     126             : #endif /* STAPPLER_THREADS_SPTHREADTASKQUEUE_H_ */