path: root/litmus/color_queue.c

#include <linux/slab.h>
#include <linux/mm.h>
#include <linux/sched.h>

#include <litmus/color_queue.h>
#include <litmus/color.h>
#include <litmus/lockdown.h>
#include <litmus/color.h>
#include <litmus/litmus.h>
#include <litmus/trace.h>

/* Uncomment to debug. */
#if 0
#define QTRACE(q, fmt, args...)	TRACE_CUR("q.phase: %llu  q.way: %d  " \
				"q.nr_cpus: %d  q.at_barrier: %d  " fmt, \
				q.phase, q.way, q.nr_cpus, q.at_barrier, \
				## args)
#endif

#ifndef QTRACE
#define QTRACE(q, fmt, args...) do { } while (0)
#endif

struct color_queue {
	struct list_head queue;
	/* the queue of work */

	int nr_cpus;
	/* number of CPUs that have outstanding requests */

	int at_barrier;
	/* number of CPUs that are currently waiting at the barrier */

	int way;
	/* which way the queue is currently reading in */

	u64 phase;
	/* monotonically increasing phase. use 64 bits so it doesn't wrap. */

	raw_spinlock_t lock;
	/* protects the above members */
};

struct cpu_entry {
	int phase;
	/* what phase a CPU is in */

	struct list_head enqueue;
	/* used as a PENDING list of work to enqueue in the color queue */

	int nr_work;
	/* outstanding work elements this CPU has */

	raw_spinlock_t lock;
	/* guard nr_work when CPU is "inside" the queue (its work is added) */
};

DEFINE_PER_CPU(struct cpu_entry, cpu_entries);
/* Per-CPU state. */

extern u32 unlocked_way[];
/* Index i means way i is unlocked. Comes from litmus/lockdown.c */

struct color_queue color_queue;
/* The queue. */

static u32 lockdown_value;
/* Cached lockdown value so that we don't need to look it up each time.
 * Should agree with color_queue.way.
 */

static u32 lockdown_value_idle = UNLOCK_ALL;
/* Lockdown state when the queue is inactive. Lets us lock down some ways
 * if we choose to.
 */

static void ***flusher_pages = NULL;
/* One page per [way, color] pair used for flushing. */

/*
 * Add work to the cpu_entry's PENDING list of work for the color queue.
 */
static void cpu_add_work(struct color_queue_request *req)
{
	struct cpu_entry *entry = &__get_cpu_var(cpu_entries);
	struct list_head *new = &req->list;
	list_add_tail(new, &entry->enqueue);
	entry->nr_work++;
	//TRACE_CUR("cpu->nr_work: %d  added work request pointer %p\n",
	//		entry->nr_work, req);
}

/*
 * Add a pending read in of a page. Does the chunking into work units.
 *
 * You should already set @way (in @color_page_info) by the time you
 * call this!
 *
 * @vaddr_start changes depending upon if this is a read-in or a "flush".
 */
static void color_page_info_add_work(struct color_page_info *info, void *vaddr_start)
{
	const int cpu = smp_processor_id();
	int i;

	//TRACE_CUR("adding work for color_page_info: 0x%p\n", info);

	for (i = 0; i < COLOR_REQUESTS_PER_PAGE; i++) {
		struct color_queue_request *req = &info->requests[i];
		void *work_vaddr = vaddr_start + i * COLOR_QUEUE_REQ_SIZE;

		WARN(req->request_type != COLOR_QUEUE_IDLE,
				"request was not idle!\n");

		req->request_type = COLOR_QUEUE_READ;
		req->cpu = cpu;
		req->request_data.read.vaddr = work_vaddr;
		cpu_add_work(req);
	}
}

static int color_queue_submit_work(void);

/*
 * Assumes this is called on the CPU that @ts ran on and that this CPU that is
 * requesting the work. Do not try and schedule work for someone else!
 *
 * Also assumes that @way set for each color_page_info before calling!
 */
int color_queue_enqueue_read(struct task_struct *ts)
{
	struct color_page_info *cur_info;

	TS_CQ_ENQUEUE_READ_START;

#if 0
	TRACE_CUR("enqueue read prev: %p  next: %p\n",
			tsk_rt(ts)->color_page_info_list.prev,
			tsk_rt(ts)->color_page_info_list.next);
#endif

	list_for_each_entry(cur_info,
			&tsk_rt(ts)->color_page_info_list,
			list)
	{
		void *vaddr_start = cur_info->vaddr;
		/* since this is a read, uses the page's address */

		color_page_info_add_work(cur_info, vaddr_start);
	}

	TS_CQ_ENQUEUE_READ_END;

	return color_queue_submit_work();
}

/*
 * Assumes this is called on the CPU that @ts ran on and that this CPU that is
 * requesting the work. Do not try and schedule work for someone else!
 *
 * Also assumes that @way is still set to the way this page was read into to
 * find the proper address to flush.
 */
