mirror of
https://gitee.com/mirrors_PX4/PX4-Autopilot.git
synced 2026-10-13 16:13:34 +08:00
Improved allocation logic
This commit is contained in:
+18
-28
@@ -24,6 +24,14 @@ namespace dynamic_node_id_server
|
||||
class IAllocationRequestHandler
|
||||
{
|
||||
public:
|
||||
/**
|
||||
* Allocation request manager uses this method to detect if it is allowed to publish follow-up responses.
|
||||
*/
|
||||
virtual bool canPublishFollowupAllocationResponse() const = 0;
|
||||
|
||||
/**
|
||||
* This method will be invoked when a new allocation request is received.
|
||||
*/
|
||||
virtual void handleAllocationRequest(const UniqueID& unique_id, NodeID preferred_node_id) = 0;
|
||||
|
||||
virtual ~IAllocationRequestHandler() { }
|
||||
@@ -42,7 +50,6 @@ class AllocationRequestManager
|
||||
|
||||
const MonotonicDuration stage_timeout_;
|
||||
|
||||
bool active_;
|
||||
MonotonicTime last_message_timestamp_;
|
||||
protocol::dynamic_node_id::Allocation::FieldTypes::unique_id current_unique_id_;
|
||||
|
||||
@@ -95,10 +102,8 @@ class AllocationRequestManager
|
||||
return InvalidStage;
|
||||
}
|
||||
|
||||
void broadcastIntermediateAllocationResponse()
|
||||
void publishFollowupAllocationResponse()
|
||||
{
|
||||
UAVCAN_ASSERT(active_);
|
||||
|
||||
protocol::dynamic_node_id::Allocation msg;
|
||||
msg.unique_id = current_unique_id_;
|
||||
UAVCAN_ASSERT(msg.unique_id.size() < msg.unique_id.capacity());
|
||||
@@ -120,11 +125,6 @@ class AllocationRequestManager
|
||||
return; // This is a response from another allocator, ignore
|
||||
}
|
||||
|
||||
if (!active_)
|
||||
{
|
||||
return; // The local node is not the leader, ignore
|
||||
}
|
||||
|
||||
/*
|
||||
* Reset the expected stage on timeout
|
||||
*/
|
||||
@@ -171,6 +171,7 @@ class AllocationRequestManager
|
||||
|
||||
/*
|
||||
* Proceeding with allocation if possible
|
||||
* Note that single-frame CAN FD allocation requests will be delivered to the server even if it's not leader.
|
||||
*/
|
||||
if (current_unique_id_.size() == current_unique_id_.capacity())
|
||||
{
|
||||
@@ -185,7 +186,14 @@ class AllocationRequestManager
|
||||
}
|
||||
else
|
||||
{
|
||||
broadcastIntermediateAllocationResponse();
|
||||
if (handler_.canPublishFollowupAllocationResponse())
|
||||
{
|
||||
publishFollowupAllocationResponse();
|
||||
}
|
||||
else
|
||||
{
|
||||
current_unique_id_.clear();
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -197,7 +205,6 @@ class AllocationRequestManager
|
||||
public:
|
||||
AllocationRequestManager(INode& node, IAllocationRequestHandler& handler)
|
||||
: stage_timeout_(MonotonicDuration::fromMSec(Allocation::FOLLOWUP_TIMEOUT_MS))
|
||||
, active_(false)
|
||||
, handler_(handler)
|
||||
, allocation_sub_(node)
|
||||
, allocation_pub_(node)
|
||||
@@ -222,25 +229,8 @@ public:
|
||||
return 0;
|
||||
}
|
||||
|
||||
void setActive(bool x)
|
||||
{
|
||||
active_ = x;
|
||||
if (!active_)
|
||||
{
|
||||
current_unique_id_.clear();
|
||||
}
|
||||
}
|
||||
|
||||
bool isActive() const { return active_; }
|
||||
|
||||
int broadcastAllocationResponse(const UniqueID& unique_id, NodeID allocated_node_id)
|
||||
{
|
||||
if (!active_)
|
||||
{
|
||||
UAVCAN_ASSERT(0);
|
||||
return -ErrLogic;
|
||||
}
|
||||
|
||||
protocol::dynamic_node_id::Allocation msg;
|
||||
|
||||
msg.unique_id.resize(msg.unique_id.capacity());
|
||||
|
||||
@@ -36,11 +36,6 @@ public:
|
||||
*/
|
||||
virtual void handleLogCommitOnLeader(const Entry& committed_entry) = 0;
|
||||
|
||||
/**
|
||||
* Assume false by default.
|
||||
*/
|
||||
virtual void handleLocalLeadershipChange(bool local_node_is_leader) = 0;
|
||||
|
||||
virtual ~IRaftLeaderMonitor() { }
|
||||
};
|
||||
|
||||
@@ -281,12 +276,6 @@ class RaftCore : private TimerBase
|
||||
int(server_state_), int(new_state));
|
||||
trace(TraceRaftStateSwitch, new_state);
|
||||
|
||||
if ((ServerStateLeader == server_state_) ||
|
||||
(ServerStateLeader == new_state))
|
||||
{
|
||||
leader_monitor_.handleLocalLeadershipChange(ServerStateLeader == new_state);
|
||||
}
|
||||
|
||||
server_state_ = new_state;
|
||||
|
||||
cluster_.resetAllServerIndices();
|
||||
@@ -782,6 +771,11 @@ public:
|
||||
*/
|
||||
Log::Index getCommitIndex() const { return commit_index_; }
|
||||
|
||||
/**
|
||||
* This essentially indicates whether the server could replicate log since last allocation.
|
||||
*/
|
||||
bool areAllLogEntriesCommitted() const { return commit_index_ == persistent_state_.getLog().getLastIndex(); }
|
||||
|
||||
/**
|
||||
* Only the leader can call @ref appendLog().
|
||||
*/
|
||||
|
||||
@@ -62,7 +62,85 @@ class Server : private IAllocationRequestHandler
|
||||
AllocationRequestManager allocation_request_manager_;
|
||||
|
||||
/*
|
||||
* Methods
|
||||
* Methods of IAllocationRequestHandler
|
||||
*/
|
||||
virtual bool canPublishFollowupAllocationResponse() const
|
||||
{
|
||||
/*
|
||||
* The server is allowed to publish follow-up allocation responses only if both conditions are met:
|
||||
* - The server is leader.
|
||||
* - The last allocation request has been completed successfully.
|
||||
*
|
||||
* Why second condition? Imagine a case when there's two Raft nodes that don't hear each other - A and B,
|
||||
* both of them are leaders (but only A can commit to the log, B is in a minor partition); then there's a
|
||||
* client X that can exchange with both leaders, and a client Y that can exchange only with A. Such a
|
||||
* situation can occur in case of a very unlikely failure of redundant interfaces.
|
||||
*
|
||||
* Both clients X and Y initially send a first-stage Allocation request; A responds to Y with a first-stage
|
||||
* response, whereas B responds to X. Both X and Y will issue a follow-up second-stage requests, which may
|
||||
* cause A to mix second-stage Allocation requests from different nodes, leading to reception of an invalid
|
||||
* unique ID. When both leaders receive full unique IDs (A will receive an invalid one, B will receive a valid
|
||||
* unique ID of X), only A will be able to make a commit, because B is in a minority. Since both clients were
|
||||
* unable to receive node ID values in this round, they will try again later.
|
||||
*
|
||||
* Now, in order to prevent B from disrupting client-server communication second time around, we introduce this
|
||||
* second restriction: the server cannot exchange with clients as long as its log contains uncommitted entries.
|
||||
*
|
||||
* Note that this restriction does not apply to allocation requests sent via CAN FD frames, as in this case
|
||||
* no follow-up responses are necessary. So only CAN FD can offer reliable Allocation exchange.
|
||||
*/
|
||||
return raft_core_.isLeader() && raft_core_.areAllLogEntriesCommitted();
|
||||
}
|
||||
|
||||
virtual void handleAllocationRequest(const UniqueID& unique_id, const NodeID preferred_node_id)
|
||||
{
|
||||
/*
|
||||
* Note that it is possible that the local node is not leader. We will still perform the log search
|
||||
* and try to find the node that requested allocation. If the node is found, response will be sent;
|
||||
* otherwise the request will be ignored because only leader can add new allocations.
|
||||
*/
|
||||
const LazyConstructor<RaftCore::LogEntryInfo> result =
|
||||
raft_core_.traverseLogFromEndUntil(UniqueIDLogPredicate(unique_id));
|
||||
|
||||
if (result.isConstructed())
|
||||
{
|
||||
if (result->committed)
|
||||
{
|
||||
tryPublishAllocationResult(result->entry);
|
||||
UAVCAN_TRACE("dynamic_node_id_server::distributed::Server",
|
||||
"Allocation request served with existing allocation; node ID %d",
|
||||
int(result->entry.node_id));
|
||||
}
|
||||
else
|
||||
{
|
||||
UAVCAN_TRACE("dynamic_node_id_server::distributed::Server",
|
||||
"Allocation request ignored - allocation exists but not committed yet; node ID %d",
|
||||
int(result->entry.node_id));
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// TODO: new allocations can be added only if the list of unidentified nodes is not empty
|
||||
if (raft_core_.isLeader())
|
||||
{
|
||||
allocateNewNode(unique_id, preferred_node_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* Methods of IRaftLeaderMonitor
|
||||
*/
|
||||
virtual void handleLogCommitOnLeader(const protocol::dynamic_node_id::server::Entry& entry)
|
||||
{
|
||||
/*
|
||||
* Maybe this node did not request allocation at all, we don't care, we publish anyway.
|
||||
*/
|
||||
tryPublishAllocationResult(entry);
|
||||
}
|
||||
|
||||
/*
|
||||
* Private methods
|
||||
*/
|
||||
bool isNodeIDTaken(const NodeID node_id) const
|
||||
{
|
||||
@@ -86,50 +164,10 @@ class Server : private IAllocationRequestHandler
|
||||
const int res = raft_core_.appendLog(unique_id, allocated_node_id);
|
||||
if (res < 0)
|
||||
{
|
||||
node_.registerInternalFailure("Raft log append");
|
||||
node_.registerInternalFailure("Raft log append new allocation");
|
||||
}
|
||||
}
|
||||
|
||||
virtual void handleAllocationRequest(const UniqueID& unique_id, const NodeID preferred_node_id)
|
||||
{
|
||||
// TODO: allocation requests must not be served if the list of unidentified nodes is not empty
|
||||
|
||||
const LazyConstructor<RaftCore::LogEntryInfo> result =
|
||||
raft_core_.traverseLogFromEndUntil(UniqueIDLogPredicate(unique_id));
|
||||
|
||||
if (result.isConstructed())
|
||||
{
|
||||
if (result->committed)
|
||||
{
|
||||
tryPublishAllocationResult(result->entry);
|
||||
UAVCAN_TRACE("dynamic_node_id_server::distributed::Server",
|
||||
"Allocation request served with existing allocation; node ID %d",
|
||||
int(result->entry.node_id));
|
||||
}
|
||||
else
|
||||
{
|
||||
UAVCAN_TRACE("dynamic_node_id_server::distributed::Server",
|
||||
"Allocation request ignored - allocation exists but not committed yet; node ID %d",
|
||||
int(result->entry.node_id));
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
allocateNewNode(unique_id, preferred_node_id);
|
||||
}
|
||||
}
|
||||
|
||||
virtual void handleLogCommitOnLeader(const protocol::dynamic_node_id::server::Entry& entry)
|
||||
{
|
||||
tryPublishAllocationResult(entry);
|
||||
}
|
||||
|
||||
virtual void handleLocalLeadershipChange(bool local_node_is_leader)
|
||||
{
|
||||
UAVCAN_TRACE("dynamic_node_id_server::distributed::Server", "I am leader: %d", int(local_node_is_leader));
|
||||
allocation_request_manager_.setActive(local_node_is_leader);
|
||||
}
|
||||
|
||||
void tryPublishAllocationResult(const protocol::dynamic_node_id::server::Entry& entry)
|
||||
{
|
||||
const int res = allocation_request_manager_.broadcastAllocationResponse(entry.unique_id, entry.node_id);
|
||||
|
||||
@@ -15,11 +15,20 @@ class AllocationRequestHandler : public uavcan::dynamic_node_id_server::IAllocat
|
||||
std::vector<std::pair<UniqueID, uavcan::NodeID> > requests_;
|
||||
|
||||
public:
|
||||
bool can_followup;
|
||||
|
||||
AllocationRequestHandler() : can_followup(false) { }
|
||||
|
||||
virtual void handleAllocationRequest(const UniqueID& unique_id, uavcan::NodeID preferred_node_id)
|
||||
{
|
||||
requests_.push_back(std::pair<UniqueID, uavcan::NodeID>(unique_id, preferred_node_id));
|
||||
}
|
||||
|
||||
virtual bool canPublishFollowupAllocationResponse() const
|
||||
{
|
||||
return can_followup;
|
||||
}
|
||||
|
||||
bool matchAndPopLastRequest(const UniqueID& unique_id, uavcan::NodeID preferred_node_id)
|
||||
{
|
||||
if (requests_.empty())
|
||||
@@ -80,15 +89,12 @@ TEST(dynamic_node_id_server_AllocationRequestManager, Basic)
|
||||
* Request manager initialization
|
||||
*/
|
||||
AllocationRequestHandler handler;
|
||||
handler.can_followup = true;
|
||||
|
||||
AllocationRequestManager manager(nodes.a, handler);
|
||||
|
||||
ASSERT_LE(0, manager.init());
|
||||
|
||||
ASSERT_FALSE(manager.isActive());
|
||||
manager.setActive(true);
|
||||
ASSERT_TRUE(manager.isActive());
|
||||
|
||||
/*
|
||||
* Allocation
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user