/*
 * Copyright (C) 2012-2020 all contributors <cmogstored-public@yhbt.net>
 * License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
 */
#include "cmogstored.h"
#include "compat_memstream.h"

/*
 * maps multiple "devXXX" directories to the device.
 * This is uncommon in real world deployments (multiple mogdevs sharing
 * the same system device), but happens frequently in testing
 */
struct mog_devlist {
	dev_t st_dev;
	Hash_table *by_mogdevid;
};

static size_t devlist_hash(const void *x, size_t tablesize)
{
	const struct mog_devlist *devlist = x;

	return devlist->st_dev % tablesize;
}

static bool devlist_cmp(const void *a, const void *b)
{
	const struct mog_devlist *devlist_a = a;
	const struct mog_devlist *devlist_b = b;

	return devlist_a->st_dev == devlist_b->st_dev;
}

static void devlist_free(void *x)
{
	struct mog_devlist *devlist = x;

	hash_free(devlist->by_mogdevid);
	free(devlist);
}

static struct mog_devlist * mog_devlist_new(dev_t st_dev)
{
	struct mog_devlist *devlist = xmalloc(sizeof(struct mog_devlist));

	devlist->st_dev = st_dev;
	devlist->by_mogdevid = hash_initialize(7, NULL,
	                                       mog_dev_hash, mog_dev_cmp,

					       /*
						* elements are freed when
						* svc->by_mog_devid is freed
						*/
					       NULL);

	mog_oom_if_null(devlist->by_mogdevid);

	return devlist;
}

/* ensures svc has a devlist, this must be called with devstats_lock held */
static struct mog_devlist * svc_devlist(struct mog_svc *svc, dev_t st_dev)
{
	struct mog_devlist *devlist;
	struct mog_devlist finder;

	assert(svc->by_st_dev && "by_st_dev unintialized in svc");

	finder.st_dev = st_dev;
	devlist = hash_lookup(svc->by_st_dev, &finder);

	if (devlist == NULL) {
		devlist = mog_devlist_new(st_dev);
		switch (hash_insert_if_absent(svc->by_st_dev, devlist, NULL)) {
		case 0:
			assert(0 && "race condition, devlist should insert "
			       "without error");
			abort();
		case 1: break; /* OK, inserted */
		default: mog_oom(); /* -1 */
		}
	}
	return devlist;
}

static void svc_init_dev_hash(struct mog_svc *svc)
{
	if (svc->by_st_dev) {
		hash_clear(svc->by_st_dev);
		return;
	}

	svc->by_st_dev = hash_initialize(7, NULL, devlist_hash,
	                                 devlist_cmp, devlist_free);
	mog_oom_if_null(svc->by_st_dev);
}

static int svc_scandev(struct mog_svc *svc, unsigned *nr, mog_scandev_cb cb)
{
	struct dirent *ent;
	int rc = 0;

	CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
	svc_init_dev_hash(svc);
	rewinddir(svc->dir);
	while ((ent = readdir(svc->dir))) {
		unsigned long mog_devid;
		char *end;
		size_t len = strlen(ent->d_name);
		struct mog_dev *dev;
		struct mog_devlist *devlist;
		Hash_table *devhash;

		if (len <= 3) continue;
		if (memcmp("dev", ent->d_name, 3) != 0) continue;

		mog_devid = strtoul(ent->d_name + 3, &end, 10);
		if (*end != 0) continue;
		if (mog_devid > MOG_DEVID_MAX) continue;

		dev = mog_dev_for(svc, (uint32_t)mog_devid, true);
		if (!dev) continue;

		devlist = svc_devlist(svc, dev->st_dev);
		devhash = devlist->by_mogdevid;

		if (cb)
			rc |= cb(dev, svc); /* mog_dev_mkusage */

		switch (hash_insert_if_absent(devhash, dev, NULL)) {
		case 0:
			/* do not free dev, it is in svc->by_mog_devid */
			syslog(LOG_ERR,
			       "%s/%s seen twice in readdir (BUG?)",
			       svc->docroot, ent->d_name);
			break;
		case 1:
			(*nr)++;
			break;
		default: mog_oom(); /* -1 */
		}
	}
	CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));

	return rc;
}

static bool write_dev_stats(void *entry, void *filep)
{
	struct mog_dev *dev = entry;
	FILE **fp = filep;
	char util[MOG_IOUTIL_LEN];

	mog_iou_read(dev->st_dev, util);

	if (fprintf(*fp, "%u\t%s\n", dev->devid, util) > 0)
		return true;

	/* stop iteration in case we get EIO/ENOSPC on systems w/o memstream */
	my_memstream_errclose(*fp);
	*fp = NULL;
	return false;
}

static bool write_devlist_stats(void *entry, void *filep)
{
	struct mog_devlist *devlist = entry;
	FILE **fp ;

	hash_do_for_each(devlist->by_mogdevid, write_dev_stats, filep);
	fp = filep;

	/* *filep becomes NULL on errors */
	return !!*fp;
}

