/*

 * Code related to writing an iolog of what a thread is doing, and to

 * later read that back and replay

 */

#include <stdio.h>

#include <stdlib.h>

#include <assert.h>

#include <sys/types.h>

#include <sys/stat.h>

#include <unistd.h>

#ifdef CONFIG_ZLIB

#include <zlib.h>

#endif

#include "flist.h"

#include "fio.h"

#include "trim.h"

#include "filelock.h"

#include "smalloc.h"

#include "blktrace.h"

#include "pshared.h"

#include <netinet/in.h>

#include <netinet/tcp.h>

#include <arpa/inet.h>

#include <sys/stat.h>

#include <sys/socket.h>

#include <sys/un.h>

static int iolog_flush(struct io_log *log);

static const char iolog_ver2[] = "fio version 2 iolog";

void queue_io_piece(struct thread_data *td, struct io_piece *ipo)

{

	flist_add_tail(&ipo->list, &td->io_log_list);

	td->total_io_size += ipo->len;

}

void log_io_u(const struct thread_data *td, const struct io_u *io_u)

{

	if (!td->o.write_iolog_file)

		return;

	fprintf(td->iolog_f, "%s %s %llu %llu\n", io_u->file->file_name,

						io_ddir_name(io_u->ddir),

						io_u->offset, io_u->buflen);

}

void log_file(struct thread_data *td, struct fio_file *f,

	      enum file_log_act what)

{

	const char *act[] = { "add", "open", "close" };

	assert(what < 3);

	if (!td->o.write_iolog_file)

		return;

	/*

	 * this happens on the pre-open/close done before the job starts

	 */

	if (!td->iolog_f)

		return;

	fprintf(td->iolog_f, "%s %s\n", f->file_name, act[what]);

}

static void iolog_delay(struct thread_data *td, unsigned long delay)

{

	uint64_t usec = utime_since_now(&td->last_issue);

	unsigned long orig_delay = delay;

	uint64_t this_delay;

	struct timespec ts;

	if (delay < td->time_offset) {

		td->time_offset = 0;

		return;

	}

	delay -= td->time_offset;

	if (delay < usec)

		return;

	delay -= usec;

	fio_gettime(&ts, NULL);

	while (delay && !td->terminate) {

		this_delay = delay;

		if (this_delay > 500000)

			this_delay = 500000;

		usec_sleep(td, this_delay);

		delay -= this_delay;

	}

	usec = utime_since_now(&ts);

	if (usec > orig_delay)

		td->time_offset = usec - orig_delay;

	else

		td->time_offset = 0;

}

static int ipo_special(struct thread_data *td, struct io_piece *ipo)

{

	struct fio_file *f;

	int ret;

	/*

	 * Not a special ipo

	 */

	if (ipo->ddir != DDIR_INVAL)

		return 0;

	f = td->files[ipo->fileno];

	switch (ipo->file_action) {

	case FIO_LOG_OPEN_FILE:

		if (td->o.replay_redirect && fio_file_open(f)) {

			dprint(FD_FILE, "iolog: ignoring re-open of file %s\n",

					f->file_name);

			break;

		}

		ret = td_io_open_file(td, f);

		if (!ret)

			break;

		td_verror(td, ret, "iolog open file");

		return -1;

	case FIO_LOG_CLOSE_FILE:

		td_io_close_file(td, f);

		break;

	case FIO_LOG_UNLINK_FILE:

		td_io_unlink_file(td, f);

		break;

	default:

		log_err("fio: bad file action %d\n", ipo->file_action);

		break;

	}

	return 1;

}

static bool read_iolog2(struct thread_data *td);

int read_iolog_get(struct thread_data *td, struct io_u *io_u)

