along with this program; if not, see <http://www.gnu.org/licenses/>.
*/
-#include <talloc.h>
-#include <tevent.h>
-
#include "replace.h"
#include "system/network.h"
+#include <talloc.h>
+#include <tevent.h>
+
#include "lib/util/debug.h"
+#include "lib/util/tevent_unix.h"
#include "protocol/protocol.h"
#include "protocol/protocol_util.h"
#include "common/db_hash.h"
-#include "common/system.h"
+#include "common/system_socket.h"
#include "common/logging.h"
-/* Contains the listening socket and the list of TCP connections to
- * kill */
-struct ctdb_kill_tcp {
+struct reset_connections_state {
+ struct tevent_context *ev;
int capture_fd;
struct tevent_fd *fde;
struct db_hash_context *connections;
void *private_data;
- void *destructor_data;
unsigned int attempts;
unsigned int max_attempts;
struct timeval retry_interval;
};
+static void reset_connections_capture_tcp_handler(struct tevent_context *ev,
+ struct tevent_fd *fde,
+ uint16_t flags,
+ void *private_data);
+static void reset_connections_batch(struct tevent_req *subreq);
+static int reset_connections_tickle_connection(
+ uint8_t *keybuf, size_t keylen,
+ uint8_t *databuf, size_t datalen,
+ void *private_data);
+
+static struct tevent_req *reset_connections_send(
+ TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ const char *iface,
+ struct ctdb_connection_list *conn_list)
+{
+ struct tevent_req *req, *subreq;
+ struct reset_connections_state *state;
+ unsigned int i;
+ int ret;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct reset_connections_state);
+ if (req == NULL) {
+ return NULL;
+ }
+
+ state->ev = ev;
+
+ if (conn_list->num == 0) {
+ /* No connections, done! */
+ tevent_req_done(req);
+ return tevent_req_post(req, ev);
+ }
+
+ ret = db_hash_init(state, "connections", 2048, DB_HASH_SIMPLE,
+ &state->connections);
+ if (ret != 0) {
+ D_ERR("Failed to initialise connection hash (%s)\n",
+ strerror(ret));
+ tevent_req_error(req, ret);
+ return tevent_req_post(req, ev);
+ }
+
+ DBG_DEBUG("Adding %u connections to hash\n", conn_list->num);
+ for (i = 0; i < conn_list->num; i++) {
+ struct ctdb_connection *c = &conn_list->conn[i];
+
+ DBG_DEBUG("Adding connection to hash: %s\n",
+ ctdb_connection_to_string(conn_list, c, true));
+
+ /* Connection is stored as a key in the connections hash */
+ ret = db_hash_add(state->connections,
+ (uint8_t *)discard_const(c), sizeof(*c),
+ NULL, 0);
+ if (ret != 0) {
+ D_ERR("Error adding connection to hash (%s)\n",
+ strerror(ret));
+ tevent_req_error(req, ret);
+ return tevent_req_post(req, ev);
+ }
+ }
+
+ state->attempts = 0;
+ state->max_attempts = 50;
+
+ state->retry_interval.tv_sec = 0;
+ state->retry_interval.tv_usec = 100 * 1000;
+
+ state->batch_count = 0;
+ state->batch_size = 300;
+
+ state->capture_fd =
+ ctdb_sys_open_capture_socket(iface, &state->private_data);
+ if (state->capture_fd == -1) {
+ D_ERR("Failed to open capture socket on iface '%s' (%s)\n",
+ iface, strerror(errno));
+ tevent_req_error(req, EIO);
+ return tevent_req_post(req, ev);
+ }
+
+ state->fde = tevent_add_fd(ev, state, state->capture_fd,
+ TEVENT_FD_READ,
+ reset_connections_capture_tcp_handler,
+ state);
+ if (tevent_req_nomem(state->fde, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_fd_set_auto_close(state->fde);
+
+ subreq = tevent_wakeup_send(state, ev, tevent_timeval_current_ofs(0,0));
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, reset_connections_batch, req);
+
+ return req;
+}
+
/*
called when we get a read event on the raw socket
*/
-static void capture_tcp_handler(struct tevent_context *ev,
- struct tevent_fd *fde,
- uint16_t flags, void *private_data)
+static void reset_connections_capture_tcp_handler(struct tevent_context *ev,
+ struct tevent_fd *fde,
+ uint16_t flags,
+ void *private_data)
{
- struct ctdb_kill_tcp *killtcp = talloc_get_type(private_data, struct ctdb_kill_tcp);
+ struct reset_connections_state *state = talloc_get_type_abort(
+ private_data, struct reset_connections_state);
/* 0 the parts that don't get set by ctdb_sys_read_tcp_packet */
struct ctdb_connection conn;
uint32_t ack_seq, seq;
uint16_t window;
int ret;
- if (ctdb_sys_read_tcp_packet(killtcp->capture_fd,
- killtcp->private_data,
- &conn.server, &conn.client,
- &ack_seq, &seq, &rst, &window) != 0) {
- /* probably a non-tcp ACK packet */
+ ret = ctdb_sys_read_tcp_packet(state->capture_fd,
+ state->private_data,
+ &conn.server, &conn.client,
+ &ack_seq, &seq, &rst, &window);
+ if (ret != 0) {
+ /* Not a TCP-ACK? Unexpected protocol? */
+ DBG_DEBUG("Failed to parse packet, errno=%d\n", ret);
return;
}
if (window == htons(1234) && (rst || seq == 0)) {
/* Ignore packets that we sent! */
- D_DEBUG("Ignoring packet: %s, "
- "seq=%"PRIu32", ack_seq=%"PRIu32", "
- "rst=%d, window=%"PRIu16"\n",
- ctdb_connection_to_string(killtcp, &conn, false),
- seq, ack_seq, rst, ntohs(window));
+ DBG_DEBUG("Ignoring sent packet: %s, "
+ "seq=%"PRIu32", ack_seq=%"PRIu32", "
+ "rst=%d, window=%"PRIu16"\n",
+ ctdb_connection_to_string(state, &conn, false),
+ seq, ack_seq, rst, ntohs(window));
return;
}
/* Check if this connection is one being reset, if found then delete */
- ret = db_hash_delete(killtcp->connections,
+ ret = db_hash_delete(state->connections,
(uint8_t*)&conn, sizeof(conn));
if (ret == ENOENT) {
/* Packet for some other connection, ignore */
+ DBG_DEBUG("Ignoring packet for unknown connection: %s\n",
+ ctdb_connection_to_string(state, &conn, true));
return;
}
if (ret != 0) {
return;
}
- D_INFO("Sending a TCP RST to kill connection %s\n",
- ctdb_connection_to_string(killtcp, &conn, true));
-
- ctdb_sys_send_tcp(&conn.server, &conn.client, ack_seq, seq, 1);
-}
-
-
-static int tickle_connection_parser(uint8_t *keybuf, size_t keylen,
- uint8_t *databuf, size_t datalen,
- void *private_data)
-{
- struct ctdb_kill_tcp *killtcp = talloc_get_type_abort(
- private_data, struct ctdb_kill_tcp);
- struct ctdb_connection *conn;
-
- if (keylen != sizeof(*conn)) {
- DBG_WARNING("Unexpected data in connection hash\n");
- return 0;
- }
-
- conn = (struct ctdb_connection *)keybuf;
+ D_INFO("Sending a TCP RST for connection %s\n",
+ ctdb_connection_to_string(state, &conn, true));
- killtcp->batch_count++;
- if (killtcp->batch_count > killtcp->batch_size) {
- /* Terminate the traverse */
- return 1;
+ ret = ctdb_sys_send_tcp(&conn.server, &conn.client, ack_seq, seq, 1);
+ if (ret != 0) {
+ DBG_ERR("Error sending TCP RST for connection\n");
}
-
- ctdb_sys_send_tcp(&conn->server, &conn->client, 0, 0, 0);
-
- return 0;
}
/*
* Called periodically until all sentenced connections have been reset
* or enough attempts have been made
*/
-static void ctdb_tickle_sentenced_connections(struct tevent_context *ev,
- struct tevent_timer *te,
- struct timeval t, void *private_data)
+static void reset_connections_batch(struct tevent_req *subreq)
{
- struct ctdb_kill_tcp *killtcp = talloc_get_type(private_data, struct ctdb_kill_tcp);
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct reset_connections_state *state = tevent_req_data(
+ req, struct reset_connections_state);
+ bool status;
int count, ret;
+ status = tevent_wakeup_recv(subreq);
+ TALLOC_FREE(subreq);
+
+ if (! status) {
+ DBG_WARNING("Unexpected error on timer expiry\n");
+ /* Keep going... */
+ }
+
/* loop over up to batch_size connections sending tickle ACKs */
- killtcp->batch_count = 0;
- ret = db_hash_traverse(killtcp->connections,
- tickle_connection_parser, killtcp, NULL);
+ state->batch_count = 0;
+ ret = db_hash_traverse(state->connections,
+ reset_connections_tickle_connection,
+ state, NULL);
if (ret != 0) {
DBG_WARNING("Unexpected error traversing connections (%s)\n",
strerror(ret));
}
- killtcp->attempts++;
+ state->attempts++;
/*
* If there are no more connections to kill or we have tried
- * too many times we can remove the entire killtcp structure
+ * too many times we're finished
*/
- ret = db_hash_traverse(killtcp->connections, NULL, NULL, &count);
+ ret = db_hash_traverse(state->connections, NULL, NULL, &count);
if (ret != 0) {
/* What now? Try again until max_attempts reached */
DBG_WARNING("Unexpected error traversing connections (%s)\n",
count = 1;
}
if (count == 0 ||
- killtcp->attempts >= killtcp->max_attempts) {
- talloc_free(killtcp);
+ state->attempts >= state->max_attempts) {
+ tevent_req_done(req);
return;
}
- /* try tickling them again in a seconds time
- */
- tevent_add_timer(ev, killtcp,
- tevent_timeval_current_ofs(
- killtcp->retry_interval.tv_sec,
- killtcp->retry_interval.tv_usec),
- ctdb_tickle_sentenced_connections, killtcp);
+ /* Schedule next attempt */
+ subreq = tevent_wakeup_send(state, state->ev,
+ tevent_timeval_current_ofs(
+ state->retry_interval.tv_sec,
+ state->retry_interval.tv_usec));
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, reset_connections_batch, req);
}
-/* Add a TCP socket to the list of connections we want to RST. The
- * list is attached to *killtcp_arg. If this is NULL then allocate
- * the structure. */
-static int ctdb_killtcp(struct tevent_context *ev,
- TALLOC_CTX *mem_ctx,
- const char *iface,
- struct ctdb_connection *conn,
- struct ctdb_kill_tcp **killtcp_arg)
+static int reset_connections_tickle_connection(
+ uint8_t *keybuf, size_t keylen,
+ uint8_t *databuf, size_t datalen,
+ void *private_data)
{
- struct ctdb_kill_tcp *killtcp;
+ struct reset_connections_state *state = talloc_get_type_abort(
+ private_data, struct reset_connections_state);
+ struct ctdb_connection *conn;
int ret;
- if (killtcp_arg == NULL) {
- DEBUG(DEBUG_ERR, (__location__ " killtcp_arg is NULL!\n"));
- return -1;
+ if (keylen != sizeof(*conn)) {
+ DBG_WARNING("Unexpected data in connection hash\n");
+ return 0;
}
- killtcp = *killtcp_arg;
-
- /* Allocate a new structure if necessary. The structure is
- * only freed when mem_ctx is freed. */
- if (killtcp == NULL) {
- killtcp = talloc_zero(mem_ctx, struct ctdb_kill_tcp);
- if (killtcp == NULL) {
- DEBUG(DEBUG_ERR, (__location__ " out of memory\n"));
- return -1;
- }
-
- killtcp->capture_fd = -1;
- ret = db_hash_init(killtcp, "connections", 2048, DB_HASH_SIMPLE,
- &killtcp->connections);
- if (ret != 0) {
- D_ERR("Failed to initialise connection hash (%s)\n",
- strerror(ret));
- talloc_free(killtcp);
- return -1;
- }
-
- killtcp->attempts = 0;
- killtcp->max_attempts = 50;
-
- killtcp->retry_interval.tv_sec = 0;
- killtcp->retry_interval.tv_usec = 100 * 1000;
-
- killtcp->batch_count = 0;
- killtcp->batch_size = 300;
+ conn = (struct ctdb_connection *)keybuf;
- *killtcp_arg = killtcp;
+ state->batch_count++;
+ if (state->batch_count > state->batch_size) {
+ /* Terminate the traverse */
+ return 1;
}
- /* Connection is stored as a key in the connections hash */
- ret = db_hash_add(killtcp->connections,
- (uint8_t *)conn, sizeof(*conn),
- NULL, 0);
+ DBG_INFO("Sending tickle ACK for connection '%s'\n",
+ ctdb_connection_to_string(state, conn, true));
+ ret = ctdb_sys_send_tcp(&conn->server, &conn->client, 0, 0, 0);
if (ret != 0) {
- D_ERR("Error adding connection to hash (%s)\n", strerror(ret));
- return -1;
- }
-
- /*
- If we don't have a socket to listen on yet we must create it
- */
- if (killtcp->capture_fd == -1) {
- killtcp->capture_fd =
- ctdb_sys_open_capture_socket(iface,
- &killtcp->private_data);
- if (killtcp->capture_fd == -1) {
- DEBUG(DEBUG_CRIT,(__location__ " Failed to open capturing "
- "socket on iface '%s' for killtcp (%s)\n",
- iface, strerror(errno)));
- return -1;
- }
- }
-
-
- if (killtcp->fde == NULL) {
- killtcp->fde = tevent_add_fd(ev, killtcp,
- killtcp->capture_fd,
- TEVENT_FD_READ,
- capture_tcp_handler, killtcp);
- tevent_fd_set_auto_close(killtcp->fde);
+ DBG_ERR("Error sending tickle ACK\n");
+ /* continue */
}
return 0;
}
-static int ctdb_killtcp_destructor(struct ctdb_kill_tcp *killtcp)
+static bool reset_connections_recv(struct tevent_req *req, int *perr)
{
- bool *done = killtcp->destructor_data;
- *done = true;
+ int err;
- return 0;
+ if (tevent_req_is_unix_error(req, &err)) {
+ if (perr != NULL) {
+ *perr = err;
+ }
+ return false;
+ }
+
+ return true;
}
static void usage(const char *prog)
int main(int argc, char **argv)
{
struct ctdb_connection conn;
- struct ctdb_kill_tcp *killtcp = NULL;
struct tevent_context *ev = NULL;
- struct TALLOC_CONTEXT *mem_ctx = NULL;
+ TALLOC_CTX *mem_ctx = NULL;
struct ctdb_connection_list *conn_list = NULL;
const char *t;
+ struct tevent_req *req;
int debug_level;
- bool done;
- int i, ret;
+ bool status;
+ bool ok;
+ int ret;
/* Set the debug level */
t = getenv("CTDB_DEBUGLEVEL");
if (t != NULL) {
- if (debug_level_parse(t, &debug_level)) {
- DEBUGLEVEL = debug_level;
- } else {
- DEBUGLEVEL = DEBUG_ERR;
+ ok = debug_level_parse(t, &debug_level);
+ if (!ok) {
+ debug_level = DEBUG_ERR;
}
+ debuglevel_set(debug_level);
}
if (argc != 2 && argc != 4) {
goto fail;
}
} else {
- ret = ctdb_connection_list_read(mem_ctx, true, &conn_list);
+ ret = ctdb_connection_list_read(mem_ctx, 0, true, &conn_list);
if (ret != 0) {
D_ERR("Unable to parse connections (%s)\n",
strerror(ret));
goto fail;
}
- ev = tevent_context_init(mem_ctx);
+ ev = tevent_context_init(mem_ctx);
if (ev == NULL) {
DEBUG(DEBUG_ERR, ("Failed to initialise tevent\n"));
goto fail;
}
- if (conn_list->num == 0) {
- /* No connections, done! */
- talloc_free(mem_ctx);
- return 0;
- }
-
- for (i = 0; i < conn_list->num; i++) {
- ret = ctdb_killtcp(ev, mem_ctx, argv[1],
- &conn_list->conn[i], &killtcp);
- if (ret != 0) {
- DEBUG(DEBUG_ERR, ("Unable to killtcp\n"));
- goto fail;
- }
+ req = reset_connections_send(mem_ctx, ev, argv[1], conn_list);
+ if (req == NULL) {
+ goto fail;
}
- done = false;
- killtcp->destructor_data = &done;
- talloc_set_destructor(killtcp, ctdb_killtcp_destructor);
-
- /* Do the initial processing of connections */
- tevent_add_timer(ev, killtcp,
- tevent_timeval_current_ofs(0, 0),
- ctdb_tickle_sentenced_connections, killtcp);
+ tevent_req_poll(req, ev);
- while (!done) {
- tevent_loop_once(ev);
+ status = reset_connections_recv(req, &ret);
+ if (! status) {
+ D_ERR("Failed to kill connections (%s)\n", strerror(ret));
+ goto fail;
}
talloc_free(mem_ctx);
git.samba.org / vlendec / samba-autobuild / .git / blobdiff