ocean.sys.SafeFork source code

1 /*******************************************************************************
2 
3     SafeFork
4 
5     Offers some wrappers for the usage of fork to call expensive blocking
6     functions without interrupting the main process and without the need to
7     synchronize.
8 
9     Useful version switches:
10         TimeFork = measures and displays the time taken by the linux fork() call
11 
12     Copyright:
13         Copyright (c) 2009-2016 dunnhumby Germany GmbH.
14         All rights reserved.
15 
16     License:
17         Boost Software License Version 1.0. See LICENSE_BOOST.txt for details.
18         Alternatively, this file may be distributed under the terms of the Tango
19         3-Clause BSD License (see LICENSE_BSD.txt for details).
20 
21 *******************************************************************************/
22 
23 module ocean.sys.SafeFork;
24 
25 import ocean.sys.ErrnoException;
26 
27 import core.stdc.errno;
28 import core.stdc.string;
29 import core.sys.posix.stdlib : exit;
30 import core.sys.posix.unistd : fork;
31 import core.sys.posix.signal;
32 import core.sys.posix.sys.wait;
33 
34 version ( TimeFork )
35 {
36     import ocean.io.Stdout;
37     import ocean.time.StopWatch;
38 }
39 
40 
41 /*******************************************************************************
42 
43     External C
44 
45     // TODO: forced to be public to be used with reflection, must be moved to
46         C bindings
47 
48 *******************************************************************************/
49 
50 extern (C)
51 {
52     enum idtype_t
53     {
54       P_ALL,        /* Wait for any child.  */
55       P_PID,        /* Wait for specified process.  */
56       P_PGID        /* Wait for members of process group.  */
57     };
58 
59     int waitid(idtype_t, id_t, siginfo_t*, int);
60 
61     static immutable WEXITED = 0x00000004;
62     static immutable WNOWAIT = 0x01000000;
63 }
64 
65 
66 
67 /*******************************************************************************
68 
69     SafeFork
70 
71     Offers some wrappers for the usage of fork to call expensive blocking
72     functions without interrupting the main process and without the need to
73     synchronize.
74 
75     Usage Example:
76     -----
77     import ocean.sys.SafeFork;
78 
79     void main ( )
80     {
81         auto dont_block = new SafeFork(&blocking_function);
82 
83         dont_block.call(); // call blocking_function
84 
85         if ( !dont_block.call() )
86         {
87             Stdout("blocking function is currently running and not done yet!");
88         }
89 
90         while ( dont_block.isRunning() )
91         {
92             Stdout("blocking function is still running!");
93         }
94 
95         if ( !dont_block.call() )
96         {
97             Stdout("blocking function is currently running and not done yet!");
98         }
99 
100         dont_block.call(true); // wait for a unfinished fork and then call
101                                // blocking_function without forking
102     }
103     -----
104 
105 *******************************************************************************/
106 
107 public class SafeFork
108 {
109     /***************************************************************************
110 
111         Exception, reusable
112 
113     ***************************************************************************/
114 
115     private ErrnoException exception;
116 
117     /***************************************************************************
118 
119         Pid of the forked child
120 
121     ***************************************************************************/
122 
123     private int child_pid = 0;
124 
125     /***************************************************************************
126 
127         Delegate to call
128 
129     ***************************************************************************/
130 
131     private void delegate () dg;
132 
133     /***************************************************************************
134 
135         Constructor
136 
137         Params:
138             dg = delegate to call
139 
140     ***************************************************************************/
141 
142     public this ( scope void delegate () dg )
143     {
144         this.dg = dg;
145 
146         this.exception = new ErrnoException;
147     }
148 
149     /***************************************************************************
150 
151         Find out whether the fork is still running or not
152 
153         Returns:
154             true if the fork is still running, else false
155 
156     ***************************************************************************/
157 
158     public bool isRunning ( )
159     {
160         return this.child_pid == 0
161             ? false
162             : this.isRunning(false, false);
163     }
164 
165     /***************************************************************************
166 
167         Call the delegate, possibly within a fork.
168         Ensures that the delegate will only be called when there is not already
169         a fork running. The fork exits after the delegate returned.
170 
171         Note that the host process is not informed about any errors in
172         the forked process.
173 
174         Params:
175             block = if true, wait for a currently running fork and don't fork
176                              when calling the delegate
177                     if false, don't do anything when a fork is currently running
178 
179         Returns:
180             true when the delegate was called
181 
182         See_Also:
183             SafeFork.isRunning
184 
185     ***************************************************************************/
186 
187     public bool call ( bool block = false )
188     {
189         if ( this.child_pid == 0 || !this.isRunning(block) )
190         {
191             if ( block )
192             {
193                 version ( TimeFork )
194                 {
195                     Stdout.formatln("Running task without forking...");
196                 }
197                 this.dg();
198 
199                 this.child_pid = 0;
200 
201                 return true;
202             }
203             else
204             {
205                 version ( TimeFork )
206                 {
207                     Stdout.formatln("Running task in fork...");
208                     StopWatch sw;
209                     sw.start;
210                 }
211 
212                 this.child_pid = fork();
213 
214                 version ( TimeFork )
215                 {
216                     Stdout.formatln("Fork took {}s",
217                         (cast(float)sw.microsec) / 1_000_000.0f);
218                 }
219 
220                 this.exception.enforce(this.child_pid >= 0, "failed to fork");
221 
222                 if ( this.child_pid == 0 )
223                 {
224                     this.dg();
225                     exit(0);
226                 }
227 
228                 return true;
229             }
230         }
231         else
232         {
233             return false;
234         }
235     }
236 
237     /***************************************************************************
238 
239         Checks whether the forked process is already running.
240 
241         Params:
242             block = if true, wait for a currently running fork
243                     if false, don't do anything when a fork is currently running
244             clear = if true, the waiting status of the forked process is cleared
245 
246         Returns:
247             true if the forked process is running
248 
249         See_Also:
250             SafeFork.isRunning
251 
252     ***************************************************************************/
253 
254     private bool isRunning ( bool block, bool clear = true )
255     {
256         if ( this.child_pid < 0 )
257             return false;
258 
259         siginfo_t siginfo;
260 
261         // In the case where we are blocking, we need to consider signals
262         // arriving while we wait, and resume the waiting if EINTR is returned
263         int waitid_ret = void;
264         do
265         {
266             waitid_ret = waitid(idtype_t.P_PID, this.child_pid, &siginfo,
267                  WEXITED | (block ? 0 : WNOHANG) | (clear ? 0 : WNOWAIT));
268         }
269         while (waitid_ret == -1 && errno == EINTR);
270 
271         this.exception.enforce(waitid_ret == 0, "waitid failed", "waitid");
272 
273         return siginfo._sifields._kill.si_pid == 0;
274     }
275 }