{

	struct io_piece *ipo;

	unsigned long elapsed;

	while (!flist_empty(&td->io_log_list)) {

		int ret;

		if (td->o.read_iolog_chunked) {

			if (td->io_log_checkmark == td->io_log_current) {

				if (!read_iolog2(td))

					return 1;

			}

			td->io_log_current--;

		}

		ipo = flist_first_entry(&td->io_log_list, struct io_piece, list);

		flist_del(&ipo->list);

		remove_trim_entry(td, ipo);

		ret = ipo_special(td, ipo);

		if (ret < 0) {

			free(ipo);

			break;

		} else if (ret > 0) {

			free(ipo);

			continue;

		}

		io_u->ddir = ipo->ddir;

		if (ipo->ddir != DDIR_WAIT) {

			io_u->offset = ipo->offset;

			io_u->buflen = ipo->len;

			io_u->file = td->files[ipo->fileno];

			get_file(io_u->file);

			dprint(FD_IO, "iolog: get %llu/%llu/%s\n", io_u->offset,

						io_u->buflen, io_u->file->file_name);

			if (ipo->delay)

				iolog_delay(td, ipo->delay);

		} else {

			elapsed = mtime_since_genesis();

			if (ipo->delay > elapsed)

				usec_sleep(td, (ipo->delay - elapsed) * 1000);

		}

		free(ipo);

		if (io_u->ddir != DDIR_WAIT)

			return 0;

	}

	td->done = 1;

	return 1;

}

void prune_io_piece_log(struct thread_data *td)

{

	struct io_piece *ipo;

	struct fio_rb_node *n;

	while ((n = rb_first(&td->io_hist_tree)) != NULL) {

		ipo = rb_entry(n, struct io_piece, rb_node);

		rb_erase(n, &td->io_hist_tree);

		remove_trim_entry(td, ipo);

		td->io_hist_len--;

		free(ipo);

	}

	while (!flist_empty(&td->io_hist_list)) {

		ipo = flist_first_entry(&td->io_hist_list, struct io_piece, list);

		flist_del(&ipo->list);

		remove_trim_entry(td, ipo);

		td->io_hist_len--;

		free(ipo);

	}

}

/*

 * log a successful write, so we can unwind the log for verify

 */

void log_io_piece(struct thread_data *td, struct io_u *io_u)

{

	struct fio_rb_node **p, *parent;

	struct io_piece *ipo, *__ipo;

	ipo = calloc(1, sizeof(struct io_piece));

	init_ipo(ipo);

	ipo->file = io_u->file;

	ipo->offset = io_u->offset;

	ipo->len = io_u->buflen;

	ipo->numberio = io_u->numberio;

	ipo->flags = IP_F_IN_FLIGHT;

	io_u->ipo = ipo;

	if (io_u_should_trim(td, io_u)) {

		flist_add_tail(&ipo->trim_list, &td->trim_list);

		td->trim_entries++;

	}

	/*

	 * Only sort writes if we don't have a random map in which case we need

	 * to check for duplicate blocks and drop the old one, which we rely on

	 * the rb insert/lookup for handling.

	 */

	if (file_randommap(td, ipo->file)) {

		INIT_FLIST_HEAD(&ipo->list);

		flist_add_tail(&ipo->list, &td->io_hist_list);

		ipo->flags |= IP_F_ONLIST;

		td->io_hist_len++;

		return;

	}

	RB_CLEAR_NODE(&ipo->rb_node);

	/*

	 * Sort the entry into the verification list

	 */

restart:

	p = &td->io_hist_tree.rb_node;

	parent = NULL;

	while (*p) {

		int overlap = 0;

		parent = *p;

		__ipo = rb_entry(parent, struct io_piece, rb_node);

		if (ipo->file < __ipo->file)

			p = &(*p)->rb_left;

		else if (ipo->file > __ipo->file)

			p = &(*p)->rb_right;

		else if (ipo->offset < __ipo->offset) {

			p = &(*p)->rb_left;

			overlap = ipo->offset + ipo->len > __ipo->offset;

		}

		else if (ipo->offset > __ipo->offset) {

			p = &(*p)->rb_right;

			overlap = __ipo->offset + __ipo->len > ipo->offset;

		}

		else

			overlap = 1;

		if (overlap) {

			dprint(FD_IO, "iolog: overlap %llu/%lu, %llu/%lu\n",

				__ipo->offset, __ipo->len,

				ipo->offset, ipo->len);

			td->io_hist_len--;

			rb_erase(parent, &td->io_hist_tree);

			remove_trim_entry(td, __ipo);

			if (!(__ipo->flags & IP_F_IN_FLIGHT))

				free(__ipo);

			goto restart;

		}

	}

	rb_link_node(&ipo->rb_node, parent, p);

	rb_insert_color(&ipo->rb_node, &td->io_hist_tree);

	ipo->flags |= IP_F_ONRB;

	td->io_hist_len++;

}

