patrick/raft-fable
1
0
Fork 0
Code Issues Pull Requests Packages Releases Activity

Use spans for Efficiency

Browse Source
This commit is contained in:
Smaug123
2022-11-13 21:44:19 +00:00
parent 2f1ae5cd7f
commit 72785eda06
12 changed files with 315 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
Raft/EfficientString.fs Normal file
View File

@@ -0,0 +1,44 @@
namespace Raft
open System
type internal EfficientString =
#if NETSTANDARD2_0
string
#else
System.ReadOnlySpan<char>
#endif
[<RequireQualifiedAccess>]
module internal EfficientString =
let inline ofString (s : string) : EfficientString =
#if NETSTANDARD2_0
s
#else
s.AsSpan ()
#endif
let inline trimStart (s : EfficientString) : EfficientString = s.TrimStart ()
let inline slice (start : int) (length : int) (s : EfficientString) : EfficientString =
#if NETSTANDARD2_0
s.[start .. start + length - 1]
#else
s.Slice (start, length)
#endif
/// Mutates the input to drop up to the first instance of the input char,
/// and returns what was dropped.
/// If the char is not present, deletes the input.
let takeUntil<'a> (c : char) (s : EfficientString byref) : EfficientString =
let first = s.IndexOf c
if first < 0 then
let toRet = s
s <- EfficientString.Empty
toRet
else
let toRet = slice 0 first s
s <- slice (first + 1) (s.Length - first - 1) s
toRet

221
Raft/InMemory.fs
View File

