server: sk_out exit error handling
[fio.git] / server.c
index ae8643bef4f20680f7dee3cbce5f49e80d28fbaf..eccd9d37e80f9b48026c9d5435111c84c55e15ca 100644 (file)
--- a/server.c
+++ b/server.c
@@ -46,7 +46,7 @@ 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 */
 };
 
@@ -151,11 +151,13 @@ static int __sk_out_drop(struct sk_out *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;
                }
        }
@@ -168,8 +170,7 @@ void sk_out_drop(void)
        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,
@@ -536,7 +537,10 @@ static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf,
                entry->buf = buf;
 
        entry->size = size;
-       entry->tagptr = tagptr;
+       if (tagptr)
+               entry->tag = *tagptr;
+       else
+               entry->tag = 0;
        entry->flags = flags;
        return entry;
 }
@@ -754,6 +758,8 @@ static int handle_run_cmd(struct sk_out *sk_out, struct flist_head *job_list,
        pid_t pid;
        int ret;
 
+       sk_out_assign(sk_out);
+
        fio_time_init();
        set_genesis_time();
 
@@ -765,6 +771,7 @@ static int handle_run_cmd(struct sk_out *sk_out, struct flist_head *job_list,
 
        ret = fio_backend(sk_out);
        free_threads_shm();
+       sk_out_drop();
        _exit(ret);
 }
 
@@ -1069,29 +1076,24 @@ 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;
@@ -1099,9 +1101,10 @@ static int send_vec_entry(struct sk_out *sk_out, struct sk_entry *first)
                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);
        }
 
@@ -1117,14 +1120,12 @@ static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry)
        if (entry->flags & SK_F_VEC)
                ret = send_vec_entry(sk_out, entry);
        else if (entry->flags & SK_F_SIMPLE) {
-               uint64_t tag = 0;
-
-               if (entry->tagptr)
-                       tag = *entry->tagptr;
-
-               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);
+               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);
+       }
 
        fio_mutex_up(&sk_out->xmit);
 
@@ -1624,7 +1625,7 @@ void fio_server_send_du(void)
        }
 }
 
-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
@@ -1677,7 +1678,7 @@ err:
        return ret;
 }
 
-static int fio_send_gz_chunks(struct sk_entry *first, struct io_log *log)
+static int fio_append_gz_chunks(struct sk_entry *first, struct io_log *log)
 {
        struct sk_entry *entry;
        struct flist_head *node;
@@ -1696,15 +1697,23 @@ static int fio_send_gz_chunks(struct sk_entry *first, struct io_log *log)
        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;
        struct sk_entry *first;
        int i, ret = 0;
 
-       if (!flist_empty(&log->chunk_list))
-               printf("log has chunks\n");
-
        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);
@@ -1744,21 +1753,17 @@ int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
        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 (!flist_empty(&log->chunk_list))
-               ret = fio_send_gz_chunks(first, log);
+               ret = fio_append_gz_chunks(first, log);
        else if (use_zlib)
-               ret = fio_send_iolog_gz(first, log);
-       else {
-               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);
-       }
+               ret = fio_append_iolog_gz(first, log);
+       else
+               ret = fio_append_text_log(first, log);
 
        fio_net_queue_entry(first);
        return ret;
@@ -1773,7 +1778,8 @@ void fio_server_send_add_job(struct thread_data *td)
        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)
@@ -1793,13 +1799,14 @@ int fio_server_get_verify_state(const char *name, int threadnumber,
        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);
@@ -1809,23 +1816,27 @@ int fio_server_get_verify_state(const char *name, int threadnumber,
        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;
        }
 
        /*
@@ -1833,12 +1844,15 @@ fail:
         * 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);
@@ -1847,7 +1861,7 @@ fail:
        sfree(rep->data);
        __fio_mutex_remove(&rep->lock);
        sfree(rep);
-       return 0;
+       return ret;
 }
 
 static int fio_init_server_ip(void)