Source file push.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
open Rresult

type configuration = { stateless : bool }

let configuration ?(stateless = false) () = { stateless }

module S = Sigs

module Make
    (Scheduler : S.SCHED)
    (IO : S.IO with type 'a t = 'a Scheduler.s)
    (Flow : S.FLOW with type 'a fiber = 'a Scheduler.s)
    (Uid : S.UID)
    (Ref : S.REF) =
struct
  let src = Logs.Src.create "push"

  module Log = (val Logs.src_log src : Logs.LOG)
  open Scheduler

  let ( >>= ) x f = IO.bind x f
  let return x = IO.return x
  let ( >>| ) x f = x >>= fun x -> return (f x)

  let sched =
    S.
      {
        bind = (fun x f -> inj (prj x >>= fun x -> prj (f x)));
        return = (fun x -> inj (return x));
      }

  let fail exn = inj (IO.fail exn)

  let io =
    S.
      {
        recv = (fun flow raw -> inj (Flow.recv flow raw));
        send = (fun flow raw -> inj (Flow.send flow raw));
        pp_error = Flow.pp_error;
      }

  let push ?(uses_git_transport = false) ~capabilities:my_caps cmds ~host path
      flow store access { stateless } pack =
    let fiber ctx =
      let open Smart in
      let* () =
        if uses_git_transport then
          send ctx proto_request
            (Proto_request.receive_pack ~host ~version:1 path)
        else return ()
      in
      let* v = recv ctx advertised_refs in
      Context.replace_their_caps ctx (Smart.Advertised_refs.capabilities v);
      return (Smart.Advertised_refs.map ~fuid:Uid.of_hex ~fref:Ref.v v)
    in
    let ctx = Smart.Context.make ~my_caps in
    Smart_flow.run sched fail io flow (fiber ctx) |> prj
    >>= fun advertised_refs ->
    Pck.commands sched ~capabilities:my_caps ~equal:Ref.equal
      ~deref:access.Sigs.deref store cmds
      (Smart.Advertised_refs.refs advertised_refs)
    |> prj
    >>= function
    | None ->
        Smart_flow.run sched fail io flow Smart.(send ctx flush ()) |> prj
        >>= fun () -> return ()
    | Some cmds -> (
        Smart_flow.run sched fail io flow
          Smart.(
            send ctx commands
              (Commands.map ~fuid:Uid.to_hex ~fref:Ref.to_string cmds))
        |> prj
        >>= fun () ->
        let exclude, sources =
          Pck.get_limits ~compare:Uid.compare
            (Smart.Advertised_refs.refs advertised_refs)
            (Smart.Commands.commands cmds)
        in
        Pck.get_uncommon_objects sched ~compare:Uid.compare access store
          ~exclude ~sources
        |> prj
        >>= fun uids ->
        Log.debug (fun m ->
            m "Prepare a pack of %d object(s)." (List.length uids));
        let stream = pack uids in
        let side_band =
          Smart.Context.is_cap_shared ctx `Side_band
          || Smart.Context.is_cap_shared ctx `Side_band_64k
        in
        let pack = Smart.send_pack ~stateless side_band in
        let rec go () =
          stream () >>= function
          | None ->
              let report_status =
                Smart.Context.is_cap_shared ctx `Report_status
              in
              Log.debug (fun m ->
                  m "report-status capability: %b." report_status);
              if report_status then
                Smart_flow.run sched fail io flow
                  Smart.(recv ctx (status side_band))
                |> prj
                >>| Smart.Status.map ~f:Ref.v
              else if uses_git_transport then
                Smart_flow.run sched fail io flow Smart.(recv ctx recv_flush)
                |> prj
                >>= fun () ->
                let cmds = List.map R.ok (Smart.Commands.commands cmds) in
                return (Smart.Status.v cmds)
              else
                let cmds = List.map R.ok (Smart.Commands.commands cmds) in
                return (Smart.Status.v cmds)
          | Some payload ->
              Smart_flow.run sched fail io flow Smart.(send ctx pack payload)
              |> prj
              >>= fun () -> go ()
        in
        go () >>= fun status ->
        match Smart.Status.to_result status with
        | Ok () ->
            Log.debug (fun m -> m "Push is done!");
            Log.info (fun m ->
                m "%a" Smart.Status.pp
                  (Smart.Status.map ~f:Ref.to_string status));
            return ()
        | Error err ->
            Log.err (fun m -> m "Push got an error: %s" err);
            return ())
end