@@ -13,17 +13,17 @@ type Cluster<'a> =
member this.SendMessage (i : int<ServerId>) (m : Message<'a>) : unit = this.SendMessageDirectly i m
member this.InactivityTimeout (i : int<ServerId>) : unit =
this.Servers.[i / 1<ServerId>].TriggerInactivityTimeout ()
this.Servers.[i / 1<ServerId>].Sync ()
this.Servers[ i / 1<ServerId> ].TriggerInactivityTimeout ()
this.Servers[ i / 1<ServerId> ].Sync ()
member this.HeartbeatTimeout (i : int<ServerId>) : unit =
this.Servers.[i / 1<ServerId>].TriggerHeartbeatTimeout ()
this.Servers.[i / 1<ServerId>].Sync ()
this.Servers[ i / 1<ServerId> ].TriggerHeartbeatTimeout ()
this.Servers[ i / 1<ServerId> ].Sync ()
member this.Status (i : int<ServerId>) : ServerStatus = this.Servers.[i / 1<ServerId>].State
member this.Status (i : int<ServerId>) : ServerStatus = this.Servers[i / 1<ServerId>].State
member this.GetCurrentInternalState (i : int<ServerId>) : ServerInternalState<'a> Async =
this.Servers.[i / 1<ServerId>].GetCurrentInternalState ()
this.Servers[ i / 1<ServerId> ].GetCurrentInternalState ()
member this.ClusterSize : int = this.Servers.Length
@@ -55,21 +55,21 @@ type Network<'a> =
}
member this.AllInboundMessages (i : int<ServerId>) : Message<'a> list =
this.CompleteMessageHistory.[i / 1<ServerId>] |> List.ofSeq
this.CompleteMessageHistory[i / 1<ServerId>] |> List.ofSeq
member this.InboundMessage (i : int<ServerId>) (id : int) : Message<'a> =
this.CompleteMessageHistory.[i / 1<ServerId>].[id]
this.CompleteMessageHistory[i / 1<ServerId>].[id]
member this.DropMessage (i : int<ServerId>) (id : int) : unit =
this.MessagesDelivered.[i / 1<ServerId>].Add id |> ignore
this.MessagesDelivered[ i / 1<ServerId> ].Add id |> ignore
member this.UndeliveredMessages (i : int<ServerId>) : (int * Message<'a>) list =
this.CompleteMessageHistory.[i / 1<ServerId>]
this.CompleteMessageHistory[i / 1<ServerId>]
|> Seq.indexed
|> Seq.filter (fun (count, _) -> this.MessagesDelivered.[i / 1<ServerId>].Contains count |> not)
|> Seq.filter (fun (count, _) -> this.MessagesDelivered[ i / 1<ServerId> ].Contains count |> not)
|> List.ofSeq
member this.AllUndeliveredMessages () : ((int * Message<'a>) list) list =
member this.AllUndeliveredMessages () : (int * Message<'a>) list list =
List.init this.CompleteMessageHistory.Length (fun i -> this.UndeliveredMessages (i * 1<ServerId>))
member this.ClusterSize = this.CompleteMessageHistory.Length
@@ -84,19 +84,19 @@ module InMemoryCluster =
let network = Network<int>.Make count
let messageChannelHold (serverId : int<ServerId>) (message : Message<'a>) : unit =
let arr = network.CompleteMessageHistory.[serverId / 1<ServerId>]
let arr = network.CompleteMessageHistory[serverId / 1<ServerId>]
lock arr (fun () -> arr.Add message)
for s in 0 .. servers.Length - 1 do
servers.[s] <- Server (count, s * 1<ServerId>, InMemoryPersistentState (), messageChannelHold)
servers[s] <- Server (count, s * 1<ServerId>, InMemoryPersistentState (), messageChannelHold)
let cluster =
{
Servers = servers
SendMessageDirectly =
fun i m ->
servers.[i / 1<ServerId>].Message m
servers.[i / 1<ServerId>].Sync ()
servers[ i / 1<ServerId> ].Message m
servers[ i / 1<ServerId> ].Sync ()
}
cluster, network
@@ -121,28 +121,33 @@ module NetworkAction =
network.DropMessage serverId messageId
| ClientRequest (server, request) -> Message.ClientRequest request |> cluster.SendMessage server
let private getMessage (clusterSize : int) (s : string) : Result<int<ServerId> * int, string> =
match s.Split ',' with
| [| serverId ; messageId |] ->
let serverId = serverId.TrimEnd ()
let messageId = messageId.Trim ()
let private getMessage (clusterSize : int) (s : EfficientString) : Result<int<ServerId> * int, string> =
let mutable messageId = s
let serverId = EfficientString.takeUntil ',' &messageId
match Int32.TryParse serverId with
| true, serverId ->
match Int32.TryParse messageId with
| true, messageId ->
if serverId >= clusterSize || serverId < 0 then
sprintf "Server ID must be between 0 and %i inclusive, got %i" (clusterSize - 1) serverId
|> Error
else
Ok (serverId * 1<ServerId>, messageId)
| false, _ -> sprintf "Non-integer input '%s' for message ID." messageId |> Error
| false, _ -> sprintf "Non-integer input '%s' for server ID." serverId |> Error
| _ -> Error "Expected a single comma."
let serverId = serverId.TrimEnd ()
let messageId = messageId.Trim ()
let private getTimeout (clusterSize : int) (serverId : string) : Result<int<ServerId>, string> =
match Int32.TryParse serverId with
| false, _ -> Error (sprintf "Expected an integer, got '%s'" serverId)
| false, _ -> sprintf "Non-integer input '%s' for server ID." (serverId.ToString ()) |> Error
| true, serverId ->
match Int32.TryParse messageId with
| false, _ ->
sprintf "Non-integer input '%s' for message ID." (messageId.ToString ())
|> Error
| true, messageId ->
if serverId >= clusterSize || serverId < 0 then
sprintf "Server ID must be between 0 and %i inclusive, got %i" (clusterSize - 1) serverId
|> Error
else
Ok (serverId * 1<ServerId>, messageId)
let private getTimeout (clusterSize : int) (serverId : EfficientString) : Result<int<ServerId>, string> =
match Int32.TryParse serverId with
| false, _ -> Error (sprintf "Expected an integer, got '%s'" (serverId.ToString ()))
| true, serverId ->
if serverId >= clusterSize || serverId < 0 then
@@ -151,9 +156,11 @@ module NetworkAction =
else
serverId * 1<ServerId> |> Ok
let private getHeartbeat (leaders : Set<int<ServerId>> option) (clusterSize : int) (serverId : string) =
let private getHeartbeat (leaders : Set<int<ServerId>> option) (clusterSize : int) (serverId : EfficientString) =
match Int32.TryParse serverId with
| false, _ -> sprintf "Expected an integer server ID, got '%s'" serverId |> Error
| false, _ ->
sprintf "Expected an integer server ID, got '%s'" (serverId.ToString ())
|> Error
| true, serverId ->
if serverId >= clusterSize || serverId < 0 then
@@ -172,9 +179,15 @@ module NetworkAction =
else
sprintf "Cannot heartbeat a non-leader (%i)." serverId |> Error
let private getNewClientTarget<'a> (clusterSize : int) (serverId : string) : Result<int<ServerId>, string> =
let private getNewClientTarget<'a>
(clusterSize : int)
(serverId : EfficientString)
: Result<int<ServerId>, string>
=
match Int32.TryParse serverId with
| false, _ -> sprintf "Expected an int for a server ID, got '%s'" serverId |> Error
| false, _ ->
sprintf "Expected an int for a server ID, got '%s'" (serverId.ToString ())
|> Error
| true, serverId ->
if serverId >= clusterSize || serverId < 0 then
@@ -183,47 +196,43 @@ module NetworkAction =
else
Ok (serverId * 1<ServerId>)
/// Mutates the input byref to contain the result.
let private getClientSubmitData<'a>
(parse : string -> Result<'a, string>)
(clusterSize : int)
(s : string)
: Result<int<ServerId> * int<ClientId> * int<ClientSequence> * 'a, string>
(s : byref<EfficientString>)
: Result<int<ServerId> * int<ClientId> * int<ClientSequence>, string>
=
match s.Split ',' |> List.ofArray with
| serverId :: clientId :: clientSequenceNumber :: (_ :: _ as rest) ->
let rest = String.concat "," rest |> fun s -> s.TrimStart ()
let serverId = EfficientString.takeUntil ',' &s
let clientId = EfficientString.takeUntil ',' &s
let clientSequenceNumber = EfficientString.takeUntil ',' &s
match Int32.TryParse (serverId.Trim ()) with
| false, _ ->
sprintf "Server ID expected as first comma-separated component, got '%s'." serverId
|> Error
| true, serverId ->
if serverId >= clusterSize || serverId < 0 then
sprintf "Server ID must be between 0 and %i inclusive, got %i." (clusterSize - 1) serverId
|> Error
else
match Int32.TryParse (clientId.Trim ()) with
| false, _ ->
sprintf "Client ID expected as second comma-separated component, got '%s'." clientId
|> Error
| true, clientId ->
match Int32.TryParse (clientSequenceNumber.Trim ()) with
| false, _ ->
sprintf
"Client sequence number expected as third comma-separated component, got '%s'."
clientSequenceNumber
|> Error
| true, clientSequenceNumber ->
match parse rest with
| Ok b -> Ok (serverId * 1<ServerId>, clientId * 1<ClientId>, clientSequenceNumber * 1<ClientSequence>, b)
| Error e -> sprintf "Failed to parse client data: %s" e |> Error
| _ ->
sprintf "Expected serverId,clientId,clientSequenceNumber,data; got '%s'" s
match Int32.TryParse (serverId.Trim ()) with
| false, _ ->
sprintf "Server ID expected as first comma-separated component, got '%s'." (serverId.ToString ())
|> Error
| true, serverId ->
if serverId >= clusterSize || serverId < 0 then
sprintf "Server ID must be between 0 and %i inclusive, got %i." (clusterSize - 1) serverId
|> Error
else
match Int32.TryParse (clientId.Trim ()) with
| false, _ ->
sprintf "Client ID expected as second comma-separated component, got '%s'." (clientId.ToString ())
|> Error
| true, clientId ->
match Int32.TryParse (clientSequenceNumber.Trim ()) with
| false, _ ->
sprintf
"Client sequence number expected as third comma-separated component, got '%s'."
(clientSequenceNumber.ToString ())
|> Error
| true, clientSequenceNumber ->
(serverId * 1<ServerId>, clientId * 1<ClientId>, clientSequenceNumber * 1<ClientSequence>)
|> Ok
/// Optionally also validates leaders against the input set of leaders.
let tryParse<'a>
@@ -238,37 +247,45 @@ module NetworkAction =
if String.IsNullOrEmpty s then
Error "Can't parse an empty string"
else
match Char.ToUpper s.[0] with
| 'T' ->
match getTimeout clusterSize (s.[1..].TrimStart ()) with
| Ok t -> t |> InactivityTimeout |> Ok
| Error e -> Error e
| 'D' ->
match getMessage clusterSize (s.[1..].TrimStart ()) with
| Ok m -> m |> DropMessage |> Ok
| Error e -> Error e
| 'M' ->
match getMessage clusterSize (s.[1..].TrimStart ()) with
| Ok m -> m |> NetworkMessage |> Ok
| Error e -> Error e
| 'H' ->
match getHeartbeat leaders clusterSize (s.[1..].TrimStart ()) with
| Ok h -> Heartbeat h |> Ok
| Error e -> Error e
| 'S' ->
match getNewClientTarget clusterSize (s.[1..].TrimStart ()) with
| Ok target ->
ClientRequest (target, ClientRequest.RegisterClient handleRegisterClientResponse)
|> Ok
| Error e -> Error e
| 'R' ->
match getClientSubmitData parse clusterSize (s.[1..].TrimStart ()) with
| Ok (server, client, sequence, data) ->
let rest = EfficientString.slice 1 (s.Length - 1) (EfficientString.ofString s)
match Char.ToUpper s[0] with
| 'T' ->
match getTimeout clusterSize (EfficientString.trimStart rest) with
| Ok t -> t |> InactivityTimeout |> Ok
| Error e -> Error e
| 'D' ->
match getMessage clusterSize (EfficientString.trimStart rest) with
| Ok m -> m |> DropMessage |> Ok
| Error e -> Error e
| 'M' ->
match getMessage clusterSize (EfficientString.trimStart rest) with
| Ok m -> m |> NetworkMessage |> Ok
| Error e -> Error e
| 'H' ->
match getHeartbeat leaders clusterSize (EfficientString.trimStart rest) with
| Ok h -> Heartbeat h |> Ok
| Error e -> Error e
| 'S' ->
match getNewClientTarget clusterSize (EfficientString.trimStart rest) with
| Ok target ->
ClientRequest (target, ClientRequest.RegisterClient handleRegisterClientResponse)
|> Ok
| Error e -> Error e
| 'R' ->
let mutable rest = EfficientString.trimStart rest
match getClientSubmitData clusterSize &rest with
| Ok (server, client, sequence) ->
match parse (rest.ToString ()) with
| Ok data ->
(server, ClientRequest.ClientRequest (client, sequence, data, handleClientDataResponse))
|> ClientRequest
|> Ok
| Error e -> Error e
| c -> Error (sprintf "unexpected start char '%c'" c)
| Error e -> Error e
| c -> Error (sprintf "unexpected start char '%c'" c)
let toString<'a> (action : NetworkAction<'a>) : string =
match action with

4
Raft/PersistentState.fs
View File

@@ -58,7 +58,7 @@ type InMemoryPersistentState<'a> () =
if log.Count = 0 then
None
else
let stored, term = log.[log.Count - 1]
let stored, term = log[log.Count - 1]
Some (
stored,
@@ -73,6 +73,6 @@ type InMemoryPersistentState<'a> () =
if log.Count < position then None
elif position <= 0 then None
else Some log.[position - 1]
else Some log[position - 1]
member this.CurrentLogIndex = log.Count * 1<LogIndex>

7
Raft/Raft.fsproj
View File

@@ -1,16 +1,21 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<TargetFrameworks>netstandard2.0;netstandard2.1</TargetFrameworks>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
</PropertyGroup>
<ItemGroup>
<Compile Include="AssemblyInfo.fs" />
<Compile Include="Domain.fs" />
<Compile Include="EfficientString.fs" />
<Compile Include="PersistentState.fs" />
<Compile Include="Server.fs" />
<Compile Include="InMemory.fs" />
</ItemGroup>
<ItemGroup>
<PackageReference Update="FSharp.Core" Version="4.3.4" />
</ItemGroup>
</Project>

45
Raft/Server.fs
View File

@@ -219,7 +219,7 @@ type private CandidateState =
static member New (count : int) (self : int<ServerId>) =
let votes = Array.zeroCreate<_> count
votes.[self / 1<ServerId>] <- Some true
votes[self / 1<ServerId>] <- Some true
{
Votes = votes
@@ -270,12 +270,12 @@ module internal ServerUtils =
if i < 0 then
result
else
let numberCounted = numberCounted + snd numberWhoCommittedIndex.[i]
let numberCounted = numberCounted + snd numberWhoCommittedIndex[i]
if numberCounted > matchIndex.Length / 2 then
fst numberWhoCommittedIndex.[i]
fst numberWhoCommittedIndex[i]
else
go numberCounted (fst numberWhoCommittedIndex.[i]) (i - 1)
go numberCounted (fst numberWhoCommittedIndex[i]) (i - 1)
go 0 0<LogIndex> (numberWhoCommittedIndex.Length - 1)
@@ -528,7 +528,7 @@ type Server<'a>
acceptRequest ()
let sendAppendEntries (leaderState : LeaderState) (j : int<ServerId>) =
let toSend = leaderState.ToSend.[j / 1<ServerId>]
let toSend = leaderState.ToSend[j / 1<ServerId>]
let prevLogTerm = persistentState.GetLogEntry (toSend - 1<LogIndex>)
{
@@ -572,14 +572,14 @@ type Server<'a>
match appendEntriesReply.Success with
| Some matchIndex ->
// They applied our request. Update our record of what we know they have applied...
leaderState.MatchIndex.[appendEntriesReply.Follower / 1<ServerId>] <- matchIndex
leaderState.MatchIndex[appendEntriesReply.Follower / 1<ServerId>] <- matchIndex
// ... and update our record of what we'll be sending them next.
leaderState.ToSend.[appendEntriesReply.Follower / 1<ServerId>] <- matchIndex + 1<LogIndex>
leaderState.ToSend[appendEntriesReply.Follower / 1<ServerId>] <- matchIndex + 1<LogIndex>
| None ->
// They failed to apply our request. Next time, we'll be trying one message further
// back in our history.
leaderState.ToSend.[appendEntriesReply.Follower / 1<ServerId>] <-
max (leaderState.ToSend.[appendEntriesReply.Follower / 1<ServerId>] - 1<LogIndex>) 1<LogIndex>
leaderState.ToSend[appendEntriesReply.Follower / 1<ServerId>] <-
max (leaderState.ToSend[appendEntriesReply.Follower / 1<ServerId>] - 1<LogIndex>) 1<LogIndex>
// Note that the decision to process this *here* means the algorithm doesn't work in clusters of
// only one node, because then there will never be any AppendEntries replies.
let maxLogAQuorumHasCommitted =
@@ -612,17 +612,18 @@ type Server<'a>
match logEntry with
| LogEntry.ClientEntry (stored, client, sequence, replyChannel) ->
let newClients =
volatileState.Clients
|> Map.change
client
(fun messageLog ->
let messages =
match messageLog with
| None -> Map.empty
| Some messageLog -> messageLog
match Map.tryFind client volatileState.Clients with
| None -> Map.add client Map.empty volatileState.Clients
| Some messageLog ->
let messageLog =
match Map.tryFind sequence messageLog with
| None -> messageLog |> Map.add sequence stored
| Some _ ->
// Here we'd assert that existing was equal to stored,
// if we had the equality constraint
messageLog
messages |> Map.change sequence (fun _ -> Some stored) |> Some
)
volatileState.Clients |> Map.add client messageLog
volatileState <-
{ volatileState with
@@ -649,7 +650,7 @@ type Server<'a>
| ServerSpecialisation.Candidate state ->
if requestVoteReply.VoterTerm = persistentState.CurrentTerm then
state.Votes.[requestVoteReply.Voter / 1<ServerId>] <- Some requestVoteReply.VoteGranted
state.Votes[requestVoteReply.Voter / 1<ServerId>] <- Some requestVoteReply.VoteGranted
// Inefficient, but :shrug:
if
@@ -714,7 +715,7 @@ type Server<'a>
(RaftOverhead (NewClientRegistered replyChannel))
persistentState.CurrentTerm
leaderState.MatchIndex.[me / 1<ServerId>] <- persistentState.CurrentLogIndex
leaderState.MatchIndex[me / 1<ServerId>] <- persistentState.CurrentLogIndex
emitHeartbeat leaderState
| ClientRequest.ClientRequest (client, sequenceNumber, toAdd, replyChannel) ->
@@ -724,7 +725,7 @@ type Server<'a>
(LogEntry.ClientEntry (toAdd, client, sequenceNumber, replyChannel))
persistentState.CurrentTerm
leaderState.MatchIndex.[me / 1<ServerId>] <- persistentState.CurrentLogIndex
leaderState.MatchIndex[me / 1<ServerId>] <- persistentState.CurrentLogIndex
emitHeartbeat leaderState
| ServerSpecialisation.Follower followerState ->