void unlog_io_piece(struct thread_data *td, struct io_u *io_u)

{

	struct io_piece *ipo = io_u->ipo;

	if (td->ts.nr_block_infos) {

		uint32_t *info = io_u_block_info(td, io_u);

		if (BLOCK_INFO_STATE(*info) < BLOCK_STATE_TRIM_FAILURE) {

			if (io_u->ddir == DDIR_TRIM)

				*info = BLOCK_INFO_SET_STATE(*info,

						BLOCK_STATE_TRIM_FAILURE);

			else if (io_u->ddir == DDIR_WRITE)

				*info = BLOCK_INFO_SET_STATE(*info,

						BLOCK_STATE_WRITE_FAILURE);

		}

	}

	if (!ipo)

		return;

	if (ipo->flags & IP_F_ONRB)

		rb_erase(&ipo->rb_node, &td->io_hist_tree);

	else if (ipo->flags & IP_F_ONLIST)

		flist_del(&ipo->list);

	free(ipo);

	io_u->ipo = NULL;

	td->io_hist_len--;

}

void trim_io_piece(const struct io_u *io_u)

{

	struct io_piece *ipo = io_u->ipo;

	if (!ipo)

		return;

	ipo->len = io_u->xfer_buflen - io_u->resid;

}

void write_iolog_close(struct thread_data *td)

{

	fflush(td->iolog_f);

	fclose(td->iolog_f);

	free(td->iolog_buf);

	td->iolog_f = NULL;

	td->iolog_buf = NULL;

}

static int64_t iolog_items_to_fetch(struct thread_data *td)

