blob: e2ffdaccb7d4d552acf4aabe73914cfdbe5d0696 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
(* Behn - Timer Driver using Eio.Time
*
* This is the first I/O driver - demonstrates the fiber-per-driver pattern!
*
* Behn manages timers:
* - SetTimer effects: schedule a timer
* - CancelTimer effects: cancel a timer
* - When timer fires: produce an ovum back to Arvo
*
* Key innovation: Uses Eio.Time.sleep for non-blocking waits!
* Multiple timers can run concurrently in separate fibers.
*)
(* Timer state *)
type timer = {
id: int64;
fire_time: float; (* Unix timestamp *)
mutable cancelled: bool;
}
(* Behn driver state *)
type t = {
timers: (int64, timer) Hashtbl.t; (* Active timers by ID *)
lock: Mutex.t;
}
(* Create behn driver *)
let create () = {
timers = Hashtbl.create 64;
lock = Mutex.create ();
}
(* Set a timer *)
let set_timer behn ~id ~fire_time =
Mutex.lock behn.lock;
let timer = { id; fire_time; cancelled = false } in
Hashtbl.replace behn.timers id timer;
Mutex.unlock behn.lock;
Printf.printf "[Behn] Set timer %Ld for %f\n%!" id fire_time
(* Cancel a timer *)
let cancel_timer behn ~id =
Mutex.lock behn.lock;
(match Hashtbl.find_opt behn.timers id with
| Some timer ->
timer.cancelled <- true;
Printf.printf "[Behn] Cancelled timer %Ld\n%!" id
| None ->
Printf.printf "[Behn] Timer %Ld not found (already fired?)\n%!" id
);
Mutex.unlock behn.lock
(* Timer fiber - waits for a timer to fire
*
* This runs as a separate fiber for each timer!
* Uses Eio.Time.sleep for non-blocking wait.
*)
let timer_fiber behn ~env ~event_stream timer =
let clock = Eio.Stdenv.clock env in
let now = Unix.gettimeofday () in
let wait_time = max 0.0 (timer.fire_time -. now) in
Printf.printf "[Behn] Timer %Ld: waiting %.3f seconds\n%!" timer.id wait_time;
(* Non-blocking sleep using Eio! *)
Eio.Time.sleep clock wait_time;
(* Check if cancelled *)
Mutex.lock behn.lock;
let is_cancelled = timer.cancelled in
Hashtbl.remove behn.timers timer.id;
Mutex.unlock behn.lock;
if not is_cancelled then begin
Printf.printf "[Behn] Timer %Ld: FIRED! 🔥\n%!" timer.id;
(* Create timer ovum and send to event stream *)
let ovum = Nock_lib.Effects.timer_ovum ~id:timer.id ~fire_time:timer.fire_time in
(* Send to runtime event stream *)
Eio.Stream.add event_stream ovum;
Printf.printf "[Behn] Timer %Ld: event sent to runtime\n%!" timer.id
end else begin
Printf.printf "[Behn] Timer %Ld: cancelled, not firing\n%!" timer.id
end
(* Behn driver fiber - processes timer effects
*
* This is the main fiber for the Behn driver.
* It reads SetTimer/CancelTimer effects and spawns timer fibers.
*)
let driver_fiber behn ~sw ~env ~effect_queue ~event_stream =
Printf.printf "[Behn] Driver fiber started 🕐\n%!";
let rec loop () =
(* Get next effect from queue *)
match Nock_lib.Effects.try_dequeue effect_queue with
| None ->
(* No effects, sleep briefly *)
Eio.Time.sleep (Eio.Stdenv.clock env) 0.01;
loop ()
| Some eff ->
(match eff with
| Nock_lib.Effects.SetTimer { id; time } ->
set_timer behn ~id ~fire_time:time;
(* Spawn a fiber for this timer *)
let timer = Hashtbl.find behn.timers id in
Eio.Fiber.fork ~sw (fun () ->
timer_fiber behn ~env ~event_stream timer
);
loop ()
| Nock_lib.Effects.CancelTimer { id } ->
cancel_timer behn ~id;
loop ()
| _ ->
(* Not a behn effect, ignore *)
loop ()
)
in
try
loop ()
with
| End_of_file ->
Printf.printf "[Behn] Driver shutting down\n%!"
(* Get active timer count *)
let active_timers behn =
Mutex.lock behn.lock;
let count = Hashtbl.length behn.timers in
Mutex.unlock behn.lock;
count
|
