dxx.sys.spawn source code

1 /*
2 
3 Taken from dlangide:
4 https://github.com/buggins/dlangide/blob/master/src/dlangide/builders/extprocess.d
5 
6 Copyright: 2018 Mark Fisher
7 
8 License:
9 Permission is hereby granted, free of charge, to any person obtaining a copy of
10 this software and associated documentation files (the "Software"), to deal in
11 the Software without restriction, including without limitation the rights to
12 use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
13 of the Software, and to permit persons to whom the Software is furnished to do
14 so, subject to the following conditions:
15 
16 The above copyright notice and this permission notice shall be included in all
17 copies or substantial portions of the Software.
18 
19 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20 IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
21 FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
22 AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
23 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
24 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
25 SOFTWARE.
26 */
27 /**
28  * Load and execute external programs buffering stdio.
29  **/
30 module dxx.sys.spawn;
31 
32 private import std.process;
33 private import std.stdio;
34 private import std.utf;
35 private import std.stdio;
36 private import core.thread;
37 private import core.sync.mutex;
38 
39 private import dxx.util;
40 
41 mixin __Text;
42 
43 /// interface to forward process output to
44 interface TextWriter {
45     /// log lines
46     void writeText(dstring text);
47 }
48 
49 /// interface to read text
50 interface TextReader {
51     /// log lines
52     dstring readText();
53 }
54 
55 /// protected text storage box to read and write text from different threads
56 class ProtectedTextStorage : TextReader, TextWriter {
57 
58     private Mutex _mutex;
59     private shared bool _closed;
60     private dchar[] _buffer;
61 
62     this() {
63         _mutex = new Mutex();
64     }
65 
66     @property bool closed() { return _closed; }
67 
68     void close() {
69         if (_closed)
70             return;
71         _closed = true;
72         _buffer = null;
73     }
74 
75     /// log lines
76     override void writeText(dstring text) {
77         if (!_closed) {
78             // if not closed
79             _mutex.lock();
80             scope(exit) _mutex.unlock();
81             // append text
82             _buffer ~= text;
83         }
84     }
85 
86     /// log lines
87     override dstring readText() {
88         if (!_closed) {
89             // if not closed
90             _mutex.lock();
91             scope(exit) _mutex.unlock();
92             if (!_buffer.length)
93                 return null;
94             dstring res = _buffer.dup;
95             _buffer = null;
96             return res;
97         } else {
98             // reading from closed
99             return null;
100         }
101     }
102 }
103 
104 enum ExternalProcessState : uint {
105     /// not initialized
106     None,
107     /// running
108     Running,
109     /// stop is requested
110     Stopping,
111     /// stopped
112     Stopped,
113     /// error occured, e.g. cannot run process
114     Error
115 }
116 
117 /// base class for text reading from std.stdio.File in background thread
118 class BackgroundReaderBase : Thread {
119     private std.stdio.File _file;
120     private shared bool _finished;
121     private ubyte[1] _byteBuffer;
122     private ubyte[] _bytes;
123     dchar[] _textbuffer;
124     private int _len;
125     private bool _utfError;
126 
127     this(std.stdio.File f) {
128         super(&run);
129         assert(f.isOpen());
130         _file = f;
131         _len = 0;
132         _finished = false;
133     }
134 
135     @property bool finished() {
136         return _finished;
137     }
138 
139     ubyte prevchar;
140     void addByte(ubyte data) {
141         if (_bytes.length < _len + 1)
142             _bytes.length = _bytes.length ? _bytes.length * 2 : 1024;
143         bool eolchar = (data == '\r' || data == '\n');
144         bool preveol = (prevchar == '\r' || prevchar == '\n');
145         _bytes[_len++] = data;
146         if (data == '\n')
147             flush();
148         //if ((eolchar && !preveol) || (!eolchar && preveol) || data == '\n') {
149         //    //Log.d("Flushing for prevChar=", prevchar, " newChar=", data);
150         //    flush();
151         //}
152         prevchar = data;
153     }
154     void flush() {
155         if (!_len)
156             return;
157         if (_textbuffer.length < _len)
158             _textbuffer.length = _len + 256;
159         size_t count = 0;
160         for(size_t i = 0; i < _len;) {
161             dchar ch = 0;
162             if (_utfError) {
163                 ch = _bytes[i++];
164             } else {
165                 try {
166                     ch = decode(cast(string)_bytes, i);
167                 } catch (UTFException e) {
168                     _utfError = true;
169                     ch = _bytes[i++];
170                     MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_NONUNICODE_PROC_OUTPUT));
171                 }
172             }
173             _textbuffer[count++] = ch;
174         }
175         _len = 0;
176 
177         if (!count)
178             return;
179 
180         // fix line endings - must be '\n'
181         count = convertLineEndings(_textbuffer[0..count]);
182 
183         // data is ready to send
184         if (count)
185             sendResult(_textbuffer[0..count].dup);
186     }
187     /// inplace convert line endings to unix format (\n)
188     size_t convertLineEndings(dchar[] text) {
189         size_t src = 0;
190         size_t dst = 0;
191         for(;src < text.length;) {
192             dchar ch = text[src++];
193             dchar nextch = src < text.length ? text[src] : 0;
194             if (ch == '\n') {
195                 if (nextch == '\r')
196                     src++;
197                 text[dst++] = '\n';
198             } else if (ch == '\r') {
199                 if (nextch == '\n')
200                     src++;
201                 text[dst++] = '\n';
202             } else {
203                 text[dst++] = ch;
204             }
205         }
206         return dst;
207     }
208     protected void sendResult(dstring text) {
209         // override to deal with ready data
210     }
211 
212     protected void handleFinish() {
213         // override to do something when thread is finishing
214     }
215 
216     private void run() {
217         //Log.d("BackgroundReaderBase run() enter");
218         // read file by bytes
219         try {
220             version (Windows) {
221                 import core.sys.windows.windows;
222                 // separate version for windows as workaround for hanging rawRead
223                 HANDLE h = _file.windowsHandle;
224                 DWORD bytesRead = 0;
225                 DWORD err;
226                 for (;;) {
227                     BOOL res = ReadFile(h, _byteBuffer.ptr, 1, &bytesRead, null);
228                     if (res) {
229                         if (bytesRead == 1)
230                             addByte(_byteBuffer[0]);
231                     } else {
232                         err = GetLastError();
233                         if (err == ERROR_MORE_DATA) {
234                             if (bytesRead == 1)
235                                 addByte(_byteBuffer[0]);
236                             continue;
237                         }
238                         //if (err == ERROR_BROKEN_PIPE || err = ERROR_INVALID_HANDLE)
239                         break;
240                     }
241                 }
242             } else {
243                 for (;;) {
244                     //Log.d("BackgroundReaderBase run() reading file");
245                     if (_file.eof)
246                         break;
247                     ubyte[] r = _file.rawRead(_byteBuffer);
248                     if (!r.length)
249                         break;
250                     //Log.d("BackgroundReaderBase run() read byte: ", r[0]);
251                     addByte(r[0]);
252                 }
253             }
254             _file.close();
255             flush();
256             //Log.d("BackgroundReaderBase run() closing file");
257             //Log.d("BackgroundReaderBase run() file closed");
258         } catch (Exception e) {
259             //Log.e("Exception occured while reading stream: ", e);
260         }
261         handleFinish();
262         _finished = true;
263         //Log.d("BackgroundReaderBase run() exit");
264     }
265 
266     void waitForFinish() {
267         static if (false) {
268             while (isRunning && !_finished)
269                 Thread.sleep( dur!("msecs")( 10 ) );
270         } else {
271             join(false);
272         }
273     }
274 
275 }
276 
277 /// reader which sends output text to TextWriter (warning: call will be made from background thread)
278 class BackgroundReader : BackgroundReaderBase {
279     protected TextWriter _destination;
280     this(std.stdio.File f, TextWriter destination) {
281         super(f);
282         assert(destination);
283         _destination = destination;
284     }
285     override protected void sendResult(dstring text) {
286         // override to deal with ready data
287         _destination.writeText(text);
288     }
289     override protected void handleFinish() {
290         // remove link to destination to help GC
291         _destination = null;
292     }
293 }
294 
295 /// runs external process, catches output, allows to stop
296 class ExternalProcess {
297 
298     protected char[][] _args;
299     protected char[] _workDir;
300     protected char[] _program;
301     protected string[string] _env;
302     protected TextWriter _stdout;
303     protected TextWriter _stderr;
304     protected BackgroundReader _stdoutReader;
305     protected BackgroundReader _stderrReader;
306     protected ProcessPipes _pipes;
307     protected ExternalProcessState _state;
308 
309     protected int _result;
310 
311     @property ExternalProcessState state() { return _state; }
312     /// returns process result for stopped process
313     @property int result() { return _result; }
314 
315     this() {
316     }
317 
318     ExternalProcessState run(string program, string[]args, string dir, TextWriter stdoutTarget, TextWriter stderrTarget = null) {
319         char[][] arguments;
320         foreach(a; args)
321             arguments ~= a.dup;
322         return run(program.dup, arguments, dir.dup, stdoutTarget, stderrTarget);
323     }
324     ExternalProcessState run(char[] program, char[][]args, char[] dir, TextWriter stdoutTarget, TextWriter stderrTarget = null) {
325         MsgLog.trace(MsgText!(DXXConfig.messages.MSG_PROC_RUN)(program,args));
326         _state = ExternalProcessState.None;
327         _program = findExecutablePath(cast(string)program).dup;
328         if (!_program) {
329             _state = ExternalProcessState.Error;
330             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_NOT_FOUND)(program));
331             return _state;
332         }
333         _args = args;
334         _workDir = dir;
335         _stdout = stdoutTarget;
336         _stderr = stderrTarget;
337         _result = 0;
338         assert(_stdout);
339         Redirect redirect;
340         char[][] params;
341         params ~= _program;
342         params ~= _args;
343         if (!_stderr)
344             redirect = Redirect.stdout | Redirect.stderrToStdout | Redirect.stdin;
345         else
346             redirect = Redirect.all;
347 //        sharedLog.info("Trying to run program ", _program, " with args ", _args);
348 //        MsgLog.trace(MsgText!(DXXConfig.messages.MSG_PROC_RUN)(program,args));
349         try {
350             _pipes = pipeProcess(params, redirect, _env, std.process.Config.suppressConsole, _workDir);
351             _state = ExternalProcessState.Running;
352             // start readers
353             _stdoutReader = new BackgroundReader(_pipes.stdout, _stdout);
354             _stdoutReader.start();
355             if (_stderr) {
356                 _stderrReader = new BackgroundReader(_pipes.stderr, _stderr);
357                 _stderrReader.start();
358             }
359         } catch (ProcessException e) {
360             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_RUN)(program,e));
361         } catch (std.stdio.StdioException e) {
362             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_REDIR)(program,e));
363         } catch (Throwable e) {
364             //sharedLog.error("Exception while trying to run program ", _program, " ", e);
365             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_UNKNOWN)(program,e));
366         }
367         return _state;
368     }
369 
370     protected void waitForReadingCompletion() {
371         try {
372             if (_stdoutReader && !_stdoutReader.finished) {
373                 _pipes.stdout.detach();
374                 //Log.d("waitForReadingCompletion - waiting for stdout");
375                 _stdoutReader.waitForFinish();
376                 //Log.d("waitForReadingCompletion - joined stdout");
377             }
378             _stdoutReader = null;
379         } catch (Exception e) {
380             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_WAITING_STDOUT)(_program,e));
381         }
382         try {
383             if (_stderrReader && !_stderrReader.finished) {
384                 _pipes.stderr.detach();
385                 //Log.d("waitForReadingCompletion - waiting for stderr");
386                 _stderrReader.waitForFinish();
387                 _stderrReader = null;
388                 //Log.d("waitForReadingCompletion - joined stderr");
389             }
390         } catch (Exception e) {
391             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_WAITING_STDERR)(_program,e));
392         }
393         //Log.d("waitForReadingCompletion - done");
394     }
395 
396     /// polls all available output from process streams
397     ExternalProcessState poll() {
398         //Log.d("ExternalProcess.poll state = ", _state);
399         bool res = true;
400         if (_state == ExternalProcessState.Error || _state == ExternalProcessState.None || _state == ExternalProcessState.Stopped)
401             return _state;
402         // check for process finishing
403         try {
404             auto pstate = std.process.tryWait(_pipes.pid);
405             if (pstate.terminated) {
406                 _state = ExternalProcessState.Stopped;
407                 _result = pstate.status;
408                 waitForReadingCompletion();
409             }
410         } catch (Exception e) {
411             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_WAITING)(_program));
412             _state = ExternalProcessState.Error;
413         }
414         return _state;
415     }
416 
417     /// waits until termination
418     ExternalProcessState wait() {
419         MsgLog.info(MsgText!(DXXConfig.messages.MSG_PROC_WAITING));
420         if (_state == ExternalProcessState.Error || _state == ExternalProcessState.None || _state == ExternalProcessState.Stopped)
421             return _state;
422         try {
423             _result = std.process.wait(_pipes.pid);
424             _state = ExternalProcessState.Stopped;
425             MsgLog.trace(MsgText!(DXXConfig.messages.MSG_PROC_READWAITING));
426             waitForReadingCompletion();
427         } catch (Exception e) {
428             MsgLog.error(MsgText!(DXXConfig.messages.MSG_ERR_PROC_UNKNOWN));
429             _state = ExternalProcessState.Error;
430         }
431         return _state;
432     }
433 
434     /// request process stop
435     ExternalProcessState kill() {
436         MsgLog.info(MsgText!(DXXConfig.messages.MSG_PROC_KILL));
437 
438         if (_state == ExternalProcessState.Error || _state == ExternalProcessState.None || _state == ExternalProcessState.Stopped)
439             return _state;
440         if (_state == ExternalProcessState.Running) {
441             std.process.kill(_pipes.pid);
442             _state = ExternalProcessState.Stopping;
443         }
444         return _state;
445     }
446 
447     bool write(string data) {
448         if (_state == ExternalProcessState.Error || _state == ExternalProcessState.None || _state == ExternalProcessState.Stopped) {
449             return false;
450         } else {
451             //Log.d("writing ", data.length, " characters to stdin");
452             _pipes.stdin.write("", data);
453             _pipes.stdin.flush();
454             //_pipes.stdin.close();
455             return true;
456         }
457     }
458 }
459 private import std.algorithm;
460 private import std.process;
461 private import std.path;
462 private import std.file;
463 private import std.utf;
464 
465 /// for executable name w/o path, find absolute path to executable
466 string findExecutablePath(string executableName) {
467     import std..string : split;
468     version (Windows) {
469         if (!executableName.endsWith(".exe"))
470             executableName = executableName ~ ".exe";
471     }
472     if(exists(executableName) && isFile(executableName)) return executableName;
473     string currentExeDir = dirName(thisExePath());
474     string inCurrentExeDir = absolutePath(buildNormalizedPath(currentExeDir, executableName));
475     if (exists(inCurrentExeDir) && isFile(inCurrentExeDir))
476         return inCurrentExeDir; // found in current directory
477     string pathVariable = environment.get("PATH");
478     if (!pathVariable)
479         return null;
480     string[] paths = pathVariable.split(pathSeparator);
481     foreach(path; paths) {
482         string pathname = absolutePath(buildNormalizedPath(path, executableName));
483         if (exists(pathname) && isFile(pathname))
484             return pathname;
485     }
486     return null;
487 }