{

	struct timespec now;

	uint64_t elapsed;

	uint64_t for_1s;

	int64_t items_to_fetch;

	if (!td->io_log_highmark)

		return 10;

	fio_gettime(&now, NULL);

	elapsed = ntime_since(&td->io_log_highmark_time, &now;;

	if (elapsed) {

		for_1s = (td->io_log_highmark - td->io_log_current) * 1000000000 / elapsed;

		items_to_fetch = for_1s - td->io_log_current;

		if (items_to_fetch < 0)

			items_to_fetch = 0;

	} else

		items_to_fetch = 0;

	td->io_log_highmark = td->io_log_current + items_to_fetch;

	td->io_log_checkmark = (td->io_log_highmark + 1) / 2;

	fio_gettime(&td->io_log_highmark_time, NULL);

	return items_to_fetch;

}

/*

 * Read version 2 iolog data. It is enhanced to include per-file logging,

 * syncs, etc.

 */

static bool read_iolog2(struct thread_data *td)

{

	unsigned long long offset;

	unsigned int bytes;

	int reads, writes, waits, fileno = 0, file_action = 0; /* stupid gcc */

	char *rfname, *fname, *act;

	char *str, *p;

	enum fio_ddir rw;

	bool realloc = false;

	int64_t items_to_fetch = 0;

	if (td->o.read_iolog_chunked) {

		items_to_fetch = iolog_items_to_fetch(td);

		if (!items_to_fetch)

			return true;

	}

	/*

	 * Read in the read iolog and store it, reuse the infrastructure

	 * for doing verifications.

	 */

	str = malloc(4096);

	rfname = fname = malloc(256+16);

	act = malloc(256+16);

	reads = writes = waits = 0;

	while ((p = fgets(str, 4096, td->io_log_rfile)) != NULL) {

		struct io_piece *ipo;

		int r;

		r = sscanf(p, "%256s %256s %llu %u", rfname, act, &offset,

									&bytes);

		if (td->o.replay_redirect)

			fname = td->o.replay_redirect;

		if (r == 4) {

			/*

			 * Check action first

			 */

			if (!strcmp(act, "wait"))

				rw = DDIR_WAIT;

			else if (!strcmp(act, "read"))

				rw = DDIR_READ;

			else if (!strcmp(act, "write"))

				rw = DDIR_WRITE;

			else if (!strcmp(act, "sync"))

				rw = DDIR_SYNC;

			else if (!strcmp(act, "datasync"))

				rw = DDIR_DATASYNC;

			else if (!strcmp(act, "trim"))

				rw = DDIR_TRIM;

			else {

				log_err("fio: bad iolog file action: %s\n",

									act);

				continue;

			}

			fileno = get_fileno(td, fname);

		} else if (r == 2) {

			rw = DDIR_INVAL;

			if (!strcmp(act, "add")) {

				if (td->o.replay_redirect &&

				    get_fileno(td, fname) != -1) {

					dprint(FD_FILE, "iolog: ignoring"

						" re-add of file %s\n", fname);

				} else {

					fileno = add_file(td, fname, td->subjob_number, 1);

					file_action = FIO_LOG_ADD_FILE;

				}

				continue;

			} else if (!strcmp(act, "open")) {

				fileno = get_fileno(td, fname);

				file_action = FIO_LOG_OPEN_FILE;

			} else if (!strcmp(act, "close")) {

				fileno = get_fileno(td, fname);

				file_action = FIO_LOG_CLOSE_FILE;

			} else {

				log_err("fio: bad iolog file action: %s\n",

									act);

				continue;

			}

		} else {

			log_err("bad iolog2: %s\n", p);

			continue;

		}

		if (rw == DDIR_READ)

			reads++;

		else if (rw == DDIR_WRITE) {

			/*

			 * Don't add a write for ro mode

			 */

			if (read_only)

				continue;

			writes++;

		} else if (rw == DDIR_WAIT) {

			if (td->o.no_stall)

				continue;

			waits++;

		} else if (rw == DDIR_INVAL) {

		} else if (!ddir_sync(rw)) {

			log_err("bad ddir: %d\n", rw);

			continue;

		}

		/*

		 * Make note of file

		 */

		ipo = calloc(1, sizeof(*ipo));

		init_ipo(ipo);

		ipo->ddir = rw;

		if (rw == DDIR_WAIT) {

			ipo->delay = offset;

		} else {

			if (td->o.replay_scale)

				ipo->offset = offset / td->o.replay_scale;

			else

				ipo->offset = offset;

			ipo_bytes_align(td->o.replay_align, ipo);

			ipo->len = bytes;

			if (rw != DDIR_INVAL && bytes > td->o.max_bs[rw]) {

				realloc = true;

				td->o.max_bs[rw] = bytes;

			}

			ipo->fileno = fileno;

			ipo->file_action = file_action;

			td->o.size += bytes;

		}

		queue_io_piece(td, ipo);

		if (td->o.read_iolog_chunked) {

			td->io_log_current++;

			items_to_fetch--;

			if (items_to_fetch == 0)

				break;

		}

	}

	free(str);

	free(act);

	free(rfname);

	if (td->o.read_iolog_chunked) {

		td->io_log_highmark = td->io_log_current;

		td->io_log_checkmark = (td->io_log_highmark + 1) / 2;

		fio_gettime(&td->io_log_highmark_time, NULL);

	}

	if (writes && read_only) {

		log_err("fio: <%s> skips replay of %d writes due to"

			" read-only\n", td->o.name, writes);

		writes = 0;

	}

	if (td->o.read_iolog_chunked) {

		if (td->io_log_current == 0) {

			return false;

		}

		td->o.td_ddir = TD_DDIR_RW;

		if (realloc && td->orig_buffer)

		{

			io_u_quiesce(td);

			free_io_mem(td);

			init_io_u_buffers(td);

		}

		return true;

	}

	if (!reads && !writes && !waits)

		return false;

	else if (reads && !writes)

		td->o.td_ddir = TD_DDIR_READ;

	else if (!reads && writes)

		td->o.td_ddir = TD_DDIR_WRITE;

	else

		td->o.td_ddir = TD_DDIR_RW;

	return true;

}

static bool is_socket(const char *path)

{

	struct stat buf;

	int r;

	r = stat(path, &buf;;

	if (r == -1)

		return false;

	return S_ISSOCK(buf.st_mode);

}

static int open_socket(const char *path)

{

	struct sockaddr_un addr;

	int ret, fd;

	fd = socket(AF_UNIX, SOCK_STREAM, 0);

	if (fd < 0)

		return fd;

	addr.sun_family = AF_UNIX;

	if (snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", path) >=

	    sizeof(addr.sun_path)) {

		log_err("%s: path name %s is too long for a Unix socket\n",

			__func__, path);

	}

	ret = connect(fd, (const struct sockaddr *)&addr, strlen(path) + sizeof(addr.sun_family));

	if (!ret)

		return fd;

	close(fd);

	return -1;

}

/*

 * open iolog, check version, and call appropriate parser

 */

static bool init_iolog_read(struct thread_data *td)

{

	char buffer[256], *p, *fname;

	FILE *f = NULL;

	fname = get_name_by_idx(td->o.read_iolog_file, td->subjob_number);

	dprint(FD_IO, "iolog: name=%s\n", fname);

	if (is_socket(fname)) {

		int fd;

		fd = open_socket(fname);

		if (fd >= 0)

			f = fdopen(fd, "r");

	} else

		f = fopen(fname, "r");

	free(fname);

	if (!f) {

		perror("fopen read iolog");

		return false;

	}

	p = fgets(buffer, sizeof(buffer), f);

	if (!p) {

		td_verror(td, errno, "iolog read");

		log_err("fio: unable to read iolog\n");

		fclose(f);

		return false;

	}

	/*

	 * version 2 of the iolog stores a specific string as the

	 * first line, check for that

	 */

	if (!strncmp(iolog_ver2, buffer, strlen(iolog_ver2))) {

		free_release_files(td);

		td->io_log_rfile = f;

		return read_iolog2(td);

	}

	log_err("fio: iolog version 1 is no longer supported\n");

	fclose(f);

	return false;

}

/*

 * Set up a log for storing io patterns.

 */

static bool init_iolog_write(struct thread_data *td)

{

	struct fio_file *ff;

	FILE *f;

	unsigned int i;

	f = fopen(td->o.write_iolog_file, "a");

	if (!f) {

		perror("fopen write iolog");

		return false;

	}

	/*

	 * That's it for writing, setup a log buffer and we're done.

	  */

	td->iolog_f = f;

	td->iolog_buf = malloc(8192);

	setvbuf(f, td->iolog_buf, _IOFBF, 8192);

	/*

	 * write our version line

	 */

	if (fprintf(f, "%s\n", iolog_ver2) < 0) {

		perror("iolog init\n");

		return false;

	}

	/*

	 * add all known files

	 */

	for_each_file(td, ff, i)

		log_file(td, ff, FIO_LOG_ADD_FILE);

	return true;

}

bool init_iolog(struct thread_data *td)

{

	bool ret;

	if (td->o.read_iolog_file) {

		int need_swap;

		/*

		 * Check if it's a blktrace file and load that if possible.

		 * Otherwise assume it's a normal log file and load that.

		 */

		if (is_blktrace(td->o.read_iolog_file, &need_swap))

			ret = load_blktrace(td, td->o.read_iolog_file, need_swap);

		else

			ret = init_iolog_read(td);

	} else if (td->o.write_iolog_file)

		ret = init_iolog_write(td);

	else

		ret = true;

	if (!ret)

		td_verror(td, EINVAL, "failed initializing iolog");

	return ret;

}

void setup_log(struct io_log **log, struct log_params *p,

	       const char *filename)

{

	struct io_log *l;

	int i;

	struct io_u_plat_entry *entry;

	struct flist_head *list;

	l = scalloc(1, sizeof(*l));

	INIT_FLIST_HEAD(&l->io_logs);

	l->log_type = p->log_type;

	l->log_offset = p->log_offset;

	l->log_gz = p->log_gz;

	l->log_gz_store = p->log_gz_store;

	l->avg_msec = p->avg_msec;

	l->hist_msec = p->hist_msec;

	l->hist_coarseness = p->hist_coarseness;

	l->filename = strdup(filename);

	l->td = p->td;

	/* Initialize histogram lists for each r/w direction,

	 * with initial io_u_plat of all zeros:

	 */

	for (i = 0; i < DDIR_RWDIR_CNT; i++) {

		list = &l->hist_window[i].list;

		INIT_FLIST_HEAD(list);

		entry = calloc(1, sizeof(struct io_u_plat_entry));

		flist_add(&entry->list, list);

	}

	if (l->td && l->td->o.io_submit_mode != IO_MODE_OFFLOAD) {

		struct io_logs *__p;

		__p = calloc(1, sizeof(*l->pending));

		__p->max_samples = DEF_LOG_ENTRIES;

		__p->log = calloc(__p->max_samples, log_entry_sz(l));

		l->pending = __p;

	}

	if (l->log_offset)

		l->log_ddir_mask = LOG_OFFSET_SAMPLE_BIT;

	INIT_FLIST_HEAD(&l->chunk_list);

	if (l->log_gz && !p->td)

		l->log_gz = 0;

	else if (l->log_gz || l->log_gz_store) {

		mutex_init_pshared(&l->chunk_lock);

		mutex_init_pshared(&l->deferred_free_lock);

		p->td->flags |= TD_F_COMPRESS_LOG;

	}

	*log = l;

}

#ifdef CONFIG_SETVBUF

static void *set_file_buffer(FILE *f)

{

	size_t size = 1048576;

	void *buf;

	buf = malloc(size);

	setvbuf(f, buf, _IOFBF, size);

	return buf;

}

static void clear_file_buffer(void *buf)

{

	free(buf);

}

#else

static void *set_file_buffer(FILE *f)

{

	return NULL;

}

static void clear_file_buffer(void *buf)

{

}

#endif

void free_log(struct io_log *log)

{

	while (!flist_empty(&log->io_logs)) {

		struct io_logs *cur_log;

		cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);

		flist_del_init(&cur_log->list);

		free(cur_log->log);

		sfree(cur_log);

	}

	if (log->pending) {

		free(log->pending->log);

		free(log->pending);

		log->pending = NULL;

	}

	free(log->pending);

	free(log->filename);

	sfree(log);

}

uint64_t hist_sum(int j, int stride, uint64_t *io_u_plat,

		uint64_t *io_u_plat_last)

{

	uint64_t sum;

	int k;

	if (io_u_plat_last) {

		for (k = sum = 0; k < stride; k++)

			sum += io_u_plat[j + k] - io_u_plat_last[j + k];

	} else {

		for (k = sum = 0; k < stride; k++)

			sum += io_u_plat[j + k];

	}

	return sum;

}

static void flush_hist_samples(FILE *f, int hist_coarseness, void *samples,

			       uint64_t sample_size)

{

	struct io_sample *s;

	int log_offset;

	uint64_t i, j, nr_samples;

	struct io_u_plat_entry *entry, *entry_before;

	uint64_t *io_u_plat;

	uint64_t *io_u_plat_before;

	int stride = 1 << hist_coarseness;

	if (!sample_size)

		return;

	s = __get_sample(samples, 0, 0);

	log_offset = (s->__ddir & LOG_OFFSET_SAMPLE_BIT) != 0;

	nr_samples = sample_size / __log_entry_sz(log_offset);

	for (i = 0; i < nr_samples; i++) {

		s = __get_sample(samples, log_offset, i);

		entry = s->data.plat_entry;

		io_u_plat = entry->io_u_plat;

		entry_before = flist_first_entry(&entry->list, struct io_u_plat_entry, list);

		io_u_plat_before = entry_before->io_u_plat;

		fprintf(f, "%lu, %u, %llu, ", (unsigned long) s->time,

						io_sample_ddir(s), (unsigned long long) s->bs);

		for (j = 0; j < FIO_IO_U_PLAT_NR - stride; j += stride) {

			fprintf(f, "%llu, ", (unsigned long long)

			        hist_sum(j, stride, io_u_plat, io_u_plat_before));

		}

		fprintf(f, "%llu\n", (unsigned long long)

		        hist_sum(FIO_IO_U_PLAT_NR - stride, stride, io_u_plat,

					io_u_plat_before));

		flist_del(&entry_before->list);

		free(entry_before);

	}

}