<commit_before>open GapiLens.Infix

type t = {
  stop_async_upload : bool;
  upload_resource_by_id : int64 -> unit;
  thread_pool : ThreadPool.t;
}
let stop_async_upload = {
  GapiLens.get = (fun x -> x.stop_async_upload);
  GapiLens.set = (fun v x -> { x with stop_async_upload = v })
}
let upload_resource_by_id = {
  GapiLens.get = (fun x -> x.upload_resource_by_id);
  GapiLens.set = (fun v x -> { x with upload_resource_by_id = v })
}
let thread_pool = {
  GapiLens.get = (fun x -> x.thread_pool);
  GapiLens.set = (fun v x -> { x with thread_pool = v })
}

module ConcurrentUploadQueue =
  ConcurrentGlobal.Make(struct type u = t let label = "upload-queue" end)

let upload_resource cache =
  let d = ConcurrentUploadQueue.get () in
  let upload = d.upload_resource_by_id in
  let upload_entry = Cache.UploadQueue.select_next_resource cache in
  let do_work e =
    let entry_id = e.CacheData.UploadEntry.id in
    let resource_id = e.CacheData.UploadEntry.resource_id in
    Utils.log_with_header
      "Uploading queued entry (id=%Ld) resource_id=%Ld.\n%!"
      entry_id
      resource_id;
    Cache.UploadQueue.update_entry_state cache
      CacheData.UploadEntry.State.Uploading entry_id;
    begin try
      upload resource_id;
    with e ->
      Utils.log_with_header
        "Upload failed for queued entry (id=%Ld).\n%!"
        entry_id;
      Cache.UploadQueue.update_entry_state cache
        CacheData.UploadEntry.State.ToUpload entry_id;
      raise e;
    end;
    Utils.log_with_header
      "Removing queued entry (id=%Ld).\n%!"
      entry_id;
    Cache.UploadQueue.delete_upload_entry cache e
  in
  match upload_entry with
  | Some e ->
    ThreadPool.add_work do_work e d.thread_pool;
  | None -> ()

let poll_upload_queue cache =
  let check () =
    let d = ConcurrentUploadQueue.get () in
    if d.stop_async_upload then begin
      let entries = Cache.UploadQueue.count_entries cache in<commit_msg>Log pending uploads during shutdown<commit_after>open GapiLens.Infix

type t = {
  stop_async_upload : bool;
  upload_resource_by_id : int64 -> unit;
  thread_pool : ThreadPool.t;
}
let stop_async_upload = {
  GapiLens.get = (fun x -> x.stop_async_upload);
  GapiLens.set = (fun v x -> { x with stop_async_upload = v })
}
let upload_resource_by_id = {
  GapiLens.get = (fun x -> x.upload_resource_by_id);
  GapiLens.set = (fun v x -> { x with upload_resource_by_id = v })
}
let thread_pool = {
  GapiLens.get = (fun x -> x.thread_pool);
  GapiLens.set = (fun v x -> { x with thread_pool = v })
}

module ConcurrentUploadQueue =
  ConcurrentGlobal.Make(struct type u = t let label = "upload-queue" end)

let upload_resource cache =
  let d = ConcurrentUploadQueue.get () in
  let upload = d.upload_resource_by_id in
  let upload_entry = Cache.UploadQueue.select_next_resource cache in
  let do_work e =
    let entry_id = e.CacheData.UploadEntry.id in
    let resource_id = e.CacheData.UploadEntry.resource_id in
    Utils.log_with_header
      "Uploading queued entry (id=%Ld) resource_id=%Ld.\n%!"
      entry_id
      resource_id;
    Cache.UploadQueue.update_entry_state cache
      CacheData.UploadEntry.State.Uploading entry_id;
    begin try
      upload resource_id;
    with e ->
      Utils.log_with_header
        "Upload failed for queued entry (id=%Ld).\n%!"
        entry_id;
      Cache.UploadQueue.update_entry_state cache
        CacheData.UploadEntry.State.ToUpload entry_id;
      raise e;
    end;
    Utils.log_with_header
      "Removing queued entry (id=%Ld).\n%!"
      entry_id;
    Cache.UploadQueue.delete_upload_entry cache e
  in
  match upload_entry with
  | Some e ->
    ThreadPool.add_work do_work e d.thread_pool;
  | None -> ()

let poll_upload_queue cache =
  let check () =
    let d = ConcurrentUploadQueue.get () in
    if d.stop_async_upload then begin
      let entries = Cache.UploadQueue.count_entries cache in
      Utils.log_with_header
        "Waiting for pending uploads (%d)\n%!"
        entries;
{'fraction_non_alphanumeric': 0.09511826874685456, 'fraction_numerical': 0.0010065425264217413, 'mean_word_length': 3.157949790794979, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4468462', 'n_tokens_mistral': 1374, 'n_tokens_neox': 1359, 'n_words': 332}