Skip to content
Newer
Older
100644 652 lines (570 sloc) 21.5 KB
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
1 (*
2 Copyright © 2011 MLstate
3
4 This file is part of OPA.
5
6 OPA is free software: you can redistribute it and/or modify it under the
7 terms of the GNU Affero General Public License, version 3, as published by
8 the Free Software Foundation.
9
10 OPA is distributed in the hope that it will be useful, but WITHOUT ANY
11 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
12 FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for
13 more details.
14
15 You should have received a copy of the GNU Affero General Public License
16 along with OPA. If not, see <http://www.gnu.org/licenses/>.
17 *)
18
19 module CR = QmlClosureRuntime
20
21 type continuation_options =
22 {
23 movable: bool;
24 atomic: bool;
25 _lazy: bool
26 }
27
28 type stack_infos =
29 { caller_info : continuation_info
30 ; call_position : string
31 ; callee_name : string
32 ; call_arguments : Obj.t } (* it is an opa tuple *)
33
34 and continuation_info =
35 { options : continuation_options ;
36 (** options for this continuation. Often, the options are shared between several continuation,
37 using syntax {[let c = { previous_c with payload = new_payload }]} *)
38 thread_context : Obj.t option ;
39 (** passing \@thread_context around.
40 The thread_context is set in Opa (or Qml) using \@with_thread_context,
41 the type is a Qml(Opa) record, the qml typer checking that the utilisation is correct
42 if it has never been set with a \@with_thread_context, it is None
43 *)
44 transaction_context : Obj.t option;
45 (** Has a behaviour similar to thread_context, but is reserved for use by
46 database and transaction handling mechanisms
47 *)
48 exn_handler : Obj.t continuation option;
49 (** continuation that can be triggered by \@raise *)
50
51 stack_infos : stack_infos option;
52 }
53
54 (* the representation of continuation is not flattened because the main operations
55 * on continuation are allocations and return, that only need the payload field
56 * this representation makes allocation much cheaper *)
57 and 'a continuation =
58 {
59 payload: CR.t ; (* the closure should have type 'a -> unit *)
60 (** the effective function of the continuation *)
61
62 continuation_info : continuation_info
63 (** contains any information associated to the current execution *)
64 }
65
66 type 'a func0 = 'a continuation -> unit
67 type ('a, 'b) func = 'a -> 'b func0
68 type ('a, 'b, 'c) func2 = 'a -> 'b -> 'c func0
69 type ('a, 'b, 'c, 'd) func3 = 'a -> 'b -> 'c -> 'd func0
70 type ('a, 'b, 'c, 'd, 'e) func4 = 'a -> 'b -> 'c -> 'd -> 'e func0
71 type ('a, 'b, 'c, 'd, 'e, 'f) func5 = 'a -> 'b -> 'c -> 'd -> 'e -> 'f func0
72
73 let debug = (* I'm a temporary handler, remove me *)
74 if (try Sys.getenv "MLSTATE_CPS_DEBUG" <> "0" with Not_found -> false)
75 then (fun fmt -> Format.eprintf ("[cps] "^^fmt^^"\n%!"))
76 else (fun fmt -> Format.ifprintf Format.err_formatter fmt)
77
78 let magic_cont = (Obj.magic : _ continuation -> _ continuation)
79
80 external make_func: ('a -> 'b continuation -> unit) -> ('a, 'b) func = "%identity"
81
82 let default_options = {movable = true; atomic = false; _lazy = false}
83 let default_thread_context = None
84 let default_transaction_context = None
85 let default_exn_handler = None
86 let default_stack_infos = None
87 let default_continuation_info =
88 { options = default_options
89 ; thread_context = default_thread_context
90 ; transaction_context = default_transaction_context
91 ; exn_handler = default_exn_handler
92 ; stack_infos = default_stack_infos }
93
94 (* inlining : these function are called every 2 lines in the generated code *)
95 let make_cont options f =
96 {payload = f; continuation_info = {default_continuation_info with options = options}}
97 let make_cont_ml options f =
98 make_cont options (CR.create_no_ident1 f)
99 let cont f =
100 {payload = f; continuation_info = default_continuation_info}
101 let cont_ml f =
102 cont (CR.create_no_ident1 f)
103 let ccont b f =
104 { b with payload = f }
105 let ccont_ml b f =
106 ccont b (CR.create_no_ident1 f)
107
108 let update_cont cont parent name position args =
109 {cont with
110 continuation_info =
111 { cont.continuation_info with
112 stack_infos =
113 Some
114 { caller_info =
115 (match parent with
116 | None -> default_continuation_info (* a little wierd *)
117 | Some cont -> cont.continuation_info)
118 ; callee_name = name
119 ; call_position = position
120 ; call_arguments = Obj.repr args } } }
121
122 let generic_trace_printer ?(first_line="*** Stack trace:") printer (cont : _ continuation) =
123 Printf.eprintf "%s\n" first_line;
124 let rec aux i infos =
125 match infos.stack_infos with
126 | None -> () (* not calling printer, because this 'infos' a the dummy one introduced above *)
127 | Some stack_infos ->
128 printer i infos stack_infos;
129 aux (i+1) stack_infos.caller_info in
130 aux 0 cont.continuation_info
131
132 let trace_printer ?(args= #<If:CPS_STACK_TRACE$contains "arg">true#<Else>false#<End>)
133 ?(thread_context= #<If:CPS_STACK_TRACE$contains "th">true#<Else>false#<End>)
134 ?(transaction_context= #<If:CPS_STACK_TRACE$contains "tr">true#<Else>false#<End>) () =
135 fun index infos stack_infos ->
136 Printf.eprintf "%3d: %20s called at %s%s%s%s\n"
137 index
138 stack_infos.callee_name
139 stack_infos.call_position
140 (if args then " with args=" ^ DebugPrint.print stack_infos.call_arguments else "")
141 (if thread_context then
142 match infos.thread_context with
143 | None -> ""
144 | Some thread_context -> " with thread_context=" ^ DebugPrint.print thread_context
145 else "")
146 (if transaction_context then
147 match infos.transaction_context with
148 | None -> ""
149 | Some transaction_context -> " with transaction_context=" ^ DebugPrint.print transaction_context
150 else "")
151 let print_trace_fl first_line cont = generic_trace_printer ~first_line (trace_printer ()) cont
152 let print_trace cont = generic_trace_printer (trace_printer ()) cont
153
154 let thread_context b : _ option = Obj.magic (b.continuation_info.thread_context : Obj.t option)
155 let with_thread_context tc b = { b with continuation_info = {b.continuation_info with thread_context = Some (Obj.repr tc) } }
156
157 (* Identical to thread_context handlers *)
158 let transaction_context b : _ option = Obj.magic (b.continuation_info.transaction_context : Obj.t option)
159 let with_transaction_context tc b = { b with continuation_info = {b.continuation_info with transaction_context = Some (Obj.repr tc) } }
160
161 (**
162 Runtime backtrace generation. Uses only constant space and time.
163 *)
164 let (bt_add, bt_take) =
165 let max_backtrace_size = 25 in
166 let cyclic_queue = Array.create max_backtrace_size "" in (*LIFO queue so far*)
167 let cursor = ref 0 in
168 let cursor_valid () = !cursor >= 0 && !cursor < max_backtrace_size in
169 let bt_add s =
170 if s <> Array.get cyclic_queue !cursor then begin (* TODO: turn this off for full backtrace *)
171 assert (cursor_valid ());
172 Array.set cyclic_queue !cursor s;
173 cursor := (!cursor + 1) mod max_backtrace_size
174 end
175 in
176 let bt_take () =
177 assert (cursor_valid ());
178 cursor := (!cursor + max_backtrace_size - 1) mod max_backtrace_size;
179 let s = Array.get cyclic_queue !cursor in
180 Array.set cyclic_queue !cursor "";
181 s
182 in
183 (bt_add, bt_take)
184
185 let fun_args2string f_string larg =
186 let larg = [larg] in (* TODO: a tmp hack *)
187 let f arg =
188 let s = DebugPrint.print arg in
189 String.sub s 0 (min (String.length s) 1000)
190 in
191 let lval = Base.String.concat_map ", " f larg in
a9f8d34 [cleanup] Base: remove sprintf
Raja authored Jun 28, 2011
192 Printf.sprintf "%s(%s, ...)" f_string lval
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
193
194 let display_backtrace s =
195 let bt_pos = ref (bt_take ()) in
196 if !bt_pos <> "" then begin
197 Printf.fprintf stderr "%s at %s\n" s !bt_pos;
198 bt_pos := bt_take ();
199 while !bt_pos <> "" do
200 Printf.fprintf stderr "Called from %s\n" !bt_pos;
201 bt_pos := bt_take ()
202 done
203 end
204
205 let report_exception exc k =
206 Printf.eprintf "Error: uncaught OPA exception %s\n" (DebugPrint.print exc);
207 (*display_backtrace "Raised"*)
208 print_trace k
209
210 let handler_cont k = match k.continuation_info.exn_handler with
211 | None -> cont_ml (fun exc -> report_exception exc k)
212 | Some h -> magic_cont h
213 let catch_ml h k = { k with continuation_info = {k.continuation_info with exn_handler = Some (ccont_ml k (fun x -> h (Obj.obj x) k)) } }
214 let catch h k = { k with continuation_info = {k.continuation_info with exn_handler = Some (ccont_ml k (fun x -> CR.args_apply2 h (Obj.obj x) k)) } }
215
216 type 'a barrier_status =
217 | Computed of 'a
218 | Exception of Obj.t
219 | Waiting of 'a continuation list ref
220
221 type 'a barrier = {
222 mutable status : 'a barrier_status;
223 (**
224 The status of the barrier. Mutable, manipulated internally by
225 + [release_barrier]
226 + [fail_barrier]
227 *)
228
229 ident : string;
230 (**
231 A human readable identifier for identifying barriers.
232 Used for debug.
233 *)
234
235 nb : int;
236 (**
237 A uniq identifier for identifying barriers.
238 *)
239 }
240
241 type task = {
242 action: unit continuation;
243 waiting: unit barrier
244 }
245
246 type 'a future = 'a barrier
247
248 let print_barrier_status f = function
249 | Computed _ -> Format.fprintf f "Computed"
250 | Exception _ -> Format.fprintf f "Exception"
251 | Waiting l -> Format.fprintf f "Waiting <%d>" (List.length !l)
252
253 let print_barrier f b =
254 Format.fprintf f "{ident : %s; nb : %d; status %a}"
255 b.ident b.nb print_barrier_status b.status
256
257 let print_task i {action = action; waiting = waiting} =
258 print_trace_fl (Printf.sprintf "** Scheduler task %d" i) action;
259 Format.eprintf "@[<2>%d - Waiting:@ %a@]\n" i print_barrier waiting
260
261 let nb_barrier = ref 0
262 let make_barrier ident =
263 incr nb_barrier;
264 {
265 nb = !nb_barrier;
266 ident = ident;
267 status = Waiting (ref [])
268 }
269
270 let make_task (f:unit continuation) : task = {
271 action = f;
272 waiting = make_barrier "make_task"
273 }
274
275 let task_of_fun = make_task
276
277 let push f = Scheduler.push Scheduler.default f
278
279 let push_cont k x =
280 push (fun () -> CR.args_apply1 k.payload x)
281
282 (* With the explicit flush of the scheduler at end of the toplevel
283 initialization, there is not need to schedule in apply or return
284 Moreover, this breaks the tail-rec optimization of ocaml code. *)
285 let nb_step_apply = 10000
286 let max_blocking_step = 1000000
287 (* cannot embbed the reference for typing problem *)
288 let applied_step = ref nb_step_apply
289
290 let check_stack_step = pred (1 lsl 10) (* <!> should be a 2^^n -1 *)
291 let stack_limit = 5000000
292 let stack_usage = BaseSys.stack_usage
293
294 let return k x =
295 #<Ifstatic:CPS_STACK_SIZE .*>
296 Printf.printf "stack-usage: %d\n%!" (stack_usage ()) ;
297 #<End>
298 let applied_step_contents = !applied_step in
299 if
300 (*
301 The stack should not be checked and nothing should be pushed
302 *)
303 (applied_step_contents land check_stack_step <> 0)
304 ||
305 (*
306 The stack is ok, and we should not push
307 *)
308 ( (stack_usage () <= stack_limit) && ( applied_step_contents <> 0 ) )
309
310 then (
311 decr applied_step;
312 CR.args_apply1 k.payload x
313 )
314 else (
315 applied_step:=nb_step_apply;
316 push_cont k x
317 )
318
319 let execute k x = CR.args_apply1 k.payload x
320
321 let apply f v k =
322 push (fun () -> f v k)
323 (* dont schedule here *)
324
325 (* used for nary application, f is the partial application of f' on every arg
326 but the final continuation *)
327 let apply0 f k =
328 push (fun () -> execute (ccont k f) ())
329 (* dont schedule here *)
330 let apply0_ml f k = apply0 (CR.create_no_ident1 f) k
331
332 let wait_barrier v k =
333 debug "Wait_barrier";
334 let payload x = push_cont k x in
335 match v.status with
336 | Computed x -> payload x
337 | Exception exc ->
338 (match k.continuation_info.exn_handler with
339 | None -> report_exception exc k
340 | Some h ->
341 push_cont (ccont_ml k (fun () -> CR.args_apply1 h.payload exc)) ())
342 | Waiting l ->
343 l := ccont_ml k payload :: !l
344
345 (* used for a non concurrency toplevel value introduction,
346 and uncps
347 *)
348 let nb_block = ref 0
349 let block_stack = ref []
350 let save_block_stack _ =
351 block_stack := !applied_step::!block_stack;
352 applied_step := max_blocking_step
353 let resume_block_stack _ =
354 match !block_stack with
355 | hd::tl ->
356 applied_step := hd ;
357 block_stack := tl
358 | [] ->
359 (* we assume that a save_block_stack is always executed before a resume_block_stack *)
360 assert false
361
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
362 let before_wait = save_block_stack
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
363
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
364 (*
365 Used internally only, not exported.
366 This wait is used for projecting cps functions into non cps functions.
367 *)
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
368 let blocking_wait (v : 'a barrier) : 'a =
369 resume_block_stack () ;
370 match v.status with
371 | Computed a -> a
372 | Exception _ -> failwith "exception outside of cps context"
373 (* TODO Obj.dump *) (* a bypass cannot raise exc, or non concurrency mode and exc at toplevel *)
374 | Waiting _ ->
375 failwith
376 (Printf.sprintf
377 "Barrier (%s : %d) was not released, don't wait anymore"
378 v.ident v.nb)
379
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
380 (*
381 This wait is exported, and used for the evaluation of toplevel expressions.
382 It is called after a [Scheduler.loop_until] enforcing that the barrier is
383 released [Computed] when this function is called.
384 *)
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
385 let toplevel_wait (v : 'a barrier) : 'a =
386 resume_block_stack () ;
387 match v.status with
388 | Computed a -> a
389 | Exception _ ->
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
390 (*
391 For the projection of the projection of the toplevel, we do not use
392 the function fail_barrier in the exception handler of the continuation
393 passed to the function releasing the barrier. So this should not happen.
394 We may want to change this behavior in the future,
395 if we want to use fail_barrier.
396 *)
397 assert false
398
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
399 | Waiting _ ->
400 (*
401 This should really not happens, because we give to the scheduler a function
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
402 for checking if the barrier was released, as the argument of a [Scheduler.loop_until]
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
403 *)
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
404 assert false
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
405
406 let release_barrier (v:'a barrier) (x:'a) =
407 debug "release_barrier";
408 match v.status with
409 | Exception _
410 | Computed _ ->
411 failwith "Internal inconsistency: this barrier has been released twice."
412 | Waiting l ->
413 v.status <- Computed x;
414 List.iter (fun f -> CR.args_apply1 f.payload x) !l
415
416 let fail_barrier (v:'a barrier) (exc:'a) =
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
417 debug "fail_barrier";
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
418 match v.status with
419 | Exception _
420 | Computed _ ->
421 failwith "Internal inconsistency: this barrier has been released twice."
422 | Waiting l ->
423 v.status <- Exception (Obj.repr exc);
424 List.iter (fun f -> return (handler_cont f) exc) !l
425
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
426 (*
427 Function to check if the barrier is still not released.
428 Used combined with a [Scheduler.loop_until].
429 *)
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
430 let is_released v =
431 match v.status with
432 | Waiting _ ->
433 let _ =
434 #<If:CPS_BLOCKING_WAIT>
435 Printf.eprintf "LOOP-UNTIL: is_released(%s)=false\n%!" v.ident
436 #<End>
437 in
438 false
439 | _ -> true
440
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
441 (*
442 Proposition using a time limit for projection cps functions.
443 We may also implement the skipping of a few second order bypass
444 call, that way we will reject code which would need such a
445 projection. Currently, this is not used.
446 *)
447 module LoopingWait =
448 struct
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
449 let max_loop_time = 40.0
450 let time_limit () = Unix.gettimeofday () +. max_loop_time
451 let false_may_fail_on_time_limit v time_limit =
452 let t = Unix.gettimeofday() in
453 if t < time_limit then false
454 else failwith
455 (Printf.sprintf
456 "Barrier (%s : %d) was not released after %1.2f seconds, don't wait anymore"
457 v.ident v.nb (t-.time_limit)
458 )
459
460 let looping_wait (v : 'a barrier) : 'a =
461 (* a first 'fast' shot without call to time *)
462 let rec one_shot i =
463 if i <> 0 then
464 if not(is_released v) then (
465 ignore (Scheduler.wait Scheduler.default ~block:false);
466 one_shot (i-1)
467 )
468 in
469 one_shot 1;
470 (* a second slower loop wait for long computation *)
471 if not(is_released v) then (
472 let loop () =
473 let time_limit = time_limit () in
474 let is_released () =
475 is_released v
476 || (false_may_fail_on_time_limit v time_limit)
477 in
478 Scheduler.loop_until Scheduler.default is_released
479 in
480 loop ();
481 );
482 toplevel_wait v
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
483 end
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
484
485 type black_future
486 external black_future : 'a future -> black_future = "%identity"
487 external unblack_future : black_future -> 'a future = "%identity"
488 let black_make_barrier str = black_future (make_barrier str)
489 let black_release_barrier v = release_barrier (unblack_future v)
490 let black_toplevel_wait v = toplevel_wait (unblack_future v)
491
492 external magic_func : ('a, 'b) func -> ('c, 'd) func = "%identity"
493 external magic_func0 : (_) func0 -> (_) func0 = "%identity"
494 external magic_func1 : (_,_) func -> (_,_) func = "%identity" (* same as magic_func, created for homogeneity *)
495 external magic_func2 : (_,_,_) func2 -> (_,_,_) func2 = "%identity"
496 external magic_func3 : (_,_,_,_) func3 -> (_,_,_,_) func3 = "%identity"
497 external magic_func4 : (_,_,_,_,_) func4 -> (_,_,_,_,_) func4 = "%identity"
498 external magic_func5 : (_,_,_,_,_,_) func5 -> (_,_,_,_,_,_) func5 = "%identity"
499 external magic_func_more : _ -> _ = "%identity"
500
501
502 let spawn (f:(unit, 'a) func) =
503 let barrier = make_barrier "spawn" in
504 let k =
505 { payload = CR.create_no_ident1 (fun x -> release_barrier barrier x);
506 continuation_info =
507 {default_continuation_info with
508 exn_handler = Some (cont_ml (fun exc -> fail_barrier barrier (Obj.obj exc)))}
509 } in
510 let action =
511 { payload = CR.create_no_ident1 (fun () -> f () k);
512 continuation_info = default_continuation_info } in
513 push_cont action ();
514 barrier
515
516 let wait v k = wait_barrier v k
517
518 let uncps ident k f =
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
519 debug "uncps: %s" ident ;
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
520 let b = make_barrier ident in
521 let c = ccont_ml k (fun z -> release_barrier b z) in
edfd810 [cleanup] cps: blocking_wait was replaced by toplevel_wait
Mathieu Barbin authored Jun 23, 2011
522 before_wait ();
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
523 let _ = CR.args_apply1 f c in
4a5e92d [fix] cps: reactivate previous behavior without re-entrance in the sc…
Mathieu Barbin authored Jun 23, 2011
524 blocking_wait b
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
525
526 let uncps_ml ident k (f:'b continuation -> unit) =
527 uncps ident k (CR.create_no_ident1 f)
528
529 let callcc_directive f k =
14c3e16 [fix] callcc: now keep threadcontext, exc_handler, ...
Hugo Heuzard authored Sep 12, 2011
530 let unit_cont = ccont_ml k (fun _ -> ()) in
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
531 f k unit_cont
532
533 (* === *)
534
535 type ('a, 'b) pair = {f_0: 'a; f_1: 'b}
536 let pair x y = {f_0=x; f_1=y}
537
538 let callback_of_fun f =
539 let barrier = make_barrier "callback_of_fun" in
540 pair (fun x -> release_barrier barrier (f x)) barrier
541
542 let callback_post () =
543 let barrier = make_barrier "callback_post" in
544 pair (fun x -> release_barrier barrier x) barrier
545
546 let wrap_async f x cb =
547 let barrier = make_barrier "wrap_async" in
548 f x (fun result_of_f ->
549 let result_of_cb = cb result_of_f in
550 release_barrier barrier result_of_cb; result_of_cb);
551 barrier
552
553
554 (* registering a debug printer *)
555 let () =
556 let module D = DebugPrint in
557 let check_continuation_options = D.tuple3 ~f1:D.bool ~f2:D.bool ~f3:D.bool in
558 let check_payload = D.true_ in (* not checking that we really have a qml closure
559 * on purpose
560 * if not, i still want to print the continuation
561 * as a continuation (even with random content) *)
562 let check_thread_context x = D.option x in
563 let check_transaction_context x = D.option x in
564 let check_exn_handler x = D.option x in (* don't do recursive checks *)
565 let check_stack_infos =
566 D.tuple_n [ D.true_
567 ; D.string
568 ; D.string
569 ; D.true_ ] in
570 let check_continuation_info =
571 D.tuple_n [ check_continuation_options
572 ; check_thread_context
573 ; check_transaction_context
574 ; check_exn_handler
575 ; check_stack_infos ] in
576 let check =
577 D.tuple_n [ check_payload
578 ; check_continuation_info ] in
579 let unsafe_print {payload = payload; continuation_info = continuation_info} =
580 Printf.sprintf "{payload = %s%s}"
581 (DebugPrint.print payload)
582 (if continuation_info = default_continuation_info then "" else
583 Printf.sprintf "continuation_info = {%s%s%s%s}"
584 (if continuation_info.options = default_options then ""
585 else
586 let options = continuation_info.options in
587 Printf.sprintf "options = {movable=%B; atomic=%B; _lazy=%B}; " options.movable options.atomic options._lazy)
588 (match continuation_info.thread_context with
589 | None -> ""
590 | Some thread_context ->
591 Printf.sprintf "thread_context = %s; " (DebugPrint.print thread_context))
592 (match continuation_info.transaction_context with
593 | None -> ""
594 | Some transaction_context ->
595 Printf.sprintf "transaction_context = %s; " (DebugPrint.print transaction_context))
596 (match continuation_info.exn_handler with
597 | None -> ""
598 | Some exn_handler ->
599 Printf.sprintf "exn_handler = %s; " (DebugPrint.print exn_handler))) in
600 let print_opt x =
601 if check (Obj.repr x) then
602 Some (unsafe_print (Obj.magic x : _ continuation))
603 else
604 None in
605 D.register {D.f = print_opt}
606
607
608 (* Useful exported functions *)
609
610 module Ops = struct
611 let (@>) f k = f k
612 let (|>) x k = return k x
613 end
614 open Ops
615
616 let rec fold_list f acc l k = match l with
617 | [] -> acc |> k
618 | hd::tl -> f acc hd @> ccont_ml k @> fun acc -> fold_list f acc tl @> k
619
0fbf161 [feature] CpsServerLib: parallel iter and map on lists
Louis Gesbert authored Jul 8, 2011
620 let rec iter_list f l k =
621 let n = ref 0 in
622 let k =
623 ccont_ml k (fun () -> decr n; if !n == 0 then () |> k)
624 in
625 List.iter
626 (fun x -> incr n; Scheduler.push Scheduler.default (fun () -> f x @> k))
627 l
628
629 let rec map_list f l k =
630 let n = ref 0 in
631 let results = ref [||] in
632 let ki =
633 fun i ->
634 ccont_ml k
635 (fun x ->
636 !results.(i) <- x;
637 decr n;
638 if !n > 0 then () else Array.to_list !results |> k)
639 in
640 List.iter
641 (fun x -> let k = ki !n in incr n; Scheduler.push Scheduler.default (fun () -> f x @> k))
642 l;
643 results := Array.make !n (Obj.magic ())
644
fccc685 Initial open-source release
MLstate authored Jun 21, 2011
645 let fold_array f acc arr k =
646 let s = Array.length arr in
647 let rec aux acc i k =
648 if i >= s then acc |> k
649 else f acc arr.(i) @> ccont_ml k @> fun acc -> aux acc (succ i) @> k
650 in
651 aux acc 0 @> k
Something went wrong with that request. Please try again.