dinu.loader.output source code

1 module dinu.loader.output;
2 
3 
4 import dinu;
5 
6 
7 shared class CommandResult {
8 	long result;
9 	size_t occurrences;
10 	this(){
11 		result = long.max;
12 	}
13 }
14 
15 
16 shared CommandResult[int] running;
17 
18 
19 class OutputLoader: ChoiceLoader {
20 
21 	int[string] loaded;
22 
23 	override void run(){
24 		auto log = options.configPath ~ ".log";
25 		while(!log.exists && active && runProgram)
26 			core.thread.Thread.sleep(10.msecs);
27 		loaded = loaded.init;
28 		auto file = new BufferedFile(log);
29 		file.seekEnd(0);
30 		size_t idx = 10000;
31 		task(&loadBackwards, idx-1000).executeInNewThread;
32 		task(&loadBackwardsExec, idx-1000).executeInNewThread;
33 		while(active && runProgram){
34 			if(!file.eof){
35 				auto line = cast(string)file.readLine;
36 				if(!line.length)
37 					continue;
38 				matchLine(line, idx++);
39 				matchLineExec(line, true);
40 			}else
41 				core.thread.Thread.sleep(5.msecs);
42 		}
43 	}
44 
45 	void loadBackwards(size_t idx){
46 		auto p = pipeProcess(["tac", options.configPath ~ ".log"], Redirect.stdout);
47 		foreach(line; p.stdout.byLine){
48 			matchLine(to!string(line), idx--);
49 			if(idx<10000-2000)
50 				core.thread.Thread.sleep(4.msecs);
51 			if(idx==0 || !active)
52 				break;
53 		}
54 		scope(exit)
55 			p.pid.kill;
56 	}
57 
58 	void loadBackwardsExec(size_t idx){
59 		auto p = pipeProcess(["tac", options.configPath ~ ".exec"], Redirect.stdout);
60 		foreach(line; p.stdout.byLine){
61 			matchLineExec(to!string(line), false);
62 			if(idx<10000-2000)
63 				core.thread.Thread.sleep(4.msecs);
64 			if(idx==0 || !active)
65 				break;
66 		}
67 		scope(exit)
68 			p.pid.kill;
69 	}
70 
71 	void matchLine(string line, size_t idx){
72 		try{
73 			foreach(match; line.matchAll(`([0-9]+) (\S+)(?: (.*))?`)){
74 				auto pid = to!int(match.captures[1]);
75 				if((match.captures[2] == "stdout" || match.captures[2] == "stderr") && match.captures[3].length){
76 					add(new immutable CommandOutput(pid, match.captures[3], idx, match.captures[2]=="stderr"));
77 				}
78 			}
79 		}catch(Exception e){
80 			writeln(e);
81 		}
82 	}
83 
84 	void matchLineExec(string line, bool forward){
85 		try{
86 			foreach(match; line.matchAll(`([0-9]+) (\S+)(?: (.*))?`)){
87 				auto pid = to!int(match.captures[1]);
88 				if(match.captures[2] == "exec"){
89 					auto cmd = match.captures[3].bangSplit;
90 					/+
91 					if(!cmd[2].length)
92 						continue;
93 					+/
94 					if(pid !in running)
95 						running[pid] = new shared CommandResult;
96 					if(!exists("/proc/%s".format(pid)) && running[pid].result == long.max)
97 						running[pid].result = 0;
98 					if(cmd[1] ~ cmd[2] in loaded){
99 						auto l = loaded[cmd[1].idup ~ cmd[2].idup];
100 						running[l].occurrences.atomicOp!"+="(1);
101 						continue;
102 					}
103 					auto history = new immutable CommandHistory(pid, to!Type(cmd[0]), cmd[1].dup, cmd[2].dup);
104 					loaded[cmd[1].idup ~ cmd[2].idup] = pid;
105 					running[pid].occurrences = 0;
106 					add(history);
107 				}else if(match.captures[2] == "exit"){
108 					if(pid !in running)
109 						running[pid] = new shared CommandResult;
110 					running[pid].result = to!int(match.captures[3]);
111 				}
112 			}
113 		}catch(Exception e){
114 			writeln(e);
115 		}
116 	}
117 }