Struct ReplicatedLog

pub struct ReplicatedLog<IO: Io> { /* private fields */ }

Raftアルゴリズムに基づく分散複製ログ.

利用者はpropose_commandメソッドを使って、コマンドをログに複製保存し、 発生するEventをハンドリングすることで、 整合性のある複製状態機械を実現することが可能となる.

ReplicatedLogはStreamトレイトを実装しているが、 これは無限ストリームであり、エラー時を除いて終了することはない.

ただし、構成変更によりノードがクラスタから切り離された場合は、 最終的には、イベントが生成されることは無くなる. this.local_history().config().is_known_node()メソッドを使うことで、 クラスタ内に属しているかどうかは判定可能なので、利用者側が明示的に確認して、 不要になったReplicatedLogインスタンスを回収することは可能.

Implementations

impl<IO: Io> ReplicatedLog<IO>

pub fn new( node_id: NodeId, members: ClusterMembers, io: IO, metric_builder: &MetricBuilder, ) -> Result<Self>

membersで指定されたクラスタに属するReplicatedLogのローカルインスタンス(ノード)を生成する.

ローカルノードのIDはnode_idで指定するが、これがmembersの含まれている必要は必ずしもない. 例えば、クラスタの構成変更に伴い、新規ノードを追加したい場合には、 membersに現行構成を指定することが望ましいが、このケースでは、 node_idはmembersの中には含まれないことになる.

なお、ノードの再起動時を除いて、node_idには対象クラスタの歴史の中でユニークなIDを 割り当てるのが望ましい. (レアケースではあるが、新規追加ノードを、以前に存在したノードと誤認識されてしまうと、 分散ログの整合性が壊れてしまう危険性があるため)

また、以前のノードを再起動したい場合でも、もし永続ストレージが壊れている等の理由で、 前回の状態を正確に復元できないのであれば、 ノード名を変更して、新規ノード追加扱いにした方が安全である.

pub fn metrics(&self) -> &Arc<RaftlogMetrics>

raftlog のメトリクスを返す。

pub fn propose_command(&mut self, command: Vec<u8>) -> Result<ProposalId>

新しいコマンドを提案する.

提案が承認(コミット)された場合には、返り値のLogPositionを含む Event::Committedイベントが返される.

もし返り値のLogPositionとは分岐したEvent::Committedが返された場合には、 この提案が棄却されたことを示している.

Errors

非リーダノードに対して、このメソッドが実行された場合には、 ErrorKind::NotLeaderを理由としたエラーが返される.

pub fn propose_config( &mut self, new_members: ClusterMembers, ) -> Result<ProposalId>

新しいクラスタ構成(新メンバ群)を提案する.

提案が承認(コミット)された場合には、返り値のLogPositionを含む Event::Committedイベントが返される. ただし、承認された場合であっても、それは新旧混合状態の構成が承認されただけであり、 新メンバのみの構成への移行完了を把握したい場合には、後続のコミットイベントの 追跡を行う必要がある.

もし返り値のLogPositionとは分岐したEvent::Committedが返された場合には、 この提案が棄却されたことを示している.

複数の構成変更を並行して実施することは可能だが、 その場合は、最後に提案されたものが最終的な構成として採用される.

Errors

非リーダノードに対して、このメソッドが実行された場合には、 ErrorKind::NotLeaderを理由としたエラーが返される.

pub fn heartbeat(&mut self) -> Result<SequenceNumber>

強制的にハートビートメッセージ(i.e., AppendEntriesCall)をブロードキャストする.

返り値は、送信メッセージのシーケンス番号.

last_heartbeat_ackメソッドを用いることで、 このハートビートに対して、過半数以上の応答を得られた タイミングを把握することが可能.

また、リーダのコミットを即座にフォロワーに伝えたい場合にも、 このメソッドが活用可能。 (Event::Committedをリーダが生成した直後にheartbeatメソッドを呼び出せば良い)

なおノードの役割が非リーダに変わった場合には、 応答待機中のハートビートは全て破棄されるので注意が必要.

Errors

非リーダノードに対して、このメソッドが実行された場合には、 ErrorKind::NotLeaderを理由としたエラーが返される.

