path: root/litmus/servers.c

/*
 * TODO: memory leaks for stopping
 */
#include <linux/hrtimer.h>
#include <linux/percpu.h>
#include <linux/sched.h>
#include <linux/uaccess.h>
#include <linux/ctype.h>

#include <litmus/bheap.h>
#include <litmus/litmus.h>
#include <litmus/litmus_proc.h>
#include <litmus/sched_trace.h>
#include <litmus/servers.h>

#define DEBUG_SERVERS

/*  Not working */
/* #define COMPLETION_ON_MASTER */

#define TIME(x)                                 \
        ({lt_t y = x;                           \
        do_div(y, NSEC_PER_MSEC);               \
        y;})

#ifdef  DEBUG_SERVERS
#define _TRACE_SUB(fmt, args...) \
        sched_trace_log_message("%d P%d     -[%s@%s:%d]: " fmt "\n", \
				TRACE_ARGS, ## args)
#define TRACE_SUB(s, fmt, args...) \
	do {\
	if ((s)->running)						\
		_TRACE_SUB("(%s/%d:%d) " SERVER_FMT " " fmt,		\
			   (s)->scheduled->comm,			\
			   (s)->scheduled->pid,			\
			   (s)->scheduled->rt_param.job_params.job_no, \
			   SERVER_ARGS(s), ##args);			\
	else								\
		_TRACE_SUB("(NULL) " SERVER_FMT " " fmt,		\
			   SERVER_ARGS(s), ##args);			\
	} while(0)

#define _TRACE_TIMER(fmt, args...)				       \
        sched_trace_log_message("%d P%d*[%s@%s:%d]: " fmt " at %d\n",  \
				TRACE_ARGS,  ## args, TIME(litmus_clock()))
#define TRACE_TIMER(s, fmt, args...)					\
	do {								\
	if ((s)->running)						\
	       _TRACE_TIMER("(%s/%d:%d) " SERVER_FMT " " fmt,		\
			   (s)->scheduled->comm,			\
			   (s)->scheduled->pid,			\
			   (s)->scheduled->rt_param.job_params.job_no, \
			    SERVER_ARGS(s), ##args);			\
	else								\
		_TRACE_TIMER("(NULL) " SERVER_FMT " " fmt,		\
			     SERVER_ARGS(s), ##args);			\
	} while(0)
#else
#define _TRACE_SUB(fmt, args...)
#define TRACE_SUB(s, fmt, args...)
#define TRACE_TIMER(s, fmt, args...)
#define _TRACE_TIMER(fmt, args...)
#endif

/* Used to run a server on a remote CPU */
DEFINE_PER_CPU(struct hrtimer_start_on_info, server_cpu_infos);

/* Memory slabs for servers */
struct kmem_cache *server_release_cache;
struct kmem_cache *server_cache;

typedef struct proc_read_args {
	char 	*page;
	int	length;
} proc_read_args_t;

typedef struct {
	admit_server_t  admit_server;
	list_servers_t	list_servers;
	stop_servers_t  stop_servers;
} methods_t;

/*
 * Okay to call if the timer is not armed.
 */
static inline int timer_cancel(struct hrtimer *timer)
{
        if (hrtimer_active(timer))
                return hrtimer_try_to_cancel(timer);
	else
		return 0;
}

static int completion_timer_arm(server_domain_t* domain, int cpu)
{
	int err = 0, on_cpu;
	lt_t now = litmus_clock();
	server_t *server = domain->running[cpu];
	lt_t budget_exhausted = now + server->budget;

	/* This happens when someone attempts to call server_run when
	 * the server completes. When this happens, we can ignore the request
	 * here because completion_timer_fire will re-arm the timer if
	 * the server is still running / was run again.
	 */
	if (hrtimer_active(&domain->completion_timers[cpu].timer)) {
		TRACE_SUB(server, "cannot arm completion, already active");
		return 0;
	}

	if (domain->completion_timers[cpu].armed) {
		TRACE_SUB(server, "cannot arm completion, waiting for arm");
		return 0;
	}

	TRACE_SUB(server, "start time: %llu", server->start_time);

#ifdef COMPLETION_ON_MASTER
	if (domain->release_master != NO_CPU)
		on_cpu = domain->release_master;
	else
#endif
		on_cpu = cpu;

	if (cpu != smp_processor_id()) {
                atomic_set(&domain->completion_timers[cpu].info.state,
                           HRTIMER_START_ON_INACTIVE);
                err = hrtimer_start_on(on_cpu,
                                       &domain->completion_timers[cpu].info,
                                       &domain->completion_timers[cpu].timer,
                                       ns_to_ktime(budget_exhausted),
                                       HRTIMER_MODE_ABS_PINNED);
                if (err) {
                        TRACE_SUB(server, "failed to arm completion");
                } else {
			TRACE_SUB(server, "success on P%d!", on_cpu);
		}
	} else {
		__hrtimer_start_range_ns(&domain->completion_timers[cpu].timer,
					 ns_to_ktime(budget_exhausted),
					 0 /* delta */,
					 HRTIMER_MODE_ABS_PINNED,
					 0 /* no wakeup */);
	}

	domain->completion_timers[cpu].armed = 1;

	TRACE_SUB(server, "completion armed to fire at %llu",
		  TIME(budget_exhausted));

	return 1;
}

void server_run(server_t *server, struct task_struct *task,
		server_domain_t *domain)
{
	int cpu = task->rt_param.linked_on;

	BUG_ON(server->scheduled);
	BUG_ON(server->running);
	BUG_ON(cpu == NO_CPU);
	BUG_ON(domain->running[cpu]);

	server->scheduled = task;
	server->running = 1;
	server->start_time = litmus_clock();

	domain->running[cpu] = server;
	domain->completion_timers[cpu].armed =completion_timer_arm(domain, cpu);

	TRACE_SUB(server, "running on cpu P%d", task->rt_param.linked_on);
}

static enum hrtimer_restart completion_timer_fire(struct hrtimer *timer)
{
	int cpu;
	unsigned long flags;
	enum hrtimer_restart rv;
	struct task_struct *was_running;
	completion_timer_t *completion_timer;
	server_domain_t *domain;
	server_t *server;
	lt_t budget_exhausted;

	rv  = HRTIMER_NORESTART;

	completion_timer = container_of(timer, completion_timer_t, timer);
	domain = completion_timer->domain;
	cpu = completion_timer->cpu;

	raw_spin_lock_irqsave(domain->timer_lock, flags);

	_TRACE_TIMER("completion timer firing on P%d");

	/* We got the lock before someone tried to re-arm. Proceed. */
	if (completion_timer->armed) {
		server = domain->running[cpu];
		TRACE_SUB(server, "completed");

		was_running = server->scheduled;

		server->budget     = 0;
		server->running    = 0;
		server->scheduled  = NULL;
		server->start_time = 0;
		domain->running[cpu] = NULL;

		domain->server_completed(server, was_running);
	}

	/* Someone either beat us to the lock or hooked up a new server
	 * when we called server_completed. Rearm the timer.
	 */
	if (domain->running[cpu] && !completion_timer->armed) {
		server = domain->running[cpu];
		TRACE_SUB(server, "rearming on P%d", cpu);
		budget_exhausted = server->start_time + server->budget;
		hrtimer_set_expires(timer, ns_to_ktime(budget_exhausted));
		completion_timer->armed = 1;
		rv = HRTIMER_RESTART;
	} else {
		completion_timer->armed = 0;
	}

	raw_spin_unlock_irqrestore(domain->timer_lock, flags);

	return rv;
}

struct kmem_cache *server_release_cache; /* In litmus.c */
static enum hrtimer_restart release_servers_fire(struct hrtimer *timer);

/*
 * Initialize heap.
 */
static server_release_heap_t* release_heap_alloc(int gfp_flags)
{
	server_release_heap_t *rh;
	rh = kmem_cache_alloc(server_release_cache, gfp_flags);
	if (rh) {
		hrtimer_init(&rh->timer, CLOCK_MONOTONIC, HRTIMER_MODE_ABS);
		rh->timer.function = release_servers_fire;
	}
	return rh;
}

static void release_heap_free(server_release_heap_t* rh)
{
	kmem_cache_free(server_release_cache, rh);
}

void server_init(server_t *server, int id, lt_t wcet, lt_t period, int grouped)
{
	server->id = id;
	server->wcet = wcet;
	server->period = period;

	server->deadline = 0;
	server->release = 0;
	server->budget = 0;
	server->start_time = 0;

	server->running = 0;
	server->job_no = 0;
	server->scheduled = NULL;

	server->data = NULL;

	server->hn = bheap_node_alloc(GFP_ATOMIC);
	bheap_node_init(&server->hn, server);
	INIT_LIST_HEAD(&server->list);
	server->release_heap = NULL;

	if (grouped) {
		server->release_heap = release_heap_alloc(GFP_ATOMIC);
		INIT_LIST_HEAD(&server->release_list);
	}
}

void server_destroy(server_t *server)
{
	bheap_node_free(server->hn);
	if (server->release_heap) {
		release_heap_free(server->release_heap);
	}
}

server_t* server_alloc(int gfp_flags)
{
	return kmem_cache_alloc(server_cache, gfp_flags);
}

void server_free(server_t *server)
{
	kmem_cache_free(server_cache, server);
}

/*
 * Handles subtraction of lt_t without underflows.
 */
static inline lt_t lt_subtract(lt_t a, lt_t b)
{
        long long sub = (long long)a - (long long)b;
        if (sub >= 0)
                return sub;
        else
                return 0;
}

void server_stop(server_t *server, server_domain_t *domain)
{
	int cpu;
	lt_t elapsed_time, now = litmus_clock();

	if (!server->running) {
		TRACE_SUB(server, "already stopped");
		return;
	}

	BUG_ON(!server->running);
	TRACE_SUB(server, "stopping server, start: %llu, end: %llu",
		  server->start_time, now);


	/* Calculate remaining budget */
	elapsed_time = lt_subtract(now, server->start_time);
	server->budget -= elapsed_time;

	TRACE_SUB(server, "new budget: %llu", TIME(server->budget));

	/* Set domain state */
	cpu = server->scheduled->rt_param.linked_on;
	domain->completion_timers[cpu].armed = 0;
	domain->running[cpu] = NULL;
	timer_cancel(&domain->completion_timers[cpu].timer);

	/* Make server inactive */
	server->running    = 0;
	server->scheduled  = NULL;
	server->start_time = 0;
}

void server_release(server_t *server)
{
	BUG_ON(!server->deadline);

	server->budget    = server->wcet;
	server->release   = server->deadline;
	server->deadline += server->period;
	++server->job_no;

	TRACE_SUB(server, "budget: %llu, release: %llu,"
		  "deadline: %llu, period: %llu, job: %d",
		  server->budget, server->release, server->deadline,
		  server->period, server->job_no);
	TRACE_SUB(server, "budget: %llu, release: %llu,"
		  "deadline: %llu, period: %llu, job: %d",
		  TIME(server->budget), TIME(server->release), TIME(server->deadline),
		  TIME(server->period), server->job_no);

	/* Need to reset for budget calculations */
	if (server->running)
		server->start_time = litmus_clock();
}

void server_release_at(server_t *server, lt_t time)
{
	server->deadline = time;
	server_release(server);

	TRACE_SUB(server, "releasing at %llu", time);
}

/******************************************************************************
 * Proc methods
 ******************************************************************************/

static int server_proc_read(char* page, char **start, off_t off,
			     int count,	int *eof, void *data)
{
	int length;
	methods_t *methods = (methods_t*)data;
	proc_read_args_t *args = kmalloc(sizeof(proc_read_args_t),
					 GFP_ATOMIC);

	args->page   = page;
	args->length = 0;
	methods->list_servers(args);

	length = args->length;
	kfree(args);
	*eof = 1;

	return length;
}

void server_proc_read_single(server_t *server, int cpu,
			     struct proc_read_args *args)
{
	if (cpu == NO_CPU) {
		args->length +=
			snprintf(args->page + args->length,
				 PAGE_SIZE - args->length,
				 "%8llu    %8llu\n",
				 server->wcet, server->period);
	} else {
		args->length +=
			snprintf(args->page + args->length,
				 PAGE_SIZE - args->length,
				 "%8llu    %8llu    %3d\n",
				 server->wcet, server->period, cpu);
	}
}

/*
 * Validate server parameters.
 */
static inline int server_param_check(unsigned long long wcet,
				     unsigned long long period,
				     int cpu)
{
	int rv = 0;

	if (wcet <= 0) {
		printk(KERN_WARNING "Invalid WCET '%llu'\n", wcet);
		rv = -EINVAL;
		goto out;
	}

	if (period < wcet) {
		printk(KERN_WARNING "Invalid period '%llu'\n", period);
		rv = -EINVAL;
		goto out;
	}

	if (cpu != NO_CPU && (cpu < 0 || cpu >= nr_cpu_ids)) {
		printk(KERN_WARNING "Invalid CPU '%d'\n", cpu);
		rv = -EINVAL;
		goto out;
	}
 out:
	return rv;
}

/* Macro to see if we are in the buffer's range and not at the null byte */
#define buf_in_range(buf, pos, max) (buf <= pos && pos < (buf + max)  && *pos)

#define find_newline(buf, pos, max)					\
	do {								\
		while (buf_in_range(buf, pos, max) &&			\
				*pos != '\n')				\
			++pos;						\
	} while (0)

static int server_proc_write(struct file *file, const char __user *input,
			      unsigned long count, void *data)
{
	methods_t *methods = (methods_t*)data;
#define SERVER_PROC_BUF 512
	char buffer[SERVER_PROC_BUF];
	unsigned long long wcet, period;
	char *pos, *newline, *space_check;
	int nums_converted, chars_seen, ret, cpu;

	/* Allow plugin to stop any running servers */
	methods->stop_servers();

	if (count >= SERVER_PROC_BUF){
		printk(KERN_WARNING "proc buffer possibly too small in %s.\n",
				__func__);
		return -ENOSPC;
	}

	memset(buffer, 0, SERVER_PROC_BUF);

	/* Input is definitely < SERVER_PROC_BUF (see above check) */
	if (copy_from_user(buffer, input, count))
		return -EFAULT;

	buffer[SERVER_PROC_BUF-1] = '\0';
	pos = buffer;

	while (buf_in_range(buffer, pos, SERVER_PROC_BUF)) {
		newline = pos;
		find_newline(buffer, newline, SERVER_PROC_BUF);
		if (buf_in_range(buffer, newline, SERVER_PROC_BUF)) {
			/* If there was a newline character */
			*newline = '\0';
		}
		nums_converted = sscanf(pos, "%llu %llu %d%n", &wcet,
					&period, &cpu, &chars_seen);
		if (nums_converted == 2)
			cpu = NO_CPU;
		if (nums_converted != 2 && nums_converted != 3) {
			printk(KERN_WARNING "Didn't see 2-3 integers for "
					"server config: %s\n", pos);
			goto loop_end;
		}
		space_check = pos + chars_seen;
		if (space_check != newline) {
			/* If the newline was not right after the numbers
			 * converted, ensure extra characters are just space
			 */
			for (; *space_check; space_check++) {
				if (!isspace(*space_check)) {
					printk(KERN_WARNING "Extra characters "
							"in line: %s\n", pos);
					goto loop_end;
				}
			}
		}

		ret = server_param_check(wcet, period, cpu);
		if (ret) goto loop_end;

		ret = methods->admit_server(wcet, period, cpu);
		if (ret) {
			printk(KERN_WARNING "Litmus plugin rejects server with "
			      "period: %llu, wcet: %llu, cpu: %d\n",
			       period, wcet, cpu);
			goto loop_end; /* Currently does nothing */
		}
loop_end:
		pos = newline + 1; /* Consider next line */
	}

	return count;
}

int server_proc_init(struct proc_dir_entry *proc_dir, char *file,
		     admit_server_t admit_server,
		     list_servers_t list_servers,
		     stop_servers_t stop_servers)
{
	int rv;
	methods_t *methods;
	struct proc_dir_entry *server_entry;

	server_entry = create_proc_entry(file, 0644, proc_dir);
	if (!server_entry) {
		printk(KERN_ERR "Could not create proc entry: %s.\n", file);
		rv = -ENOENT;
		goto out;
 	}

	/* Ack memory leak TODO */
	methods = kmalloc(sizeof(methods_t), GFP_ATOMIC);
	methods->admit_server = admit_server;
	methods->list_servers = list_servers;
	methods->stop_servers = stop_servers;
	server_entry->data = methods;

	server_entry->read_proc  = server_proc_read;
	server_entry->write_proc = server_proc_write;

 out:
	return rv;
}

void server_proc_exit(struct proc_dir_entry *proc_dir, char *file)
{
	remove_proc_entry(file, proc_dir);
}

/******************************************************************************
 * Domain methods
 ******************************************************************************/

void server_domain_init(server_domain_t *domain,
			servers_released_t servers_released,
			server_completed_t server_completed,
			int release_master, raw_spinlock_t *timer_lock)
{
	int i;
	BUG_ON(!servers_released || !server_completed);

	INIT_LIST_HEAD(&domain->tobe_released);
	for (i = 0; i < SERVER_RELEASE_QUEUE_SLOTS; i++)
		INIT_LIST_HEAD(&domain->release_queue[i]);

	raw_spin_lock_init(&domain->release_lock);
	raw_spin_lock_init(&domain->tobe_lock);


	domain->release_master = release_master;
	domain->timer_lock = timer_lock;
	domain->server_completed = server_completed;
	domain->servers_released = servers_released;

	domain->completion_timers = kmalloc(NR_CPUS*sizeof(completion_timer_t),
					    GFP_ATOMIC);
	domain->running = kmalloc(NR_CPUS*sizeof(server_t*), GFP_ATOMIC);

	for_each_online_cpu(i) {
		domain->running[i] = NULL;
		domain->completion_timers[i].armed = 0;
		domain->completion_timers[i].cpu = i;
		hrtimer_init(&domain->completion_timers[i].timer,
			     CLOCK_MONOTONIC,
			     HRTIMER_MODE_ABS);
		domain->completion_timers[i].timer.function =
			completion_timer_fire;
		hrtimer_start_on_info_init(&domain->completion_timers[i].info);
		domain->completion_timers[i].domain = domain;
	}
}

void server_domain_destroy(server_domain_t *domain)
{
	kfree(domain->completion_timers);
	kfree(domain->running);
}

static unsigned int time2slot(lt_t time)
{
	return (unsigned int) time2quanta(time, FLOOR) %
		SERVER_RELEASE_QUEUE_SLOTS;
}

/*
 * Send a list of servers to a client callback.
 */
static enum hrtimer_restart release_servers_fire(struct hrtimer *timer)
{
	unsigned long flags;
	server_release_heap_t *rh;

	_TRACE_SUB("on_release_timer(0x%p) starts.", timer);

	rh = container_of(timer, server_release_heap_t, timer);

	raw_spin_lock_irqsave(&rh->domain->release_lock, flags);
	_TRACE_SUB("CB has the release_lock");

	/* Remove from release queue */
	list_del(&rh->list);

	raw_spin_unlock_irqrestore(&rh->domain->release_lock, flags);
	_TRACE_SUB("CB returned release_lock");

	/* Call release callback */
	rh->domain->servers_released(&rh->servers);
	/* WARNING: rh can be referenced from other CPUs from now on. */

	_TRACE_SUB("on_release_timer(0x%p) ends.", timer);

	return HRTIMER_NORESTART;
}

/*
 * Caller must hold release lock.
 * Will return heap for given time. If no such heap exists prior to
 * the invocation it will be created.
 */
static server_release_heap_t* get_release_heap(server_domain_t *rt,
					       server_t *server,
					       int use_server_heap)
{
	struct list_head *pos;
	server_release_heap_t *heap = NULL;
	server_release_heap_t *rh;
	lt_t release_time = server->release;
	unsigned int slot = time2slot(release_time);

	_TRACE_SUB("searching for release time %llu", release_time);

	/* Initialize pos for the case that the list is empty */
	pos = rt->release_queue[slot].next;
	list_for_each(pos, &rt->release_queue[slot]) {
		rh = list_entry(pos, server_release_heap_t, list);
		if (release_time == rh->release_time) {
			/* Perfect match -- this happens on hyperperiod
			 * boundaries
			 */
			heap = rh;
			break;
		} else if (lt_before(release_time, rh->release_time)) {
			/* We need to insert a new node since rh is
			 * already in the future
			 */
			break;
		}
	}
	if (!heap && use_server_heap) {
		/* Use pre-allocated release heap */
		rh = server->release_heap;
		rh->domain = rt;
		rh->release_time = release_time;

		/* Add to release queue */
		list_add(&rh->list, pos->prev);
		heap = rh;
	}
	return heap;
}

/*
 * Prepare a server's release_heap for use.
 */
static int reinit_release_heap(server_t *server)
{
	int rv = 0;
	server_release_heap_t* rh;

	/* Use pre-allocated release heap */
	rh = server->release_heap;

	/* WARNING: If the CPU still holds the release_lock at this point,
	 *          deadlock may occur!
	 */
	rv = hrtimer_try_to_cancel(&rh->timer);

	/* The timer callback is running, it is useless to add
	 * to the release heap now.
	 */
	if (rv == -1) {
		rv = 0;
		goto out;
	}

	/* Under no cirumstances should the timer have been active
	 * but not running.
	 */
	/* TODO: stop living dangerously */
	//BUG_ON(rv == 1);
	rv = 1;

	/* initialize */
	INIT_LIST_HEAD(&rh->servers);
	atomic_set(&rh->info.state, HRTIMER_START_ON_INACTIVE);
 out:
	return rv;
}

/*
 * Arm the release timer for the next set of servers.
 */
static int arm_release_timer(server_domain_t *domain)
{
	int rv = 1;
	struct list_head list;
	struct list_head *pos, *safe;
	server_t *server;
	server_release_heap_t *rh;

	_TRACE_SUB("arm_release_timer() at %llu", litmus_clock());
	list_replace_init(&domain->tobe_released, &list);

	list_for_each_safe(pos, safe, &list) {
		/* Pick server from work list */
		server = list_entry(pos, server_t, release_list);
		list_del(pos);

		/* Put into release heap while holding release_lock */
		raw_spin_lock(&domain->release_lock);
		TRACE_SUB(server, "I have the release_lock");

		rh = get_release_heap(domain, server, 0);
		if (!rh) {
			/* Need to use our own, but drop lock first */
			raw_spin_unlock(&domain->release_lock);
			TRACE_SUB(server, "Dropped release_lock");

			rv = reinit_release_heap(server);

			/* Bail! We missed the release time */
			if (!rv) {
				TRACE_SUB(server, "missed release");
				rv = 0;
				goto out;
			}

			TRACE_SUB(server, "release_heap ready");

			raw_spin_lock(&domain->release_lock);
			TRACE_SUB(server, "Re-acquired release_lock");

			rh = get_release_heap(domain, server, 1);
		}

		list_add(&server->release_list, &rh->servers);
		TRACE_SUB(server, "arm_release_timer(): added to release heap");

		raw_spin_unlock(&domain->release_lock);
		TRACE_SUB(server, "Returned the release_lock");

		/* To avoid arming the timer multiple times, we only let the
		 * owner do the arming (which is the "first" task to reference
		 * this release_heap anyway).
		 */
		if (rh == server->release_heap) {
			TRACE_SUB(server, "arming timer 0x%p at %llu on P%d",
				  &rh->timer, rh->release_time,
				  domain->release_master);
			/* We cannot arm the timer using hrtimer_start()
			 * as it may deadlock on rq->lock
			 *
			 * PINNED mode is ok on both local and remote CPU
			 */
			if (domain->release_master == NO_CPU) {
				__hrtimer_start_range_ns(&rh->timer,
						ns_to_ktime(rh->release_time),
						0, HRTIMER_MODE_ABS_PINNED, 0);
			} else {
				hrtimer_start_on(domain->release_master,
					&rh->info, &rh->timer,
					ns_to_ktime(rh->release_time),
					HRTIMER_MODE_ABS_PINNED);
			}
		} else
			TRACE_SUB(server, "0x%p is not my timer", &rh->timer);
	}
 out:
	return rv;
}

int add_server_release(server_t *server, server_domain_t *domain)
{
	TRACE_SUB(server, "adding to release at %llu", server->release);
	list_add(&server->release_list, &domain->tobe_released);
	return arm_release_timer(domain);
}

static void init_servers(void)
{
	server_cache = KMEM_CACHE(server, SLAB_PANIC);
	server_release_cache = KMEM_CACHE(server_release_heap, SLAB_PANIC);
}

static void exit_servers(void)
{
	kmem_cache_destroy(server_cache);
	kmem_cache_destroy(server_release_cache);
}


module_init(init_servers);
module_exit(exit_servers);