src/rt/rust_scheduler.cpp - rust - Git at Google


 #include <stdarg.h>
 #include <cassert>
 #include <pthread.h>
 #include "rust_internal.h"
 #include "rust_util.h"
 #include "globals.h"

 #ifndef _WIN32
 pthread_key_t rust_scheduler::task_key;
 #else
 DWORD rust_scheduler::task_key;
 #endif

 bool rust_scheduler::tls_initialized = false;

 rust_scheduler::rust_scheduler(rust_kernel *kernel,
                                rust_srv *srv,
                                int id) :
     ref_count(1),
     interrupt_flag(0),
     _log(srv, this),
     log_lvl(log_debug),
     srv(srv),
     // TODO: calculate a per scheduler name.
     name("main"),
     newborn_tasks(this, "newborn"),
     running_tasks(this, "running"),
     blocked_tasks(this, "blocked"),
     dead_tasks(this, "dead"),
     cache(this),
     kernel(kernel),
     id(id),
     min_stack_size(kernel->env->min_stack_size),
     env(kernel->env)
 {
     LOGPTR(this, "new dom", (uintptr_t)this);
     isaac_init(this, &rctx);
 #ifndef __WIN32__
     pthread_attr_init(&attr);
     pthread_attr_setstacksize(&attr, 1024 * 1024);
     pthread_attr_setdetachstate(&attr, true);
 #endif

     if (!tls_initialized)
         init_tls();
 }

 rust_scheduler::~rust_scheduler() {
     DLOG(this, dom, "~rust_scheduler %s @0x%" PRIxPTR, name, (uintptr_t)this);

     newborn_tasks.delete_all();
     running_tasks.delete_all();
     blocked_tasks.delete_all();
     dead_tasks.delete_all();
 #ifndef __WIN32__
     pthread_attr_destroy(&attr);
 #endif
 }

 void
 rust_scheduler::activate(rust_task *task) {
     task->ctx.next = &c_context;
     DLOG(this, task, "descheduling...");
     lock.unlock();
     task->ctx.swap(c_context);
     lock.lock();
     DLOG(this, task, "task has returned");
 }

 void
 rust_scheduler::log(rust_task* task, uint32_t level, char const *fmt, ...) {
     char buf[BUF_BYTES];
     va_list args;
     va_start(args, fmt);
     vsnprintf(buf, sizeof(buf), fmt, args);
     _log.trace_ln(task, level, buf);
     va_end(args);
 }

 void
 rust_scheduler::fail() {
     log(NULL, log_err, "domain %s @0x%" PRIxPTR " root task failed",
         name, this);
     kernel->fail();
 }

 void
 rust_scheduler::kill_all_tasks() {
     I(this, !lock.lock_held_by_current_thread());
     scoped_lock with(lock);

     for (size_t i = 0; i < running_tasks.length(); i++) {
         // We don't want the failure of these tasks to propagate back
         // to the kernel again since we're already failing everything
         running_tasks[i]->unsupervise();
         running_tasks[i]->kill();
     }

     for (size_t i = 0; i < blocked_tasks.length(); i++) {
         blocked_tasks[i]->unsupervise();
         blocked_tasks[i]->kill();
     }
 }

 size_t
 rust_scheduler::number_of_live_tasks() {
     return running_tasks.length() + blocked_tasks.length();
 }

 /**
  * Delete any dead tasks.
  */
 void
 rust_scheduler::reap_dead_tasks(int id) {
     I(this, lock.lock_held_by_current_thread());
     if (dead_tasks.length() == 0) {
         return;
     }

     // First make a copy of the dead_task list with the lock held
     size_t dead_tasks_len = dead_tasks.length();
     rust_task **dead_tasks_copy = (rust_task**)
         srv->malloc(sizeof(rust_task*) * dead_tasks_len);
     for (size_t i = 0; i < dead_tasks_len; ++i) {
         rust_task *task = dead_tasks[i];
         dead_tasks_copy[i] = task;
     }

     // Now drop the lock and futz with the tasks. This avoids establishing
     // a sched->lock then task->lock locking order, which would be devestating
     // to performance.
     lock.unlock();

     for (size_t i = 0; i < dead_tasks_len; ++i) {
         rust_task *task = dead_tasks_copy[i];
         task->lock.lock();
         // Make sure this task isn't still running somewhere else...
         if (task->can_schedule(id)) {
             DLOG(this, task,
                 "deleting unreferenced dead task %s @0x%" PRIxPTR,
                 task->name, task);
             task->lock.unlock();
         } else {
             task->lock.unlock();
             dead_tasks_copy[i] = NULL;
         }
     }

     // Now grab the lock again and remove the tasks that were truly dead
     lock.lock();

     for (size_t i = 0; i < dead_tasks_len; ++i) {
         rust_task *task = dead_tasks_copy[i];
         if (task) {
             dead_tasks.remove(task);
         }
     }

     // Now unlock again because we have to actually free the dead tasks,
     // and that may end up wanting to lock the task and sched locks
     // again (via target->send)
     lock.unlock();

     for (size_t i = 0; i < dead_tasks_len; ++i) {
         rust_task *task = dead_tasks_copy[i];
         if (task) {
             task->deref();
             sync::decrement(kernel->live_tasks);
             kernel->wakeup_schedulers();
         }
     }
     srv->free(dead_tasks_copy);

     lock.lock();
 }

 /**
  * Schedules a running task for execution. Only running tasks can be
  * activated.  Blocked tasks have to be unblocked before they can be
  * activated.
  *
  * Returns NULL if no tasks can be scheduled.
  */
 rust_task *
 rust_scheduler::schedule_task(int id) {
     I(this, this);
     // FIXME: in the face of failing tasks, this is not always right.
     // I(this, n_live_tasks() > 0);
     if (running_tasks.length() > 0) {
         size_t k = isaac_rand(&rctx);
         // Look around for a runnable task, starting at k.
         for(size_t j = 0; j < running_tasks.length(); ++j) {
             size_t  i = (j + k) % running_tasks.length();
             if (running_tasks[i]->can_schedule(id)) {
                 return (rust_task *)running_tasks[i];
             }
         }
     }
     return NULL;
 }

 void
 rust_scheduler::log_state() {
     if (log_rt_task < log_debug) return;

     if (!running_tasks.is_empty()) {
         log(NULL, log_debug, "running tasks:");
         for (size_t i = 0; i < running_tasks.length(); i++) {
             log(NULL, log_debug, "\t task: %s @0x%" PRIxPTR
                 " remaining: %" PRId64 " us",
                 running_tasks[i]->name,
                 running_tasks[i],
                 running_tasks[i]->yield_timer.remaining_us());
         }
     }

     if (!blocked_tasks.is_empty()) {
         log(NULL, log_debug, "blocked tasks:");
         for (size_t i = 0; i < blocked_tasks.length(); i++) {
             log(NULL, log_debug, "\t task: %s @0x%" PRIxPTR ", blocked on: 0x%"
                 PRIxPTR " '%s'",
                 blocked_tasks[i]->name, blocked_tasks[i],
                 blocked_tasks[i]->cond, blocked_tasks[i]->cond_name);
         }
     }

     if (!dead_tasks.is_empty()) {
         log(NULL, log_debug, "dead tasks:");
         for (size_t i = 0; i < dead_tasks.length(); i++) {
             log(NULL, log_debug, "\t task: %s 0x%" PRIxPTR,
                 dead_tasks[i]->name, dead_tasks[i]);
         }
     }
 }
 /**
  * Starts the main scheduler loop which performs task scheduling for this
  * domain.
  *
  * Returns once no more tasks can be scheduled and all task ref_counts
  * drop to zero.
  */
 void
 rust_scheduler::start_main_loop() {
     lock.lock();

     // Make sure someone is watching, to pull us out of infinite loops.
     //
     // FIXME: time-based interruption is not presently working; worked
     // in rustboot and has been completely broken in rustc.
     //
     // rust_timer timer(this);

     DLOG(this, dom, "started domain loop %d", id);

     while (kernel->live_tasks > 0) {
         A(this, kernel->is_deadlocked() == false, "deadlock");

         DLOG(this, dom, "worker %d, number_of_live_tasks = %d, total = %d",
              id, number_of_live_tasks(), kernel->live_tasks);

         rust_task *scheduled_task = schedule_task(id);

         if (scheduled_task == NULL) {
             log_state();
             DLOG(this, task,
                  "all tasks are blocked, scheduler id %d yielding ...",
                  id);
             lock.timed_wait(10);
             reap_dead_tasks(id);
             DLOG(this, task,
                  "scheduler %d resuming ...", id);
             continue;
         }

         I(this, scheduled_task->running());

         DLOG(this, task,
              "activating task %s 0x%" PRIxPTR
              ", sp=0x%" PRIxPTR
              ", state: %s",
              scheduled_task->name,
              (uintptr_t)scheduled_task,
              scheduled_task->user.rust_sp,
              scheduled_task->state->name);

         place_task_in_tls(scheduled_task);

         interrupt_flag = 0;

         DLOG(this, task,
              "Running task %p on worker %d",
              scheduled_task, id);
         scheduled_task->running_on = id;
         activate(scheduled_task);
         scheduled_task->running_on = -1;

         DLOG(this, task,
              "returned from task %s @0x%" PRIxPTR
              " in state '%s', sp=0x%x, worker id=%d" PRIxPTR,
              scheduled_task->name,
              (uintptr_t)scheduled_task,
              scheduled_task->state->name,
              scheduled_task->user.rust_sp,
              id);

         reap_dead_tasks(id);
     }

     DLOG(this, dom,
          "terminated scheduler loop, reaping dead tasks ...");

     while (dead_tasks.length() > 0) {
         DLOG(this, dom,
              "waiting for %d dead tasks to become dereferenced, "
              "scheduler yielding ...",
              dead_tasks.length());
         log_state();
         lock.unlock();
         sync::yield();
         lock.lock();
         reap_dead_tasks(id);
     }

     DLOG(this, dom, "finished main-loop %d", id);

     lock.unlock();
 }

 rust_crate_cache *
 rust_scheduler::get_cache() {
     return &cache;
 }

 rust_task *
 rust_scheduler::create_task(rust_task *spawner, const char *name) {
     rust_task *task =
         new (this->kernel, "rust_task")
         rust_task (this, &newborn_tasks, spawner, name);
     DLOG(this, task, "created task: " PTR ", spawner: %s, name: %s",
                         task, spawner ? spawner->name : "null", name);
     if(spawner) {
         task->pin(spawner->pinned_on);
     }

     {
         scoped_lock with(lock);
         newborn_tasks.append(task);
     }

     sync::increment(kernel->live_tasks);

     return task;
 }

 void rust_scheduler::run() {
     this->start_main_loop();
 }

 #ifndef _WIN32
 void
 rust_scheduler::init_tls() {
     int result = pthread_key_create(&task_key, NULL);
     assert(!result && "Couldn't create the TLS key!");
     tls_initialized = true;
 }

 void
 rust_scheduler::place_task_in_tls(rust_task *task) {
     int result = pthread_setspecific(task_key, task);
     assert(!result && "Couldn't place the task in TLS!");
     task->record_stack_limit();
 }

 rust_task *
 rust_scheduler::get_task() {
     if (!tls_initialized)
         return NULL;
     rust_task *task = reinterpret_cast<rust_task *>
         (pthread_getspecific(task_key));
     assert(task && "Couldn't get the task from TLS!");
     return task;
 }
 #else
 void
 rust_scheduler::init_tls() {
     task_key = TlsAlloc();
     assert(task_key != TLS_OUT_OF_INDEXES && "Couldn't create the TLS key!");
     tls_initialized = true;
 }

 void
 rust_scheduler::place_task_in_tls(rust_task *task) {
     BOOL result = TlsSetValue(task_key, task);
     assert(result && "Couldn't place the task in TLS!");
     task->record_stack_limit();
 }

 rust_task *
 rust_scheduler::get_task() {
     if (!tls_initialized)
         return NULL;
     rust_task *task = reinterpret_cast<rust_task *>(TlsGetValue(task_key));
     assert(task && "Couldn't get the task from TLS!");
     return task;
 }
 #endif

 //
 // Local Variables:
 // mode: C++
 // fill-column: 70;
 // indent-tabs-mode: nil
 // c-basic-offset: 4
 // buffer-file-coding-system: utf-8-unix
 // End:
 //

	#include <stdarg.h>
	#include <cassert>
	#include <pthread.h>
	#include "rust_internal.h"
	#include "rust_util.h"
	#include "globals.h"

	#ifndef _WIN32
	pthread_key_t rust_scheduler::task_key;
	#else
	DWORD rust_scheduler::task_key;
	#endif

	bool rust_scheduler::tls_initialized = false;

	rust_scheduler::rust_scheduler(rust_kernel *kernel,
	rust_srv *srv,
	int id) :
	ref_count(1),
	interrupt_flag(0),
	_log(srv, this),
	log_lvl(log_debug),
	srv(srv),
	// TODO: calculate a per scheduler name.
	name("main"),
	newborn_tasks(this, "newborn"),
	running_tasks(this, "running"),
	blocked_tasks(this, "blocked"),
	dead_tasks(this, "dead"),
	cache(this),
	kernel(kernel),
	id(id),
	min_stack_size(kernel->env->min_stack_size),
	env(kernel->env)
	{
	LOGPTR(this, "new dom", (uintptr_t)this);
	isaac_init(this, &rctx);
	#ifndef __WIN32__
	pthread_attr_init(&attr);
	pthread_attr_setstacksize(&attr, 1024 * 1024);
	pthread_attr_setdetachstate(&attr, true);
	#endif

	if (!tls_initialized)
	init_tls();
	}

	rust_scheduler::~rust_scheduler() {
	DLOG(this, dom, "~rust_scheduler %s @0x%" PRIxPTR, name, (uintptr_t)this);

	newborn_tasks.delete_all();
	running_tasks.delete_all();
	blocked_tasks.delete_all();
	dead_tasks.delete_all();
	#ifndef __WIN32__
	pthread_attr_destroy(&attr);
	#endif
	}

	void
	rust_scheduler::activate(rust_task *task) {
	task->ctx.next = &c_context;
	DLOG(this, task, "descheduling...");
	lock.unlock();
	task->ctx.swap(c_context);
	lock.lock();
	DLOG(this, task, "task has returned");
	}

	void
	rust_scheduler::log(rust_task* task, uint32_t level, char const *fmt, ...) {
	char buf[BUF_BYTES];
	va_list args;
	va_start(args, fmt);
	vsnprintf(buf, sizeof(buf), fmt, args);
	_log.trace_ln(task, level, buf);
	va_end(args);
	}

	void
	rust_scheduler::fail() {
	log(NULL, log_err, "domain %s @0x%" PRIxPTR " root task failed",
	name, this);
	kernel->fail();
	}

	void
	rust_scheduler::kill_all_tasks() {
	I(this, !lock.lock_held_by_current_thread());
	scoped_lock with(lock);

	for (size_t i = 0; i < running_tasks.length(); i++) {
	// We don't want the failure of these tasks to propagate back
	// to the kernel again since we're already failing everything
	running_tasks[i]->unsupervise();
	running_tasks[i]->kill();
	}

	for (size_t i = 0; i < blocked_tasks.length(); i++) {
	blocked_tasks[i]->unsupervise();
	blocked_tasks[i]->kill();
	}
	}

	size_t
	rust_scheduler::number_of_live_tasks() {
	return running_tasks.length() + blocked_tasks.length();
	}

	/**
	* Delete any dead tasks.
	*/
	void
	rust_scheduler::reap_dead_tasks(int id) {
	I(this, lock.lock_held_by_current_thread());
	if (dead_tasks.length() == 0) {
	return;
	}

	// First make a copy of the dead_task list with the lock held
	size_t dead_tasks_len = dead_tasks.length();
	rust_task dead_tasks_copy = (rust_task)
	srv->malloc(sizeof(rust_task) dead_tasks_len);
	for (size_t i = 0; i < dead_tasks_len; ++i) {
	rust_task *task = dead_tasks[i];
	dead_tasks_copy[i] = task;
	}

	// Now drop the lock and futz with the tasks. This avoids establishing
	// a sched->lock then task->lock locking order, which would be devestating
	// to performance.
	lock.unlock();

	for (size_t i = 0; i < dead_tasks_len; ++i) {
	rust_task *task = dead_tasks_copy[i];
	task->lock.lock();
	// Make sure this task isn't still running somewhere else...
	if (task->can_schedule(id)) {
	DLOG(this, task,
	"deleting unreferenced dead task %s @0x%" PRIxPTR,
	task->name, task);
	task->lock.unlock();
	} else {
	task->lock.unlock();
	dead_tasks_copy[i] = NULL;
	}
	}

	// Now grab the lock again and remove the tasks that were truly dead
	lock.lock();

	for (size_t i = 0; i < dead_tasks_len; ++i) {
	rust_task *task = dead_tasks_copy[i];
	if (task) {
	dead_tasks.remove(task);
	}
	}

	// Now unlock again because we have to actually free the dead tasks,
	// and that may end up wanting to lock the task and sched locks
	// again (via target->send)
	lock.unlock();

	for (size_t i = 0; i < dead_tasks_len; ++i) {
	rust_task *task = dead_tasks_copy[i];
	if (task) {
	task->deref();
	sync::decrement(kernel->live_tasks);
	kernel->wakeup_schedulers();
	}
	}
	srv->free(dead_tasks_copy);

	lock.lock();
	}

	/**
	* Schedules a running task for execution. Only running tasks can be
	* activated. Blocked tasks have to be unblocked before they can be
	* activated.
	*
	* Returns NULL if no tasks can be scheduled.
	*/
	rust_task *
	rust_scheduler::schedule_task(int id) {
	I(this, this);
	// FIXME: in the face of failing tasks, this is not always right.
	// I(this, n_live_tasks() > 0);
	if (running_tasks.length() > 0) {
	size_t k = isaac_rand(&rctx);
	// Look around for a runnable task, starting at k.
	for(size_t j = 0; j < running_tasks.length(); ++j) {
	size_t i = (j + k) % running_tasks.length();
	if (running_tasks[i]->can_schedule(id)) {
	return (rust_task *)running_tasks[i];
	}
	}
	}
	return NULL;
	}

	void
	rust_scheduler::log_state() {
	if (log_rt_task < log_debug) return;

	if (!running_tasks.is_empty()) {
	log(NULL, log_debug, "running tasks:");
	for (size_t i = 0; i < running_tasks.length(); i++) {
	log(NULL, log_debug, "\t task: %s @0x%" PRIxPTR
	" remaining: %" PRId64 " us",
	running_tasks[i]->name,
	running_tasks[i],
	running_tasks[i]->yield_timer.remaining_us());
	}
	}

	if (!blocked_tasks.is_empty()) {
	log(NULL, log_debug, "blocked tasks:");
	for (size_t i = 0; i < blocked_tasks.length(); i++) {
	log(NULL, log_debug, "\t task: %s @0x%" PRIxPTR ", blocked on: 0x%"
	PRIxPTR " '%s'",
	blocked_tasks[i]->name, blocked_tasks[i],
	blocked_tasks[i]->cond, blocked_tasks[i]->cond_name);
	}
	}

	if (!dead_tasks.is_empty()) {
	log(NULL, log_debug, "dead tasks:");
	for (size_t i = 0; i < dead_tasks.length(); i++) {
	log(NULL, log_debug, "\t task: %s 0x%" PRIxPTR,
	dead_tasks[i]->name, dead_tasks[i]);
	}
	}
	}
	/**
	* Starts the main scheduler loop which performs task scheduling for this
	* domain.
	*
	* Returns once no more tasks can be scheduled and all task ref_counts
	* drop to zero.
	*/
	void
	rust_scheduler::start_main_loop() {
	lock.lock();

	// Make sure someone is watching, to pull us out of infinite loops.
	//
	// FIXME: time-based interruption is not presently working; worked
	// in rustboot and has been completely broken in rustc.
	//
	// rust_timer timer(this);

	DLOG(this, dom, "started domain loop %d", id);

	while (kernel->live_tasks > 0) {
	A(this, kernel->is_deadlocked() == false, "deadlock");

	DLOG(this, dom, "worker %d, number_of_live_tasks = %d, total = %d",
	id, number_of_live_tasks(), kernel->live_tasks);

	rust_task *scheduled_task = schedule_task(id);

	if (scheduled_task == NULL) {
	log_state();
	DLOG(this, task,
	"all tasks are blocked, scheduler id %d yielding ...",
	id);
	lock.timed_wait(10);
	reap_dead_tasks(id);
	DLOG(this, task,
	"scheduler %d resuming ...", id);
	continue;
	}

	I(this, scheduled_task->running());

	DLOG(this, task,
	"activating task %s 0x%" PRIxPTR
	", sp=0x%" PRIxPTR
	", state: %s",
	scheduled_task->name,
	(uintptr_t)scheduled_task,
	scheduled_task->user.rust_sp,
	scheduled_task->state->name);

	place_task_in_tls(scheduled_task);

	interrupt_flag = 0;

	DLOG(this, task,
	"Running task %p on worker %d",
	scheduled_task, id);
	scheduled_task->running_on = id;
	activate(scheduled_task);
	scheduled_task->running_on = -1;

	DLOG(this, task,
	"returned from task %s @0x%" PRIxPTR
	" in state '%s', sp=0x%x, worker id=%d" PRIxPTR,
	scheduled_task->name,
	(uintptr_t)scheduled_task,
	scheduled_task->state->name,
	scheduled_task->user.rust_sp,
	id);

	reap_dead_tasks(id);
	}

	DLOG(this, dom,
	"terminated scheduler loop, reaping dead tasks ...");

	while (dead_tasks.length() > 0) {
	DLOG(this, dom,
	"waiting for %d dead tasks to become dereferenced, "
	"scheduler yielding ...",
	dead_tasks.length());
	log_state();
	lock.unlock();
	sync::yield();
	lock.lock();
	reap_dead_tasks(id);
	}

	DLOG(this, dom, "finished main-loop %d", id);

	lock.unlock();
	}

	rust_crate_cache *
	rust_scheduler::get_cache() {
	return &cache;
	}

	rust_task *
	rust_scheduler::create_task(rust_task spawner, const char name) {
	rust_task *task =
	new (this->kernel, "rust_task")
	rust_task (this, &newborn_tasks, spawner, name);
	DLOG(this, task, "created task: " PTR ", spawner: %s, name: %s",
	task, spawner ? spawner->name : "null", name);
	if(spawner) {
	task->pin(spawner->pinned_on);
	}

	{
	scoped_lock with(lock);
	newborn_tasks.append(task);
	}

	sync::increment(kernel->live_tasks);

	return task;
	}

	void rust_scheduler::run() {
	this->start_main_loop();
	}

	#ifndef _WIN32
	void
	rust_scheduler::init_tls() {
	int result = pthread_key_create(&task_key, NULL);
	assert(!result && "Couldn't create the TLS key!");
	tls_initialized = true;
	}

	void
	rust_scheduler::place_task_in_tls(rust_task *task) {
	int result = pthread_setspecific(task_key, task);
	assert(!result && "Couldn't place the task in TLS!");
	task->record_stack_limit();
	}

	rust_task *
	rust_scheduler::get_task() {
	if (!tls_initialized)
	return NULL;
	rust_task task = reinterpret_cast<rust_task >
	(pthread_getspecific(task_key));
	assert(task && "Couldn't get the task from TLS!");
	return task;
	}
	#else
	void
	rust_scheduler::init_tls() {
	task_key = TlsAlloc();
	assert(task_key != TLS_OUT_OF_INDEXES && "Couldn't create the TLS key!");
	tls_initialized = true;
	}

	void
	rust_scheduler::place_task_in_tls(rust_task *task) {
	BOOL result = TlsSetValue(task_key, task);
	assert(result && "Couldn't place the task in TLS!");
	task->record_stack_limit();
	}

	rust_task *
	rust_scheduler::get_task() {
	if (!tls_initialized)
	return NULL;
	rust_task task = reinterpret_cast<rust_task >(TlsGetValue(task_key));
	assert(task && "Couldn't get the task from TLS!");
	return task;
	}
	#endif

	//
	// Local Variables:
	// mode: C++
	// fill-column: 70;
	// indent-tabs-mode: nil
	// c-basic-offset: 4
	// buffer-file-coding-system: utf-8-unix
	// End:
	//