1 // The MIT License (MIT)
2 //
3 // 	Copyright (c) 2015 Sergey Makeev, Vadim Slyusarev
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 #pragma once
24 
25 #include <MTTools.h>
26 #include <MTPlatform.h>
27 #include <MTConcurrentQueueLIFO.h>
28 #include <MTStackArray.h>
29 #include <MTArrayView.h>
30 #include <MTThreadContext.h>
31 #include <MTFiberContext.h>
32 #include <MTTaskBase.h>
33 
34 #ifdef MT_INSTRUMENTED_BUILD
35 #include <MTMicroWebSrv.h>
36 #endif
37 
38 namespace MT
39 {
40 	const uint32 MT_MAX_THREAD_COUNT = 64;
41 	const uint32 MT_MAX_FIBERS_COUNT = 128;
42 	const uint32 MT_MAX_GROUPS_COUNT = 256;
43 	const uint32 MT_SCHEDULER_STACK_SIZE = 131072;
44 	const uint32 MT_FIBER_STACK_SIZE = 32768;
45 
46 	namespace internal
47 	{
48 		struct ThreadContext;
49 	}
50 
51 	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
52 	// Task scheduler
53 	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
54 	class TaskScheduler
55 	{
56 		friend class FiberContext;
57 		friend struct internal::ThreadContext;
58 
59 
60 
61 		////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
62 		// Task group description
63 		////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
64 		// Application can assign task group to task and later wait until group was finished.
65 		class TaskGroupDescription
66 		{
67 			AtomicInt inProgressTaskCount;
68 			Event allDoneEvent;
69 
70 			//Tasks awaiting group through FiberContext::WaitGroupAndYield call
71 			ConcurrentQueueLIFO<FiberContext*> waitTasksQueue;
72 
73 		public:
74 
75 			bool debugIsFree;
76 
77 
78 		private:
79 
80 			TaskGroupDescription(TaskGroupDescription& ) {}
81 			void operator=(const TaskGroupDescription&) {}
82 
83 		public:
84 
85 			TaskGroupDescription()
86 			{
87 				inProgressTaskCount.Set(0);
88 				allDoneEvent.Create( EventReset::MANUAL, true );
89 				debugIsFree = true;
90 			}
91 
92 			int GetTaskCount() const { return inProgressTaskCount.Get(); }
93 			ConcurrentQueueLIFO<FiberContext*> & GetWaitQueue() { return waitTasksQueue; }
94 			int Dec() { return inProgressTaskCount.Dec(); }
95 			int Inc() { return inProgressTaskCount.Inc(); }
96 			int Add(int sum) { return inProgressTaskCount.Add(sum); }
97 			void Signal() { allDoneEvent.Signal(); }
98 			void Reset() { allDoneEvent.Reset(); }
99 			bool Wait(uint32 milliseconds) { return allDoneEvent.Wait(milliseconds); }
100 		};
101 
102 
103 		// Thread index for new task
104 		AtomicInt roundRobinThreadIndex;
105 
106 		// Started threads count
107 		AtomicInt startedThreadsCount;
108 
109 		// Threads created by task manager
110 		volatile uint32 threadsCount;
111 		internal::ThreadContext threadContext[MT_MAX_THREAD_COUNT];
112 
113 		// All groups task statistic
114 		TaskGroupDescription allGroups;
115 
116 		// Groups pool
117 		ConcurrentQueueLIFO<TaskGroup> availableGroups;
118 
119 		//
120 		TaskGroupDescription groupStats[MT_MAX_GROUPS_COUNT];
121 
122 		// Fibers pool
123 		ConcurrentQueueLIFO<FiberContext*> availableFibers;
124 
125 		// Fibers context
126 		FiberContext fiberContext[MT_MAX_FIBERS_COUNT];
127 
128 #ifdef MT_INSTRUMENTED_BUILD
129 		int32 webServerPort;
130 		profile::MicroWebServer profilerWebServer;
131 		int64 startTime;
132 #endif
133 
134 		FiberContext* RequestFiberContext(internal::GroupedTask& task);
135 		void ReleaseFiberContext(FiberContext* fiberExecutionContext);
136 		void RunTasksImpl(ArrayView<internal::TaskBucket>& buckets, FiberContext * parentFiber, bool restoredFromAwaitState);
137 		TaskGroupDescription & GetGroupDesc(TaskGroup group);
138 
139 		static void ThreadMain( void* userData );
140 		static void FiberMain( void* userData );
141 		static bool TryStealTask(internal::ThreadContext& threadContext, internal::GroupedTask & task, uint32 workersCount);
142 
143 		static FiberContext* ExecuteTask (internal::ThreadContext& threadContext, FiberContext* fiberContext);
144 
145 	public:
146 
147 		/// \brief Initializes a new instance of the TaskScheduler class.
148 		/// \param workerThreadsCount Worker threads count. Automatically determines the required number of threads if workerThreadsCount set to 0
149 		TaskScheduler(uint32 workerThreadsCount = 0);
150 		~TaskScheduler();
151 
152 		template<class TTask>
153 		void RunAsync(TaskGroup group, TTask* taskArray, uint32 taskCount);
154 
155 		bool WaitGroup(TaskGroup group, uint32 milliseconds);
156 		bool WaitAll(uint32 milliseconds);
157 
158 		TaskGroup CreateGroup();
159 		void ReleaseGroup(TaskGroup group);
160 
161 		bool IsEmpty();
162 
163 		uint32 GetWorkerCount() const;
164 
165 		bool IsWorkerThread() const;
166 
167 #ifdef MT_INSTRUMENTED_BUILD
168 
169 		size_t GetProfilerEvents(uint32 workerIndex, MT::ProfileEventDesc * dstBuffer, size_t dstBufferSize);
170 		void UpdateProfiler();
171 		int32 GetWebServerPort() const;
172 
173 		inline int64 GetStartTime() const
174 		{
175 			return startTime;
176 		}
177 
178 #endif
179 	};
180 }
181 
182 #include "MTScheduler.inl"
183 #include "MTFiberContext.inl"
184