forked from mirage/mirage-tcpip
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pcb.ml
641 lines (581 loc) · 23 KB
/
pcb.ml
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
(*
* Copyright (c) 2010-2012 Anil Madhavapeddy <anil@recoil.org>
* Copyright (c) 2012 Balraj Singh <bs375@cl.cam.ac.uk>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*)
open Lwt.Infix
type error = [`Bad_state of State.tcpstate]
type 'a result = [`Ok of 'a | `Error of error]
let ok x = Lwt.return (`Ok x)
let error s = Lwt.return (`Error (`Bad_state s))
let (>+=) x f =
x >>= function
| `Ok x -> f x
| `Error _ as e -> Lwt.return e
let iter_s f l =
let rec aux = function
| [] -> ok ()
| h::t -> f h >+= fun () -> aux t
in
aux l
let debug = Log.create "PCB"
let info = Log.create ~enabled:true ~stats:false "PCB"
module Tcp_wire = Wire_structs.Tcp_wire
cstruct pseudo_header {
uint32_t src;
uint32_t dst;
uint8_t res;
uint8_t proto;
uint16_t len
} as big_endian
module Make(Ip:V1_LWT.IP)(Time:V1_LWT.TIME)(Clock:V1.CLOCK)(Random:V1.RANDOM) =
struct
module RXS = Segment.Rx(Time)
module TXS = Segment.Tx(Time)(Clock)
module ACK = Ack.Immediate
module UTX = User_buffer.Tx(Time)(Clock)
module WIRE = Wire.Make(Ip)
module STATE = State.Make(Time)
type pcb = {
id: WIRE.id;
wnd: Window.t; (* Window information *)
rxq: RXS.t; (* Received segments queue for out-of-order data *)
txq: TXS.t; (* Transmit segments queue *)
ack: ACK.t; (* Ack state *)
state: State.t; (* Connection state *)
urx: User_buffer.Rx.t; (* App rx buffer *)
urx_close_t: unit Lwt.t; (* App rx close thread *)
urx_close_u: unit Lwt.u; (* App rx connection close wakener *)
utx: UTX.t; (* App tx buffer *)
}
type connection = pcb * unit Lwt.t
type connection_result = [ `Ok of connection | `Rst | `Timeout ]
type t = {
ip : Ip.t;
mutable localport : int;
channels: (WIRE.id, connection) Hashtbl.t;
(* server connections the process of connecting - SYN-ACK sent
waiting for ACK *)
listens: (WIRE.id, (Sequence.t * ((pcb -> unit Lwt.t) * connection)))
Hashtbl.t;
(* clients in the process of connecting *)
connects: (WIRE.id, (connection_result Lwt.u * Sequence.t)) Hashtbl.t;
}
let pp_stats fmt t =
Log.pf fmt "[channels=%d listens=%d connects=%d]"
(Hashtbl.length t.channels)
(Hashtbl.length t.listens)
(Hashtbl.length t.connects)
let with_stats name t fmt = Log.pf fmt "%s: %a" name pp_stats t
let ip { ip; _ } = ip
let verify_checksum _ _ _ = true
let wscale_default = 2
module Tx = struct
(* Output a TCP packet, and calculate some settings from a state descriptor *)
let xmit_pcb ip id ~flags ~wnd ~options ~seq datav =
let window = Int32.to_int (Window.rx_wnd_unscaled wnd) in
let rx_ack = Some (Window.rx_nxt wnd) in
let syn = match flags with Segment.Syn -> true | _ -> false in
let fin = match flags with Segment.Fin -> true | _ -> false in
let rst = match flags with Segment.Rst -> true | _ -> false in
let psh = match flags with Segment.Psh -> true | _ -> false in
WIRE.xmit ~ip ~id ~syn ~fin ~rst ~psh ~rx_ack ~seq ~window ~options datav
(* Output an RST response when we dont have a PCB *)
let send_rst { ip; _ } id ~sequence ~ack_number ~syn ~fin =
let datalen = Int32.add (if syn then 1l else 0l) (if fin then 1l else 0l) in
let window = 0 in
let options = [] in
let seq = Sequence.of_int32 ack_number in
let rx_ack = Some (Sequence.of_int32 (Int32.add sequence datalen)) in
WIRE.xmit ~ip ~id ~rst:true ~rx_ack ~seq ~window ~options []
(* Output a SYN packet *)
let send_syn { ip; _ } id ~tx_isn ~options ~window =
WIRE.xmit ~ip ~id ~syn:true ~rx_ack:None ~seq:tx_isn ~window ~options []
(* Queue up an immediate close segment *)
let close pcb =
Log.s debug "TX.close";
match State.state pcb.state with
| State.Established | State.Close_wait ->
UTX.wait_for_flushed pcb.utx >>= fun () ->
(let { wnd; _ } = pcb in
STATE.tick pcb.state (State.Send_fin (Window.tx_nxt wnd));
TXS.output ~flags:Segment.Fin pcb.txq []
)
| _ ->
Log.f debug (fun fmt ->
Log.pf fmt "TX.close: skipping, state=%a" State.pp pcb.state);
Lwt.return_unit
(* Thread that transmits ACKs in response to received packets,
thus telling the other side that more can be sent, and
also data from the user transmit queue *)
let thread t pcb ~send_ack ~rx_ack =
let { wnd; ack; _ } = pcb in
(* Transmit an empty ack when prompted by the Ack thread *)
let rec send_empty_ack () =
Lwt_mvar.take send_ack >>= fun _ ->
let ack_number = Window.rx_nxt wnd in
let flags = Segment.No_flags in
let options = [] in
let seq = Window.tx_nxt wnd in
ACK.transmit ack ack_number >>= fun () ->
xmit_pcb t.ip pcb.id ~flags ~wnd ~options ~seq [] >>= fun () ->
send_empty_ack () in
(* When something transmits an ACK, tell the delayed ACK thread *)
let rec notify () =
Lwt_mvar.take rx_ack >>= fun ack_number ->
ACK.transmit ack ack_number >>= fun () ->
notify () in
send_empty_ack () <&> (notify ())
end
module Rx = struct
(* Process an incoming TCP packet that has an active PCB *)
let input _t pkt (pcb,_) =
let sequence = Sequence.of_int32 (Tcp_wire.get_tcp_sequence pkt) in
let ack_number =
Sequence.of_int32 (Tcp_wire.get_tcp_ack_number pkt)
in
let fin = Tcp_wire.get_fin pkt in
let syn = Tcp_wire.get_syn pkt in
let ack = Tcp_wire.get_ack pkt in
let rst = Tcp_wire.get_rst pkt in
let window = Tcp_wire.get_tcp_window pkt in
let data = Wire.get_payload pkt in
let seg =
RXS.segment ~sequence ~fin ~syn ~rst ~ack ~ack_number ~window ~data
in
let { rxq; _ } = pcb in
(* Coalesce any outstanding segments and retrieve ready segments *)
RXS.input rxq seg
(* Thread that spools the data into an application receive buffer,
and notifies the ACK subsystem that new data is here *)
let thread (pcb:pcb) ~rx_data =
let { wnd; ack; urx; urx_close_u; _ } = pcb in
(* Thread to monitor application receive and pass it up *)
let rec rx_application_t () =
Lwt_mvar.take rx_data >>= fun (data, winadv) ->
begin match winadv with
| None -> Lwt.return_unit
| Some winadv ->
if (winadv > 0) then (
Window.rx_advance wnd winadv;
ACK.receive ack (Window.rx_nxt wnd)
) else (
Window.rx_advance wnd winadv;
ACK.pushack ack (Window.rx_nxt wnd)
)
end >>= fun _ ->
begin match data with
| None ->
STATE.tick pcb.state State.Recv_fin;
Lwt.wakeup urx_close_u ();
User_buffer.Rx.add_r urx None >>= fun () ->
rx_application_t ()
| Some data ->
let rec queue = function
| [] -> Lwt.return_unit
| hd::tl ->
User_buffer.Rx.add_r urx (Some hd) >>= fun () ->
queue tl
in
queue data >>= fun _ ->
rx_application_t ()
end
in
rx_application_t ()
end
module Wnd = struct
let thread ~urx:_ ~utx ~wnd:_ ~state ~tx_wnd_update =
(* Monitor our transmit window when updates are received
remotely, and tell the application that new space is
available when it is blocked *)
let rec tx_window_t () =
Lwt_mvar.take tx_wnd_update >>= fun tx_wnd ->
begin match State.state state with
| State.Reset -> UTX.reset utx
| _ -> UTX.free utx tx_wnd
end >>= fun () ->
tx_window_t ()
in
tx_window_t ()
end
(* Helper function to apply function with contents of hashtbl, or
take default action *)
let with_hashtbl h k fn default =
try fn (Hashtbl.find h k) with Not_found -> default k
let hashtbl_find h k =
try Some (Hashtbl.find h k) with Not_found -> None
let clearpcb t id tx_isn =
(* TODO: add more info to log msgs *)
Log.f debug (with_stats "removing pcb from tables" t);
match hashtbl_find t.channels id with
| Some _ ->
Log.s debug "removed from channels!!";
Hashtbl.remove t.channels id;
Stats.decr_channel ();
| None ->
match hashtbl_find t.listens id with
| Some (isn, _) ->
if isn = tx_isn then (
Log.s debug "removing incomplete listen pcb";
Hashtbl.remove t.listens id;
Stats.decr_listen ();
)
| None ->
Log.s debug "error in removing pcb - no such connection"
let pcb_allocs = ref 0
let th_allocs = ref 0
let pcb_frees = ref 0
let th_frees = ref 0
let resolve_wnd_scaling options rx_wnd_scaleoffer =
let tx_wnd_scale = List.fold_left (fun a ->
function Options.Window_size_shift m -> Some m | _ -> a
) None options in
match tx_wnd_scale with
| None -> (0, 0), []
| Some tx_f ->
(rx_wnd_scaleoffer, tx_f),
(Options.Window_size_shift rx_wnd_scaleoffer :: [])
type pcb_params =
{ tx_wnd: int;
sequence: int32;
options: Options.t list;
tx_isn: Sequence.t;
rx_wnd: int;
rx_wnd_scaleoffer: int }
let new_pcb t params id =
let { tx_wnd; sequence; options; tx_isn; rx_wnd; rx_wnd_scaleoffer } =
params
in
let tx_mss = List.fold_left (fun a ->
function Options.MSS m -> Some m | _ -> a
) None options
in
let (rx_wnd_scale, tx_wnd_scale), opts =
resolve_wnd_scaling options rx_wnd_scaleoffer
in
(* Set up the windowing variables *)
let rx_isn = Sequence.of_int32 sequence in
(* Initialise the window handler *)
let wnd =
Window.t ~rx_wnd_scale ~tx_wnd_scale ~rx_wnd ~tx_wnd ~rx_isn ~tx_mss
~tx_isn
in
(* When we transmit an ACK for a received segment, rx_ack is written to *)
let rx_ack = MProf.Trace.named_mvar_empty "rx_ack" in
(* When we receive an ACK for a transmitted segment, tx_ack is written to *)
let tx_ack = MProf.Trace.named_mvar_empty "tx_ack" in
(* When new data is received, rx_data is written to *)
let rx_data = MProf.Trace.named_mvar_empty "rx_data" in
(* Write to this mvar to transmit an empty ACK to the remote side *)
let send_ack = MProf.Trace.named_mvar_empty "send_ack" in
(* The user application receive buffer and close notification *)
let rx_buf_size = Window.rx_wnd wnd in
let urx = User_buffer.Rx.create ~max_size:rx_buf_size ~wnd in
let urx_close_t, urx_close_u = MProf.Trace.named_task "urx_close" in
(* The window handling thread *)
let tx_wnd_update = MProf.Trace.named_mvar_empty "tx_wnd_update" in
(* Set up transmit and receive queues *)
let on_close () = clearpcb t id tx_isn in
let state = State.t ~on_close in
let txq, _tx_t =
TXS.create ~xmit:(Tx.xmit_pcb t.ip id) ~wnd ~state ~rx_ack ~tx_ack ~tx_wnd_update
in
(* The user application transmit buffer *)
let utx = UTX.create ~wnd ~txq ~max_size:16384l in
let rxq = RXS.create ~rx_data ~wnd ~state ~tx_ack in
(* Set up ACK module *)
let ack = ACK.t ~send_ack ~last:(Sequence.incr rx_isn) in
(* Construct basic PCB in Syn_received state *)
let pcb = { state; rxq; txq; wnd; id; ack; urx; urx_close_t; urx_close_u; utx } in
(* Compose the overall thread from the various tx/rx threads
and the main listener function *)
let th =
(Tx.thread t pcb ~send_ack ~rx_ack) <?>
(Rx.thread pcb ~rx_data) <?>
(Wnd.thread ~utx ~urx ~wnd ~state ~tx_wnd_update)
in
pcb_allocs := !pcb_allocs + 1;
th_allocs := !th_allocs + 1;
let fnpcb = fun _ -> pcb_frees := !pcb_frees + 1 in
let fnth = fun _ -> th_frees := !th_frees + 1 in
Gc.finalise fnpcb pcb;
Gc.finalise fnth th;
Lwt.return (pcb, th, opts)
let new_server_connection t params id pushf =
Log.f debug (with_stats "new-server-connection" t);
new_pcb t params id >>= fun (pcb, th, opts) ->
STATE.tick pcb.state State.Passive_open;
STATE.tick pcb.state (State.Send_synack params.tx_isn);
(* Add the PCB to our listens table *)
if Hashtbl.mem t.listens id then (
Log.s info "WARNING: connection already being attempted";
Hashtbl.remove t.listens id;
Stats.decr_listen ();
);
Hashtbl.add t.listens id (params.tx_isn, (pushf, (pcb, th)));
Stats.incr_listen ();
(* Queue a SYN ACK for transmission *)
let options = Options.MSS 1460 :: opts in
TXS.output ~flags:Segment.Syn ~options pcb.txq [] >>= fun () ->
Lwt.return (pcb, th)
let new_client_connection t params id ack_number =
Log.f debug (with_stats "new-client-connection" t);
let tx_isn = params.tx_isn in
let params = { params with tx_isn = Sequence.incr tx_isn } in
new_pcb t params id >>= fun (pcb, th, _) ->
(* A hack here because we create the pcb only after the SYN-ACK is rx-ed*)
STATE.tick pcb.state (State.Send_syn tx_isn);
(* Add the PCB to our connection table *)
Hashtbl.add t.channels id (pcb, th);
Stats.incr_channel ();
STATE.tick pcb.state (State.Recv_synack (Sequence.of_int32 ack_number));
(* xmit ACK *)
TXS.output pcb.txq [] >>= fun () ->
Lwt.return (pcb, th)
let process_reset t id =
Log.f debug (with_stats "process-reset" t);
match hashtbl_find t.connects id with
| Some (wakener, _) ->
(* URG_TODO: check if RST ack num is valid before it is accepted *)
Hashtbl.remove t.connects id;
Stats.decr_connect ();
Lwt.wakeup wakener `Rst;
Lwt.return_unit
| None ->
match hashtbl_find t.listens id with
| Some (_, (_, (pcb, th))) ->
Hashtbl.remove t.listens id;
Stats.decr_listen ();
STATE.tick pcb.state State.Recv_rst;
Lwt.cancel th;
Lwt.return_unit
| None ->
(* Incoming RST possibly to listen port - ignore per RFC793 pg65 *)
Lwt.return_unit
let process_synack t id ~pkt ~ack_number ~sequence ~options ~syn ~fin =
Log.f debug (with_stats "process-synack" t);
match hashtbl_find t.connects id with
| Some (wakener, tx_isn) ->
if Sequence.(to_int32 (incr tx_isn)) = ack_number then (
Hashtbl.remove t.connects id;
Stats.decr_connect ();
let tx_wnd = Tcp_wire.get_tcp_window pkt in
let rx_wnd = 65535 in
(* TODO: fix hardcoded value - it assumes that this value was
sent in the SYN *)
let rx_wnd_scaleoffer = wscale_default in
new_client_connection t
{ tx_wnd; sequence; options; tx_isn; rx_wnd; rx_wnd_scaleoffer }
id ack_number
>>= fun (pcb, th) ->
Lwt.wakeup wakener (`Ok (pcb, th));
Lwt.return_unit
) else
(* Normally sending a RST reply to a random pkt would be in
order but here we stay quiet since we are actively trying
to connect this id *)
Lwt.return_unit
| None ->
(* Incomming SYN-ACK with no pending connect and no matching pcb
- send RST *)
Tx.send_rst t id ~sequence ~ack_number ~syn ~fin
let process_syn t id ~listeners ~pkt ~ack_number ~sequence ~options ~syn ~fin =
Log.f debug (with_stats "process-syn" t);
match listeners id.WIRE.local_port with
| Some pushf ->
let tx_isn = Sequence.of_int ((Random.int 65535) + 0x1AFE0000) in
let tx_wnd = Tcp_wire.get_tcp_window pkt in
(* TODO: make this configurable per listener *)
let rx_wnd = 65535 in
let rx_wnd_scaleoffer = wscale_default in
new_server_connection t
{ tx_wnd; sequence; options; tx_isn; rx_wnd; rx_wnd_scaleoffer }
id pushf
>>= fun _ ->
Lwt.return_unit
| None ->
Tx.send_rst t id ~sequence ~ack_number ~syn ~fin
let process_ack t id ~pkt ~ack_number ~sequence ~syn ~fin =
Log.f debug (with_stats "process-ack" t);
match hashtbl_find t.listens id with
| Some (tx_isn, (pushf, newconn)) ->
if Sequence.(to_int32 (incr tx_isn)) = ack_number then (
(* Established connection - promote to active channels *)
Hashtbl.remove t.listens id;
Stats.decr_listen ();
Hashtbl.add t.channels id newconn;
Stats.incr_channel ();
(* Finish processing ACK, so pcb.state is correct *)
Rx.input t pkt newconn >>= fun () ->
(* send new connection up to listener *)
pushf (fst newconn)
) else
(* No RST because we are trying to connect on this id *)
Lwt.return_unit
| None ->
match hashtbl_find t.connects id with
| Some _ ->
(* No RST because we are trying to connect on this id *)
Lwt.return_unit
| None ->
(* ACK but no matching pcb and no listen - send RST *)
Tx.send_rst t id ~sequence ~ack_number ~syn ~fin
let input_no_pcb t listeners pkt id =
match Tcp_wire.get_rst pkt with
| true -> process_reset t id
| false ->
let sequence = Tcp_wire.get_tcp_sequence pkt in
let options = Wire.get_options pkt in
let ack_number = Tcp_wire.get_tcp_ack_number pkt in
let syn = Tcp_wire.get_syn pkt in
let ack = Tcp_wire.get_ack pkt in
let fin = Tcp_wire.get_fin pkt in
match syn, ack with
| true , true -> process_synack t id ~pkt ~ack_number ~sequence
~options ~syn ~fin
| true , false -> process_syn t id ~listeners ~pkt ~ack_number ~sequence
~options ~syn ~fin
| false, true -> process_ack t id ~pkt ~ack_number ~sequence ~syn ~fin
| false, false ->
(* What the hell is this packet? No SYN,ACK,RST *)
Log.s debug "input-no-pcb: unknown packet";
Lwt.return_unit
(* Main input function for TCP packets *)
let input t ~listeners ~src ~dst data =
match verify_checksum src dst data with
| false ->
Log.s debug "RX.input: checksum error";
Lwt.return_unit
| true ->
let source_port = Tcp_wire.get_tcp_src_port data in
let dest_port = Tcp_wire.get_tcp_dst_port data in
let id =
{ WIRE.local_port = dest_port;
dest_ip = src;
local_ip = dst;
dest_port = source_port }
in
(* Lookup connection from the active PCB hash *)
with_hashtbl t.channels id
(* PCB exists, so continue the connection state machine in tcp_input *)
(Rx.input t data)
(* No existing PCB, so check if it is a SYN for a listening function *)
(input_no_pcb t listeners data)
(* Blocking read on a PCB *)
let read pcb =
User_buffer.Rx.take_l pcb.urx
(* Maximum allowed write *)
let write_available pcb =
(* Our effective outgoing MTU is what can fit in a page *)
min 4000 (min (Window.tx_mss pcb.wnd)
(Int32.to_int (UTX.available pcb.utx)))
(* Wait for more write space *)
let write_wait_for pcb sz =
UTX.wait_for pcb.utx (Int32.of_int sz)
let rec writefn pcb wfn data =
let len = Cstruct.len data in
match write_available pcb with
| 0 ->
write_wait_for pcb 1 >>= fun () ->
writefn pcb wfn data
| av_len when av_len < len ->
let first_bit = Cstruct.sub data 0 av_len in
let remaing_bit = Cstruct.sub data av_len (len - av_len) in
writefn pcb wfn first_bit >+= fun () ->
writefn pcb wfn remaing_bit
| _ ->
match State.state pcb.state with
| State.Established | State.Close_wait -> wfn [data] >>= ok
| e -> error e
(* Blocking write on a PCB *)
let write pcb data = writefn pcb (UTX.write pcb.utx) data
let writev pcb data = iter_s (write pcb) data
let write_nodelay pcb data = writefn pcb (UTX.write_nodelay pcb.utx) data
let writev_nodelay pcb data = iter_s (write_nodelay pcb) data
(* Close - no more will be written *)
let close pcb = Tx.close pcb
let get_dest pcb = pcb.id.WIRE.dest_ip, pcb.id.WIRE.dest_port
let getid t dest_ip dest_port =
(* TODO: make this more robust and recognise when all ports are gone *)
let islistener _t _port =
(* TODO keep a list of active listen ports *)
false in
let idinuse t id =
Hashtbl.mem t.channels id ||
Hashtbl.mem t.connects id ||
Hashtbl.mem t.listens id
in
let inuse t id = islistener t id.WIRE.local_port || idinuse t id in
let rec bumpport t =
(match t.localport with
| 65535 -> t.localport <- 10000
| _ -> t.localport <- t.localport + 1);
let id =
{ WIRE.local_port = t.localport;
dest_ip = dest_ip;
local_ip = Ip.get_source t.ip dest_ip;
dest_port = dest_port }
in
if inuse t id then bumpport t else id
in
bumpport t
(* SYN retransmission timer *)
let rec connecttimer t id tx_isn options window count =
let rxtime = match count with
| 0 -> 3. | 1 -> 6. | 2 -> 12. | 3 -> 24. | _ -> 48.
in
Time.sleep rxtime >>= fun () ->
match hashtbl_find t.connects id with
| None -> Lwt.return_unit
| Some (wakener, isn) ->
if isn = tx_isn then
if count > 3 then (
Hashtbl.remove t.connects id;
Stats.decr_connect ();
Lwt.wakeup wakener `Timeout;
Lwt.return_unit
) else (
Tx.send_syn t id ~tx_isn ~options ~window >>= fun () ->
connecttimer t id tx_isn options window (count + 1)
)
else Lwt.return_unit
let connect t ~dest_ip ~dest_port =
let id = getid t dest_ip dest_port in
let tx_isn = Sequence.of_int ((Random.int 65535) + 0x1BCD0000) in
(* TODO: This is hardcoded for now - make it configurable *)
let rx_wnd_scaleoffer = wscale_default in
let options =
Options.MSS 1460 :: Options.Window_size_shift rx_wnd_scaleoffer :: []
in
let window = 5840 in
let th, wakener = MProf.Trace.named_task "TCP connect" in
if Hashtbl.mem t.connects id then (
Log.s info "WARNING: connection already being attempted";
Hashtbl.remove t.connects id;
Stats.decr_connect ();
);
Hashtbl.add t.connects id (wakener, tx_isn);
Stats.incr_connect ();
Tx.send_syn t id ~tx_isn ~options ~window >>= fun () ->
Lwt.async (fun () -> connecttimer t id tx_isn options window 0);
th
(* Construct the main TCP thread *)
let create ip =
Random.self_init ();
let localport = 10000 + (Random.int 10000) in
let listens = Hashtbl.create 1 in
let connects = Hashtbl.create 1 in
let channels = Hashtbl.create 7 in
{ ip; localport; channels; listens; connects }
end