int color_queue_enqueue_flush(struct task_struct *ts)
{
	struct color_page_info *cur_info;

	TS_CQ_ENQUEUE_FLUSH_START;

	list_for_each_entry(cur_info,
			&tsk_rt(ts)->color_page_info_list,
			list)
	{
		void *vaddr_start;
		vaddr_start = flusher_pages[cur_info->way][cur_info->color];
		/* now we use the corresponding flusher page */
		color_page_info_add_work(cur_info, vaddr_start);
	}

	TS_CQ_ENQUEUE_FLUSH_END;

	return color_queue_submit_work();
}

static void do_work_read(struct color_queue_request *request)
{
	void *vaddr = request->request_data.read.vaddr;

	/* Don't know which CPU is first to do work in a given phase, so have
	 * all CPUs set the lockdown register to the same value. Also, the
	 * "unlock_val" is also the lock value, since we don't know if other
	 * CPUs are still reading. We could take the unlock out of the read_in
	 * function, but it's one store operation and probably takes a few
	 * nanoseconds...
	 */
	color_read_in_mem(lockdown_value, lockdown_value,
			vaddr, vaddr + COLOR_QUEUE_REQ_SIZE);
}

/*
 * This just does a unit of work, ... son.
 */
static void do_work_son(struct color_queue_request *request)
{
	struct cpu_entry *entry;

	TS_CQ_WORK_DO_WORK_START;
	switch (request->request_type) {
		case COLOR_QUEUE_IDLE:
			TRACE_CUR("idle work in the queue!\n");
			WARN(1, "Idle work in the queue!\n");
			break;
		case COLOR_QUEUE_READ:
			//TRACE_CUR("doing work for CPU %d\n", request->cpu);
			do_work_read(request);
			break;
	}
	TS_CQ_WORK_DO_WORK_END;

	TS_CQ_WORK_NOTIFY_START;
	/* Tell the (possibly remote) CPU that we're a bro and helped out. */
	entry = &per_cpu(cpu_entries, request->cpu);
	raw_spin_lock(&entry->lock);
	entry->nr_work--;
	//TRACE_CUR("after doing work, cpu->nr_work: %d\n", entry->nr_work);
	raw_spin_unlock(&entry->lock);

	/* work is done, set it idle */
	request->request_type = COLOR_QUEUE_IDLE;
	TS_CQ_WORK_NOTIFY_END;
}

static void wait_next_phase(void)
{
	struct cpu_entry *entry = &__get_cpu_var(cpu_entries);
	struct color_queue_request *request;

	TS_CQ_PHASE_WAIT_START;

	for (;;) {
		raw_spin_lock(&color_queue.lock);
		if (entry->phase < color_queue.phase) {
			/* Move on to the next phase (or later). Another CPU already
			 * set up the lockdown value and updated the queue phase.
			 */
			entry->phase = color_queue.phase;
			QTRACE(color_queue,
					"cpu->phase: %d advances to a higher phase\n",
					entry->phase);
			raw_spin_unlock(&color_queue.lock);
			TS_CQ_PHASE_WAIT_END;
			return;
		}

		if (color_queue.nr_cpus == color_queue.at_barrier) {
			int next_way;

			QTRACE(color_queue, "everyone reached the barrier\n");

			/* Ready to start the next phase. */
			if (unlikely(list_empty(&color_queue.queue))) {
				QTRACE(color_queue, "The queue was empty (bad)\n");
				/* This should not happen! Will loop forever? */
				WARN(1, "color queue list was empty!\n");
				raw_spin_unlock(&color_queue.lock);
				continue;
			}
			request = list_first_entry(&color_queue.queue,
					struct color_queue_request, list);
			next_way = request->color_page_info->way;
			lockdown_value = unlocked_way[next_way];
			color_queue.way = next_way;
			color_queue.at_barrier = 0;
			color_queue.phase++;
			QTRACE(color_queue, "bumped the phase and way\n");
			raw_spin_unlock(&color_queue.lock);
		} else if (color_queue.nr_cpus < color_queue.at_barrier) {
			BUG();
		} else {
			/* Wait for work from the last phase to complete. */
			QTRACE(color_queue, "still waiting for others\n");
			raw_spin_unlock(&color_queue.lock);
			cpu_relax();
		}
	}
}

