SK_F_COPY = 2,
SK_F_SIMPLE = 4,
SK_F_VEC = 8,
+ SK_F_INLINE = 16,
};
struct sk_entry {
int opcode; /* Actual command fields */
void *buf;
off_t size;
- uint64_t *tagptr;
+ uint64_t tag;
struct flist_head next; /* Other sk_entry's, if linked command */
};
* protected by below ->lock */
int sk; /* socket fd to talk to client */
- struct fio_mutex *lock; /* protects ref and below list */
+ struct fio_mutex lock; /* protects ref and below list */
struct flist_head list; /* list of pending transmit work */
- struct fio_mutex *wait; /* wake backend when items added to list */
+ struct fio_mutex wait; /* wake backend when items added to list */
+ struct fio_mutex xmit; /* held while sending data */
};
static char *fio_server_arg;
static void sk_lock(struct sk_out *sk_out)
{
- fio_mutex_down(sk_out->lock);
+ fio_mutex_down(&sk_out->lock);
}
static void sk_unlock(struct sk_out *sk_out)
{
- fio_mutex_up(sk_out->lock);
+ fio_mutex_up(&sk_out->lock);
}
void sk_out_assign(struct sk_out *sk_out)
static void sk_out_free(struct sk_out *sk_out)
{
- fio_mutex_remove(sk_out->lock);
- fio_mutex_remove(sk_out->wait);
+ __fio_mutex_remove(&sk_out->lock);
+ __fio_mutex_remove(&sk_out->wait);
+ __fio_mutex_remove(&sk_out->xmit);
sfree(sk_out);
}
int refs;
sk_lock(sk_out);
+ assert(sk_out->refs != 0);
refs = --sk_out->refs;
sk_unlock(sk_out);
if (!refs) {
sk_out_free(sk_out);
+ pthread_setspecific(sk_out_key, NULL);
return 0;
}
}
struct sk_out *sk_out;
sk_out = pthread_getspecific(sk_out_key);
- if (!__sk_out_drop(sk_out))
- pthread_setspecific(sk_out_key, NULL);
+ __sk_out_drop(sk_out);
}
static void __fio_init_net_cmd(struct fio_net_cmd *cmd, uint16_t opcode,
return fio_sendv_data(sk, &iov, 1);
}
-static int fio_recv_data(int sk, void *p, unsigned int len)
+static int fio_recv_data(int sk, void *p, unsigned int len, bool wait)
{
+ int flags;
+
+ if (wait)
+ flags = MSG_WAITALL;
+ else
+ flags = OS_MSG_DONTWAIT;
+
do {
- int ret = recv(sk, p, len, MSG_WAITALL);
+ int ret = recv(sk, p, len, flags);
if (ret > 0) {
len -= ret;
continue;
} else if (!ret)
break;
- else if (errno == EAGAIN || errno == EINTR)
- continue;
- else
+ else if (errno == EAGAIN || errno == EINTR) {
+ if (wait)
+ continue;
+ break;
+ } else
break;
} while (!exit_backend);
/*
* Read (and defragment, if necessary) incoming commands
*/
-struct fio_net_cmd *fio_net_recv_cmd(int sk)
+struct fio_net_cmd *fio_net_recv_cmd(int sk, bool wait)
{
struct fio_net_cmd cmd, *tmp, *cmdret = NULL;
size_t cmd_size = 0, pdu_offset = 0;
void *pdu = NULL;
do {
- ret = fio_recv_data(sk, &cmd, sizeof(cmd));
+ ret = fio_recv_data(sk, &cmd, sizeof(cmd), wait);
if (ret)
break;
/* There's payload, get it */
pdu = (void *) cmdret->payload + pdu_offset;
- ret = fio_recv_data(sk, pdu, cmd.pdu_len);
+ ret = fio_recv_data(sk, pdu, cmd.pdu_len, wait);
if (ret)
break;
return ret;
}
-static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf, off_t size,
- uint64_t *tagptr, int flags)
+static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf,
+ size_t size, uint64_t *tagptr,
+ int flags)
{
struct sk_entry *entry;
memcpy(entry->buf, buf, size);
} else
entry->buf = buf;
+
entry->size = size;
- entry->tagptr = tagptr;
+ if (tagptr)
+ entry->tag = *tagptr;
+ else
+ entry->tag = 0;
entry->flags = flags;
-
return entry;
}
+static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry);
+
static void fio_net_queue_entry(struct sk_entry *entry)
{
struct sk_out *sk_out = pthread_getspecific(sk_out_key);
- sk_lock(sk_out);
- flist_add_tail(&entry->list, &sk_out->list);
- sk_unlock(sk_out);
+ if (entry->flags & SK_F_INLINE)
+ handle_sk_entry(sk_out, entry);
+ else {
+ sk_lock(sk_out);
+ flist_add_tail(&entry->list, &sk_out->list);
+ sk_unlock(sk_out);
- fio_mutex_up(sk_out->wait);
+ fio_mutex_up(&sk_out->wait);
+ }
}
static int fio_net_queue_cmd(uint16_t opcode, void *buf, off_t size,
pid_t pid;
int ret;
+ sk_out_assign(sk_out);
+
fio_time_init();
set_genesis_time();
ret = fio_backend(sk_out);
free_threads_shm();
+ sk_out_drop();
_exit(ret);
}
struct all_io_list state;
state.threads = cpu_to_le64((uint64_t) 0);
- fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, &state, sizeof(state), NULL, SK_F_COPY);
+ fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, &state, sizeof(state), NULL, SK_F_COPY | SK_F_INLINE);
} else
- fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE);
+ fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE | SK_F_INLINE);
exec_trigger(buf);
return 0;
{
struct fio_net_cmd cmd;
struct iovec iov[2];
+ size_t this_len;
+ int ret;
iov[0].iov_base = (void *) &cmd;
iov[0].iov_len = sizeof(cmd);
- iov[1].iov_base = (void *) buf;
- iov[1].iov_len = size;
- __fio_init_net_cmd(&cmd, opcode, size, tag);
- cmd.flags = __cpu_to_le32(flags);
- fio_net_cmd_crc_pdu(&cmd, buf);
+ do {
+ uint32_t this_flags = flags;
- return fio_sendv_data(sk, iov, 2);
+ this_len = size;
+ if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
+ this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
+
+ if (this_len < size)
+ this_flags |= FIO_NET_CMD_F_MORE;
+
+ __fio_init_net_cmd(&cmd, opcode, this_len, tag);
+ cmd.flags = __cpu_to_le32(this_flags);
+ fio_net_cmd_crc_pdu(&cmd, buf);
+
+ iov[1].iov_base = (void *) buf;
+ iov[1].iov_len = this_len;
+
+ ret = fio_sendv_data(sk, iov, 2);
+ size -= this_len;
+ buf += this_len;
+ } while (!ret && size);
+
+ return ret;
}
static void finish_entry(struct sk_entry *entry)
sfree(entry);
}
-static void entry_set_flags_tag(struct sk_entry *entry, struct flist_head *list,
- unsigned int *flags, uint64_t *tag)
+static void entry_set_flags(struct sk_entry *entry, struct flist_head *list,
+ unsigned int *flags)
{
if (!flist_empty(list))
*flags = FIO_NET_CMD_F_MORE;
else
*flags = 0;
-
- if (entry->tagptr)
- *tag = *entry->tagptr;
- else
- *tag = 0;
}
static int send_vec_entry(struct sk_out *sk_out, struct sk_entry *first)
{
unsigned int flags;
- uint64_t tag;
int ret;
- entry_set_flags_tag(first, &first->next, &flags, &tag);
+ entry_set_flags(first, &first->next, &flags);
- ret = fio_send_cmd_ext_pdu(sk_out->sk, first->opcode, first->buf, first->size, tag, flags);
+ ret = fio_send_cmd_ext_pdu(sk_out->sk, first->opcode, first->buf,
+ first->size, first->tag, flags);
while (!flist_empty(&first->next)) {
struct sk_entry *next;
next = flist_first_entry(&first->next, struct sk_entry, list);
flist_del_init(&next->list);
- entry_set_flags_tag(next, &first->next, &flags, &tag);
+ entry_set_flags(next, &first->next, &flags);
- ret += fio_send_cmd_ext_pdu(sk_out->sk, next->opcode, next->buf, next->size, tag, flags);
+ ret += fio_send_cmd_ext_pdu(sk_out->sk, next->opcode, next->buf,
+ next->size, next->tag, flags);
finish_entry(next);
}
{
int ret;
+ fio_mutex_down(&sk_out->xmit);
+
if (entry->flags & SK_F_VEC)
ret = send_vec_entry(sk_out, entry);
- if (entry->flags & SK_F_SIMPLE) {
- uint64_t tag = 0;
-
- if (entry->tagptr)
- tag = *entry->tagptr;
+ else if (entry->flags & SK_F_SIMPLE) {
+ ret = fio_net_send_simple_cmd(sk_out->sk, entry->opcode,
+ entry->tag, NULL);
+ } else {
+ ret = fio_net_send_cmd(sk_out->sk, entry->opcode, entry->buf,
+ entry->size, &entry->tag, NULL);
+ }
- ret = fio_net_send_simple_cmd(sk_out->sk, entry->opcode, tag, NULL);
- } else
- ret = fio_net_send_cmd(sk_out->sk, entry->opcode, entry->buf, entry->size, entry->tagptr, NULL);
+ fio_mutex_up(&sk_out->xmit);
if (ret)
log_err("fio: failed handling cmd %s\n", fio_server_op(entry->opcode));
break;
} else if (!ret) {
fio_server_check_jobs(&job_list);
- fio_mutex_down_timeout(sk_out->wait, timeout);
+ fio_mutex_down_timeout(&sk_out->wait, timeout);
continue;
}
if (ret < 0)
break;
- cmd = fio_net_recv_cmd(sk_out->sk);
+ cmd = fio_net_recv_cmd(sk_out->sk, true);
if (!cmd) {
ret = -1;
break;
sk_out = smalloc(sizeof(*sk_out));
sk_out->sk = sk;
INIT_FLIST_HEAD(&sk_out->list);
- sk_out->lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
- sk_out->wait = fio_mutex_init(FIO_MUTEX_LOCKED);
+ __fio_mutex_init(&sk_out->lock, FIO_MUTEX_UNLOCKED);
+ __fio_mutex_init(&sk_out->wait, FIO_MUTEX_LOCKED);
+ __fio_mutex_init(&sk_out->xmit, FIO_MUTEX_UNLOCKED);
pid = fork();
if (pid) {
flist_for_each(entry, opt_list) {
struct print_option *p;
+ size_t len;
p = flist_entry(entry, struct print_option, list);
memset(&pdu, 0, sizeof(pdu));
+
if (groupid == -1U) {
pdu.global = __cpu_to_le16(1);
pdu.groupid = 0;
} else {
pdu.global = 0;
- pdu.groupid = __cpu_to_le16(groupid);
+ pdu.groupid = cpu_to_le32(groupid);
+ }
+ len = strlen(p->name);
+ if (len >= sizeof(pdu.name)) {
+ len = sizeof(pdu.name) - 1;
+ pdu.truncated = __cpu_to_le16(1);
+ }
+ memcpy(pdu.name, p->name, len);
+ if (p->value) {
+ len = strlen(p->value);
+ if (len >= sizeof(pdu.value)) {
+ len = sizeof(pdu.value) - 1;
+ pdu.truncated = __cpu_to_le16(1);
+ }
+ memcpy(pdu.value, p->value, len);
}
- memcpy(pdu.name, p->name, strlen(p->name));
- if (p->value)
- memcpy(pdu.value, p->value, strlen(p->value));
fio_net_queue_cmd(FIO_NET_CMD_JOB_OPT, &pdu, sizeof(pdu), NULL, SK_F_COPY);
}
}
}
}
-static int fio_send_iolog_gz(struct sk_entry *first, struct io_log *log)
+static int fio_append_iolog_gz(struct sk_entry *first, struct io_log *log)
{
int ret = 0;
#ifdef CONFIG_ZLIB
this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream.avail_out;
entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, out_pdu, this_len,
- NULL, SK_F_FREE | SK_F_VEC);
+ NULL, SK_F_VEC | SK_F_INLINE | SK_F_FREE);
+ out_pdu = NULL;
flist_add_tail(&entry->list, &first->next);
} while (stream.avail_in);
return ret;
}
+static int fio_append_gz_chunks(struct sk_entry *first, struct io_log *log)
+{
+ struct sk_entry *entry;
+ struct flist_head *node;
+
+ pthread_mutex_lock(&log->chunk_lock);
+ flist_for_each(node, &log->chunk_list) {
+ struct iolog_compress *c;
+
+ c = flist_entry(node, struct iolog_compress, list);
+ entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, c->buf, c->len,
+ NULL, SK_F_VEC | SK_F_INLINE);
+ flist_add_tail(&entry->list, &first->next);
+ }
+ pthread_mutex_unlock(&log->chunk_lock);
+
+ return 0;
+}
+
+static int fio_append_text_log(struct sk_entry *first, struct io_log *log)
+{
+ struct sk_entry *entry;
+ size_t size = log->nr_samples * log_entry_sz(log);
+
+ entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, log->log, size,
+ NULL, SK_F_VEC | SK_F_INLINE);
+ flist_add_tail(&entry->list, &first->next);
+ return 0;
+}
+
int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
{
struct cmd_iolog_pdu pdu;
pdu.nr_samples = cpu_to_le64(log->nr_samples);
pdu.thread_number = cpu_to_le32(td->thread_number);
pdu.log_type = cpu_to_le32(log->log_type);
- pdu.compressed = cpu_to_le32(use_zlib);
+
+ if (!flist_empty(&log->chunk_list))
+ pdu.compressed = __cpu_to_le32(STORE_COMPRESSED);
+ else if (use_zlib)
+ pdu.compressed = __cpu_to_le32(XMIT_COMPRESSED);
+ else
+ pdu.compressed = 0;
strncpy((char *) pdu.name, name, FIO_NET_NAME_MAX);
pdu.name[FIO_NET_NAME_MAX - 1] = '\0';
+ /*
+ * We can't do this for a pre-compressed log, but for that case,
+ * log->nr_samples is zero anyway.
+ */
for (i = 0; i < log->nr_samples; i++) {
struct io_sample *s = get_sample(log, i);
/*
* Assemble header entry first
*/
- first = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, &pdu, sizeof(pdu), NULL, SK_F_COPY | SK_F_VEC);
+ first = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, &pdu, sizeof(pdu), NULL, SK_F_VEC | SK_F_INLINE | SK_F_COPY);
/*
- * Now append actual log entries. Compress if we can, otherwise just
- * plain text output.
+ * Now append actual log entries. If log compression was enabled on
+ * the job, just send out the compressed chunks directly. If we
+ * have a plain log, compress if we can, then send. Otherwise, send
+ * the plain text output.
*/
- if (use_zlib)
- ret = fio_send_iolog_gz(first, log);
- else {
- struct sk_entry *entry;
-
- entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, log->log,
- log->nr_samples * log_entry_sz(log),
- NULL, SK_F_FREE | SK_F_VEC);
- flist_add_tail(&entry->list, &first->next);
- }
+ if (!flist_empty(&log->chunk_list))
+ ret = fio_append_gz_chunks(first, log);
+ else if (use_zlib)
+ ret = fio_append_iolog_gz(first, log);
+ else
+ ret = fio_append_text_log(first, log);
+ fio_net_queue_entry(first);
return ret;
}
pdu.groupid = cpu_to_le32(td->groupid);
convert_thread_options_to_net(&pdu.top, &td->o);
- fio_net_queue_cmd(FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL, SK_F_COPY);
+ fio_net_queue_cmd(FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL,
+ SK_F_COPY);
}
void fio_server_send_start(struct thread_data *td)
struct cmd_reply *rep;
uint64_t tag;
void *data;
+ int ret;
dprint(FD_NET, "server: request verify state\n");
rep = smalloc(sizeof(*rep));
if (!rep) {
log_err("fio: smalloc pool too small\n");
- return 1;
+ return ENOMEM;
}
__fio_mutex_init(&rep->lock, FIO_MUTEX_LOCKED);
verify_state_gen_name((char *) out.path, sizeof(out.path), name, me,
threadnumber);
tag = (uint64_t) (uintptr_t) rep;
- fio_net_queue_cmd(FIO_NET_CMD_SENDFILE, &out, sizeof(out), &tag, SK_F_COPY);
+ fio_net_queue_cmd(FIO_NET_CMD_SENDFILE, &out, sizeof(out), &tag,
+ SK_F_COPY);
/*
* Wait for the backend to receive the reply
*/
if (fio_mutex_down_timeout(&rep->lock, 10000)) {
log_err("fio: timed out waiting for reply\n");
+ ret = ETIMEDOUT;
goto fail;
}
if (rep->error) {
- log_err("fio: failure on receiving state file: %s\n", strerror(rep->error));
+ log_err("fio: failure on receiving state file %s: %s\n",
+ out.path, strerror(rep->error));
+ ret = rep->error;
fail:
*datap = NULL;
sfree(rep);
fio_net_queue_quit();
- return 1;
+ return ret;
}
/*
* the header, and the thread_io_list checksum
*/
s = rep->data + sizeof(struct verify_state_hdr);
- if (verify_state_hdr(rep->data, s, version))
+ if (verify_state_hdr(rep->data, s, version)) {
+ ret = EILSEQ;
goto fail;
+ }
/*
* Don't need the header from now, copy just the thread_io_list
*/
+ ret = 0;
rep->size -= sizeof(struct verify_state_hdr);
data = malloc(rep->size);
memcpy(data, s, rep->size);
sfree(rep->data);
__fio_mutex_remove(&rep->lock);
sfree(rep);
- return 0;
+ return ret;
}
static int fio_init_server_ip(void)
return -1;
}
#ifdef SO_REUSEPORT
- if (setsockopt(sk, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
- log_err("fio: setsockopt(REUSEPORT): %s\n", strerror(errno));
- close(sk);
- return -1;
- }
+ /*
+ * Not fatal if fails, so just ignore it if that happens
+ */
+ setsockopt(sk, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
#endif
if (use_ipv6) {
sigaction(SIGINT, &act, NULL);
}
-static int fio_server(void)
+void fio_server_destroy_sk_key(void)
{
- int sk, ret;
+ pthread_key_delete(sk_out_key);
+}
+int fio_server_create_sk_key(void)
+{
if (pthread_key_create(&sk_out_key, NULL)) {
log_err("fio: can't create sk_out backend key\n");
- return -1;
+ return 1;
}
pthread_setspecific(sk_out_key, NULL);
+ return 0;
+}
+
+static int fio_server(void)
+{
+ int sk, ret;
dprint(FD_NET, "starting server\n");
projects / fio.git / blobdiff