<commit_before>open Core.Std
open Common
module RaftMonitor = RaftMonitorWrapper

module PureState  = 
  functor (MonoTime: Clock.TIME) ->
  functor (Mach: Statemach.MACHINE) -> struct

  (* Split this record down into sections, seperating general statem *)
        type t = {
          (** Generic state as specified by the protocol *) 
          term : Index.t;
          mode : role;
          votedFor : IntID.t option;
          log : Mach.cmd Log.t;
          lastlogIndex : Index.t;
          lastlogTerm : Index.t;
          commitIndex : Index.t;
          votesResponded : IntID.t list;
          votesGranted : IntID.t list;
          nextIndex : (IntID.t * Index.t) list;
          matchIndex : (IntID.t * Index.t) list;
          (** Simulation specfic state, need removing/altering for real
           * implementation *)
          time : unit -> MonoTime.t;
          (* This timer is used in all 3 states and has different meanings depending on state *)
          timer : int; 
          (** this flag is used to indicate if event of a timer
          has happened since last checked, a better method for this should be
          used *)
          id : IntID.t;
          allNodes : IntID.t list;
          leader : IntID.t option;
          state_mach : Mach.t;
          outstanding_request : (Index.t * Rpcs.ClientArg.t) option;
          safety_monitor : RaftMonitor.t
        }
   



  type statecall = 
   | IncrementTerm
   | Reset | Set
   | Vote of IntID.t
   | StepDown of Index.t
   | VoteFrom of IntID.t
   | StartCandidate
   | StartLeader
   | SetTime of MonoTime.t
   | SetLeader of IntID.t
   | SetTerm of Index.t
   | Restart
   | Commit of Index.t
   | AppendEntries of (Index.t * Index.t * Mach.cmd) list
   | AppendEntry of (Index.t * Index.t * Mach.cmd)
   | RemoveEntries of Index.t * Index.t
   | ReplicationFailure of IntID.t * Index.t
   | ReplicationSuccess of IntID.t * Index.t
   | AddClientRequest of Index.t * Rpcs.ClientArg.t
   | RemoveClientRes 


  let init me all =
    { term = Index.init();
      mode = Follower;
      time = MonoTime.init;
      timer = 0;
      votedFor = None;
      log = Log.init(); 
      lastlogIndex = Index.init();
      lastlogTerm = Index.init();
      commitIndex = Index.init();
      votesResponded = [];
      votesGranted = [];
      nextIndex  = [];
      matchIndex = [];
      id = me;
      allNodes = all;
      leader = None;
      state_mach = Mach.init();
      outstanding_request = None;
      safety_monitor = RaftMonitor.tick (RaftMonitor.init()) `Startup ;
    } 

  let refresh s:t =
    { term = s.term;
      mode = Follower;
      time = s.time ;
      timer = s.timer;
      votedFor = None;
      log = s.log; 
      lastlogIndex = Index.init();
      lastlogTerm = Index.init();
      commitIndex = Index.init();
      votesResponded = [];
      votesGranted = [];
      nextIndex  = [];
      matchIndex = [];
      id = s.id; 
      allNodes = s.allNodes;
      leader = None;
      state_mach = s.state_mach;
      outstanding_request = None;
      safety_monitor =  RaftMonitor.tick (RaftMonitor.init()) `Recover;
    } 


  let id_print = function  None -> "none" | Some x -> IntID.to_string x
  let id_index_print (id,index) = (" ID: "^(IntID.to_string id)^"Index: "^(Index.to_string index) )

  let print s = 
    "-- NODE STATE ------------------------------------------------------------------------\n"^
    " | Time: "^(MonoTime.to_string (s.time()))^
    " | ID: "^(IntID.to_string s.id)^
    " | Term: "^(Index.to_string s.term)^
    " | Mode: "^(string_of_role s.mode)^"\n"^
    " | VotedFor: "^(string_of_option IntID.to_string s.votedFor)^
    " | All Nodes: "^(List.to_string ~f:IntID.to_string s.allNodes)^
    " | Votes Recieved: "^ (List.to_string ~f:IntID.to_string s.votesGranted)^
    " | Leader: "^(string_of_option (IntID.to_string) s.leader)^"\n"^
    " | State Machine: "^(Mach.to_string s.state_mach)^
    " | Match Index: "^(List.to_string s.matchIndex ~f:id_index_print)^"\n"^
    " | Next Index: "^(List.to_string s.matchIndex ~f:id_index_print)^"\n"^
    " | Last Log Index: "^(Index.to_string s.lastlogIndex)^
    " | Last Log Term: "^(Index.to_string s.lastlogTerm)^
    " | Commit Index: "^(Index.to_string s.commitIndex)^
    " | Replicated Log: "^(Log.to_string ~cmd_to_string:Mach.cmd_to_string s.log)^
    "\n-------------------------------------------------------------------------------------"
 (* sexp_of_t s |> Sexp.to_string *)

 (* let rec log_add original new_list = 
   (*combine logs, if two items with same index, give proirity to new *)
   match new_list with
   | (index,term,cmd)::rest -> log_add ((index,term,cmd)::(List.filter original ~f:(fun (i,_,_) -> not(index=i)))) rest
   | [] -> original *)


 let update_commitIndex match_list commitIndex log curr_term =
    let magority = (List.length match_list +1 / 2) in
    let discard (_,index) = 
      (if index > commitIndex then Some index else None) in
    let new_list = 
      List.sort ~cmp:(fun x y -> -1* (Index.compare x y))
        (List.filter_map ~f:discard match_list) in
    match List.nth new_list magority with 
    | Some index -> 
      let (_,term) = Log.specific_index_term index log in
      if (curr_term=term) then index else commitIndex
    | None -> commitIndex

  let tick tk s =
  match tk with
    | Set -> 
        {s with timer = Int.succ (s.timer)}
    | IncrementTerm -> 
        let t = Index.succ s.term in
       { s with term=t }
    | Vote id -> 
         assert(s.votedFor = None);
         assert(s.mode=Follower);
        { s with votedFor = Some id}
    | VoteFrom id ->
        { s with votesGranted = id::s.votesGranted }
    | StepDown tm ->
        { s with mode=Follower;
        term = tm;
        votedFor = None;
        votesResponded=[];
        votesGranted=[];
        nextIndex  = [];
        matchIndex = []; 
        safety_monitor = (
            match s.mode with 
            | Candidate -> RaftMonitor.tick s.safety_monitor `StepDown_from_Candidate
            | Leader -> RaftMonitor.tick s.safety_monitor `StepDown_from_Leader 
            | Follower -> s.safety_monitor );
        }
    | StartCandidate -> 
        { s with mode=Candidate;
        votedFor = Some s.id;
        votesResponded=[];
        votesGranted=[s.id];
        term = (Index.succ s.term);
        leader = None;
        safety_monitor =
            match s.mode with
            | Candidate -> RaftMonitor.tick s.safety_monitor `RestartElection<commit_msg>further fix for leader election<commit_after>open Core.Std
open Common
module RaftMonitor = RaftMonitorWrapper

module PureState  = 
  functor (MonoTime: Clock.TIME) ->
  functor (Mach: Statemach.MACHINE) -> struct

  (* Split this record down into sections, seperating general statem *)
        type t = {
          (** Generic state as specified by the protocol *) 
          term : Index.t;
          mode : role;
          votedFor : IntID.t option;
          log : Mach.cmd Log.t;
          lastlogIndex : Index.t;
          lastlogTerm : Index.t;
          commitIndex : Index.t;
          votesResponded : IntID.t list;
          votesGranted : IntID.t list;
          nextIndex : (IntID.t * Index.t) list;
          matchIndex : (IntID.t * Index.t) list;
          (** Simulation specfic state, need removing/altering for real
           * implementation *)
          time : unit -> MonoTime.t;
          (* This timer is used in all 3 states and has different meanings depending on state *)
          timer : int; 
          (** this flag is used to indicate if event of a timer
          has happened since last checked, a better method for this should be
          used *)
          id : IntID.t;
          allNodes : IntID.t list;
          leader : IntID.t option;
          state_mach : Mach.t;
          outstanding_request : (Index.t * Rpcs.ClientArg.t) option;
          safety_monitor : RaftMonitor.t
        }
   



  type statecall = 
   | IncrementTerm
   | Reset | Set
   | Vote of IntID.t
   | StepDown of Index.t
   | VoteFrom of IntID.t
   | StartCandidate
   | StartLeader
   | SetTime of MonoTime.t
   | SetLeader of IntID.t
   | SetTerm of Index.t
   | Restart
   | Commit of Index.t
   | AppendEntries of (Index.t * Index.t * Mach.cmd) list
   | AppendEntry of (Index.t * Index.t * Mach.cmd)
   | RemoveEntries of Index.t * Index.t
   | ReplicationFailure of IntID.t * Index.t
   | ReplicationSuccess of IntID.t * Index.t
   | AddClientRequest of Index.t * Rpcs.ClientArg.t
   | RemoveClientRes 


  let init me all =
    { term = Index.init();
      mode = Follower;
      time = MonoTime.init;
      timer = 0;
      votedFor = None;
      log = Log.init(); 
      lastlogIndex = Index.init();
      lastlogTerm = Index.init();
      commitIndex = Index.init();
      votesResponded = [];
      votesGranted = [];
      nextIndex  = [];
      matchIndex = [];
      id = me;
      allNodes = all;
      leader = None;
      state_mach = Mach.init();
      outstanding_request = None;
      safety_monitor = RaftMonitor.tick (RaftMonitor.init()) `Startup ;
    } 

  let refresh s:t =
    { term = s.term;
      mode = Follower;
      time = s.time ;
      timer = s.timer;
      votedFor = None;
      log = s.log; 
      lastlogIndex = Index.init();
      lastlogTerm = Index.init();
      commitIndex = Index.init();
      votesResponded = [];
      votesGranted = [];
      nextIndex  = [];
      matchIndex = [];
      id = s.id; 
      allNodes = s.allNodes;
      leader = None;
      state_mach = s.state_mach;
      outstanding_request = None;
      safety_monitor =  RaftMonitor.tick (RaftMonitor.init()) `Recover;
    } 


  let id_print = function  None -> "none" | Some x -> IntID.to_string x
  let id_index_print (id,index) = (" ID: "^(IntID.to_string id)^"Index: "^(Index.to_string index) )

  let print s = 
    "-- NODE STATE ------------------------------------------------------------------------\n"^
    " | Time: "^(MonoTime.to_string (s.time()))^
    " | ID: "^(IntID.to_string s.id)^
    " | Term: "^(Index.to_string s.term)^
    " | Mode: "^(string_of_role s.mode)^"\n"^
    " | VotedFor: "^(string_of_option IntID.to_string s.votedFor)^
    " | All Nodes: "^(List.to_string ~f:IntID.to_string s.allNodes)^
    " | Votes Recieved: "^ (List.to_string ~f:IntID.to_string s.votesGranted)^
    " | Leader: "^(string_of_option (IntID.to_string) s.leader)^"\n"^
    " | State Machine: "^(Mach.to_string s.state_mach)^
    " | Match Index: "^(List.to_string s.matchIndex ~f:id_index_print)^"\n"^
    " | Next Index: "^(List.to_string s.matchIndex ~f:id_index_print)^"\n"^
    " | Last Log Index: "^(Index.to_string s.lastlogIndex)^
    " | Last Log Term: "^(Index.to_string s.lastlogTerm)^
    " | Commit Index: "^(Index.to_string s.commitIndex)^
    " | Replicated Log: "^(Log.to_string ~cmd_to_string:Mach.cmd_to_string s.log)^
    "\n-------------------------------------------------------------------------------------"
 (* sexp_of_t s |> Sexp.to_string *)

 (* let rec log_add original new_list = 
   (*combine logs, if two items with same index, give proirity to new *)
   match new_list with
   | (index,term,cmd)::rest -> log_add ((index,term,cmd)::(List.filter original ~f:(fun (i,_,_) -> not(index=i)))) rest
   | [] -> original *)


 let update_commitIndex match_list commitIndex log curr_term =
    let magority = (List.length match_list +1 / 2) in
    let discard (_,index) = 
      (if index > commitIndex then Some index else None) in
    let new_list = 
      List.sort ~cmp:(fun x y -> -1* (Index.compare x y))
        (List.filter_map ~f:discard match_list) in
    match List.nth new_list magority with 
    | Some index -> 
      let (_,term) = Log.specific_index_term index log in
      if (curr_term=term) then index else commitIndex
    | None -> commitIndex

  let tick tk s =
  match tk with
    | Set -> 
        {s with timer = Int.succ (s.timer)}
    | IncrementTerm -> 
        let t = Index.succ s.term in
       { s with term=t }
    | Vote id -> 
         assert(s.votedFor = None);
         assert(s.mode=Follower);
        { s with votedFor = Some id}
    | VoteFrom id ->
        { s with votesGranted = id::s.votesGranted }
    | StepDown tm ->
        { s with mode=Follower;
        term = tm;
        votedFor = None;
        votesResponded=[];
        votesGranted=[];
        nextIndex  = [];
        matchIndex = []; 
        leader = None;
        safety_monitor = (
            match s.mode with 
            | Candidate -> RaftMonitor.tick s.safety_monitor `StepDown_from_Candidate
            | Leader -> RaftMonitor.tick s.safety_monitor `StepDown_from_Leader 
            | Follower -> s.safety_monitor );
        }
    | StartCandidate -> 
        { s with mode=Candidate;
        votedFor = Some s.id;
        votesResponded=[];
        votesGranted=[s.id];
        term = (Index.succ s.term);
        leader = None;
        safety_monitor =
            match s.mode with
            | Candidate -> RaftMonitor.tick s.safety_monitor `RestartElection
{'fraction_non_alphanumeric': 0.14313079457512834, 'fraction_numerical': 0.0007662248103593594, 'mean_word_length': 2.03111936832327, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 49, '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': '6216612', 'n_tokens_mistral': 4300, 'n_tokens_neox': 4170, 'n_words': 1073}