static int color_queue_loop(void)
{
	struct cpu_entry *entry = &__get_cpu_var(cpu_entries);
	struct color_queue_request *request;
	int nr_work, nr_work_done_by_task = 0;

	for (;;) {
		TS_CQ_LOOP_WORK_CHECK_START;
		raw_spin_lock(&entry->lock);
		nr_work = entry->nr_work;
		raw_spin_unlock(&entry->lock);
		TS_CQ_LOOP_WORK_CHECK_END;

		if (0 == nr_work) {
			TS_CQ_LOOP_PEACE_OUT_START;
			/* All the work is done for this CPU. We can leave. */
			raw_spin_lock(&color_queue.lock);
			color_queue.nr_cpus--;
			if (0 == color_queue.nr_cpus) {
				/* Queue is going idle. Restore lockdown. */
				set_lockdown(lockdown_value_idle);
				QTRACE(color_queue, "last CPU in queue.\n");
			}
			QTRACE(color_queue, "just left the queue\n");
			raw_spin_unlock(&color_queue.lock);
			TS_CQ_LOOP_PEACE_OUT_END;
			return nr_work_done_by_task;
		}

		/* Our work is not done, so continue processing more work. */

		TS_CQ_LOOP_BRANCH_START;
		raw_spin_lock(&color_queue.lock);
		if (unlikely(list_empty(&color_queue.queue))) {
			/* can this happen? */
			QTRACE(color_queue, "color queue was empty?\n");
			WARN(1, "color queue list was empty...\n");
			raw_spin_unlock(&color_queue.lock);
			TS_CQ_LOOP_BRANCH_END;
			continue;
		}
		request = list_first_entry(&color_queue.queue,
				struct color_queue_request, list);
		if (color_queue.way == request->color_page_info->way) {
			/* we're going to do this work */
			list_del(&request->list);
			QTRACE(color_queue, "found a piece of work to do: %p\n", request);
			raw_spin_unlock(&color_queue.lock);
			TS_CQ_LOOP_BRANCH_END;
			do_work_son(request);
			++nr_work_done_by_task;
		} else {
			/* we need to wait for the next phase */
			color_queue.at_barrier++;
			QTRACE(color_queue, "begins to wait at barrier...\n");
			raw_spin_unlock(&color_queue.lock);
			TS_CQ_LOOP_BRANCH_END;
			wait_next_phase();
		}
	}
}

/*
 * Actually enqueues the work on the color queue and enters the work loop.
 */
static int color_queue_submit_work(void)
{
	struct cpu_entry *entry;

	TS_CQ_SUBMIT_WORK_START;

	entry = &__get_cpu_var(cpu_entries);

	raw_spin_lock(&color_queue.lock);
	QTRACE(color_queue, "start submit work\n");
	entry->phase = color_queue.phase;
	color_queue.nr_cpus++;
	list_splice_tail_init(&entry->enqueue, &color_queue.queue);
	QTRACE(color_queue, "end submit work\n");
	raw_spin_unlock(&color_queue.lock);

	TS_CQ_SUBMIT_WORK_END;

	return color_queue_loop();
}

void cleanup_color_page_infos(struct list_head *head)
{
	struct color_page_info *cur, *tmp;

	if (NULL == head) {
		TRACE("Head was null.\n");
		return;
	}

	if (NULL == head->next || NULL == head->prev) {
		TRACE_CUR("next or prev was null\n");
		return;
	}

	list_for_each_entry_safe(cur, tmp, head, list) {
		TRACE_CUR("cleanup color_queue_info %p\n", cur);
		list_del(&cur->list);
		kfree(cur);
	}
}

/* called when user does add_pages proc handler */
int setup_flusher_array(void)
{
	int color, way, ret = 0;
	struct page *page;

	if (0 == color_cache_info.ways || 0 == color_cache_info.nr_colors) {
		WARN(1, "Cache information not initialized!\n");
		ret = -EINVAL;
		goto out;
	}

	if (flusher_pages != NULL)
		goto out;

	flusher_pages = (void***) kmalloc(color_cache_info.ways
			* sizeof(*flusher_pages), GFP_KERNEL);
	if (!flusher_pages) {
		printk(KERN_WARNING "No memory for flusher array!\n");
		ret = -EINVAL;
		goto out;
	}

	for (way = 0; way < color_cache_info.ways; way++) {
		void **flusher_color_arr;
		flusher_color_arr = (void**) kmalloc(sizeof(**flusher_pages)
				* color_cache_info.nr_colors, GFP_KERNEL);
		if (!flusher_color_arr) {
			printk(KERN_WARNING "No memory for flusher array!\n");
			ret = -ENOMEM;
			goto out_free;
		}

		flusher_pages[way] = flusher_color_arr;

		for (color = 0; color < color_cache_info.nr_colors; color++) {
			page = get_colored_page(color);
			if (!page) {
				printk(KERN_WARNING "no more colored pages\n");
				ret = -EINVAL;
				goto out_free;
			}
			flusher_pages[way][color] = page_address(page);
			if (!flusher_pages[way][color]) {
				printk(KERN_WARNING "bad page address\n");
				ret = -EINVAL;
				goto out_free;
			}
		}
	}
out:
	return ret;
out_free:
	for (way = 0; way < color_cache_info.ways; way++) {
		for (color = 0; color < color_cache_info.nr_colors; color++) {
			/* not bothering to try and give back colored pages */
		}
		kfree(flusher_pages[way]);
	}
	kfree(flusher_pages);
	flusher_pages = NULL;
	return ret;
}

static int __init init_color_queue(void)
{
	struct cpu_entry *cpu_entry;
	int cpu;

	BUILD_BUG_ON((PAGE_SIZE % COLOR_QUEUE_REQ_SIZE) != 0);

	for_each_online_cpu(cpu) {
		cpu_entry = &per_cpu(cpu_entries, cpu);
		INIT_LIST_HEAD(&cpu_entry->enqueue);
		cpu_entry->nr_work = 0;
		raw_spin_lock_init(&cpu_entry->lock);
	}

	raw_spin_lock_init(&color_queue.lock);
	INIT_LIST_HEAD(&color_queue.queue);

	return 0;
}

module_init(init_color_queue);