futur.blue / pegasus
objective categorical abstract machine language personal data server
fork atom

Try to avoid lock contention in sequencer

futur.blue 350eca3e 8eec7e70

verified
+20 -15
+1 -1
pegasus/lib/api/sync/subscribeRepos.ml
··· 4 4 let cursor = 5 5 match Dream.query req "cursor" with 6 6 | Some s -> 7 - Option.value (int_of_string_opt s) ~default:0 7 + max 0 (Option.value (int_of_string_opt s) ~default:0) 8 8 | None -> 9 9 0 10 10 in
+19 -14
pegasus/lib/sequencer.ml
··· 518 518 (Uri.to_string crawler) (Printexc.to_string exn) ) ) ) 519 519 Env.crawlers 520 520 end ; 521 + let to_remove = ref [] in 521 522 Hashtbl.iter 522 523 (fun _ s -> 523 524 if not s.closed then ( ··· 525 526 if Queue.length s.q > queue_max then ( 526 527 s.closed <- true ; 527 528 s.close_reason <- Some "ConsumerTooSlow" ; 528 - Hashtbl.remove subs s.id ; 529 + to_remove := s.id :: !to_remove ; 529 530 Lwt_condition.broadcast s.cond () ) 530 531 else Lwt_condition.broadcast s.cond () ) ) 531 532 subs ; 533 + List.iter (Hashtbl.remove subs) !to_remove ; 532 534 Lwt.return_unit ) 533 535 534 536 let latest_seq () = !head_seq ··· 552 554 Hashtbl.remove subs s.id ; 553 555 Lwt.return_unit ) 554 556 555 - let ring_after (after : int) : item list = 556 - if !head_seq <= after then [] 557 - else 558 - let first = max (!head_seq - !count + 1) (after + 1) in 559 - if first > !head_seq then [] 560 - else 561 - let rec collect acc seq = 562 - if seq > !head_seq then List.rev acc 557 + let ring_after (after : int) : item list Lwt.t = 558 + Lwt_mutex.with_lock lock (fun () -> 559 + let head = !head_seq in 560 + let cnt = !count in 561 + if head <= after then Lwt.return [] 562 + else 563 + let first = max (head - cnt + 1) (after + 1) in 564 + if first > head then Lwt.return [] 563 565 else 564 - let it = ring.(seq mod ring_size) in 565 - collect (it :: acc) (seq + 1) 566 - in 567 - collect [] first 566 + let rec collect acc seq = 567 + if seq > head then List.rev acc 568 + else 569 + let it = ring.(seq mod ring_size) in 570 + collect (it :: acc) (seq + 1) 571 + in 572 + Lwt.return (collect [] first) ) 568 573 569 574 let rec wait_next (s : subscriber) : item Lwt.t = 570 575 if s.closed then failwith "subscriber closed" ··· 634 639 let%lwt head_db = DB.latest_seq conn in 635 640 let cutoff = head_db in 636 641 (* try backfill from buffer first *) 637 - let ring = Bus.ring_after cursor in 642 + let%lwt ring = Bus.ring_after cursor in 638 643 let ring_covers = 639 644 match ring with 640 645 | [] ->