<commit_before>//
// XXX
//  - workers should have a uwq
//  - pin workers

#include "types.h"
#include "amd64.h"
#include "kernel.hh"
#include "cpu.hh"
#include "gc.hh"
#include "percpu.hh"
#include "spinlock.h"
#include "condvar.h"
#include "proc.hh"
#include "uwq.hh"
#include "vm.hh"
#include "kalloc.hh"
#include "bits.hh"
extern "C" {
#include "kern_c.h"
}

int
uwq_trywork(void)
{
  u64 i, k;

  // A "random" victim CPU
  k = rdtsc();
  for (i = 0; i < NCPU; i++) {
    u64 j = (i+k) % NCPU;

    if (j == mycpuid())
      continue;
    struct cpu *c = &cpus[j];

    scoped_gc_epoch xgc();
    barrier();
    struct proc *p = c->proc;
    if (p == nullptr || p->uwq == nullptr)
      continue;
    uwq* uwq = p->uwq;

    if (uwq->haswork()) {
      if (uwq->trywork())
        return 1;
      // XXX(sbw) start a worker thread..
      break;
    }
  }

  return 0;
}

long
sys_wqwait(void)
{
  uwq_worker* w = myproc()->worker;
  if (w == nullptr)
    return -1;

  return w->wait();
}

//
// uwq_worker
//
long
uwq_worker::wait(void)
{
  acquire(&lock);
  uwq->tryexit(this);
  cv_sleep(&cv, &lock);
  uwq->tryexit(this);
  release(&lock);
  return 0;
}

//
// uwq
//
uwq*
uwq::alloc(vmap* vmap, filetable *ftable)
{
  padded_length* len;
  uwq* u;

  len = (padded_length*) ksalloc(slab_userwq);  
  if (len == nullptr)
    return nullptr;

  ftable->incref();
  vmap->incref();
    
  u = new uwq(vmap, ftable, len);
  if (u == nullptr) {
    ftable->decref();
    vmap->decref();
    ksfree(slab_userwq, len);
    return nullptr;
  }
  u->inc();

  if (mapkva(vmap->pml4, (char*)len, USERWQ, USERWQSIZE)) {
    ftable->decref();
    vmap->decref();
    ksfree(slab_userwq, len);
    u->dec();
    return nullptr;
  }

  return u;
}

uwq::uwq(vmap* vmap, filetable *ftable, padded_length *len) 
  : rcu_freed("uwq"),
    vmap_(vmap), ftable_(ftable), len_(len),
    uentry_(0), ustack_(UWQSTACK), uref_(0)
{
  for (int i = 0; i < NCPU; i++)
    len_[i].v_ = 0;

  initlock(&lock_, "uwq_lock", 0);

  // XXX(sbw) move to uwq_worker constructor
  for (int i = 0; i < NCPU; i++) {
    initlock(&worker_[i].lock, "worker_lock", 0);
    initcondvar(&worker_[i].cv, "worker_cv");
    worker_[i].uwq = this;
    worker_[i].running = false;
    worker_[i].proc = nullptr;
  }
}

uwq::~uwq(void)
{ 
  if (len_ != nullptr)
    ksfree(slab_userwq, len_);
  vmap_->decref();
  ftable_->decref();
}

void
uwq::tryexit(uwq_worker *w)
{
  if (ref() == 0) {
    if (--uref_ == 0)
      gc_delayed(this);
    release(&w->lock);
    w->proc = nullptr;
    exit();
  }
}

bool
uwq::haswork(void)
{
  if (len_ == nullptr)
    return false;

  for (int i = 0; i < NCPU; i++) {
    if (len_[i].v_ > 0) {
      return true;
    }
  }
  return false;
}

