Recent changes (master)

index c8554bc..c9875f4 100644
@@ -1358,19 +1358,29 @@ static uint64_t do_dry_run(struct thread_data *td)
 	return td->bytes_done[DDIR_WRITE] + td->bytes_done[DDIR_TRIM];
 }
 
+struct fork_data {
+	struct thread_data *td;
+	struct sk_out *sk_out;
+};
+
 /*
  * Entry point for the thread based jobs. The process based jobs end up
  * here as well, after a little setup.
  */
 static void *thread_main(void *data)
 {
+	struct fork_data *fd = data;
 	unsigned long long elapsed_us[DDIR_RWDIR_CNT] = { 0, };
-	struct thread_data *td = data;
+	struct thread_data *td = fd->td;
 	struct thread_options *o = &td->o;
+	struct sk_out *sk_out = fd->sk_out;
 	pthread_condattr_t attr;
 	int clear_state;
 	int ret;
 
+	sk_out_assign(sk_out);
+	free(fd);
+
 	if (!o->use_thread) {
 		setsid();
 		td->pid = getpid();

@@ -1550,12 +1560,12 @@ static void *thread_main(void *data)
 			goto err;
 	}
 
-	if (iolog_compress_init(td))
+	if (iolog_compress_init(td, sk_out))
 		goto err;
 
 	fio_verify_init(td);
 
-	if (rate_submit_init(td))
+	if (rate_submit_init(td, sk_out))
 		goto err;
 
 	fio_gettime(&td->epoch, NULL);

@@ -1702,6 +1712,7 @@ err:
 	 */
 	check_update_rusage(td);
 
+	sk_out_drop();
 	return (void *) (uintptr_t) td->error;
 }
 

@@ -1710,9 +1721,9 @@ err:
  * We cannot pass the td data into a forked process, so attach the td and
  * pass it to the thread worker.
  */
-static int fork_main(int shmid, int offset)
+static int fork_main(struct sk_out *sk_out, int shmid, int offset)
 {
-	struct thread_data *td;
+	struct fork_data *fd;
 	void *data, *ret;
 
 #if !defined(__hpux) && !defined(CONFIG_NO_SHM)

@@ -1730,8 +1741,10 @@ static int fork_main(int shmid, int offset)
 	data = threads;
 #endif
 
-	td = data + offset * sizeof(struct thread_data);
-	ret = thread_main(td);
+	fd = calloc(1, sizeof(*fd));
+	fd->td = data + offset * sizeof(struct thread_data);
+	fd->sk_out = sk_out;
+	ret = thread_main(fd);
 	shmdt(data);
 	return (int) (uintptr_t) ret;
 }

@@ -1956,7 +1969,7 @@ mounted:
 /*
  * Main function for kicking off and reaping jobs, as needed.
  */
-static void run_threads(void)
+static void run_threads(struct sk_out *sk_out)
 {
 	struct thread_data *td;
 	unsigned int i, todo, nr_running, m_rate, t_rate, nr_started;

@@ -2090,14 +2103,20 @@ reap:
 			nr_started++;
 
 			if (td->o.use_thread) {
+				struct fork_data *fd;
 				int ret;
 
+				fd = calloc(1, sizeof(*fd));
+				fd->td = td;
+				fd->sk_out = sk_out;
+
 				dprint(FD_PROCESS, "will pthread_create\n");
 				ret = pthread_create(&td->thread, NULL,
-							thread_main, td);
+							thread_main, fd);
 				if (ret) {
 					log_err("pthread_create: %s\n",
 							strerror(ret));
+					free(fd);
 					nr_started--;
 					break;
 				}

@@ -2110,7 +2129,7 @@ reap:
 				dprint(FD_PROCESS, "will fork\n");
 				pid = fork();
 				if (!pid) {
-					int ret = fork_main(shm_id, i);
+					int ret = fork_main(sk_out, shm_id, i);
 
 					_exit(ret);
 				} else if (i == fio_debug_jobno)

@@ -2220,11 +2239,10 @@ static void free_disk_util(void)
 
 static void *helper_thread_main(void *data)
 {
-	struct backend_data *d = data;
+	struct sk_out *sk_out = data;
 	int ret = 0;
 
-	if (d)
-		pthread_setspecific(d->key, d->ptr);
+	sk_out_assign(sk_out);
 
 	fio_mutex_up(startup_mutex);
 

@@ -2256,10 +2274,11 @@ static void *helper_thread_main(void *data)
 			print_thread_status();
 	}
 
+	sk_out_drop();
 	return NULL;
 }
 