/* updates per-svc device stats from the global mount list */
static ssize_t devstats_stringify(struct mog_svc *svc, char **dst)
{
	FILE *fp;
	size_t bytes;

	assert(svc->by_st_dev && "need to scan devices first");

	/* open_memstream() may fail on EIO/EMFILE/ENFILE on fake memstream */
	fp = open_memstream(dst, &bytes);
	if (!fp)
		return -1;

	/*
	 * write_devlist_stats->write_dev_stats may fclose and NULL fp
	 * to indicate error:
	 */
	hash_do_for_each(svc->by_st_dev, write_devlist_stats, &fp);
	if (!fp)
		return -1;

	if (fprintf(fp, ".\n") == 2) {
		CHECK(int, 0, my_memstream_close(fp, dst, &bytes));
		return bytes;
	}

	my_memstream_errclose(fp);
	return -1;
}

void mog_svc_devstats_subscribe(struct mog_mgmt *mgmt)
{
	struct mog_svc *svc = mgmt->svc;

	CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
	LIST_INSERT_HEAD(&svc->devstats_subscribers, mgmt, subscribed);
	CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));
}

/* called while iterating through all mog_svc objects */
bool mog_svc_devstats_broadcast(void *ent, void *ignored)
{
	struct mog_svc *svc = ent;
	struct mog_mgmt *mgmt, *tmp;
	struct iovec iov;
	char *buf = NULL;
	ssize_t len;
	struct mog_fd *mfd;

	CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));

	len = devstats_stringify(svc, &buf);
	if (len < 0)
		goto out;

	LIST_FOREACH_SAFE(mgmt, &svc->devstats_subscribers, subscribed, tmp) {
		assert(mgmt->wbuf == NULL && "wbuf not null");
		iov.iov_base = buf;
		iov.iov_len = (size_t)len;
		mog_mgmt_writev(mgmt, &iov, 1);

		if (mgmt->wbuf == NULL) continue; /* success */

		LIST_REMOVE(mgmt, subscribed);
		mfd = mog_fd_of(mgmt);
		if (mgmt->wbuf == MOG_WR_ERROR) {
			assert(mgmt->rbuf == NULL && "would leak rbuf");
			mog_fd_put(mfd);
		} else { /* blocked on write */
			mog_idleq_push(mgmt->svc->queue, mfd, MOG_QEV_WR);
		}
	}
out:
	CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));

	free(buf);

	return true;
}

static bool devstats_shutdown_i(void *svcptr, void *ignored)
{
	struct mog_svc *svc = svcptr;
	struct mog_mgmt *mgmt, *tmp;
	struct mog_fd *mfd;

	CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
	LIST_FOREACH_SAFE(mgmt, &svc->devstats_subscribers, subscribed, tmp) {
		assert(mgmt->wbuf == NULL && "wbuf not null");
		assert(mgmt->rbuf == NULL && "would leak rbuf");
		LIST_REMOVE(mgmt, subscribed);
		mfd = mog_fd_of(mgmt);
		mog_fd_put(mfd);
	}
	CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));

	return true;
}

void mog_svc_dev_shutdown(void)
{
	mog_svc_each(devstats_shutdown_i, NULL);
}

static bool svc_mkusage_each(void *svcptr, void *ignored)
{
	struct mog_svc *svc = svcptr;
	unsigned ndev = 0;

	svc_scandev(svc, &ndev, mog_dev_mkusage);

	if (svc->queue && (svc->nmogdev != ndev))
		mog_svc_thrpool_rescale(svc, ndev);
	svc->nmogdev = ndev;

	return true;
}

void mog_mkusage_all(void)
{
	mog_svc_each(svc_mkusage_each, NULL);
}

/* we should never set ioq_max == 0 */
static void svc_rescale_warn_fix_capa(struct mog_svc *svc, unsigned ndev_new)
{
	if (svc->thr_per_dev != 0)
		return;

	syslog(LOG_WARNING,
	       "serving %s with fewer aio_threads(%u) than devices(%u)",
	       svc->docroot, svc->user_set_aio_threads, ndev_new);
	syslog(LOG_WARNING,
	       "set \"server aio_threads = %u\" or higher via sidechannel",
	       ndev_new);

	svc->thr_per_dev = 1;
}

static void mog_svc_dev_rescale_all(struct mog_svc *svc)
{
	/* iterate through each device of this svc */
	CHECK(int, 0, pthread_mutex_lock(&svc->by_mog_devid_lock));
	hash_do_for_each(svc->by_mog_devid, mog_dev_user_rescale_i, svc);
	CHECK(int, 0, pthread_mutex_unlock(&svc->by_mog_devid_lock));
}

void mog_svc_dev_requeue_prepare(struct mog_svc *svc)
{
	/* iterate through each device of this svc */
	CHECK(int, 0, pthread_mutex_lock(&svc->by_mog_devid_lock));
	hash_do_for_each(svc->by_mog_devid, mog_dev_requeue_prepare, svc);
	CHECK(int, 0, pthread_mutex_unlock(&svc->by_mog_devid_lock));
}

/* rescaling only happens in the main thread */
void mog_svc_dev_user_rescale(struct mog_svc *svc, size_t ndev_new)
{
	assert(svc->user_set_aio_threads &&
	       "user did not set aio_threads via sidechannel");

	svc->thr_per_dev = svc->user_set_aio_threads / ndev_new;

	svc_rescale_warn_fix_capa(svc, ndev_new);
	mog_svc_dev_rescale_all(svc);
}

/*
 * this forces ioqs to detect contention and yield,
 * leading to quicker shutdown
 */
void mog_svc_dev_quit_prepare(struct mog_svc *svc)
{
	svc->thr_per_dev = 1;
	mog_svc_dev_rescale_all(svc);
}