bool
uwq::trywork(void)
{
  scoped_acquire lock0(&lock_);

  if (ref() == 0)
    return false;

  int slot = -1;
  for (int i = 0; i < NCPU; i++) {
    if (worker_[i].running)
      continue;
    else if (worker_[i].proc != nullptr) {
      scoped_acquire lock1(&worker_[i].lock);
      proc* p = worker_[i].proc;

      panic("uwq::trywork: untested");
      acquire(&p->lock);
      p->cpuid = mycpuid();
      release(&p->lock);

      worker_[i].running = true;
      cv_wakeup(&worker_[i].cv);
      return true;
    } else if (slot == -1) {
      slot = i;
    }
  }

  if (slot != -1) {
    proc* p = allocworker();
    if (p != nullptr) {
      ++uref_;
      p->worker = &worker_[slot];
      worker_[slot].proc = p;
      worker_[slot].running = true;
<commit_msg>Oops, clear running flag when a worker waits<commit_after>//
// XXX
//  - workers should have a uwq
//  - pin workers

#include "types.h"
#include "amd64.h"
#include "kernel.hh"
#include "cpu.hh"
#include "gc.hh"
#include "percpu.hh"
#include "spinlock.h"
#include "condvar.h"
#include "proc.hh"
#include "uwq.hh"
#include "vm.hh"
#include "kalloc.hh"
#include "bits.hh"
extern "C" {
#include "kern_c.h"
}

int
uwq_trywork(void)
{
  u64 i, k;

  // A "random" victim CPU
  k = rdtsc();
  for (i = 0; i < NCPU; i++) {
    u64 j = (i+k) % NCPU;

    if (j == mycpuid())
      continue;
    struct cpu *c = &cpus[j];

    scoped_gc_epoch xgc();
    barrier();
    struct proc *p = c->proc;
    if (p == nullptr || p->uwq == nullptr)
      continue;
    uwq* uwq = p->uwq;

    if (uwq->haswork()) {
      if (uwq->trywork())
        return 1;
      // XXX(sbw) start a worker thread..
      break;
    }
  }

  return 0;
}

long
sys_wqwait(void)
{
  uwq_worker* w = myproc()->worker;
  if (w == nullptr)
    return -1;

  return w->wait();
}

//
// uwq_worker
//
long
uwq_worker::wait(void)
{
  acquire(&lock);
  uwq->tryexit(this);
  running = false;
  cv_sleep(&cv, &lock);
  uwq->tryexit(this);
  release(&lock);
  return 0;
}

//
// uwq
//
uwq*
uwq::alloc(vmap* vmap, filetable *ftable)
{
  padded_length* len;
  uwq* u;

  len = (padded_length*) ksalloc(slab_userwq);  
  if (len == nullptr)
    return nullptr;

  ftable->incref();
  vmap->incref();
    
  u = new uwq(vmap, ftable, len);
  if (u == nullptr) {
    ftable->decref();
    vmap->decref();
    ksfree(slab_userwq, len);
    return nullptr;
  }
  u->inc();

  if (mapkva(vmap->pml4, (char*)len, USERWQ, USERWQSIZE)) {
    ftable->decref();
    vmap->decref();
    ksfree(slab_userwq, len);
    u->dec();
    return nullptr;
  }

  return u;
}

uwq::uwq(vmap* vmap, filetable *ftable, padded_length *len) 
  : rcu_freed("uwq"),
    vmap_(vmap), ftable_(ftable), len_(len),
    uentry_(0), ustack_(UWQSTACK), uref_(0)
{
  for (int i = 0; i < NCPU; i++)
    len_[i].v_ = 0;

  initlock(&lock_, "uwq_lock", 0);

  // XXX(sbw) move to uwq_worker constructor
  for (int i = 0; i < NCPU; i++) {
    initlock(&worker_[i].lock, "worker_lock", 0);
    initcondvar(&worker_[i].cv, "worker_cv");
    worker_[i].uwq = this;
    worker_[i].running = false;
    worker_[i].proc = nullptr;
  }
}

uwq::~uwq(void)
{ 
  if (len_ != nullptr)
    ksfree(slab_userwq, len_);
  vmap_->decref();
  ftable_->decref();
}

void
uwq::tryexit(uwq_worker *w)
{
  if (ref() == 0) {
    if (--uref_ == 0)
      gc_delayed(this);
    release(&w->lock);
    w->proc = nullptr;
    exit();
  }
}

bool
uwq::haswork(void)
{
  if (len_ == nullptr)
    return false;

  for (int i = 0; i < NCPU; i++) {
    if (len_[i].v_ > 0) {
      return true;
    }
  }
  return false;
}

bool
uwq::trywork(void)
{
  scoped_acquire lock0(&lock_);

  if (ref() == 0)
    return false;

  int slot = -1;
  for (int i = 0; i < NCPU; i++) {
    if (worker_[i].running)
      continue;
    else if (worker_[i].proc != nullptr) {
      scoped_acquire lock1(&worker_[i].lock);
      proc* p = worker_[i].proc;

      //cprintf("uwq::trywork: untested\n");
      acquire(&p->lock);
      p->cpuid = mycpuid();
      release(&p->lock);

      worker_[i].running = true;
      cv_wakeup(&worker_[i].cv);
      return true;
    } else if (slot == -1) {
      slot = i;
    }
  }

  if (slot != -1) {
    proc* p = allocworker();
    if (p != nullptr) {
      ++uref_;
      p->worker = &worker_[slot];
      worker_[slot].proc = p;
      worker_[slot].running = true;

{'fraction_non_alphanumeric': 0.19069111424541607, 'fraction_numerical': 0.00846262341325811, 'mean_word_length': 2.599492385786802, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2429270', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2955, 'n_words': 659}