-static int create_helper_thread(struct backend_data *data)
+static int create_helper_thread(struct sk_out *sk_out)
 {
 	int ret;
 

@@ -2268,7 +2287,7 @@ static int create_helper_thread(struct backend_data *data)
 	pthread_cond_init(&helper_cond, NULL);
 	pthread_mutex_init(&helper_lock, NULL);
 
-	ret = pthread_create(&helper_thread, NULL, helper_thread_main, data);
+	ret = pthread_create(&helper_thread, NULL, helper_thread_main, sk_out);
 	if (ret) {
 		log_err("Can't create helper thread: %s\n", strerror(ret));
 		return 1;

@@ -2280,7 +2299,7 @@ static int create_helper_thread(struct backend_data *data)
 	return 0;
 }
 
-int fio_backend(struct backend_data *data)
+int fio_backend(struct sk_out *sk_out)
 {
 	struct thread_data *td;
 	int i;

@@ -2310,12 +2329,12 @@ int fio_backend(struct backend_data *data)
 
 	set_genesis_time();
 	stat_init();
-	create_helper_thread(data);
+	create_helper_thread(sk_out);
 
 	cgroup_list = smalloc(sizeof(*cgroup_list));
 	INIT_FLIST_HEAD(cgroup_list);
 
-	run_threads();
+	run_threads(sk_out);
 
 	wait_for_helper_thread_exit();
 

187 lines not on this page · raw message

index cfb0b4d..035e606 100644
@@ -41,6 +41,8 @@ struct fio_client {
 
 	char *name;
 
+	struct flist_head *opt_lists;
+
 	int state;
 
 	int skip_newline;

@@ -60,6 +62,7 @@ struct fio_client {
 
 	struct flist_head eta_list;
 	struct client_eta *eta_in_flight;
+	unsigned int eta_timeouts;
 
 	struct flist_head cmd_list;
 

index 63778b6..ddc29db 100644
@@ -111,16 +111,16 @@ enum {
  * Per-thread/process specific data. Only used for the network client
  * for now.
  */
-struct backend_data {
-	pthread_key_t key;
-	void *ptr;
-};
+struct sk_out;
+void sk_out_assign(struct sk_out *);
+void sk_out_drop(void);
 
 /*
  * This describes a single thread/process executing a fio job.
  */
 struct thread_data {
 	struct thread_options o;
+	struct flist_head opt_list;
 	unsigned long flags;
 	void *eo;
 	char verror[FIO_VERROR_SIZE];

@@ -477,10 +477,10 @@ extern int __must_check fio_init_options(void);
 extern int __must_check parse_options(int, char **);
 extern int parse_jobs_ini(char *, int, int, int);
 extern int parse_cmd_line(int, char **, int);
-extern int fio_backend(struct backend_data *);
+extern int fio_backend(struct sk_out *);
 extern void reset_fio_state(void);
 extern void clear_io_state(struct thread_data *, int);
-extern int fio_options_parse(struct thread_data *, char **, int, int);
+extern int fio_options_parse(struct thread_data *, char **, int);
 extern void fio_keywords_init(void);
 extern void fio_keywords_exit(void);
 extern int fio_cmd_option_parse(struct thread_data *, const char *, char *);

@@ -499,6 +499,7 @@ extern int parse_dryrun(void);
 extern int fio_running_or_pending_io_threads(void);
 extern int fio_set_fd_nonblocking(int, const char *);
 extern void sig_show_status(int sig);
+extern struct thread_data *get_global_options(void);
 
 extern uintptr_t page_mask;
 extern uintptr_t page_size;

127 lines not on this page · raw message

index e674171..d4a1017 100644
@@ -1133,15 +1133,15 @@ static int gz_init_worker(struct submit_worker *sw)
 static struct workqueue_ops log_compress_wq_ops = {
 	.fn		= gz_work,
 	.init_worker_fn	= gz_init_worker,
-	.nice	= 1,
+	.nice		= 1,
 };
 
-int iolog_compress_init(struct thread_data *td)
+int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
 {
 	if (!(td->flags & TD_F_COMPRESS_LOG))
 		return 0;
 
-	workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1);
+	workqueue_init(td, &td->log_compress_wq, &log_compress_wq_ops, 1, sk_out);
 	return 0;
 }
 

@@ -1164,6 +1164,8 @@ int iolog_flush(struct io_log *log, int wait)
 {
 	struct iolog_flush_data *data;
 
+	io_u_quiesce(log->td);
+
 	data = malloc(sizeof(*data));
 	if (!data)
 		return 1;

@@ -1205,7 +1207,7 @@ int iolog_flush(struct io_log *log, int wait)
 	return 1;
 }
 
-int iolog_compress_init(struct thread_data *td)
+int iolog_compress_init(struct thread_data *td, struct sk_out *sk_out)
 {
 	return 0;
 }

index 6f027ca..b99329a 100644
@@ -184,7 +184,7 @@ extern void trim_io_piece(struct thread_data *, const struct io_u *);
 extern void queue_io_piece(struct thread_data *, struct io_piece *);
 extern void prune_io_piece_log(struct thread_data *);
 extern void write_iolog_close(struct thread_data *);
-extern int iolog_compress_init(struct thread_data *);
+extern int iolog_compress_init(struct thread_data *, struct sk_out *);
 extern void iolog_compress_exit(struct thread_data *);
 
 #ifdef CONFIG_ZLIB

index 46d5fb9..964e263 100644
@@ -4079,8 +4079,7 @@ static void show_closest_option(const char *opt)
 	free(name);
 }
 
-int fio_options_parse(struct thread_data *td, char **opts, int num_opts,
-			int dump_cmdline)
+int fio_options_parse(struct thread_data *td, char **opts, int num_opts)
 {
 	int i, ret, unknown;
 	char **opts_copy;

@@ -4091,7 +4090,7 @@ int fio_options_parse(struct thread_data *td, char **opts, int num_opts,
 	for (ret = 0, i = 0, unknown = 0; i < num_opts; i++) {
 		struct fio_option *o;
 		int newret = parse_option(opts_copy[i], opts[i], fio_options,
-						&o, td, dump_cmdline);
+						&o, td, &td->opt_list);
 
 		if (!newret && o)
 			fio_option_mark_set(&td->o, o);

@@ -4124,7 +4123,7 @@ int fio_options_parse(struct thread_data *td, char **opts, int num_opts,
 			if (td->eo)
 				newret = parse_option(opts_copy[i], opts[i],
 						      td->io_ops->options, &o,
-						      td->eo, dump_cmdline);
+						      td->eo, &td->opt_list);
 
 			ret |= newret;
 			if (!o) {

index e330dea..df42e22 100644
@@ -978,7 +978,7 @@ int parse_cmd_option(const char *opt, const char *val,
 
 int parse_option(char *opt, const char *input,
 		 struct fio_option *options, struct fio_option **o, void *data,
-		 int dump_cmdline)
+		 struct flist_head *dump_list)
 {
 	char *post;
 

@@ -1004,17 +1004,16 @@ int parse_option(char *opt, const char *input,
 		return 1;
 	}
 
-	if (dump_cmdline) {
-		const char *delim;
+	if (dump_list) {
+		struct print_option *p = malloc(sizeof(*p));
 
-		if (!strcmp("description", (*o)->name))
-			delim = "\"";
+		p->name = strdup((*o)->name);
+		if (post)
+			p->value = strdup(post);
 		else
-			delim = "";
+			p->value = NULL;
 
-		log_info("--%s%s", (*o)->name, post ? "" : " ");
-		if (post)
-			log_info("=%s%s%s ", delim, post, delim);
+		flist_add_tail(&p->list, dump_list);
 	}
 
 	return 0;

index 0a813b0..1882810 100644
@@ -80,7 +80,7 @@ struct fio_option {
 
 typedef int (str_cb_fn)(void *, char *);
 
-extern int parse_option(char *, const char *, struct fio_option *, struct fio_option **, void *, int);
+extern int parse_option(char *, const char *, struct fio_option *, struct fio_option **, void *, struct flist_head *);
 extern void sort_options(char **, struct fio_option *, int);
 extern int parse_cmd_option(const char *t, const char *l, struct fio_option *, void *);
 extern int show_cmd_help(struct fio_option *, const char *);

@@ -124,4 +124,10 @@ static inline int parse_is_percent(unsigned long long val)
 	return val <= -1ULL && val >= (-1ULL - 100ULL);
 }
 
+struct print_option {
+	struct flist_head list;
+	char *name;
+	char *value;
+};
+
 #endif

index 39b552d..92cb622 100644
@@ -7,6 +7,7 @@
 #include "fio.h"
 #include "ioengine.h"
 #include "lib/getrusage.h"
+#include "rate-submit.h"
 
 static int io_workqueue_fn(struct submit_worker *sw,
 			   struct workqueue_work *work)

@@ -235,12 +236,12 @@ static struct workqueue_ops rated_wq_ops = {
 	.exit_worker_fn		= io_workqueue_exit_worker_fn,
 };
 
-int rate_submit_init(struct thread_data *td)
+int rate_submit_init(struct thread_data *td, struct sk_out *sk_out)
 {
 	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
 		return 0;
 
-	return workqueue_init(td, &td->io_wq, &rated_wq_ops, td->o.iodepth);
+	return workqueue_init(td, &td->io_wq, &rated_wq_ops, td->o.iodepth, sk_out);
 }
 
 void rate_submit_exit(struct thread_data *td)

index b4ca129..19fde3a 100644
@@ -1,7 +1,7 @@
 #ifndef FIO_RATE_SUBMIT
 #define FIO_RATE_SUBMIT
 
-int rate_submit_init(struct thread_data *);
+int rate_submit_init(struct thread_data *, struct sk_out *);
 void rate_submit_exit(struct thread_data *);
 
 #endif

961 lines not on this page · raw message

84 lines not on this page · raw message

130 lines not on this page · raw message

index dda88fc..9c3f192 100644
@@ -247,7 +247,7 @@ extern struct jobs_eta *get_jobs_eta(bool force, size_t *size);
 extern void stat_init(void);
 extern void stat_exit(void);
 
-extern struct json_object * show_thread_status(struct thread_stat *ts, struct group_run_stats *rs, struct buf_output *);
+extern struct json_object * show_thread_status(struct thread_stat *ts, struct group_run_stats *rs, struct flist_head *, struct buf_output *);
 extern void show_group_stats(struct group_run_stats *rs, struct buf_output *);
 extern int calc_thread_status(struct jobs_eta *je, int force);
 extern void display_thread_status(struct jobs_eta *je);

index 5fd95b9..6e67f3e 100644
@@ -133,6 +133,8 @@ static void *worker_thread(void *data)
 	unsigned int eflags = 0, ret = 0;
 	FLIST_HEAD(local_list);
 
+	sk_out_assign(sw->sk_out);
+
 	if (wq->ops.nice) {
 		if (nice(wq->ops.nice) < 0) {
 			log_err("workqueue: nice %s\n", strerror(errno));

@@ -206,6 +208,7 @@ done:
 	pthread_mutex_lock(&sw->lock);
 	sw->flags |= (SW_F_EXITED | eflags);
 	pthread_mutex_unlock(&sw->lock);
+	sk_out_drop();
 	return NULL;
 }
 

@@ -267,7 +270,8 @@ void workqueue_exit(struct workqueue *wq)
 	pthread_mutex_destroy(&wq->stat_lock);
 }
 
-static int start_worker(struct workqueue *wq, unsigned int index)
+static int start_worker(struct workqueue *wq, unsigned int index,
+			struct sk_out *sk_out)
 {
 	struct submit_worker *sw = &wq->workers[index];
 	int ret;

@@ -277,6 +281,7 @@ static int start_worker(struct workqueue *wq, unsigned int index)
 	pthread_mutex_init(&sw->lock, NULL);
 	sw->wq = wq;
 	sw->index = index;
+	sw->sk_out = sk_out;
 
 	if (wq->ops.alloc_worker_fn) {
 		ret = wq->ops.alloc_worker_fn(sw);

@@ -297,12 +302,13 @@ static int start_worker(struct workqueue *wq, unsigned int index)
 }
 
 int workqueue_init(struct thread_data *td, struct workqueue *wq,
-		   struct workqueue_ops *ops, unsigned max_pending)
+		   struct workqueue_ops *ops, unsigned int max_workers,
+		   struct sk_out *sk_out)
 {
 	unsigned int running;
 	int i, error;
 
-	wq->max_workers = max_pending;
+	wq->max_workers = max_workers;
 	wq->td = td;
 	wq->ops = *ops;
 	wq->work_seq = 0;

@@ -314,7 +320,7 @@ int workqueue_init(struct thread_data *td, struct workqueue *wq,
 	wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
 
 	for (i = 0; i < wq->max_workers; i++)
-		if (start_worker(wq, i))
+		if (start_worker(wq, i, sk_out))
 			break;
 
 	wq->max_workers = i;

index 46a3979..1961b2a 100644
@@ -17,6 +17,7 @@ struct submit_worker {
 	uint64_t seq;
 	struct workqueue *wq;
 	void *private;
+	struct sk_out *sk_out;
 };
 
 typedef int (workqueue_work_fn)(struct submit_worker *, struct workqueue_work *);

@@ -60,7 +61,7 @@ struct workqueue {
 	volatile int wake_idle;
 };
 
-int workqueue_init(struct thread_data *td, struct workqueue *wq, struct workqueue_ops *ops, unsigned int max_workers);
+int workqueue_init(struct thread_data *td, struct workqueue *wq, struct workqueue_ops *ops, unsigned int max_workers, struct sk_out *sk_out);
 void workqueue_exit(struct workqueue *wq);
 
 void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work);