pub fn install_snapshot( &mut self, new_head: LogIndex, snapshot: Vec<u8>, ) -> Result<()>

ローカルログにスナップショットをインストールする.

new_headが新しいローカルログの先頭位置となり、 snapshotはその地点までのコマンド群が適用済みの状態機械のスナップショット、となる.

Errors

既にローカルログに対するスナップショットのインストールが進行中の場合には、 ErrorKind::Busyを理由としてエラーが返される.

また現在のログの先頭よりも前の地点のスナップショットをインストールしようとした場合には、 ErrorKind::InvalidInputを理由としたエラーが返される.

pub fn start_election(&mut self)

新しい選挙を開始する.

何らかの手段で現在のリーダのダウンを検知した場合に呼び出される.

pub fn local_node(&self) -> &Node

ローカルノードの情報を返す.

pub fn local_history(&self) -> &LogHistory

ローカルログの履歴を返す.

pub fn proposal_queue_len(&self) -> usize

ローカルログへの書き込み待ちの状態の提案群の数を返す.

この値は、ローカルストレージの詰まり具合を把握するために有用である.

「ローカルログへは追記完了 and コミット待ち」の個数は 知りたい場合にはlocal_historyメソッド経由で取得可能.

ローカルノードが非リーダである場合には、常に0が返される.

pub fn is_snapshot_installing(&self) -> bool

スナップショットをインストール中の場合にはtrueを返す.

このメソッドがtrueを返している間は、 新しいスナップショットのインストールを行うことはできない.

pub fn last_heartbeat_ack(&self) -> SequenceNumber

過半数以上の応答を得られた最新のハートビート(i.e., AppendEntriesCall) のシーケンス番号を返す.

この値は、同じ選挙期間に関しては減少することはないことが保証されている.

注意

ハートビートを行うのはリーダノードのみなので、それ以外のノードに関しては、 このメソッドが返す値は意味を持たない.

pub fn cluster_config(&self) -> &ClusterConfig

現在のクラスタ構成を返す.

pub fn io(&self) -> &IO

I/O実装に対する参照を返す.

pub unsafe fn io_mut(&mut self) -> &mut IO

I/O実装に対する破壊的な参照を返す.

Safety

破壊的な操作は、Raftの管理外の挙動となり、 整合性を崩してしまう可能性もあるので、 注意を喚起する意味でunsafeと設定されている.

Trait Implementations

impl<IO: Io> Stream for ReplicatedLog<IO>

type Item = Event

The type of item this stream will yield on success.

type Error = Error

The type of error this stream may generate.

fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>

Attempt to pull out the next value of this stream, returning None if the stream is finished.

fn wait(self) -> Wait<Self>

where Self: Sized,

Creates an iterator which blocks the current thread until each item of this stream is resolved.

fn into_future(self) -> StreamFuture<Self>

where Self: Sized,

Converts this stream into a Future.

fn map<U, F>(self, f: F) -> Map<Self, F>

where F: FnMut(Self::Item) -> U, Self: Sized,

Converts a stream of type T to a stream of type U.

fn map_err<U, F>(self, f: F) -> MapErr<Self, F>

where F: FnMut(Self::Error) -> U, Self: Sized,

Converts a stream of error type T to a stream of error type U.

fn filter<F>(self, f: F) -> Filter<Self, F>

where F: FnMut(&Self::Item) -> bool, Self: Sized,

Filters the values produced by this stream according to the provided predicate.

fn filter_map<F, B>(self, f: F) -> FilterMap<Self, F>

where F: FnMut(Self::Item) -> Option<B>, Self: Sized,

Filters the values produced by this stream while simultaneously mapping them to a different type.

fn then<F, U>(self, f: F) -> Then<Self, F, U>

where F: FnMut(Result<Self::Item, Self::Error>) -> U, U: IntoFuture, Self: Sized,

Chain on a computation for when a value is ready, passing the resulting item to the provided closure f.

fn and_then<F, U>(self, f: F) -> AndThen<Self, F, U>

where F: FnMut(Self::Item) -> U, U: IntoFuture<Error = Self::Error>, Self: Sized,

