//////////////////////////////////////////////////////////////////////////////// /// DISCLAIMER /// /// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany /// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany /// /// 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. /// /// Copyright holder is ArangoDB GmbH, Cologne, Germany /// /// @author Dr. Frank Celler /// @author Achim Brandt //////////////////////////////////////////////////////////////////////////////// #ifndef ARANGOD_SCHEDULER_SCHEDULER_H #define ARANGOD_SCHEDULER_SCHEDULER_H 1 #include "Basics/Common.h" #include "Basics/socket-utils.h" #include "Scheduler/TaskManager.h" #include "Basics/Mutex.h" struct TRI_json_t; namespace triagens { namespace basics { class ConditionVariable; } namespace rest { class SchedulerThread; class TaskData; //////////////////////////////////////////////////////////////////////////////// /// @brief input-output scheduler //////////////////////////////////////////////////////////////////////////////// class Scheduler : private TaskManager { Scheduler(Scheduler const&) = delete; Scheduler& operator=(Scheduler const&) = delete; public: //////////////////////////////////////////////////////////////////////////////// /// @brief scheduler singleton //////////////////////////////////////////////////////////////////////////////// static std::unique_ptr SCHEDULER; public: //////////////////////////////////////////////////////////////////////////////// /// @brief constructor /// /// If the number of threads is one, then the scheduler is single-threaded. /// In this case the only methods, which can be called from a different thread /// are beginShutdown, isShutdownInProgress, and isRunning. The method /// registerTask must be called before the Scheduler is started or from /// within the Scheduler thread. /// /// If the number of threads is greater than one, then the scheduler is /// multi-threaded. In this case the method registerTask can be called from /// threads other than the scheduler. //////////////////////////////////////////////////////////////////////////////// explicit Scheduler(size_t nrThreads); virtual ~Scheduler(); public: //////////////////////////////////////////////////////////////////////////////// /// @brief starts scheduler, keeps running /// /// The functions returns true, if the scheduler has been started. In this /// case the condition variable is signal as soon as at least one of the /// scheduler threads stops. //////////////////////////////////////////////////////////////////////////////// bool start(basics::ConditionVariable*); //////////////////////////////////////////////////////////////////////////////// /// @brief checks if the scheduler threads are up and running //////////////////////////////////////////////////////////////////////////////// bool isStarted(); //////////////////////////////////////////////////////////////////////////////// /// @brief opens the scheduler for business //////////////////////////////////////////////////////////////////////////////// bool open(); //////////////////////////////////////////////////////////////////////////////// /// @brief checks if scheduler is still running //////////////////////////////////////////////////////////////////////////////// bool isRunning(); //////////////////////////////////////////////////////////////////////////////// /// @brief starts shutdown sequence //////////////////////////////////////////////////////////////////////////////// void beginShutdown(); //////////////////////////////////////////////////////////////////////////////// /// @brief checks if scheduler is shuting down //////////////////////////////////////////////////////////////////////////////// bool isShutdownInProgress(); //////////////////////////////////////////////////////////////////////////////// /// @brief shuts down the scheduler //////////////////////////////////////////////////////////////////////////////// void shutdown(); //////////////////////////////////////////////////////////////////////////////// /// @brief get all user tasks //////////////////////////////////////////////////////////////////////////////// struct TRI_json_t* getUserTasks(); //////////////////////////////////////////////////////////////////////////////// /// @brief get a single user task //////////////////////////////////////////////////////////////////////////////// struct TRI_json_t* getUserTask(std::string const&); //////////////////////////////////////////////////////////////////////////////// /// @brief unregister and delete a user task by id //////////////////////////////////////////////////////////////////////////////// int unregisterUserTask(std::string const&); //////////////////////////////////////////////////////////////////////////////// /// @brief unregister all user tasks //////////////////////////////////////////////////////////////////////////////// int unregisterUserTasks(); //////////////////////////////////////////////////////////////////////////////// /// @brief registers a new task //////////////////////////////////////////////////////////////////////////////// int registerTask(Task*); //////////////////////////////////////////////////////////////////////////////// /// @brief registers a new task and returns the chosen threads number //////////////////////////////////////////////////////////////////////////////// int registerTask(Task*, ssize_t* n); //////////////////////////////////////////////////////////////////////////////// /// @brief registers a new task with a pre-selected thread number //////////////////////////////////////////////////////////////////////////////// int registerTaskInThread(Task* task, ssize_t tn); //////////////////////////////////////////////////////////////////////////////// /// @brief unregisters a task /// /// Note that this method is called by the task itself when cleanupTask is /// executed. If a Task failed in setupTask, it must not call unregisterTask. //////////////////////////////////////////////////////////////////////////////// int unregisterTask(Task*); //////////////////////////////////////////////////////////////////////////////// /// @brief destroys task /// /// Even if a Task failed in setupTask, it can still call destroyTask. The /// methods will delete the task. //////////////////////////////////////////////////////////////////////////////// int destroyTask(Task*); //////////////////////////////////////////////////////////////////////////////// /// @brief called to display current status //////////////////////////////////////////////////////////////////////////////// void reportStatus(); //////////////////////////////////////////////////////////////////////////////// /// @brief whether or not the scheduler is active //////////////////////////////////////////////////////////////////////////////// bool isActive() const { return (bool)_active; } //////////////////////////////////////////////////////////////////////////////// /// @brief set the scheduler activity //////////////////////////////////////////////////////////////////////////////// void setActive(bool value) { _active = value ? 1 : 0; } //////////////////////////////////////////////////////////////////////////////// /// @brief sets the process affinity //////////////////////////////////////////////////////////////////////////////// void setProcessorAffinity(size_t i, size_t c); //////////////////////////////////////////////////////////////////////////////// /// @brief returns the task for a task id /// /// Warning: ONLY call this from within the scheduler task! Otherwise, the /// task /// MIGHT already be deleted. //////////////////////////////////////////////////////////////////////////////// Task* lookupTaskById(uint64_t); public: //////////////////////////////////////////////////////////////////////////////// /// @brief main event loop //////////////////////////////////////////////////////////////////////////////// virtual void eventLoop(EventLoop) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief wakes up an event loop //////////////////////////////////////////////////////////////////////////////// virtual void wakeupLoop(EventLoop) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief called to register a socket descriptor event //////////////////////////////////////////////////////////////////////////////// virtual EventToken installSocketEvent(EventLoop, EventType, Task*, TRI_socket_t) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief re-starts the socket events //////////////////////////////////////////////////////////////////////////////// virtual void startSocketEvents(EventToken) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief stops the socket events //////////////////////////////////////////////////////////////////////////////// virtual void stopSocketEvents(EventToken) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief called to register a timer event //////////////////////////////////////////////////////////////////////////////// virtual EventToken installTimerEvent(EventLoop, Task*, double timeout) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief clears a timer without removing it //////////////////////////////////////////////////////////////////////////////// virtual void clearTimer(EventToken) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief rearms a timer //////////////////////////////////////////////////////////////////////////////// virtual void rearmTimer(EventToken, double timeout) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief called to register a periodic event //////////////////////////////////////////////////////////////////////////////// virtual EventToken installPeriodicEvent(EventLoop, Task*, double offset, double interval) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief rearms a periodic timer //////////////////////////////////////////////////////////////////////////////// virtual void rearmPeriodic(EventToken, double offset, double timeout) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief called to register a signal event //////////////////////////////////////////////////////////////////////////////// virtual EventToken installSignalEvent(EventLoop, Task*, int signal) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief called to unregister an event handler //////////////////////////////////////////////////////////////////////////////// virtual void uninstallEvent(EventToken) = 0; //////////////////////////////////////////////////////////////////////////////// /// @brief sends data to a task //////////////////////////////////////////////////////////////////////////////// virtual void signalTask(std::unique_ptr&) = 0; private: //////////////////////////////////////////////////////////////////////////////// /// @brief registers a new task //////////////////////////////////////////////////////////////////////////////// int registerTask(Task* task, ssize_t* got, ssize_t want); //////////////////////////////////////////////////////////////////////////////// /// @brief check whether a task can be inserted /// /// the caller must ensure the schedulerLock is held //////////////////////////////////////////////////////////////////////////////// int checkInsertTask(Task const*); protected: //////////////////////////////////////////////////////////////////////////////// /// @brief number of scheduler threads //////////////////////////////////////////////////////////////////////////////// size_t nrThreads; //////////////////////////////////////////////////////////////////////////////// /// @brief scheduler threads //////////////////////////////////////////////////////////////////////////////// SchedulerThread** threads; private: //////////////////////////////////////////////////////////////////////////////// /// @brief initializes the signal handlers for the scheduler //////////////////////////////////////////////////////////////////////////////// static void initializeSignalHandlers(); private: //////////////////////////////////////////////////////////////////////////////// /// @brief true if scheduler is shutting down //////////////////////////////////////////////////////////////////////////////// volatile sig_atomic_t stopping; // TODO(fc) XXX make this atomic //////////////////////////////////////////////////////////////////////////////// /// @brief true if scheduler is multi-threaded //////////////////////////////////////////////////////////////////////////////// bool multiThreading; //////////////////////////////////////////////////////////////////////////////// /// @brief round-robin for event loops //////////////////////////////////////////////////////////////////////////////// size_t nextLoop; //////////////////////////////////////////////////////////////////////////////// /// @brief lock for scheduler threads //////////////////////////////////////////////////////////////////////////////// basics::Mutex schedulerLock; //////////////////////////////////////////////////////////////////////////////// /// @brief tasks to thread //////////////////////////////////////////////////////////////////////////////// std::unordered_map task2thread; //////////////////////////////////////////////////////////////////////////////// /// @brief active tasks //////////////////////////////////////////////////////////////////////////////// std::unordered_map taskRegistered; //////////////////////////////////////////////////////////////////////////////// /// @brief scheduler activity flag //////////////////////////////////////////////////////////////////////////////// bool _active; }; } } #endif