snippet of pcb.ml

pcb_snippet.ml

337     (* Compose the overall thread from the various tx/rx threads                
338        and the main listener function *)                                        
339     let tx_thread = (Tx.thread t pcb ~send_ack ~rx_ack) in                      
340     let rx_thread = (Rx.thread pcb ~rx_data) in                                 
341     let wnd_thread = (Wnd.thread ~utx ~urx ~wnd ~state ~tx_wnd_update) in       
342     let threads = [ tx_thread; rx_thread; wnd_thread ] in                       
343     let catch_and_cancel = function                                             
344       | Lwt.Canceled -> ()                                                      
345       | ex ->                                                                   
346         (* cancel the other threads *)                                          
347         List.iter Lwt.cancel threads;                                           
348         Log.s info "ERROR: thread failure; terminating threads and closing connection";
349         on_close ();                                                            
350         !Lwt.async_exception_hook ex                                            
351     in                                                                          
352     List.iter (fun t -> Lwt.on_failure t catch_and_cancel) threads;             
353     let th = Lwt.join threads in                                                
354     pcb_allocs := !pcb_allocs + 1;                                              
355     th_allocs := !th_allocs + 1;                                                
356     let fnpcb = fun _ -> pcb_frees := !pcb_frees + 1 in                         
357     let fnth = fun _ -> th_frees := !th_frees + 1 in                            
358     Gc.finalise fnpcb pcb;                                                      
359     Gc.finalise fnth th;                                                        
360     Lwt.return (pcb, th, opts)