diggler.commandqueue source code

1 module diggler.commandqueue;
2 
3 final class CommandQueue
4 {
5 	import core.thread : Fiber;
6 
7 	import std.algorithm;
8 	import std.array;
9 	import std.exception : enforce;
10 	debug(CommandQueue) import std.stdio;
11 
12 	import diggler.context;
13 	import diggler.command;
14 
15 	private: // TODO: use std.container.Array?
16 	static struct Task
17 	{
18 		ICommandSet cmdSet;
19 		Context context;
20 		void delegate() entryPoint;
21 
22 		void prepare()
23 		{
24 			cmdSet.context = context;
25 		}
26 	}
27 
28 	Task[] queue;
29 	CommandFiber[] fibers;
30 
31 	public:
32 	class CommandFiber : Fiber
33 	{
34 		import std.typecons : Nullable;
35 		Nullable!Task currentTask;
36 		bool ready = true;
37 
38 		void run() // Never to be called directly
39 		{
40 			ready = false;
41 			scope(exit) ready = true;
42 
43 			if(!currentTask.isNull) // Can be set directly before calling fiber
44 			{
45 				currentTask.prepare();
46 				currentTask.entryPoint();
47 			}
48 
49 			while(!queue.empty)
50 			{
51 				auto task = queue.front;
52 				queue.popFront();
53 
54 				currentTask = task;
55 				task.prepare();
56 				task.entryPoint();
57 			}
58 		}
59 
60 		void resume()
61 		{
62 			debug(CommandQueue) writefln("resuming a fiber %s %s", fibers.map!(fiber => fiber.state).array, fibers.map!(fiber => fiber.ready).array);
63 			if(state == State.HOLD)
64 			{
65 				currentTask.prepare();
66 				call();
67 			}
68 		}
69 
70 		this()
71 		{
72 			super(&run);
73 		}
74 	}
75 
76 	this(size_t numFibers = 4)
77 	{
78 		fibers = new CommandFiber[](numFibers);
79 
80 		foreach(ref fiber; fibers) // TODO: grow lazily up to a max?
81 			fiber = new CommandFiber();
82 	}
83 
84 	CommandFiber fiber() @property
85 	{
86 		return enforce(cast(CommandFiber)Fiber.getThis());
87 	}
88 
89 	void post(ICommandSet cmdSet, Context ctx, void delegate() cb)
90 	{
91 		auto task = Task(cmdSet, ctx, cb);
92 
93 		import std.stdio;
94 
95 		auto availableFibers = fibers.find!(fiber => fiber.ready)();//fibers.find!(fiber => fiber.state == Fiber.State.TERM);
96 		if(!availableFibers.empty)
97 		{
98 			debug(CommandQueue) writefln("Found available fiber (#%s), running command right away %s", fibers.length - availableFibers.length + 1,
99 					 fibers.map!(fiber => fiber.ready).array);
100 			auto availableFiber = availableFibers.front;
101 			availableFiber.currentTask = task;
102 			availableFiber.reset();
103 			availableFiber.call();
104 		}
105 		else
106 		{
107 			queue ~= task;
108 			debug(CommandQueue) writeln("no available fiber, queued command");
109 		}
110 	}
111 }