void notify_all() const
    {
        spinlock_.lock();

        broadcasting_ = waiters_ > 0;

        if ( broadcasting_ )
        {
            ZI_VERIFY( win32::ReleaseSemaphore( semaphore_, waiters_, 0 ) );
            spinlock_.unlock();
            ZI_VERIFY_0( win32::WaitForSingleObject( last_event_, win32::forever ) );
            broadcasting_ = 0;
        }
        else
        {
            spinlock_.unlock();
        }
    }

 explicit scoped_lock( spinlock & sp ): sp_( sp )
 {
     sp.lock();
 }

void OSThread::suspend(spinlock& lock)
{
    lock.unlock();
    suspend();
}

/**
 * Body of the main profiler thread
 */
void profiler::profiler_thread(spinlock& l) {
  // Open the output file
  ofstream output;
  output.open(_output_filename, ios_base::app);
  output.rdbuf()->pubsetbuf(0, 0);
  output.setf(ios::fixed, ios::floatfield);
  output.precision(2);

  // Initialize the delay size RNG
  default_random_engine generator(get_time());
  uniform_int_distribution<size_t> delay_dist(0, ZeroSpeedupWeight + SpeedupDivisions);

  // Initialize the experiment duration
  size_t experiment_length = ExperimentMinTime;

  // Get the starting time for the profiler
  size_t start_time = get_time();

  // Log the start of this execution
  output << "startup\t"
         << "time=" << start_time << "\n";

  // Unblock the main thread
  l.unlock();

  // Wait until there is at least one progress point
  _throughput_points_lock.lock();
  _latency_points_lock.lock();
  while(_throughput_points.size() == 0 && _latency_points.size() == 0 && _running) {
    _throughput_points_lock.unlock();
    _latency_points_lock.unlock();
    wait(ExperimentCoolOffTime);
    _throughput_points_lock.lock();
    _latency_points_lock.lock();
  }
  _throughput_points_lock.unlock();
  _latency_points_lock.unlock();

  // Log sample counts after this many experiments (doubles each time)
  size_t sample_log_interval = 32;
  size_t sample_log_countdown = sample_log_interval;

  // Main experiment loop
  while(_running) {
    // Select a line
    line* selected;
    if(_fixed_line) {   // If this run has a fixed line, use it
      selected = _fixed_line;
    } else {            // Otherwise, wait for the next line to be selected
      selected = _next_line.load();
      while(_running && selected == nullptr) {
        wait(SamplePeriod * SampleBatchSize);
        selected = _next_line.load();
      }

      // If we're no longer running, exit the experiment loop
      if(!_running) break;

      _selected_line.store(selected);
    }

    // Choose a delay size
    size_t delay_size;
    if(_fixed_delay_size >= 0) {
      delay_size = _fixed_delay_size;
    } else {
      size_t r = delay_dist(generator);
      if(r <= ZeroSpeedupWeight) {
        delay_size = 0;
      } else {
        delay_size = (r - ZeroSpeedupWeight) * SamplePeriod / SpeedupDivisions;
      }
    }

    _delay_size.store(delay_size);

    // Save the starting time and sample count
    size_t start_time = get_time();
    size_t starting_samples = selected->get_samples();
    size_t starting_delay_time = _global_delay.load();
    
    // Was arrival speedup enabled?
    int arrival_speedup_percent;
    // Yes. Choose a random arrival speedup size using the same procedure as for line speedup size
    if(_enable_arrival_speedup) {
      size_t r = delay_dist(generator);
      if(r <= ZeroSpeedupWeight) {
        arrival_speedup_percent = 0;
      } else {
        arrival_speedup_percent = (r - ZeroSpeedupWeight) * 100 / SpeedupDivisions;
      }
    }

    // Save throughput point values at the start of the experiment
    vector<unique_ptr<throughput_point::saved>> saved_throughput_points;
    _throughput_points_lock.lock();
    for(pair<const std::string, throughput_point*>& p : _throughput_points) {
//.........这里部分代码省略.........

namespace exlib
{

#define FB_STK_ALIGN 256

void init_timer();

static bool s_service_inited;
static Service* s_service = NULL;

void Service::init(int32_t workers)
{
    if (!s_service)
    {
        static Service _srv(workers);
        s_service = &_srv;
        s_service->m_main.saveStackGuard();
        s_service->bindCurrent();
    }
}

Thread_base* Thread_base::current()
{
    if (!s_service_inited)
        return 0;

    OSThread* thread_ = OSThread::current();

    if (thread_ == 0)
        return 0;

    if (thread_->is(Service::type))
        return ((Service*)thread_)->running();
    return thread_;
}

Service *Service::current()
{
    OSThread* thread_ = OSThread::current();

    assert(s_service_inited);
    assert(thread_ != 0);
    assert(thread_->is(Service::type));

    return (Service*)thread_;
}

#ifdef DEBUG

static LockedList<linkitem> s_fibers;
static spinlock s_locker;

void Service::forEach(void (*func)(Fiber*))
{
    s_locker.lock();

    linkitem* p = s_fibers.head();

    while (p)
    {
        Fiber* zfb = 0;
        func((Fiber*)((intptr_t)p - (intptr_t)(&zfb->m_link)));

        p = s_fibers.next(p);
    }

    s_locker.unlock();
}

#endif

Service::Service() :
    m_master(s_service), m_main(this, NULL), m_running(&m_main), m_cb(NULL)
{
    m_main.set_name("main");
    m_main.Ref();
}

Service::Service(int32_t workers) :
    m_master(NULL), m_main(this, NULL), m_running(&m_main), m_cb(NULL), m_workers(workers - 1)
{
    m_main.set_name("main");
    m_main.Ref();

    if (!s_service_inited)
    {
        s_service_inited = true;
        init_timer();
    }
}

void Service::fiber_proc(void *(*func)(void *), Fiber *fb)
{
    class cb : public Service::switchConextCallback
    {
    public:
        cb(Fiber* fb) : m_fb(fb)
        {}

    public:
//.........这里部分代码省略.........

 /**
  * Checks in the message.  If the message was checked out for
  * writing, the buffer will be committed and returned to the
  * object_allocator if allow_simultaneous_rw is not set, the
  * buffer will be committing regardless of whether there is an
  * existing reader.  If it is set, the thread will wait until
  * there are no readers before committing the buffer.  Returns the
  * norm of the change of value. (using the norm() function) if its
  * a write checkin. Returns 0 otherwise.
  *
  * \warning writes may block for extended periods if there are
  * readers
  */
 double checkin(object_allocator_tls<F> &pool, 
                const F *msg, const factor_norm<F>& norm, 
                const bool allow_simultaneous_rw) {
   double residual = -1;
   lock.lock();
   // if it matches message address, this was a read request
   if (msg == &message) {
     assert(readercount > 0);
     readercount--;
   }
   //this was a write request
   else if (msg == writebuffer) {
     //Optional: Wait for readers to complete
     if (!allow_simultaneous_rw) {
       while(readercount > 0) {
         lock.unlock();
         sched_yield();
         lock.lock();
       }
     }
     // compute the delta
     residual = norm(message , *(writebuffer));
     message = *(writebuffer);
     pool.checkin(writebuffer);
     writebuffer = NULL;
     lastresidual = residual;
   }
   else {
     //Erroneous message!
     assert(0);
   }
   lock.unlock();
   return residual;
 }

// k_heap_compress - Merge free blocks
// This function "compresses" the linked list, consolidating as many adjacent free blocks as possible, to reduce fragmentation.
int k_heap_compress() {
    k_heap_blk* blk = heap.start;
    k_heap_blk* next = NULL;
    int blks_deleted = 0;
    
    __dynmem_lock.lock_cli();
    while((blk != NULL) && (blk->next != NULL)) { // Don't compress the first or last block in the heap.
        next = blk->next;
        if(blk->prev != NULL) {
            if(!blk->used && !blk->prev->used) { // If both the current block and the one before it are free...
                // ...then _delete_ this block. 
                k_heap_delete(blk);
                blks_deleted++;
            }
        }
        blk = next;
    }
    
    if((heap.end->prev != NULL) && (!heap.end->prev->used)) {
        k_heap_delete(heap.end);
        blks_deleted++;
    }
    __dynmem_lock.unlock_cli();
    return blks_deleted;
}

// kfree - free memory block
// This function frees a block of memory given by kmalloc(), allowing other kernel tasks to use it.
void kfree(char* ptr) {
    // given a pointer to a block of memory:
    // find the header for that block of memory
    // set the free bit
    // compress the list
    k_heap_blk *header_ptr = (k_heap_blk*)((size_t)(ptr-sizeof(k_heap_blk)-1));
    __dynmem_lock.lock();
#ifdef DYNMEM_CHECK_FREE_CALLS
    if(header_ptr->magic == HEAP_MAGIC_NUMBER) {
#endif
        header_ptr->used = false;
        if(header_ptr->prev != NULL) {
            if( !(header_ptr->prev->used) ) {
                // Just delete this block.
                k_heap_delete(header_ptr);
            } 
        }
        if(header_ptr->next != NULL) {
            if( !(header_ptr->next->used) ) {
                // Delete header_ptr->next.
                k_heap_blk *next = header_ptr->next;
                k_heap_delete(next);
            }
        }
        
        //k_heap_compress();
#ifdef DYNMEM_CHECK_FREE_CALLS
    } else {
        // We're freeing an invalid pointer.
        panic("dynmem: bad free() call -- could not find magic number\ndynmem: Pointer points to: 0x%x.\n", (unsigned long long int)((size_t)ptr) );
    }
#endif
    __dynmem_lock.unlock();
}

char* kmalloc(size_t size) {
    // iterate over every block in the heap list except for the last one
    // and look for a block where the space between the block and the next in the list is at least size bytes...
    // if we can't find one, extend the heap.
    int n_blks = (size / HEAP_BLK_SIZE)+1;
    char* ptr = NULL;
    k_heap_blk* blk = heap.start;
    __dynmem_lock.lock();
    while(blk->next != NULL) {
        if(!blk->used) {
            int blk_sz = ((size_t)blk->next - (size_t)blk);
            if( blk_sz-sizeof(k_heap_blk) >= size ) {
                blk->used = true;
                char* ptr = NULL;
                if( (blk_sz / HEAP_BLK_SIZE) > n_blks ) {
                    k_heap_add_at_offset(blk, n_blks);
                    blk->next->used = false;
                    ptr = (char*)((size_t)blk+sizeof(k_heap_blk)+1);
                } else {
                    ptr = (char*)((size_t)blk+sizeof(k_heap_blk)+1);
                }
                break;
            }
        }
        blk = blk->next;
    }
    if(ptr == NULL) { // if we still haven't allocated memory, then make a new block.
        k_heap_add_at_offset(heap.end, n_blks);
        heap.end->prev->used = true;
        ptr = (char*)((size_t)(heap.end->prev)+sizeof(k_heap_blk)+1);
    }
    __dynmem_lock.unlock();
    return ptr;
}

// terminal_writestring - print a string to screen
// this function prints a line of text to screen, wrapping and scrolling if necessary.
void terminal_writestring(char* data)
{
    __vga_write_lock.lock();
	size_t datalen = strlen(data);
	for ( size_t i = 0; i < datalen; i++ ) {
		terminal_putchar(data[i]);
    }
    __vga_write_lock.unlock();
}

        virtual void invoke()
        {
#ifdef DEBUG
            s_locker.lock();
            s_fibers.remove(&m_fb->m_link);
            s_locker.unlock();
#endif

            m_fb->m_joins.set();
            m_fb->Unref();
        }

void terminal_backspace() {
    __vga_write_lock.lock();
    if(--terminal_column > VGA_WIDTH) {
        if(--terminal_row > VGA_HEIGHT) {
            terminal_scroll(-1);
            terminal_row = 0;
        }
    }
    terminal_putentryat(' ', terminal_color, terminal_column, terminal_row);
    __vga_write_lock.unlock();
}

 bool find(data item) {
     size_t bucket = hashfn(item) % buckets.size();
     lock.lock();
     for (auto it = buckets[bucket].begin(); it != buckets[bucket].end(); it++) {
         if (*it == item) {
             lock.unlock();
             return true;
         }
     }
     lock.unlock();
     return false;
 }

void Service::Create(fiber_func func, void *data, int32_t stacksize, const char* name, Fiber** retVal)
{
    Fiber *fb;
    void **stack;

    stacksize = (stacksize + FB_STK_ALIGN - 1) & ~(FB_STK_ALIGN - 1);
#ifdef WIN32
    fb = (Fiber *) VirtualAlloc(NULL, stacksize, MEM_COMMIT | MEM_TOP_DOWN, PAGE_READWRITE);
#else
    fb = (Fiber *) malloc(stacksize);
#endif
    if (fb == NULL)
        return;

    stack = (void **) fb + stacksize / sizeof(void *) - 5;

    new(fb) Fiber(s_service, data);

    fb->m_cntxt.ip = (intptr_t) fiber_proc;
    fb->m_cntxt.sp = (intptr_t) stack;

#if defined(x64)
#ifdef _WIN32
    fb->m_cntxt.Rcx = (intptr_t) func;
    fb->m_cntxt.Rdx = (intptr_t) fb;
#else
    fb->m_cntxt.Rdi = (intptr_t) func;
    fb->m_cntxt.Rsi = (intptr_t) fb;
#endif
#elif defined(I386)
    stack[1] = (void *)func;
    stack[2] = fb;
#elif defined(arm)
    fb->m_cntxt.r0 = (intptr_t) func;
    fb->m_cntxt.r1 = (intptr_t) fb;
#endif

#ifdef DEBUG
    s_locker.lock();
    s_fibers.putTail(&fb->m_link);
    s_locker.unlock();
#endif

    if (retVal)
    {
        *retVal = fb;
        fb->Ref();
    }

    fb->Ref();
    fb->resume();
}

// k_heap_add_at_offset - Add a new heap block
// This function places a new heap block in memory, linked to an "origin" block.
void k_heap_add_at_offset(k_heap_blk* origin_blk, int block_offset) {
    __dynmem_lock.lock_cli();
    k_heap_blk* blk = (k_heap_blk*)((size_t)origin_blk+(block_offset*HEAP_BLK_SIZE));
    blk->prev = origin_blk;
    blk->next = origin_blk->next;
    blk->magic = HEAP_MAGIC_NUMBER;
    
    blk->prev->next = blk;
    if(blk->next != NULL)
        blk->next->prev = blk;
    else
        heap.end = blk;
    __dynmem_lock.unlock_cli();
}

void Service::forEach(void (*func)(Fiber*))
{
    s_locker.lock();

    linkitem* p = s_fibers.head();

    while (p)
    {
        Fiber* zfb = 0;
        func((Fiber*)((intptr_t)p - (intptr_t)(&zfb->m_link)));

        p = s_fibers.next(p);
    }

    s_locker.unlock();
}

 /**
  * Checks the message.  If its a write request, a buffer will be
  * taken from the object_allocator.  If the object is currently
  * allocated for writing then this function will return NULL
  */
 F *trycheckout(object_allocator_tls<F> &pool, 
             const ReadWrite rw) {
   if (rw == Reading) {
     return checkout(pool,rw);
   }
   else {
     lock.lock();
     if (writebuffer != NULL) {
       lock.unlock();
       return NULL;
     }
     writebuffer = pool.checkout();
     lock.unlock();
     (*writebuffer) = message;
     return writebuffer;
   }
 }

// terminal_scroll - scroll the console
// Positive values scroll down (adding new lines to the bottom); negative values do the inverse.
void terminal_scroll(int num_rows)
{
    __vga_write_lock.lock();
    if(num_rows > 0) { // scroll down
        for(size_t y=0;y<VGA_HEIGHT-1;y++)
            for(size_t x=0;x<VGA_WIDTH;x++)
                terminal_buffer[y*VGA_WIDTH+x] = terminal_buffer[(y+1)*VGA_WIDTH+x];
        for(size_t x=0;x<VGA_WIDTH;x++)
            terminal_buffer[(VGA_HEIGHT-1)*VGA_WIDTH+x] = make_vgaentry(' ', make_color(COLOR_LIGHT_GREY, COLOR_BLACK));
    } else if(num_rows < 0) { // scroll up
        for(size_t y=VGA_HEIGHT-1;y>0;y--)
            for(size_t x=0;x<VGA_WIDTH;x++)
                terminal_buffer[y*VGA_WIDTH+x] = terminal_buffer[(y-1)*VGA_WIDTH+x];
        for(size_t x=0;x<VGA_WIDTH;x++)
            terminal_buffer[x] = make_vgaentry(' ', make_color(COLOR_LIGHT_GREY, COLOR_BLACK));
    }
    __vga_write_lock.unlock();
}

    /**
     * Checks the message.  If its a write request, a buffer will be
     * taken from the object_allocator
     */
    F *checkout(object_allocator_tls<F> &pool, 
                const ReadWrite rw) {
      if (rw == Reading) {
        lock.lock();
        readercount++;
        lock.unlock();
        return &message;
      }
      else {
        lock.lock();
        // we only support 1 writer. 
        assert(writebuffer == NULL);
        writebuffer = pool.checkout();
        lock.unlock();
        (*writebuffer) = message;

        return writebuffer;
      }
    }

// k_heap_delete - Unlink a block
// This function removes a block from the linked list, effectively "deleting" it.
void k_heap_delete(k_heap_blk* blk) {
    __dynmem_lock.lock_cli();
    if(blk == NULL) {
        panic("dynmem: Attempted to delete NULL block!");
    }
    if(blk->prev != NULL) {
        blk->prev->next = blk->next;
    } else {
        return; // can't delete the first block
    }
    if(blk->next != NULL) {
        blk->next->prev = blk->prev;
    } else {
        heap.end = blk->prev;
    }
    blk->next = NULL;
    blk->prev = NULL;
    blk->magic = 0;
    blk->used = false;
    __dynmem_lock.unlock_cli();
}