diff --git a/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp b/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp new file mode 100644 index 0000000000..bd5d311152 --- /dev/null +++ b/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp @@ -0,0 +1,365 @@ +/* + * Copyright (C) 2015 Pavel Kirienko + */ + +#ifndef UAVCAN_PROTOCOL_DYNAMIC_NODE_ID_ALLOCATION_SERVER_HPP_INCLUDED +#define UAVCAN_PROTOCOL_DYNAMIC_NODE_ID_ALLOCATION_SERVER_HPP_INCLUDED + +#include +#include +#include +#include +#include +#include +// Types used by the server +#include +#include +#include +#include +#include +#include +#include + +namespace uavcan +{ +/** + * + */ +class DynamicNodeIDAllocationServer +{ + /* + * Public type definitions + */ +public: + /** + * This interface is used by the server to read and write stable storage. + * The storage is represented as a key-value container, where keys and values are ASCII strings up to 32 + * characters long, not including the termination byte. Fixed block size allows for absolutely straightforward + * and efficient implementation of storage backends, e.g. based on text files. + */ + class IStorageBackend + { + public: + /** + * Maximum length of keys and values. One pair takes twice as much space. + */ + enum { MaxStringLength = 32 }; + + /** + * It is guaranteed that the server will never require more than this number of key/value pairs. + * Total storage space needed is (MaxKeyValuePairs * MaxStringLength * 2), not including storage overhead. + */ + enum { MaxKeyValuePairs = 400 }; + + /** + * This type is used to exchange data chunks with the backend. + * It doesn't use any dynamic memory; please refer to the Array<> class for details. + */ + typedef Array, ArrayModeDynamic, MaxStringLength> String; + + /** + * Read one value from the storage. + * If such key does not exist, or if read failed, an empty string will be returned. + * This method should not block for more than 50 ms. + */ + virtual String get(const String& key) const = 0; + + /** + * Create or update value for the given key. + * This method should not block for more than 50 ms. + * Failures will be ignored. + */ + virtual void set(const String& key, const String& value) = 0; + + virtual ~IStorageBackend() { } + }; + +private: + /* + * Callback type definitions + */ + typedef MethodBinder&)> + AllocationCallback; + + typedef MethodBinder&)> NodeStatusCallback; + + typedef MethodBinder AppendEntriesCallback; + + typedef MethodBinder RequestVoteCallback; + + /* + * Internal type definitions + */ + typedef Map PendingGetNodeInfoAttemptsMap; + + typedef StorageType::Type Term; + + /** + * This class extends the storage backend interface with serialization/deserialization functionality. + */ + class StorageMarshallingDecorator + { + IStorageBackend& storage_; + + public: + StorageMarshallingDecorator(IStorageBackend& storage) + : storage_(storage) + { + // Making sure that there will be no data loss during serialization. + StaticAssert<(sizeof(Term) <= sizeof(uint32_t))>::check(); + } + + /** + * Setters do the following: + * 1. Serialize the value. + * 2. Update the value on the backend. + * 3. Read the value back from the backend; return false if read fails. + * 4. Deserealize the newly read value. + * 5. Update the argument with deserialized value. + * 6. Return true. + * The caller then is supposed to check whether the argument has the desired value. + */ + bool setAndGetBack(const IStorageBackend::String& key, uint32_t& inout_value); + bool setAndGetBack(const IStorageBackend::String& key, + protocol::dynamic_node_id::server::Entry::FieldTypes::unique_id& inout_value); + + /** + * Getters simply read and deserialize the value. + */ + bool get(const IStorageBackend::String& key, uint32_t& out_value) const; + bool get(const IStorageBackend::String& key, + protocol::dynamic_node_id::server::Entry::FieldTypes::unique_id& out_value) const; + }; + + /** + * Raft log. + * This class transparently replicates its state to the storage backend, keeping the most recent state in memory. + * Writes are slow, reads are instantaneous. + */ + class Log + { + public: + typedef uint8_t Index; + + private: + enum { Capacity = NodeID::Max + 1 }; + + IStorageBackend& storage_; + protocol::dynamic_node_id::server::Entry entries_[Capacity]; + Index max_index_; // Index zero always contains an empty entry + + public: + Log(IStorageBackend& storage) + : storage_(storage) + , max_index_(0) + { } + + /** + * This method invokes storage IO. + */ + int init(); + + /** + * This method invokes storage IO. + */ + void append(const protocol::dynamic_node_id::server::Entry& entry); + + /** + * This method invokes storage IO. + */ + void removeEntriesWhereIndexGreaterOrEqual(Index index); + + /** + * Returns nullptr if there's no such index. + * This method does not use storage IO. + */ + const protocol::dynamic_node_id::server::Entry* getEntryAtIndex(Index index) const; + + Index getMaxIndex() const { return max_index_; } + + bool isOtherLogUpToDate(Index other_last_index, Term other_last_term) const; + }; + + /** + * This class is a convenient container for persistent state variables defined by Raft. + * Writes are slow, reads are instantaneous. + */ + class PersistentState + { + IStorageBackend& storage_; + + Term current_term_; + NodeID voted_for_; + Log log_; + + public: + PersistentState(IStorageBackend& storage) + : storage_(storage) + , current_term_(0) + { } + + int init(); + + Term getCurrentTerm() const { return current_term_; } + + NodeID getVotedFor() const { return voted_for_; } + + Log& getLog() { return log_; } + const Log& getLog() const { return log_; } + + /** + * Invokes storage IO. + */ + void setCurrentTerm(Term term); + + /** + * Invokes storage IO. + */ + void setVotedFor(NodeID node_id); + }; + + /** + * This class maintains the cluster state. + */ + class ClusterManager : private TimerBase + { + typedef MethodBinder&)> + DiscoveryCallback; + + struct Server + { + const NodeID node_id; + Log::Index next_index; + Log::Index match_index; + + Server() + : next_index(0) + , match_index(0) + { } + }; + + enum { MaxServers = protocol::dynamic_node_id::server::Discovery::FieldTypes::known_nodes::MaxSize }; + + const IStorageBackend& storage_; + const Log& log_; + + Subscriber discovery_sub_; + mutable Publisher discovery_pub_; + + Server servers_[MaxServers - 1]; ///< Minus one because the local server is not listed there. + + uint8_t cluster_size_; + uint8_t num_known_servers_; + + virtual void handleTimerEvent(const TimerEvent& event); + + void handleDiscovery(const ReceivedDataStructure& msg); + + void publishDiscovery() const; + + public: + enum { ClusterSizeUnknown = 0 }; + + /** + * @param node Needed to publish and subscribe to Discovery message + * @param storage Needed to read the cluster size parameter from the storage + * @param log Needed to initialize nextIndex[] values after elections + */ + ClusterManager(INode& node, const IStorageBackend& storage, const Log& log) + : storage_(storage) + , log_(log) + , discovery_sub_(node) + , discovery_pub_(node) + , cluster_size_(0) + , num_known_servers_(0) + { } + + /** + * If cluster_size is set to ClusterSizeUnknown, the class will try to read this parameter from the + * storage backend using key 'cluster_size'. + * Returns negative error code. + */ + int init(uint8_t cluster_size = ClusterSizeUnknown); + + /** + * An invalid node ID will be returned if there's no such server. + * The local server is not listed there. + */ + NodeID getRemoteServerNodeIDAtIndex(uint8_t index) const; + + /** + * See next_index[] in Raft paper. + */ + Log::Index getServerNextIndex(NodeID server_node_id) const; + void incrementServerNextIndexBy(NodeID server_node_id, Log::Index increment); + void decrementServerNextIndex(NodeID server_node_id); + + /** + * See match_index[] in Raft paper. + */ + Log::Index getServerMatchIndex(NodeID server_node_id) const; + void setServerMatchIndex(NodeID server_node_id, Log::Index match_index); + + /** + * This method must be called when the current server becomes leader. + */ + void resetAllServerIndices(); + + uint8_t getNumKnownServers() const { return num_known_servers_; } + uint8_t getConfiguredClusterSize() const { return cluster_size_; } + uint8_t getQuorumSize() const { return static_cast(cluster_size_ / 2U + 1U); } + }; + + enum ServerState + { + ServerStateFollower, + ServerStateCandidate, + ServerStateLeader + }; + + /* + * Raft-related states + */ + PersistentState persistent_state_; ///< Modifications of this state are slow as they involve storage IO. + + Log::Index commit_index_; + + ClusterManager cluster_; + + /* + * Implementation-specific states + */ + PendingGetNodeInfoAttemptsMap pending_get_node_info_attempts_; + + MonotonicTime last_activity_timestamp_; + bool active_mode_; + + ServerState server_state_; + + /* + * Transport + */ + Subscriber allocation_sub_; + Publisher allocation_pub_; + + Subscriber node_status_sub_; + + ServiceServer append_entries_srv_; + ServiceServer request_vote_srv_; +}; + +} + +#endif // UAVCAN_PROTOCOL_DYNAMIC_NODE_ID_ALLOCATION_SERVER_HPP_INCLUDED