oysteikt/heimdal
0
0
Fork 0
Code Issues 5 Pull Requests Releases Activity

ipropd-master: use async I/O

Browse Source 
If a slave is slow to consume what the master sends it, the master can
block and all iprop operations with it.

With minimal effort we make the master async oriented.
This commit is contained in:
Nicolas Williams
2019-09-17 01:26:25 -05:00
committed by Nico Williams
parent 9de1728850
commit 0334472ab5
1 changed files with 339 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

420
lib/kadm5/ipropd_master.c
View File

@@ -122,6 +122,7 @@ make_listen_socket (krb5_context context, const char *port_str)
return fd;
}
struct slave {
krb5_socket_t fd;
struct sockaddr_in addr;
@@ -135,6 +136,29 @@ struct slave {
#define SLAVE_F_DEAD 0x1
#define SLAVE_F_AYT 0x2
#define SLAVE_F_READY 0x4
/*
* We'll use non-blocking I/O so no slave can hold us back.
*
* We call the state left over from a partial write a "tail".
*
* The krb5_data holding an KRB-PRIV will be the write buffer.
*/
struct {
/* Every message we send is a KRB-PRIV with a 4-byte length prefixed */
uint8_t header_buf[4];
krb5_data header;
krb5_data packet;
size_t packet_off;
/* For send_complete() we need an sp as part of the tail */
krb5_storage *dump;
uint32_t vno;
} tail;
struct {
uint8_t header_buf[4];
krb5_data packet;
size_t offset;
int hlen;
} input;
struct slave *next;
};
@@ -223,6 +247,11 @@ remove_slave (krb5_context context, slave *s, slave **root)
if (s->ac)
krb5_auth_con_free (context, s->ac);
/* Free any pending input/output state */
krb5_data_free(&s->input.packet);
krb5_data_free(&s->tail.packet);
krb5_storage_free(s->tail.dump);
for (p = root; *p; p = &(*p)->next)
if (*p == s) {
*p = s->next;
@@ -242,13 +271,17 @@ add_slave (krb5_context context, krb5_keytab keytab, slave **root,
krb5_ticket *ticket = NULL;
char hostname[128];
s = malloc(sizeof(*s));
s = calloc(1, sizeof(*s));
if (s == NULL) {
krb5_warnx (context, "add_slave: no memory");
return;
}
s->name = NULL;
s->ac = NULL;
s->input.packet.data = NULL;
s->tail.header.data = NULL;
s->tail.packet.data = NULL;
s->tail.dump = NULL;
addr_len = sizeof(s->addr);
s->fd = accept (fd, (struct sockaddr *)&s->addr, &addr_len);
@@ -256,6 +289,7 @@ add_slave (krb5_context context, krb5_keytab keytab, slave **root,
krb5_warn (context, rk_SOCK_ERRNO, "accept");
goto error;
}
if (master_hostname)
strlcpy(hostname, master_hostname, sizeof(hostname));
else
@@ -270,6 +304,21 @@ add_slave (krb5_context context, krb5_keytab keytab, slave **root,
ret = krb5_recvauth (context, &s->ac, &s->fd,
IPROP_VERSION, server, 0, keytab, &ticket);
/*
* We'll be doing non-blocking I/O only after authentication. We don't
* want to get stuck talking to any one slave.
*
* If we get a partial write, we'll finish writing when the socket becomes
* writable.
*
* Partial reads will be treated as EOF, causing the slave to be marked
* dead.
*
* To do non-blocking I/O for authentication we'll have to implement our
* own krb5_recvauth().
*/
socket_set_nonblocking(s->fd, 1);
krb5_free_principal (context, server);
if (ret) {
krb5_warn (context, ret, "krb5_recvauth");
@@ -455,6 +504,132 @@ write_dump (krb5_context context, krb5_storage *dump,
return ret;
}
static int
mk_priv_tail(krb5_context context, slave *s, krb5_data *data)
{
uint32_t len;
int ret;
ret = krb5_mk_priv(context, s->ac, data, &s->tail.packet, NULL);
if (ret)
return ret;
len = s->tail.packet.length;
_krb5_put_int(s->tail.header_buf, len, sizeof(s->tail.header_buf));
s->tail.header.length = sizeof(s->tail.header_buf);
s->tail.header.data = s->tail.header_buf;
return 0;
}
static int
have_tail(slave *s)
{
return s->tail.header.length || s->tail.packet.length || s->tail.dump;
}
#define SEND_COMPLETE_MAX_RECORDS 50
static int
send_tail(krb5_context context, slave *s)
{
krb5_data data;
ssize_t bytes = 0;
size_t rem = 0;
size_t n;
int ret;
if (! have_tail(s))
return 0;
/*
* For the case where we're continuing a send_complete() send up to
* SEND_COMPLETE_MAX_RECORDS records now, and the rest asynchronously
* later. This ensures that sending a complete dump to a slow-to-drain
* client does not prevent others from getting serviced.
*/
for (n = 0; n < SEND_COMPLETE_MAX_RECORDS; n++) {
if (! have_tail(s))
return 0;
if (s->tail.header.length) {
bytes = krb5_net_write(context, &s->fd,
s->tail.header.data,
s->tail.header.length);
if (bytes < 0)
goto err;
s->tail.header.length -= bytes;
s->tail.header.data = (char *)s->tail.header.data + bytes;
rem = s->tail.header.length;
if (rem)
goto ewouldblock;
}
if (s->tail.packet.length) {
bytes = krb5_net_write(context, &s->fd,
(char *)s->tail.packet.data + s->tail.packet_off,
s->tail.packet.length - s->tail.packet_off);
if (bytes < 0)
goto err;
s->tail.packet_off += bytes;
if (bytes)
slave_seen(s);
rem = s->tail.packet.length - s->tail.packet_off;
if (rem)
goto ewouldblock;
krb5_data_free(&s->tail.packet);
s->tail.packet_off = 0;
}
if (s->tail.dump == NULL)
return 0;
/*
* We're in the middle of a send_complete() that was interrupted by
* EWOULDBLOCK. Continue the sending of the dump.
*/
ret = krb5_ret_data(s->tail.dump, &data);
if (ret == HEIM_ERR_EOF) {
krb5_storage_free(s->tail.dump);
s->tail.dump = NULL;
s->version = s->tail.vno;
return 0;
}
if (ret) {
krb5_warn(context, ret, "failed to read entry from dump!");
} else {
ret = mk_priv_tail(context, s, &data);
krb5_data_free(&data);
if (ret == 0)
continue;
krb5_warn(context, ret, "failed to make and send a KRB-PRIV to %s",
s->name);
}
slave_dead(context, s);
return ret;
}
if (ret == 0 && s->tail.dump != NULL)
return EWOULDBLOCK;
err:
if (errno != EAGAIN && errno != EWOULDBLOCK) {
krb5_warn(context, ret = errno,
"error sending diffs to now-dead slave %s", s->name);
slave_dead(context, s);
return ret;
}
ewouldblock:
if (verbose)
krb5_warnx(context, "would block writing %llu bytes to slave %s",
(unsigned long long)rem, s->name);
return EWOULDBLOCK;
}
static int
send_complete (krb5_context context, slave *s, const char *database,
uint32_t current_version, uint32_t oldest_version,
@@ -463,7 +638,6 @@ send_complete (krb5_context context, slave *s, const char *database,
krb5_error_code ret;
krb5_storage *dump = NULL;
uint32_t vno = 0;
krb5_data data;
int fd = -1;
struct stat st;
char *dfn;
@@ -519,7 +693,6 @@ send_complete (krb5_context context, slave *s, const char *database,
* If the current dump has an appropriate version, then we can
* break out of the loop and send the file below.
*/
if (ret == 0 && vno != 0 && st.st_mtime > initial_log_tstamp &&
vno >= oldest_version && vno <= current_version)
break;
@@ -585,38 +758,18 @@ send_complete (krb5_context context, slave *s, const char *database,
/*
* Leaving the above loop, dump should have a ptr right after the initial
* 4 byte DB version number and we should have a shared lock on the file
* (which we may have just created), so we are reading to simply blast
* (which we may have just created), so we are reading to start sending
* the data down the wire.
*
* Note: (krb5_storage_from_fd() dup()'s the fd)
*/
for (;;) {
ret = krb5_ret_data(dump, &data);
if (ret == HEIM_ERR_EOF) {
ret = 0; /* EOF is not an error, it's success */
goto done;
}
if (ret) {
krb5_warn(context, ret, "krb5_ret_data(dump, &data)");
slave_dead(context, s);
goto done;
}
ret = krb5_write_priv_message(context, s->ac, &s->fd, &data);
krb5_data_free(&data);
if (ret) {
krb5_warn (context, ret, "krb5_write_priv_message");
slave_dead(context, s);
goto done;
}
}
s->tail.dump = dump;
s->tail.vno = vno;
dump = NULL;
ret = send_tail(context, s);
done:
if (!ret) {
s->version = vno;
slave_seen(s);
}
if (fd != -1)
close(fd);
if (dump)
@@ -635,6 +788,14 @@ send_are_you_there (krb5_context context, slave *s)
if (s->flags & (SLAVE_F_DEAD|SLAVE_F_AYT))
return 0;
/*
* Write any remainder of previous write, if we can. If we'd block we'll
* return EWOULDBLOCK.
*/
ret = send_tail(context, s);
if (ret)
return ret;
krb5_warnx(context, "slave %s missing, sending AYT", s->name);
s->flags |= SLAVE_F_AYT;
@@ -646,22 +807,20 @@ send_are_you_there (krb5_context context, slave *s)
if (sp == NULL) {
krb5_warnx (context, "are_you_there: krb5_data_alloc");
slave_dead(context, s);
return 1;
return ENOMEM;
}
ret = krb5_store_uint32(sp, ARE_YOU_THERE);
krb5_storage_free (sp);
if (ret == 0) {
ret = krb5_write_priv_message(context, s->ac, &s->fd, &data);
if (ret) {
krb5_warn(context, ret, "are_you_there: krb5_write_priv_message");
slave_dead(context, s);
return 1;
}
if (ret == 0)
ret = mk_priv_tail(context, s, &data);
if (ret == 0)
ret = send_tail(context, s);
if (ret && ret != EWOULDBLOCK) {
krb5_warn(context, ret, "are_you_there");
slave_dead(context, s);
}
return 0;
return ret;
}
static int
@@ -692,25 +851,28 @@ send_diffs (kadm5_server_context *server_context, slave *s, int log_fd,
return 0;
}
if (s->version == current_version) {
char buf[4];
/*
* Write any remainder of previous write, if we can. If we'd block we'll
* return EWOULDBLOCK.
*/
ret = send_tail(context, s);
if (ret)
return ret;
sp = krb5_storage_from_mem(buf, 4);
if (s->version == current_version) {
krb5_warnx(context, "slave %s version %ld already sent", s->name,
(long)s->version);
sp = krb5_storage_emem();
if (sp == NULL)
krb5_errx(context, IPROPD_RESTART, "krb5_storage_from_mem");
ret = krb5_store_uint32(sp, YOU_HAVE_LAST_VERSION);
if (ret == 0)
ret = krb5_storage_to_data(sp, &data);
krb5_storage_free(sp);
data.data = buf;
data.length = 4;
if (ret == 0) {
ret = krb5_write_priv_message(context, s->ac, &s->fd, &data);
if (ret) {
krb5_warn(context, ret, "send_diffs: failed to send to slave");
slave_dead(context, s);
}
krb5_warnx(context, "slave %s version %ld already sent",
s->name, (long)s->version);
}
if (ret == 0)
ret = mk_priv_tail(context, s, &data);
if (ret == 0)
ret = send_tail(context, s);
return ret;
}
@@ -858,11 +1020,14 @@ send_diffs (kadm5_server_context *server_context, slave *s, int log_fd,
krb5_store_uint32 (sp, FOR_YOU);
krb5_storage_free(sp);
ret = krb5_write_priv_message(context, s->ac, &s->fd, &data);
ret = mk_priv_tail(context, s, &data);
krb5_data_free(&data);
if (ret == 0)
ret = send_tail(context, s);
if (ret) {
krb5_warn (context, ret, "send_diffs: krb5_write_priv_message");
if (ret && ret != EWOULDBLOCK) {
krb5_warn(context, ret, "send_diffs: making or sending "
"KRB-PRIV message");
slave_dead(context, s);
return 1;
}
@@ -875,6 +1040,76 @@ send_diffs (kadm5_server_context *server_context, slave *s, int log_fd,
return 0;
}
/* Sensible bound on slave message size */
#define SLAVE_MSG_MAX 65536
static int
fill_input(krb5_context context, slave *s)
{
krb5_error_code ret;
if (s->input.hlen < 4) {
uint8_t *buf = s->input.header_buf + s->input.hlen;
size_t len = 4 - s->input.hlen;
krb5_ssize_t bytes = krb5_net_read(context, &s->fd, buf, len);
if (bytes == 0)
return HEIM_ERR_EOF;
if (bytes < 0) {
if (errno == EWOULDBLOCK || errno == EAGAIN)
return EWOULDBLOCK;
return errno ? errno : EIO;
}
s->input.hlen += bytes;
if (bytes < len)
return EWOULDBLOCK;
buf = s->input.header_buf;
len = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
if (len > SLAVE_MSG_MAX)
return EINVAL;
ret = krb5_data_alloc(&s->input.packet, len);
if (ret != 0)
return ret;
}
if (s->input.offset < s->input.packet.length) {
u_char *buf = (u_char *)s->input.packet.data + s->input.offset;
size_t len = s->input.packet.length - s->input.offset;
krb5_ssize_t bytes = krb5_net_read(context, &s->fd, buf, len);
if (bytes == 0)
return HEIM_ERR_EOF;
if (bytes < 0) {
if (errno == EWOULDBLOCK || errno == EAGAIN)
return EWOULDBLOCK;
return errno ? errno : EIO;
}
s->input.offset += bytes;
if (bytes != len)
return EWOULDBLOCK;
}
return 0;
}
static int
read_msg(krb5_context context, slave *s, krb5_data *out)
{
int ret = fill_input(context, s);
if (ret != 0)
return ret;
ret = krb5_rd_priv(context, s->ac, &s->input.packet, out, NULL);
/* Prepare for next packet */
krb5_data_free(&s->input.packet);
s->input.offset = 0;
s->input.hlen = 0;
return ret;
}
static int
process_msg (kadm5_server_context *server_context, slave *s, int log_fd,
const char *database, uint32_t current_version,
@@ -886,10 +1121,11 @@ process_msg (kadm5_server_context *server_context, slave *s, int log_fd,
krb5_storage *sp;
uint32_t tmp;
ret = krb5_read_priv_message(context, s->ac, &s->fd, &out);
if(ret) {
krb5_warn(context, ret, "error reading message from %s", s->name);
return 1;
ret = read_msg(context, s, &out);
if (ret) {
if (ret != EWOULDBLOCK)
krb5_warn(context, ret, "error reading message from %s", s->name);
return ret;
}
sp = krb5_storage_from_mem(out.data, out.length);
@@ -910,11 +1146,15 @@ process_msg (kadm5_server_context *server_context, slave *s, int log_fd,
krb5_warnx(context, "process_msg: client send too little I_HAVE data");
break;
}
/*
* XXX Make the slave send the timestamp as well, and try to get it
* here, and pass it to send_diffs().
*/
/*
* New slave whose version number we've not yet seen. If the version
* number is zero, the slave has no data, and we'll send a complete
* database. Otherwise, we'll record a non-zero initial version and
* attempt an incremental update.
* database (that happens in send_diffs()). Otherwise, we'll record a
* non-zero initial version and attempt an incremental update.
*
* NOTE!: Once the slave is "ready" (its first I_HAVE has conveyed its
* initial version), we MUST NOT update s->version to the slave's
@@ -930,6 +1170,8 @@ process_msg (kadm5_server_context *server_context, slave *s, int log_fd,
krb5_warnx(context, "Slave %s (version %u) has later version "
"than the master (version %u) OUT OF SYNC",
s->name, tmp, current_version);
/* Force send_complete() */
tmp = 0;
}
/*
* Mark the slave as ready for updates based on incoming signals.
@@ -1239,7 +1481,7 @@ main(int argc, char **argv)
while (exit_flag == 0){
slave *p;
fd_set readset;
fd_set readset, writeset;
int max_fd = 0;
struct timeval to = {30, 0};
uint32_t vers;
@@ -1252,6 +1494,7 @@ main(int argc, char **argv)
#endif
FD_ZERO(&readset);
FD_ZERO(&writeset);
FD_SET(signal_fd, &readset);
max_fd = max(max_fd, signal_fd);
FD_SET(listen_fd, &readset);
@@ -1265,11 +1508,12 @@ main(int argc, char **argv)
if (p->flags & SLAVE_F_DEAD)
continue;
FD_SET(p->fd, &readset);
if (have_tail(p))
FD_SET(p->fd, &writeset);
max_fd = max(max_fd, p->fd);
}
ret = select (max_fd + 1,
&readset, NULL, NULL, &to);
ret = select(max_fd + 1, &readset, &writeset, NULL, &to);
if (ret < 0) {
if (errno == EINTR)
continue;
@@ -1287,7 +1531,7 @@ main(int argc, char **argv)
log_fd = open(server_context->log_context.log_file, O_RDONLY, 0);
if (log_fd < 0)
krb5_err(context, 1, IPROPD_RESTART_SLOW, "open %s",
krb5_err(context, IPROPD_RESTART_SLOW, errno, "open %s",
server_context->log_context.log_file);
if (fstat(log_fd, &st) == -1)
@@ -1315,10 +1559,11 @@ main(int argc, char **argv)
flock(log_fd, LOCK_UN);
if (current_version > old_version) {
krb5_warnx(context,
"Missed a signal, updating slaves %lu to %lu",
(unsigned long)old_version,
(unsigned long)current_version);
if (verbose)
krb5_warnx(context,
"Missed a signal, updating slaves %lu to %lu",
(unsigned long)old_version,
(unsigned long)current_version);
for (p = slaves; p != NULL; p = p->next) {
if (p->flags & SLAVE_F_DEAD)
continue;
@@ -1369,10 +1614,11 @@ main(int argc, char **argv)
* breaking backwards compatibility for the protocol or
* adding new messages to it.
*/
krb5_warnx(context,
"Got a signal, updating slaves %lu to %lu",
(unsigned long)old_version,
(unsigned long)current_version);
if (verbose)
krb5_warnx(context,
"Got a signal, updating slaves %lu to %lu",
(unsigned long)old_version,
(unsigned long)current_version);
for (p = slaves; p != NULL; p = p->next) {
if (p->flags & SLAVE_F_DEAD)
continue;
@@ -1380,20 +1626,32 @@ main(int argc, char **argv)
current_version, current_tstamp);
}
} else {
krb5_warnx(context,
"Got a signal, but no update in log version %lu",
(unsigned long)current_version);
if (verbose)
krb5_warnx(context,
"Got a signal, but no update in log version %lu",
(unsigned long)current_version);
}
}
for (p = slaves; p != NULL; p = p->next) {
if (!(p->flags & SLAVE_F_DEAD) &&
FD_ISSET(p->fd, &writeset) &&
have_tail(p) &&
send_tail(context, p) == 0) {
(void) send_diffs(server_context, p, log_fd, database,
current_version, current_tstamp);
}
}
for(p = slaves; p != NULL; p = p->next) {
if (p->flags & SLAVE_F_DEAD)
continue;
if (ret && FD_ISSET(p->fd, &readset)) {
--ret;
assert(ret >= 0);
if(process_msg (server_context, p, log_fd, database,
current_version, current_tstamp))
ret = process_msg(server_context, p, log_fd, database,
current_version, current_tstamp);
if (ret && ret != EWOULDBLOCK)
slave_dead(context, p);
} else if (slave_gone_p (p))
slave_dead(context, p);