Skip to content

Commit

Permalink
Make Agent Register request asynchronous
Browse files Browse the repository at this point in the history 
The "Register" agent->controller method call could get stuck
when the IP address configured as ControllerHost does not exist.
This timeout of about 2 seconds could completely occupy the event
loop and would not allow other calls on the DBus API of the Agent.

Making the "Register" method call asynchronous we allow the event
loop to stay unoccupied and be able to accept SwitchController
request.

Resolves: #950
Signed-off-by: Mark Kemel <[email protected]>
  • Loading branch information
@mkemel
mkemel committed Oct 16, 2024
1 parent 0e8c2df commit 4fb58b2
Show file tree
Hide file tree
Showing 2 changed files with 90 additions and 45 deletions.
131 changes: 87 additions & 44 deletions src/agent/agent.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,13 @@ static int agent_heartbeat_timer_callback(sd_event_source *event_source, UNUSED
Agent *agent = (Agent *) userdata;

int r = 0;
if (agent->connection_state == AGENT_CONNECTION_STATE_CONNECTING) {
bc_log_error("Agent connection attempt failed, retrying");
if (agent->register_call_slot != NULL) {
sd_bus_slot_unrefp(&agent->register_call_slot);
}
agent->connection_state = AGENT_CONNECTION_STATE_RETRY;
}
if (agent->connection_state == AGENT_CONNECTION_STATE_CONNECTED &&
agent_check_controller_liveness(agent)) {
r = sd_bus_emit_signal(
Expand Down Expand Up @@ -532,6 +539,9 @@ void agent_unref(Agent *agent) {
sd_event_unrefp(&agent->event);
}

if (agent->register_call_slot != NULL) {
sd_bus_slot_unrefp(&agent->register_call_slot);
}
if (agent->metrics_slot != NULL) {
sd_bus_slot_unrefp(&agent->metrics_slot);
}
Expand Down Expand Up @@ -2684,53 +2694,21 @@ void agent_stop(Agent *agent) {
}
}

static bool agent_connect(Agent *agent) {
peer_bus_close(agent->peer_dbus);

bc_log_infof("Connecting to controller on %s", agent->orch_addr);

agent->peer_dbus = peer_bus_open(agent->event, "peer-bus-to-controller", agent->orch_addr);
if (agent->peer_dbus == NULL) {
bc_log_error("Failed to open peer dbus");
return false;
}

bus_socket_set_options(agent->peer_dbus, agent->peer_socket_options);

int r = sd_bus_add_object_vtable(
agent->peer_dbus,
NULL,
INTERNAL_AGENT_OBJECT_PATH,
INTERNAL_AGENT_INTERFACE,
internal_agent_vtable,
agent);
if (r < 0) {
bc_log_errorf("Failed to add agent vtable: %s", strerror(-r));
return false;
}
static int agent_process_register_callback(sd_bus_message *m, Agent *agent) {
int r = 0;

_cleanup_sd_bus_message_ sd_bus_message *bus_msg = NULL;
sd_bus_error error = SD_BUS_ERROR_NULL;
r = sd_bus_call_method(
agent->peer_dbus,
BC_DBUS_NAME,
INTERNAL_CONTROLLER_OBJECT_PATH,
INTERNAL_CONTROLLER_INTERFACE,
"Register",
&error,
&bus_msg,
"s",
agent->name);
if (r < 0) {
bc_log_errorf("Registering as '%s' failed: %s", agent->name, error.message);
sd_bus_error_free(&error);
return false;
if (sd_bus_message_is_method_error(m, NULL)) {
bc_log_errorf("Registering as '%s' failed: %s",
agent->name,
sd_bus_message_get_error(m)->message);
return -EPERM;
}

r = sd_bus_message_read(bus_msg, "");
bc_log_info("Register call response received");
r = sd_bus_message_read(m, "");
if (r < 0) {
bc_log_errorf("Failed to parse response message: %s", strerror(-r));
return false;
return r;
}

bc_log_infof("Connected to controller as '%s'", agent->name);
Expand Down Expand Up @@ -2759,7 +2737,7 @@ static bool agent_connect(Agent *agent) {
agent);
if (r < 0) {
bc_log_errorf("Failed to add heartbeat signal match: %s", strerror(-r));
return false;
return r;
}

r = sd_bus_match_signal_async(
Expand All @@ -2774,7 +2752,7 @@ static bool agent_connect(Agent *agent) {
agent);
if (r < 0) {
bc_log_errorf("Failed to request match for Disconnected signal: %s", strerror(-r));
return false;
return r;
}

/* re-emit ProxyNew signals */
Expand All @@ -2791,6 +2769,71 @@ static bool agent_connect(Agent *agent) {
}
}

return 0;
}

static int agent_register_callback(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
Agent *agent = (Agent *) userdata;

if (agent->connection_state != AGENT_CONNECTION_STATE_CONNECTING) {
bc_log_error("Agent is not in CONNECTING state, dropping Register callback");
if (agent->register_call_slot != NULL) {
sd_bus_slot_unrefp(&agent->register_call_slot);
}
return -EPERM;
}
int r = 0;
r = agent_process_register_callback(m, agent);
if (r < 0) {
agent->connection_state = AGENT_CONNECTION_STATE_RETRY;
return r;
}
return 0;
}

static bool agent_connect(Agent *agent) {
peer_bus_close(agent->peer_dbus);

bc_log_infof("Connecting to controller on %s", agent->orch_addr);
agent->connection_state = AGENT_CONNECTION_STATE_CONNECTING;

agent->peer_dbus = peer_bus_open(agent->event, "peer-bus-to-controller", agent->orch_addr);
if (agent->peer_dbus == NULL) {
bc_log_error("Failed to open peer dbus");
return false;
}

bus_socket_set_options(agent->peer_dbus, agent->peer_socket_options);

int r = sd_bus_add_object_vtable(
agent->peer_dbus,
NULL,
INTERNAL_AGENT_OBJECT_PATH,
INTERNAL_AGENT_INTERFACE,
internal_agent_vtable,
agent);
if (r < 0) {
bc_log_errorf("Failed to add agent vtable: %s", strerror(-r));
return false;
}

_cleanup_sd_bus_message_ sd_bus_message *bus_msg = NULL;
r = sd_bus_call_method_async(
agent->peer_dbus,
&agent->register_call_slot,
BC_DBUS_NAME,
INTERNAL_CONTROLLER_OBJECT_PATH,
INTERNAL_CONTROLLER_INTERFACE,
"Register",
agent_register_callback,
agent,
"s",
agent->name);
if (r < 0) {
bc_log_errorf("Registering as '%s' failed: %s", agent->name, strerror(-r));
return false;
}

return true;
}

Expand Down
4 changes: 3 additions & 1 deletion src/agent/agent.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,8 @@ typedef struct JobTracker JobTracker;
typedef enum {
AGENT_CONNECTION_STATE_DISCONNECTED,
AGENT_CONNECTION_STATE_CONNECTED,
AGENT_CONNECTION_STATE_RETRY
AGENT_CONNECTION_STATE_RETRY,
AGENT_CONNECTION_STATE_CONNECTING
} AgentConnectionState;

struct Agent {
Expand Down Expand Up @@ -82,6 +83,7 @@ struct Agent {
sd_bus *systemd_dbus;
sd_bus *peer_dbus;

sd_bus_slot *register_call_slot;
sd_bus_slot *metrics_slot;

LIST_HEAD(SystemdRequest, outstanding_requests);
Expand Down

0 comments on commit 4fb58b2

Please sign in to comment.