Add downloading lock for files

src/buffering.ml

@@ -49,7 +49,7 @@ struct
     mutex : Mutex.t;
   }
 
-  let create ?(n = 64) block_size = {
+  let create ?(n = Utils.hashtable_initial_size) block_size = {
     blocks = Hashtbl.create n;
     files = Hashtbl.create n;
     block_size;

src/concurrentGlobal.ml

@@ -7,14 +7,7 @@ struct
   let mutex = Mutex.create ()
 
   let with_lock f =
-    try
-      let () = Mutex.lock mutex in
-      let result = f () in
-        Mutex.unlock mutex;
-        result
-    with e ->
-      Mutex.unlock mutex;
-      raise e
+    Utils.with_lock mutex f
 
   let get_no_lock () =
     Global.get global

src/context.ml

@@ -26,6 +26,8 @@ type t = {
   skip_trash : bool;
   (* Memory buffers *)
   memory_buffers : Buffering.MemoryBuffers.t;
+  (* File locks *)
+  file_locks : (string, Mutex.t) Hashtbl.t;
 }
 
 let app_dir = {
@@ -68,6 +70,10 @@ let memory_buffers = {
   GapiLens.get = (fun x -> x.memory_buffers);
   GapiLens.set = (fun v x -> { x with memory_buffers = v })
 }
+let file_locks = {
+  GapiLens.get = (fun x -> x.file_locks);
+  GapiLens.set = (fun v x -> { x with file_locks = v })
+}
 
 let config_lens =
   config_store |-- ConfigFileStore.data
@@ -101,6 +107,8 @@ let clear_ctx = ConcurrentContext.clear
 
 let update_ctx = ConcurrentContext.update
 
+let with_ctx_lock = ConcurrentContext.with_lock
+
 let save_state_store state_store =
   Utils.log_with_header "BEGIN: Saving application state in %s\n"
     state_store.StateFileStore.path;

src/drive.ml

@@ -176,7 +176,7 @@ let build_resource_tables parent_path trashed =
   let resources =
     Cache.Resource.select_resources_with_parent_path
       cache parent_path trashed in
-  let filename_table = Hashtbl.create 64 in
+  let filename_table = Hashtbl.create Utils.hashtable_initial_size in
   let remote_id_table = Hashtbl.create (List.length resources) in
   List.iter
     (fun resource ->
@@ -456,26 +456,33 @@ let delete_memory_buffers memory_buffers resource =
     )
     resource.Cache.Resource.remote_id
 
+let delete_from_context context resource =
+  let memory_buffers = context.Context.memory_buffers in
+  delete_memory_buffers memory_buffers resource;
+  Context.with_ctx_lock
+    (fun () ->
+       let remote_id = resource.Cache.Resource.remote_id |> Option.get in
+       Hashtbl.remove context.Context.file_locks remote_id
+    )
+
 let delete_cached_resource resource =
   let context = Context.get_ctx () in
   let cache = context.Context.cache in
-  let memory_buffers = context.Context.memory_buffers in
   Cache.Resource.delete_resource cache resource;
   let total_size =
     Cache.delete_files_from_cache cache [resource] in
   let metadata = context |. Context.metadata_lens in
   update_cache_size (Int64.neg total_size) metadata cache;
-  delete_memory_buffers memory_buffers resource
+  delete_from_context context resource
 
 let delete_cached_resources metadata cache resources =
   Cache.Resource.delete_resources cache resources;
   let total_size =
     Cache.delete_files_from_cache cache resources in
   update_cache_size (Int64.neg total_size) metadata cache;
   let context = Context.get_ctx () in
-  let memory_buffers = context.Context.memory_buffers in
   List.iter
-    (delete_memory_buffers memory_buffers)
+    (delete_from_context context)
     resources
 
 let update_cache_size_for_documents cache resource content_path op =
@@ -1018,7 +1025,7 @@ let download_resource resource =
       GapiMediaResource.destination;
       range_spec = "";
     } in
-    let fileId = resource |. Cache.Resource.remote_id |> Option.get in
+    let fileId = resource.Cache.Resource.remote_id |> Option.get in
     if Cache.Resource.is_document resource then begin
       let fmt = Cache.Resource.get_format resource config in
       let mimeType = Cache.Resource.mime_type_of_format fmt in
@@ -1041,7 +1048,8 @@ let download_resource resource =
       SessionM.return ()
     end
   in
-  let do_download () =
+  let do_download =
+    SessionM.return () >>= fun () ->
     Utils.log_with_header
       "BEGIN: Downloading resource (id=%Ld) to %s\n%!"
       resource.Cache.Resource.id content_path;
@@ -1070,6 +1078,23 @@ let download_resource resource =
       Cache.Resource.State.Synchronized resource.Cache.Resource.id;
     SessionM.return ()
   in
+  let get_lock () =
+    let context = Context.get_ctx () in
+    Context.with_ctx_lock
+      (fun () ->
+         let remote_id = resource.Cache.Resource.remote_id |> Option.get in
+         match Utils.safe_find context.Context.file_locks remote_id with
+         | None ->
+           let mutex = Mutex.create () in
+           Hashtbl.add context.Context.file_locks remote_id mutex;
+           mutex
+         | Some mutex -> mutex
+      )
+  in
+  let do_download_with_lock () =
+    let mutex = get_lock () in
+    Utils.with_lock_m mutex do_download
+  in
   let rec check_state n =
     let reloaded_resource = Option.map_default
       (Cache.Resource.select_resource_with_remote_id cache)
@@ -1087,7 +1112,7 @@ let download_resource resource =
           Cache.Resource.State.Synchronized resource.Cache.Resource.id;
         SessionM.return ()
       end else
-        do_download ()
+        do_download_with_lock ()
     in
     begin match reloaded_state with
         Cache.Resource.State.Synchronized
@@ -1096,7 +1121,7 @@ let download_resource resource =
           if Sys.file_exists content_path then
             SessionM.return ()
           else
-            do_download ()
+            do_download_with_lock ()
       | Cache.Resource.State.ToDownload ->
           download_if_not_updated ()
       | Cache.Resource.State.Downloading ->

src/gdfuse.ml

@@ -232,6 +232,7 @@ let setup_application params =
     skip_trash = params.skip_trash;
     memory_buffers =
       Buffering.MemoryBuffers.create config.Config.memory_buffer_size;
+    file_locks = Hashtbl.create Utils.hashtable_initial_size;
   } in
   Context.set_ctx context;
   let refresh_token = context |. Context.refresh_token_lens in

src/utils.ml

@@ -9,6 +9,7 @@ let verbose = ref false
 let log_channel = ref stdout
 let max_retries = ref 10
 let mb = 1048576L
+let hashtable_initial_size = 64
 
 (* Threads *)
 let get_thread_id () =