1 /**
2 Copyright: Copyright (c) 2020, Joakim Brännström. All rights reserved.
3 License: $(LINK2 http://www.boost.org/LICENSE_1_0.txt, Boost Software License 1.0)
4 Author: Joakim Brännström (joakim.brannstrom@gmx.com)
5 */
6 module proc.process;
7 
8 import core.sys.posix.signal : SIGKILL;
9 import core.thread : Thread;
10 import core.time : dur, Duration;
11 import logger = std.experimental.logger;
12 import std.algorithm : filter, count, joiner, map;
13 import std.array : appender, empty, array;
14 import std.exception : collectException;
15 import std.stdio : File, fileno, writeln;
16 import std.typecons : Flag, Yes;
17 static import std.process;
18 static import std.stdio;
19 
20 import my.gc.refc;
21 import my.from_;
22 import my.path;
23 import my.named_type;
24 
25 public import proc.channel;
26 public import proc.pid;
27 public import proc.tty;
28 
29 version (unittest) {
30     import std.file : remove;
31 }
32 
33 /** Manage a process by reference counting so that it is terminated when the it
34  * stops being used such as the instance going out of scope.
35  */
36 auto rcKill(T)(T p, int signal = SIGKILL) {
37     return ScopeKill!T(p, signal);
38 }
39 
40 // backward compatibility.
41 alias scopeKill = rcKill;
42 
43 struct ScopeKill(T) {
44     private {
45         static struct Payload {
46             T process;
47             int signal = SIGKILL;
48             bool hasProcess;
49 
50             ~this() @safe {
51                 if (hasProcess)
52                     process.dispose();
53             }
54         }
55 
56         RefCounted!Payload payload_;
57     }
58 
59     alias process this;
60     ref T process() {
61         return payload_.get.process;
62     }
63 
64     this(T process, int signal) @safe {
65         payload_ = refCounted(Payload(process, signal, true));
66     }
67 }
68 
69 /// Async process wrapper for a std.process SpawnProcess
70 struct SpawnProcess {
71     import core.sys.posix.signal : SIGKILL;
72     import std.algorithm : among;
73 
74     private {
75         enum State {
76             running,
77             terminated,
78             exitCode
79         }
80 
81         std.process.Pid process;
82         RawPid pid;
83         int status_;
84         State st;
85     }
86 
87     this(std.process.Pid process) @safe {
88         this.process = process;
89         this.pid = process.osHandle.RawPid;
90     }
91 
92     ~this() @safe {
93     }
94 
95     /// Returns: The raw OS handle for the process ID.
96     RawPid osHandle() nothrow @safe {
97         return pid;
98     }
99 
100     /// Kill and cleanup the process.
101     void dispose() @safe {
102         final switch (st) {
103         case State.running:
104             this.kill;
105             this.wait;
106             break;
107         case State.terminated:
108             this.wait;
109             break;
110         case State.exitCode:
111             break;
112         }
113 
114         st = State.exitCode;
115     }
116 
117     /** Send `signal` to the process.
118      *
119      * Param:
120      *  signal = a signal from `core.sys.posix.signal`
121      */
122     void kill(int signal = SIGKILL) nothrow @trusted {
123         final switch (st) {
124         case State.running:
125             break;
126         case State.terminated:
127             goto case;
128         case State.exitCode:
129             return;
130         }
131 
132         try {
133             std.process.kill(process, signal);
134         } catch (Exception e) {
135         }
136 
137         st = State.terminated;
138     }
139 
140     /// Blocking wait for the process to terminated.
141     /// Returns: the exit status.
142     int wait() @safe {
143         final switch (st) {
144         case State.running:
145             status_ = std.process.wait(process);
146             break;
147         case State.terminated:
148             status_ = std.process.wait(process);
149             break;
150         case State.exitCode:
151             break;
152         }
153 
154         st = State.exitCode;
155 
156         return status_;
157     }
158 
159     /// Non-blocking wait for the process termination.
160     /// Returns: `true` if the process has terminated.
161     bool tryWait() @safe {
162         final switch (st) {
163         case State.running:
164             auto s = std.process.tryWait(process);
165             if (s.terminated) {
166                 st = State.exitCode;
167                 status_ = s.status;
168             }
169             break;
170         case State.terminated:
171             status_ = std.process.wait(process);
172             st = State.exitCode;
173             break;
174         case State.exitCode:
175             break;
176         }
177 
178         return st.among(State.terminated, State.exitCode) != 0;
179     }
180 
181     /// Returns: The exit status of the process.
182     int status() @safe {
183         if (st != State.exitCode) {
184             throw new Exception(
185                     "Process has not terminated and wait/tryWait been called to collect the exit status");
186         }
187         return status_;
188     }
189 
190     /// Returns: If the process has terminated.
191     bool terminated() @safe {
192         return st.among(State.terminated, State.exitCode) != 0;
193     }
194 }
195 
196 /// Async process that do not block on read from stdin/stderr.
197 struct PipeProcess {
198     import std.algorithm : among;
199     import core.sys.posix.signal : SIGKILL;
200 
201     private {
202         enum State {
203             running,
204             terminated,
205             exitCode
206         }
207 
208         std.process.ProcessPipes process;
209         std.process.Pid pid;
210 
211         FileReadChannel stderr_;
212         FileReadChannel stdout_;
213         FileWriteChannel stdin_;
214         int status_;
215         State st;
216     }
217 
218     this(std.process.Pid pid, File stdin, File stdout, File stderr) @safe {
219         this.pid = pid;
220 
221         this.stdin_ = FileWriteChannel(stdin);
222         this.stdout_ = FileReadChannel(stdout);
223         this.stderr_ = FileReadChannel(stderr);
224     }
225 
226     this(std.process.ProcessPipes process, std.process.Redirect r) @safe {
227         this.process = process;
228         this.pid = process.pid;
229 
230         if (r & std.process.Redirect.stdin) {
231             stdin_ = FileWriteChannel(this.process.stdin);
232         }
233         if (r & std.process.Redirect.stdout) {
234             stdout_ = FileReadChannel(this.process.stdout);
235         }
236         if (r & std.process.Redirect.stderr) {
237             this.stderr_ = FileReadChannel(this.process.stderr);
238         }
239     }
240 
241     /// Returns: The raw OS handle for the process ID.
242     RawPid osHandle() nothrow @safe {
243         return pid.osHandle.RawPid;
244     }
245 
246     /// Access to stdout.
247     FileWriteChannel stdin() nothrow @safe {
248         return stdin_;
249     }
250 
251     /// Access to stdout.
252     FileReadChannel stdout() nothrow @safe {
253         return stdout_;
254     }
255 
256     /// Access stderr.
257     FileReadChannel stderr() nothrow @safe {
258         return stderr_;
259     }
260 
261     /// Kill and cleanup the process.
262     void dispose() @safe {
263         final switch (st) {
264         case State.running:
265             this.kill;
266             this.wait;
267             .destroy(process);
268             break;
269         case State.terminated:
270             this.wait;
271             .destroy(process);
272             break;
273         case State.exitCode:
274             break;
275         }
276 
277         st = State.exitCode;
278     }
279 
280     /** Send `signal` to the process.
281      *
282      * Param:
283      *  signal = a signal from `core.sys.posix.signal`
284      */
285     void kill(int signal = SIGKILL) nothrow @trusted {
286         final switch (st) {
287         case State.running:
288             break;
289         case State.terminated:
290             return;
291         case State.exitCode:
292             return;
293         }
294 
295         try {
296             std.process.kill(pid, signal);
297         } catch (Exception e) {
298         }
299 
300         st = State.terminated;
301     }
302 
303     /// Blocking wait for the process to terminated.
304     /// Returns: the exit status.
305     int wait() @safe {
306         final switch (st) {
307         case State.running:
308             status_ = std.process.wait(pid);
309             break;
310         case State.terminated:
311             status_ = std.process.wait(pid);
312             break;
313         case State.exitCode:
314             break;
315         }
316 
317         st = State.exitCode;
318 
319         return status_;
320     }
321 
322     /// Non-blocking wait for the process termination.
323     /// Returns: `true` if the process has terminated.
324     bool tryWait() @safe {
325         final switch (st) {
326         case State.running:
327             auto s = std.process.tryWait(pid);
328             if (s.terminated) {
329                 st = State.exitCode;
330                 status_ = s.status;
331             }
332             break;
333         case State.terminated:
334             status_ = std.process.wait(pid);
335             st = State.exitCode;
336             break;
337         case State.exitCode:
338             break;
339         }
340 
341         return st.among(State.terminated, State.exitCode) != 0;
342     }
343 
344     /// Returns: The exit status of the process.
345     int status() @safe {
346         if (st != State.exitCode) {
347             throw new Exception(
348                     "Process has not terminated and wait/tryWait been called to collect the exit status");
349         }
350         return status_;
351     }
352 
353     /// Returns: If the process has terminated.
354     bool terminated() @safe {
355         return st.among(State.terminated, State.exitCode) != 0;
356     }
357 }
358 
359 SpawnProcess spawnProcess(scope const(char[])[] args, File stdin = std.stdio.stdin,
360         File stdout = std.stdio.stdout, File stderr = std.stdio.stderr,
361         const string[string] env = null, std.process.Config config = std.process.Config.none,
362         scope const char[] workDir = null) {
363     return SpawnProcess(std.process.spawnProcess(args, stdin, stdout, stderr,
364             env, config, workDir));
365 }
366 
367 SpawnProcess spawnProcess(scope const(char[])[] args, const string[string] env,
368         std.process.Config config = std.process.Config.none, scope const(char)[] workDir = null) {
369     return SpawnProcess(std.process.spawnProcess(args, std.stdio.stdin,
370             std.stdio.stdout, std.stdio.stderr, env, config, workDir));
371 }
372 
373 SpawnProcess spawnProcess(scope const(char)[] program,
374         File stdin = std.stdio.stdin, File stdout = std.stdio.stdout,
375         File stderr = std.stdio.stderr, const string[string] env = null,
376         std.process.Config config = std.process.Config.none, scope const(char)[] workDir = null) {
377     return SpawnProcess(std.process.spawnProcess((&program)[0 .. 1], stdin,
378             stdout, stderr, env, config, workDir));
379 }
380 
381 SpawnProcess spawnShell(scope const(char)[] command, File stdin = std.stdio.stdin,
382         File stdout = std.stdio.stdout, File stderr = std.stdio.stderr,
383         scope const string[string] env = null, std.process.Config config = std.process.Config.none,
384         scope const(char)[] workDir = null, scope string shellPath = std.process.nativeShell) {
385     return SpawnProcess(std.process.spawnShell(command, stdin, stdout, stderr,
386             env, config, workDir, shellPath));
387 }
388 
389 /// ditto
390 SpawnProcess spawnShell(scope const(char)[] command, scope const string[string] env,
391         std.process.Config config = std.process.Config.none,
392         scope const(char)[] workDir = null, scope string shellPath = std.process.nativeShell) {
393     return SpawnProcess(std.process.spawnShell(command, env, config, workDir, shellPath));
394 }
395 
396 PipeProcess pipeProcess(scope const(char[])[] args,
397         std.process.Redirect redirect = std.process.Redirect.all,
398         const string[string] env = null, std.process.Config config = std.process.Config.none,
399         scope const(char)[] workDir = null) @safe {
400     return PipeProcess(std.process.pipeProcess(args, redirect, env, config, workDir), redirect);
401 }
402 
403 PipeProcess pipeShell(scope const(char)[] command,
404         std.process.Redirect redirect = std.process.Redirect.all,
405         const string[string] env = null, std.process.Config config = std.process.Config.none,
406         scope const(char)[] workDir = null, string shellPath = std.process.nativeShell) @safe {
407     return PipeProcess(std.process.pipeShell(command, redirect, env, config,
408             workDir, shellPath), redirect);
409 }
410 
411 /** Moves the process to a separate process group and on exit kill it and all
412  * its children.
413  */
414 @safe struct Sandbox(ProcessT) {
415     import core.sys.posix.signal : SIGKILL;
416 
417     private {
418         ProcessT p;
419         RawPid pid;
420     }
421 
422     this(ProcessT p) @safe {
423         import core.sys.posix.unistd : setpgid;
424 
425         this.p = p;
426         this.pid = p.osHandle;
427         setpgid(pid, 0);
428     }
429 
430     RawPid osHandle() nothrow @safe {
431         return pid;
432     }
433 
434     static if (__traits(hasMember, ProcessT, "stdin")) {
435         ref FileWriteChannel stdin() nothrow @safe {
436             return p.stdin;
437         }
438     }
439 
440     static if (__traits(hasMember, ProcessT, "stdout")) {
441         ref FileReadChannel stdout() nothrow @safe {
442             return p.stdout;
443         }
444     }
445 
446     static if (__traits(hasMember, ProcessT, "stderr")) {
447         ref FileReadChannel stderr() nothrow @safe {
448             return p.stderr;
449         }
450     }
451 
452     void dispose() @safe {
453         // this also reaps the children thus cleaning up zombies
454         this.kill;
455         p.dispose;
456     }
457 
458     /** Send `signal` to the process.
459      *
460      * Param:
461      *  signal = a signal from `core.sys.posix.signal`
462      */
463     void kill(int signal = SIGKILL) nothrow @safe {
464         // must first retrieve the submap because after the process is killed
465         // its children may have changed.
466         auto pmap = makePidMap.getSubMap(pid);
467 
468         p.kill(signal);
469 
470         // only kill and reap the children
471         pmap.remove(pid);
472         proc.pid.kill(pmap, Yes.onlyCurrentUser, signal).reap;
473     }
474 
475     int wait() @safe {
476         return p.wait;
477     }
478 
479     bool tryWait() @safe {
480         return p.tryWait;
481     }
482 
483     int status() @safe {
484         return p.status;
485     }
486 
487     bool terminated() @safe {
488         return p.terminated;
489     }
490 }
491 
492 auto sandbox(T)(T p) @safe {
493     return Sandbox!T(p);
494 }
495 
496 @("shall terminate a group of processes")
497 unittest {
498     import std.datetime.stopwatch : StopWatch, AutoStart;
499 
500     immutable scriptName = makeScript(`#!/bin/bash
501 sleep 10m &
502 sleep 10m &
503 sleep 10m
504 `);
505     scope (exit)
506         remove(scriptName);
507 
508     auto p = pipeProcess([scriptName]).sandbox.rcKill;
509     waitUntilChildren(p.osHandle, 3);
510     const preChildren = makePidMap.getSubMap(p.osHandle).remove(p.osHandle).length;
511     p.kill;
512     Thread.sleep(500.dur!"msecs"); // wait for the OS to kill the children
513     const postChildren = makePidMap.getSubMap(p.osHandle).remove(p.osHandle).length;
514 
515     assert(p.wait == -9);
516     assert(p.terminated);
517     assert(preChildren == 3);
518     assert(postChildren == 0);
519 }
520 
521 /** dispose the process after the timeout.
522  */
523 @safe struct Timeout(ProcessT) {
524     import core.sys.posix.signal : SIGKILL;
525     import core.thread;
526     import std.algorithm : among;
527     import std.datetime : Clock, Duration;
528 
529     private {
530         enum Msg {
531             none,
532             stop,
533             status,
534         }
535 
536         enum Reply {
537             none,
538             running,
539             normalDeath,
540             killedByTimeout,
541         }
542 
543         static struct Payload {
544             ProcessT p;
545             RawPid pid;
546             Background background;
547             Reply backgroundReply;
548         }
549 
550         RefCounted!Payload rc;
551     }
552 
553     this(ProcessT p, Duration timeout) @trusted {
554         import std.algorithm : move;
555 
556         auto pid = p.osHandle;
557         rc = refCounted(Payload(move(p), pid));
558         rc.get.background = new Background(&rc.get.p, timeout);
559         rc.get.background.isDaemon = true;
560         rc.get.background.start;
561     }
562 
563     ~this() @trusted {
564         rc.release;
565     }
566 
567     private static class Background : Thread {
568         import core.sync.condition : Condition;
569         import core.sync.mutex : Mutex;
570 
571         Duration timeout;
572         ProcessT* p;
573         Mutex mtx;
574         Msg[] msg;
575         Reply reply_;
576         RawPid pid;
577         int signal = SIGKILL;
578 
579         this(ProcessT* p, Duration timeout) {
580             this.p = p;
581             this.timeout = timeout;
582             this.mtx = new Mutex();
583             this.pid = p.osHandle;
584 
585             super(&run);
586         }
587 
588         void run() {
589             checkProcess(this.pid, this.timeout, this);
590         }
591 
592         void put(Msg msg) @trusted nothrow {
593             this.mtx.lock_nothrow();
594             scope (exit)
595                 this.mtx.unlock_nothrow();
596             this.msg ~= msg;
597         }
598 
599         Msg popMsg() @trusted nothrow {
600             this.mtx.lock_nothrow();
601             scope (exit)
602                 this.mtx.unlock_nothrow();
603             if (msg.empty)
604                 return Msg.none;
605             auto rval = msg[$ - 1];
606             msg = msg[0 .. $ - 1];
607             return rval;
608         }
609 
610         void setReply(Reply reply_) @trusted nothrow {
611             this.mtx.lock_nothrow();
612             scope (exit)
613                 this.mtx.unlock_nothrow();
614             this.reply_ = reply_;
615         }
616 
617         Reply reply() @trusted nothrow {
618             this.mtx.lock_nothrow();
619             scope (exit)
620                 this.mtx.unlock_nothrow();
621             return reply_;
622         }
623 
624         void setSignal(int signal) @trusted nothrow {
625             this.mtx.lock_nothrow();
626             scope (exit)
627                 this.mtx.unlock_nothrow();
628             this.signal = signal;
629         }
630 
631         void kill() @trusted nothrow {
632             this.mtx.lock_nothrow();
633             scope (exit)
634                 this.mtx.unlock_nothrow();
635             p.kill(signal);
636         }
637     }
638 
639     private static void checkProcess(RawPid p, Duration timeout, Background bg) nothrow {
640         import std.algorithm : max, min;
641         import std.variant : Variant;
642         static import core.sys.posix.signal;
643 
644         const stopAt = Clock.currTime + timeout;
645         // the purpose is to poll the process often "enough" that if it
646         // terminates early `Process` detects it fast enough. 1000 is chosen
647         // because it "feels good". the purpose
648         auto sleepInterval = min(50, max(1, timeout.total!"msecs" / 1000)).dur!"msecs";
649 
650         bool forceStop;
651         bool running = true;
652         while (running && Clock.currTime < stopAt) {
653             const msg = bg.popMsg;
654 
655             final switch (msg) {
656             case Msg.none:
657                 () @trusted { Thread.sleep(sleepInterval); }();
658                 break;
659             case Msg.stop:
660                 forceStop = true;
661                 running = false;
662                 break;
663             case Msg.status:
664                 bg.setReply(Reply.running);
665                 break;
666             }
667 
668             () @trusted {
669                 if (core.sys.posix.signal.kill(p, 0) == -1) {
670                     running = false;
671                 }
672             }();
673         }
674 
675         // may be children alive thus must ensure that the whole process tree
676         // is killed if this is a sandbox with a timeout.
677         bg.kill();
678 
679         if (!forceStop && Clock.currTime >= stopAt) {
680             bg.setReply(Reply.killedByTimeout);
681         } else {
682             bg.setReply(Reply.normalDeath);
683         }
684     }
685 
686     RawPid osHandle() nothrow @trusted {
687         return rc.get.pid;
688     }
689 
690     static if (__traits(hasMember, ProcessT, "stdin")) {
691         ref FileWriteChannel stdin() nothrow @safe {
692             return rc.get.p.stdin;
693         }
694     }
695 
696     static if (__traits(hasMember, ProcessT, "stdout")) {
697         ref FileReadChannel stdout() nothrow @safe {
698             return rc.get.p.stdout;
699         }
700     }
701 
702     static if (__traits(hasMember, ProcessT, "stderr")) {
703         ref FileReadChannel stderr() nothrow @trusted {
704             return rc.get.p.stderr;
705         }
706     }
707 
708     void dispose() @trusted {
709         if (rc.get.backgroundReply.among(Reply.none, Reply.running)) {
710             rc.get.background.put(Msg.stop);
711             rc.get.background.join;
712             rc.get.backgroundReply = rc.get.background.reply;
713         }
714         rc.get.p.dispose;
715     }
716 
717     /** Send `signal` to the process.
718      *
719      * Param:
720      *  signal = a signal from `core.sys.posix.signal`
721      */
722     void kill(int signal = SIGKILL) nothrow @trusted {
723         rc.get.background.setSignal(signal);
724         rc.get.background.kill();
725     }
726 
727     int wait() @trusted {
728         while (!this.tryWait) {
729             Thread.sleep(20.dur!"msecs");
730         }
731         return rc.get.p.wait;
732     }
733 
734     bool tryWait() @trusted {
735         return rc.get.p.tryWait;
736     }
737 
738     int status() @trusted {
739         return rc.get.p.status;
740     }
741 
742     bool terminated() @trusted {
743         return rc.get.p.terminated;
744     }
745 
746     bool timeoutTriggered() @trusted {
747         if (rc.get.backgroundReply.among(Reply.none, Reply.running)) {
748             rc.get.background.put(Msg.status);
749             rc.get.backgroundReply = rc.get.background.reply;
750         }
751         return rc.get.backgroundReply == Reply.killedByTimeout;
752     }
753 }
754 
755 auto timeout(T)(T p, Duration timeout_) @trusted {
756     return Timeout!T(p, timeout_);
757 }
758 
759 /// Returns when the process has pending data.
760 void waitForPendingData(ProcessT)(Process p) {
761     while (!p.pipe.hasPendingData || !p.stderr.hasPendingData) {
762         Thread.sleep(20.dur!"msecs");
763     }
764 }
765 
766 @("shall kill the process after the timeout")
767 unittest {
768     import std.datetime.stopwatch : StopWatch, AutoStart;
769 
770     auto p = pipeProcess(["sleep", "1m"]).timeout(100.dur!"msecs").rcKill;
771     auto sw = StopWatch(AutoStart.yes);
772     p.wait;
773     sw.stop;
774 
775     assert(sw.peek >= 100.dur!"msecs");
776     assert(sw.peek <= 500.dur!"msecs");
777     assert(p.wait == -9);
778     assert(p.terminated);
779     assert(p.status == -9);
780     assert(p.timeoutTriggered);
781 }
782 
783 struct DrainElement {
784     enum Type {
785         stdout,
786         stderr,
787     }
788 
789     Type type;
790     const(ubyte)[] data;
791 
792     /// Returns: iterates the data as an input range.
793     auto byUTF8() @safe pure nothrow const @nogc {
794         static import std.utf;
795 
796         return std.utf.byUTF!(const(char))(cast(const(char)[]) data);
797     }
798 
799     bool empty() @safe pure nothrow const @nogc {
800         return data.length == 0;
801     }
802 }
803 
804 /** A range that drains a process stdout/stderr until it terminates.
805  *
806  * There may be `DrainElement` that are empty.
807  */
808 struct DrainRange(ProcessT) {
809     private {
810         enum State {
811             start,
812             draining,
813             lastStdout,
814             lastStderr,
815             lastElement,
816             empty,
817         }
818 
819         ProcessT p;
820         DrainElement front_;
821         State st;
822         ubyte[] buf;
823     }
824 
825     this(ProcessT p) {
826         this.p = p;
827         this.buf = new ubyte[4096];
828     }
829 
830     DrainElement front() @safe pure nothrow const @nogc {
831         assert(!empty, "Can't get front of an empty range");
832         return front_;
833     }
834 
835     void popFront() @safe {
836         assert(!empty, "Can't pop front of an empty range");
837 
838         static bool isAnyPipeOpen(ref ProcessT p) {
839             return p.stdout.isOpen || p.stderr.isOpen;
840         }
841 
842         DrainElement readData(ref ProcessT p) @safe {
843             if (p.stderr.hasPendingData) {
844                 return DrainElement(DrainElement.Type.stderr, p.stderr.read(buf));
845             } else if (p.stdout.hasPendingData) {
846                 return DrainElement(DrainElement.Type.stdout, p.stdout.read(buf));
847             }
848             return DrainElement.init;
849         }
850 
851         DrainElement waitUntilData() @safe {
852             import std.datetime : Clock;
853 
854             // may livelock if the process never terminates and never writes to
855             // the terminal. timeout ensure that it sooner or later is break
856             // the loop. This is important if the drain is part of a timeout
857             // wrapping.
858 
859             const timeout = 100.dur!"msecs";
860             const stopAt = Clock.currTime + timeout;
861             const sleepFor = timeout / 20;
862             const useSleep = Clock.currTime + sleepFor;
863             bool running = true;
864             while (running) {
865                 const now = Clock.currTime;
866 
867                 running = (now < stopAt) && isAnyPipeOpen(p);
868 
869                 auto bufRead = readData(p);
870 
871                 if (!bufRead.empty) {
872                     return DrainElement(bufRead.type, bufRead.data.dup);
873                 } else if (running && now > useSleep && bufRead.empty) {
874                     import core.thread : Thread;
875 
876                     () @trusted { Thread.sleep(sleepFor); }();
877                 }
878             }
879 
880             return DrainElement.init;
881         }
882 
883         front_ = DrainElement.init;
884 
885         final switch (st) {
886         case State.start:
887             st = State.draining;
888             front_ = waitUntilData;
889             break;
890         case State.draining:
891             if (p.terminated) {
892                 st = State.lastStdout;
893             } else if (isAnyPipeOpen(p)) {
894                 front_ = waitUntilData();
895             } else {
896                 st = State.lastStdout;
897             }
898             break;
899         case State.lastStdout:
900             if (p.stdout.hasPendingData) {
901                 front_ = DrainElement(DrainElement.Type.stdout, p.stdout.read(buf).dup);
902             } else {
903                 st = State.lastStderr;
904             }
905             break;
906         case State.lastStderr:
907             if (p.stderr.hasPendingData) {
908                 front_ = DrainElement(DrainElement.Type.stderr, p.stderr.read(buf).dup);
909             } else {
910                 st = State.lastElement;
911             }
912             break;
913         case State.lastElement:
914             st = State.empty;
915             break;
916         case State.empty:
917             break;
918         }
919     }
920 
921     bool empty() @safe pure nothrow const @nogc {
922         return st == State.empty;
923     }
924 }
925 
926 /// Drain a process pipe until empty.
927 auto drain(T)(T p) {
928     return DrainRange!T(p);
929 }
930 
931 /// Read the data from a ReadChannel by line.
932 struct DrainByLineCopyRange(ProcessT) {
933     private {
934         enum State {
935             start,
936             draining,
937             lastLine,
938             lastBuf,
939             empty,
940         }
941 
942         ProcessT process;
943         DrainRange!ProcessT range;
944         State st;
945         const(ubyte)[] buf;
946         const(char)[] line;
947     }
948 
949     this(ProcessT p) {
950         process = p;
951         range = p.drain;
952     }
953 
954     string front() @trusted pure nothrow const @nogc {
955         import std.exception : assumeUnique;
956 
957         assert(!empty, "Can't get front of an empty range");
958         return line.assumeUnique;
959     }
960 
961     void popFront() @safe {
962         assert(!empty, "Can't pop front of an empty range");
963         import std.algorithm : countUntil;
964         import std.array : array;
965         static import std.utf;
966 
967         const(ubyte)[] updateBuf(size_t idx) {
968             const(ubyte)[] tmp;
969             if (buf.empty) {
970                 // do nothing
971             } else if (idx == -1) {
972                 tmp = buf;
973                 buf = null;
974             } else {
975                 idx = () {
976                     if (idx < buf.length) {
977                         return idx + 1;
978                     }
979                     return idx;
980                 }();
981                 tmp = buf[0 .. idx];
982                 buf = buf[idx .. $];
983             }
984 
985             if (!tmp.empty && tmp[$ - 1] == '\n') {
986                 tmp = tmp[0 .. $ - 1];
987             }
988             return tmp;
989         }
990 
991         void drainLine() {
992             void fillBuf() {
993                 if (!range.empty) {
994                     range.popFront;
995                 }
996                 if (!range.empty) {
997                     buf ~= range.front.data;
998                 }
999             }
1000 
1001             size_t idx;
1002             () {
1003                 int cnt;
1004                 do {
1005                     fillBuf();
1006                     idx = buf.countUntil('\n');
1007                     // 2 is a magic number which mean that it at most wait 2x timeout for data
1008                 }
1009                 while (!range.empty && idx == -1 && cnt++ < 2);
1010             }();
1011 
1012             if (idx != -1) {
1013                 auto tmp = updateBuf(idx);
1014                 line = std.utf.byUTF!(const(char))(cast(const(char)[]) tmp).array;
1015             }
1016         }
1017 
1018         bool lastLine() {
1019             size_t idx = buf.countUntil('\n');
1020             if (idx == -1)
1021                 return true;
1022 
1023             auto tmp = updateBuf(idx);
1024             line = std.utf.byUTF!(const(char))(cast(const(char)[]) tmp).array;
1025             return false;
1026         }
1027 
1028         line = null;
1029         final switch (st) {
1030         case State.start:
1031             drainLine;
1032             st = State.draining;
1033             break;
1034         case State.draining:
1035             drainLine;
1036             if (range.empty)
1037                 st = State.lastLine;
1038             break;
1039         case State.lastLine:
1040             if (lastLine)
1041                 st = State.lastBuf;
1042             break;
1043         case State.lastBuf:
1044             line = std.utf.byUTF!(const(char))(cast(const(char)[]) buf).array;
1045             st = State.empty;
1046             break;
1047         case State.empty:
1048             break;
1049         }
1050     }
1051 
1052     bool empty() @safe pure nothrow const @nogc {
1053         return st == State.empty;
1054     }
1055 }
1056 
1057 @("shall drain the process output by line")
1058 unittest {
1059     import std.algorithm : filter, joiner, map;
1060     import std.array : array;
1061 
1062     auto p = pipeProcess(["dd", "if=/dev/zero", "bs=10", "count=3"]).rcKill;
1063     auto res = p.process.drainByLineCopy.filter!"!a.empty".array;
1064 
1065     assert(res.length == 3);
1066     assert(res.joiner.count >= 30);
1067     assert(p.wait == 0);
1068     assert(p.terminated);
1069 }
1070 
1071 auto drainByLineCopy(T)(T p) {
1072     return DrainByLineCopyRange!T(p);
1073 }
1074 
1075 /// Drain the process output until it is done executing.
1076 auto drainToNull(T)(T p) {
1077     foreach (l; p.drain()) {
1078     }
1079     return p;
1080 }
1081 
1082 /// Drain the output from the process into an output range.
1083 auto drain(ProcessT, T)(ProcessT p, ref T range) {
1084     foreach (l; p.drain()) {
1085         range.put(l);
1086     }
1087     return p;
1088 }
1089 
1090 @("shall drain the output of a process while it is running with a separation of stdout and stderr")
1091 unittest {
1092     auto p = pipeProcess(["dd", "if=/dev/urandom", "bs=10", "count=3"]).rcKill;
1093     auto res = p.drain.array;
1094 
1095     // this is just a sanity check. It has to be kind a high because there is
1096     // some wiggleroom allowed
1097     assert(res.count > 1 && res.count <= 50);
1098 
1099     assert(res.filter!(a => a.type == DrainElement.Type.stdout)
1100             .map!(a => a.data)
1101             .joiner
1102             .count == 30);
1103     assert(p.wait == 0);
1104     assert(p.terminated);
1105 }
1106 
1107 @("shall kill the process tree when the timeout is reached")
1108 unittest {
1109     immutable script = makeScript(`#!/bin/bash
1110 sleep 10m
1111 `);
1112     scope (exit)
1113         remove(script);
1114 
1115     auto p = pipeProcess([script]).sandbox.timeout(1.dur!"seconds").rcKill;
1116     waitUntilChildren(p.osHandle, 1);
1117     const preChildren = makePidMap.getSubMap(p.osHandle).remove(p.osHandle).length;
1118     const res = p.process.drain.array;
1119     const postChildren = makePidMap.getSubMap(p.osHandle).remove(p.osHandle).length;
1120 
1121     assert(p.wait == -9);
1122     assert(p.terminated);
1123     assert(preChildren == 1);
1124     assert(postChildren == 0);
1125 }
1126 
1127 string makeScript(string script, string file = __FILE__, uint line = __LINE__) {
1128     import core.sys.posix.sys.stat;
1129     import std.file : getAttributes, setAttributes, thisExePath;
1130     import std.stdio : File;
1131     import std.path : baseName;
1132     import std.conv : to;
1133 
1134     immutable fname = thisExePath ~ "_" ~ file.baseName ~ line.to!string ~ ".sh";
1135 
1136     File(fname, "w").writeln(script);
1137     setAttributes(fname, getAttributes(fname) | S_IXUSR | S_IXGRP | S_IXOTH);
1138     return fname;
1139 }
1140 
1141 /// Wait for p to have num children or fail after 10s.
1142 void waitUntilChildren(RawPid p, int num) {
1143     import std.datetime : Clock;
1144 
1145     const failAt = Clock.currTime + 10.dur!"seconds";
1146     do {
1147         Thread.sleep(50.dur!"msecs");
1148         if (Clock.currTime > failAt)
1149             break;
1150     }
1151     while (makePidMap.getSubMap(p).remove(p).length < num);
1152 }
1153 
1154 alias Address = NamedType!(ulong, Tag!"Address", ulong.init, TagStringable);
1155 struct AddressMap {
1156     Address begin;
1157     Address end;
1158     NamedType!(string, Tag!"Permission", string.init, TagStringable) perm;
1159 }
1160 
1161 /** An assoc array mapping the path of each shared library loaded by
1162  * the process to the address it is loaded at in the process address space.
1163  */
1164 AddressMap[][AbsolutePath] libs(RawPid pid) nothrow {
1165     import std.stdio : File;
1166     import std.format : formattedRead, format;
1167     import std.algorithm : countUntil;
1168     import std.typecons : tuple;
1169     import std.string : strip;
1170 
1171     typeof(return) rval;
1172 
1173     try {
1174         foreach (l; File(format!"/proc/%s/maps"(pid)).byLine
1175                 .map!(a => tuple(a, a.countUntil('/')))
1176                 .filter!(a => a[1] != -1)) {
1177             try {
1178                 auto p = AbsolutePath(l[0][l[1] .. $].idup);
1179                 auto m = l[0][0 .. l[1]].strip;
1180                 ulong begin;
1181                 ulong end;
1182                 char[] perm;
1183                 if (formattedRead!"%x-%x %s "(m, begin, end, perm) != 3) {
1184                     continue;
1185                 }
1186 
1187                 auto amap = AddressMap(Address(begin), Address(end),
1188                         typeof(AddressMap.init.perm)(perm.idup));
1189                 if (auto v = p in rval) {
1190                     *v ~= amap;
1191                 } else {
1192                     rval[p] = [amap];
1193                 }
1194             } catch (Exception e) {
1195             }
1196         }
1197     } catch (Exception e) {
1198         logger.trace(e.msg).collectException;
1199     }
1200 
1201     return rval;
1202 }
1203 
1204 @("shall read the libraries the process is using")
1205 unittest {
1206     immutable scriptName = makeScript(`#!/bin/bash
1207 sleep 10m &
1208 sleep 10m &
1209 sleep 10m
1210 `);
1211     scope (exit)
1212         remove(scriptName);
1213 
1214     auto p = pipeProcess([scriptName]).sandbox.rcKill;
1215     waitUntilChildren(p.osHandle, 3);
1216     auto res = libs(p.osHandle);
1217     p.kill;
1218 
1219     assert(res.length != 0);
1220     assert(AbsolutePath("/bin/bash") in res);
1221 }
1222 
1223 /** Parses the output from a run of 'ldd' on a binary.
1224  *
1225  * Params:
1226  *  input = an input range of lines which is the output from ldd
1227  *
1228  * Example
1229  * ---
1230  * writeln(parseLddOutput(`
1231  *     linux-vdso.so.1 =>  (0x00007fffbf5fe000)
1232  *     libtinfo.so.5 => /lib/x86_64-linux-gnu/libtinfo.so.5 (0x00007fe28117f000)
1233  *     libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007fe280f7b000)
1234  *     libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007fe280bb4000)
1235  *     /lib64/ld-linux-x86-64.so.2 (0x00007fe2813dd000)
1236  * `))
1237  * ---
1238  *
1239  * Returns: a dictionary of {path: address} for each library required by the
1240  * specified binary.
1241  */
1242 Address[AbsolutePath] parseLddOutput(T)(T input) if (std_.range.isInputRange!T) {
1243     import std.regex : regex, matchFirst;
1244     import std.format : formattedRead;
1245 
1246     typeof(return) rval;
1247     const reLinux = regex(`\s(?P<lib>\S?/\S+)\s+\((?P<addr>0x.+)\)`);
1248 
1249     foreach (l; input) {
1250         auto m = matchFirst(l, reLinux);
1251         if (m.empty || m.length < 3) {
1252             continue;
1253         }
1254 
1255         try {
1256             ulong addr;
1257             formattedRead!"0x%x"(m["addr"], addr);
1258             rval[AbsolutePath(m["lib"])] = Address(addr);
1259         } catch (Exception e) {
1260         }
1261     }
1262 
1263     return rval;
1264 }
1265 
1266 @("shall parse the output of ldd")
1267 unittest {
1268     auto res = parseLddOutput([
1269             "linux-vdso.so.1 =>  (0x00007fffbf5fe000)",
1270             "libtinfo.so.5 => /lib/x86_64-linux-gnu/libtinfo.so.5 (0x00007fe28117f000)",
1271             "libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007fe280f7b000)",
1272             "libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007fe280bb4000)",
1273             "/lib64/ld-linux-x86-64.so.2 (0x00007fe2813dd000)"
1274             ]);
1275     assert(res.length == 3);
1276     assert(res[AbsolutePath("/lib/x86_64-linux-gnu/libtinfo.so.5")].get == 140610805166080);
1277 }