server: rewrite message handling

[fio.git] / iolog.c

diff --git a/iolog.c b/iolog.c
index 961c9171f72f21654f17cc84f7ef07c8683b7991..650a21b0beb9143c46916a07675df1f4c6ded915 100644 (file)
--- a/iolog.c
+++ b/iolog.c
@@ -996,8 +996,13 @@ static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
 
 #ifdef CONFIG_ZLIB
 
-static void drop_data(struct iolog_flush_data *data, int refs)
+static void drop_data_unlock(struct iolog_flush_data *data)
 {
+       int refs;
+
+       refs = --data->refs;
+       pthread_mutex_unlock(&data->lock);
+
        if (!refs) {
                free(data);
                pthread_mutex_destroy(&data->lock);
@@ -1092,14 +1097,11 @@ static int gz_work(struct submit_worker *sw, struct workqueue_work *work)
        ret = 0;
 done:
        if (data->wait) {
-               int refs;
-
                pthread_mutex_lock(&data->lock);
                data->done = 1;
                pthread_cond_signal(&data->cv);
-               refs = --data->refs;
-               pthread_mutex_unlock(&data->lock);
-               drop_data(data, refs);
+
+               drop_data_unlock(data);
        } else
                free(data);
        return ret;
@@ -1113,8 +1115,24 @@ err:
        goto done;
 }
 
+static int gz_init_worker(struct submit_worker *sw)
+{
+       struct thread_data *td = sw->wq->td;
+
+       if (!fio_option_is_set(&td->o, log_gz_cpumask))
+               return 0;
+
+       if (fio_setaffinity(gettid(), td->o.log_gz_cpumask) == -1) {
+               log_err("gz: failed to set CPU affinity\n");
+               return 1;
+       }
+
+       return 0;
+}
+
 static struct workqueue_ops log_compress_wq_ops = {
-       .fn     = gz_work,
+       .fn             = gz_work,
+       .init_worker_fn = gz_init_worker,
        .nice   = 1,
 };
 
@@ -1136,15 +1154,17 @@ void iolog_compress_exit(struct thread_data *td)
 }
 
 /*
- * Queue work item to compress the existing log entries. We copy the
- * samples, and reset the log sample count to 0 (so the logging will
- * continue to use the memory associated with the log). If called with
- * wait == 1, will not return until the log compression has completed.
+ * Queue work item to compress the existing log entries. We reset the
+ * current log to a small size, and reference the existing log in the
+ * data that we queue for compression. Once compression has been done,
+ * this old log is freed. If called with wait == 1, will not return until
+ * the log compression has completed.
  */
 int iolog_flush(struct io_log *log, int wait)
 {
        struct iolog_flush_data *data;
-       size_t sample_size;
+
+       io_u_quiesce(log->td);
 
        data = malloc(sizeof(*data));
        if (!data)
@@ -1152,16 +1172,12 @@ int iolog_flush(struct io_log *log, int wait)
 
        data->log = log;
 
-       sample_size = log->nr_samples * log_entry_sz(log);
-       data->samples = malloc(sample_size);
-       if (!data->samples) {
-               free(data);
-               return 1;
-       }
-
-       memcpy(data->samples, log->log, sample_size);
+       data->samples = log->log;
        data->nr_samples = log->nr_samples;
+
        log->nr_samples = 0;
+       log->max_samples = 128;
+       log->log = malloc(log->max_samples * log_entry_sz(log));
 
        data->wait = wait;
        if (data->wait) {
@@ -1174,14 +1190,11 @@ int iolog_flush(struct io_log *log, int wait)
        workqueue_enqueue(&log->td->log_compress_wq, &data->work);
 
        if (wait) {
-               int refs;
-
                pthread_mutex_lock(&data->lock);
                while (!data->done)
                        pthread_cond_wait(&data->cv, &data->lock);
-               refs = --data->refs;
-               pthread_mutex_unlock(&data->lock);
-               drop_data(data, refs);
+
+               drop_data_unlock(data);
        }
 
        return 0;
@@ -1205,54 +1218,37 @@ void iolog_compress_exit(struct thread_data *td)
 
 #endif
 
-static int write_iops_log(struct thread_data *td, int try)
+static int __write_log(struct thread_data *td, struct io_log *log, int try)
 {
-       struct io_log *log = td->iops_log;
+       if (log)
+               return finish_log(td, log, try);
 
-       if (!log)
-               return 0;
+       return 0;
+}
 
-       return finish_log(td, log, try);
+static int write_iops_log(struct thread_data *td, int try)
+{
+       return __write_log(td, td->iops_log, try);
 }
 
 static int write_slat_log(struct thread_data *td, int try)
 {
-       struct io_log *log = td->slat_log;
-
-       if (!log)
-               return 0;
-
-       return finish_log(td, log, try);
+       return __write_log(td, td->slat_log, try);
 }
 
 static int write_clat_log(struct thread_data *td, int try)
 {
-       struct io_log *log = td->clat_log;
-
-       if (!log)
-               return 0;
-
-       return finish_log(td, log, try);
+       return __write_log(td, td->clat_log, try);
 }
 
 static int write_lat_log(struct thread_data *td, int try)
 {
-       struct io_log *log = td->lat_log;
-
-       if (!log)
-               return 0;
-
-       return finish_log(td, log, try);
+       return __write_log(td, td->lat_log, try);
 }
 
 static int write_bandw_log(struct thread_data *td, int try)
 {
-       struct io_log *log = td->bw_log;
-
-       if (!log)
-               return 0;
-
-       return finish_log(td, log, try);
+       return __write_log(td, td->bw_log, try);
 }
 
 enum {