Chain on a computation for when a value is ready, passing the successful results to the provided closure f.

fn or_else<F, U>(self, f: F) -> OrElse<Self, F, U>

where F: FnMut(Self::Error) -> U, U: IntoFuture<Item = Self::Item>, Self: Sized,

Chain on a computation for when an error happens, passing the erroneous result to the provided closure f.

fn collect(self) -> Collect<Self>

where Self: Sized,

Collect all of the values of this stream into a vector, returning a future representing the result of that computation.

fn fold<F, T, Fut>(self, init: T, f: F) -> Fold<Self, F, Fut, T>

where F: FnMut(T, Self::Item) -> Fut, Fut: IntoFuture<Item = T>, Self::Error: From<<Fut as IntoFuture>::Error>, Self: Sized,

Execute an accumulating computation over a stream, collecting all the values into one final result.

fn skip_while<P, R>(self, pred: P) -> SkipWhile<Self, P, R>

where P: FnMut(&Self::Item) -> R, R: IntoFuture<Item = bool, Error = Self::Error>, Self: Sized,

Skip elements on this stream while the predicate provided resolves to true.

fn take_while<P, R>(self, pred: P) -> TakeWhile<Self, P, R>

where P: FnMut(&Self::Item) -> R, R: IntoFuture<Item = bool, Error = Self::Error>, Self: Sized,

Take elements from this stream while the predicate provided resolves to true.

fn for_each<F, U>(self, f: F) -> ForEach<Self, F, U>

where F: FnMut(Self::Item) -> U, U: IntoFuture<Item = (), Error = Self::Error>, Self: Sized,

Runs this stream to completion, executing the provided closure for each element on the stream.

fn from_err<E>(self) -> FromErr<Self, E>

where E: From<Self::Error>, Self: Sized,

Map this stream’s error to any error implementing From for this stream’s Error, returning a new stream.

fn take(self, amt: u64) -> Take<Self>

where Self: Sized,

Creates a new stream of at most amt items of the underlying stream.

fn skip(self, amt: u64) -> Skip<Self>

where Self: Sized,

Creates a new stream which skips amt items of the underlying stream.

fn fuse(self) -> Fuse<Self>

where Self: Sized,

Fuse a stream such that poll will never again be called once it has finished.

fn by_ref(&mut self) -> &mut Self

where Self: Sized,

Borrows a stream, rather than consuming it.

fn catch_unwind(self) -> CatchUnwind<Self>

where Self: Sized + UnwindSafe,

Catches unwinding panics while polling the stream.

fn merge<S>(self, other: S) -> Merge<Self, S>

where S: Stream<Error = Self::Error>, Self: Sized,

👎Deprecated: functionality provided by select now

An adapter for merging the output of two streams.

fn zip<S>(self, other: S) -> Zip<Self, S>

where S: Stream<Error = Self::Error>, Self: Sized,

An adapter for zipping two streams together.

fn chain<S>(self, other: S) -> Chain<Self, S>

where S: Stream<Item = Self::Item, Error = Self::Error>, Self: Sized,

Adapter for chaining two stream.

fn peekable(self) -> Peekable<Self>

where Self: Sized,

Creates a new stream which exposes a peek method.

fn chunks(self, capacity: usize) -> Chunks<Self>

where Self: Sized,

An adaptor for chunking up items of the stream inside a vector.

fn select<S>(self, other: S) -> Select<Self, S>

where S: Stream<Item = Self::Item, Error = Self::Error>, Self: Sized,

Creates a stream that selects the next element from either this stream or the provided one, whichever is ready first.

fn forward<S>(self, sink: S) -> Forward<Self, S>

where S: Sink<SinkItem = Self::Item>, Self::Error: From<<S as Sink>::SinkError>, Self: Sized,

A future that completes after the given stream has been fully processed into the sink, including flushing.

fn inspect<F>(self, f: F) -> Inspect<Self, F>

where F: FnMut(&Self::Item), Self: Sized,

Do something with each item of this stream, afterwards passing it on.

fn inspect_err<F>(self, f: F) -> InspectErr<Self, F>

where F: FnMut(&Self::Error), Self: Sized,

Do something with the error of this stream, afterwards passing it on.