<commit_before>/*
 * Copyright 2012-2013 Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "phenom/job.h"
#include "phenom/log.h"
#include "phenom/timerwheel.h"
#include "phenom/sysutil.h"
#include "phenom/memory.h"
#include "phenom/counter.h"
#include "phenom/configuration.h"
#include "corelib/job.h"
#include <ck_epoch.h>

#ifdef USE_GIMLI
#include <libgimli.h>
GIMLI_DECLARE_TRACER_MODULE("gimli_libphenom");
#endif

static ph_memtype_def_t ajob_def = {
  "nbio", "affine_job", sizeof(struct ph_nbio_affine_job), PH_MEM_FLAGS_ZERO
};
static ph_memtype_t mt_ajob;
static ph_counter_scope_t *counter_scope = NULL;
static const char *counter_names[] = {
  "dispatched",     // number of jobs dispatched
  "timer_ticks",    // how many times the timer has ticked
  "timer_busy",     // couldn't claim from timer
};

static uint32_t num_schedulers;
int _ph_run_loop = 1;
static struct ph_nbio_emitter *emitters;
static ph_job_t gc_job;
static int gc_interval;
#ifdef USE_GIMLI
static volatile struct gimli_heartbeat *hb = NULL;
#endif

static inline struct ph_nbio_emitter *emitter_for_affinity(uint32_t n)
{
  return &emitters[n % num_schedulers];
}

static inline struct ph_nbio_emitter *emitter_for_job(ph_job_t *job)
{
  return emitter_for_affinity(job->emitter_affinity);
}

struct ph_nbio_emitter *ph_nbio_emitter_for_job(ph_job_t *job) {
  if (!emitters) {
    // We can be called during TLS teardown
    return NULL;
  }
  return emitter_for_job(job);
}

static void process_deferred(ph_thread_t *me, void *impl);

void ph_nbio_emitter_dispatch_immediate(
    struct ph_nbio_emitter *emitter,
    ph_job_t *job, ph_iomask_t why)
{
  if (why != PH_IOMASK_TIME &&
      ph_timerwheel_disable(&emitter->wheel, &job->timer) == PH_BUSY) {
    // timer is currently dispatching this: it wins
    ph_counter_block_add(emitter->cblock, SLOT_BUSY, 1);
    return;
  }
  ph_counter_block_add(emitter->cblock, SLOT_DISP, 1);
  job->callback(job, why, job->data);
}

// map the timer address back to that of its containing
// work item
static inline ph_job_t *job_from_timer(struct ph_timerwheel_timer *timer)
{
  return ph_container_of(timer, ph_job_t, timer);
}

static bool before_dispatch_timer(
    ph_timerwheel_t *w,
    struct ph_timerwheel_timer *timer,
    struct timeval now,
    void *arg)
{
  ph_job_t *job;
  struct ph_nbio_emitter *emitter = arg;

  ph_unused_parameter(w);
  ph_unused_parameter(now);
  ph_unused_parameter(arg);
  ph_unused_parameter(emitter);

  job = job_from_timer(timer);

  if (job->fd != -1) {
    // Turn off any pending kernel notification
    ph_nbio_emitter_apply_io_mask(emitter, job, 0);
  }

  return true;
}

static void dispatch_timer(
    ph_timerwheel_t *w,
    struct ph_timerwheel_timer *timer,
    struct timeval now,
    void *arg)
{
  ph_job_t *job;
  struct ph_nbio_emitter *emitter = arg;

  ph_unused_parameter(w);
  ph_unused_parameter(now);

  // map the timer address back to that of its containing
  // work item
  job = job_from_timer(timer);

  ph_nbio_emitter_dispatch_immediate(emitter, job, PH_IOMASK_TIME);
}

void ph_nbio_emitter_timer_tick(struct ph_nbio_emitter *emitter)
{
  struct timeval now = ph_time_now();
  while (timercmp(&emitter->wheel.next_run, &now, <)) {
    ph_counter_block_add(emitter->cblock, SLOT_TIMER_TICK, 1);
    ph_timerwheel_tick(&emitter->wheel, now,
        before_dispatch_timer, dispatch_timer, emitter);
  }
}

void ph_nbio_process_affine_jobs(struct ph_nbio_emitter *emitter)
{
  struct ph_nbio_affine_job *ajob, *tmp;

  if (ph_pingfd_consume_all(&emitter->affine_ping)) {
    ph_nbio_affine_job_stailq_t list;

    PH_STAILQ_INIT(&list);
    ck_rwlock_write_lock(&emitter->wheel.lock);
    PH_STAILQ_SWAP(&list, &emitter->affine_jobs, ph_nbio_affine_job);
    ck_rwlock_write_unlock(&emitter->wheel.lock);

    PH_STAILQ_FOREACH_SAFE(ajob, &list, ent, tmp) {
      PH_STAILQ_REMOVE(&list, ajob, ph_nbio_affine_job, ent);

      ajob->func(ajob->code, ajob->arg);
      ph_mem_free(mt_ajob, ajob);
    }
  }
}

ph_result_t ph_nbio_queue_affine_func(uint32_t emitter_affinity,
    ph_nbio_affine_func func, intptr_t code, void *arg)
{
  struct ph_nbio_affine_job *ajob;
  struct ph_nbio_emitter *emitter;

  ajob = ph_mem_alloc(mt_ajob);
  if (!ajob) {
    return PH_NOMEM;
  }

  ajob->func = func;
  ajob->code = code;
  ajob->arg = arg;

  emitter = emitter_for_affinity(emitter_affinity);
  ck_rwlock_write_lock(&emitter->wheel.lock);
  PH_STAILQ_INSERT_TAIL(&emitter->affine_jobs, ajob, ent);
  ck_rwlock_write_unlock(&emitter->wheel.lock);

  ph_pingfd_ping(&emitter->affine_ping);
  return PH_OK;
}<commit_msg>coalesce pings when queueing affine function calls<commit_after>/*
 * Copyright 2012-2013 Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "phenom/job.h"
#include "phenom/log.h"
#include "phenom/timerwheel.h"
#include "phenom/sysutil.h"
#include "phenom/memory.h"
#include "phenom/counter.h"
#include "phenom/configuration.h"
#include "corelib/job.h"
#include <ck_epoch.h>

#ifdef USE_GIMLI
#include <libgimli.h>
GIMLI_DECLARE_TRACER_MODULE("gimli_libphenom");
#endif

static ph_memtype_def_t ajob_def = {
  "nbio", "affine_job", sizeof(struct ph_nbio_affine_job), PH_MEM_FLAGS_ZERO
};
static ph_memtype_t mt_ajob;
static ph_counter_scope_t *counter_scope = NULL;
static const char *counter_names[] = {
  "dispatched",     // number of jobs dispatched
  "timer_ticks",    // how many times the timer has ticked
  "timer_busy",     // couldn't claim from timer
};

static uint32_t num_schedulers;
int _ph_run_loop = 1;
static struct ph_nbio_emitter *emitters;
static ph_job_t gc_job;
static int gc_interval;
#ifdef USE_GIMLI
static volatile struct gimli_heartbeat *hb = NULL;
#endif

static inline struct ph_nbio_emitter *emitter_for_affinity(uint32_t n)
{
  return &emitters[n % num_schedulers];
}

static inline struct ph_nbio_emitter *emitter_for_job(ph_job_t *job)
{
  return emitter_for_affinity(job->emitter_affinity);
}

struct ph_nbio_emitter *ph_nbio_emitter_for_job(ph_job_t *job) {
  if (!emitters) {
    // We can be called during TLS teardown
    return NULL;
  }
  return emitter_for_job(job);
}

static void process_deferred(ph_thread_t *me, void *impl);

void ph_nbio_emitter_dispatch_immediate(
    struct ph_nbio_emitter *emitter,
    ph_job_t *job, ph_iomask_t why)
{
  if (why != PH_IOMASK_TIME &&
      ph_timerwheel_disable(&emitter->wheel, &job->timer) == PH_BUSY) {
    // timer is currently dispatching this: it wins
    ph_counter_block_add(emitter->cblock, SLOT_BUSY, 1);
    return;
  }
  ph_counter_block_add(emitter->cblock, SLOT_DISP, 1);
  job->callback(job, why, job->data);
}

// map the timer address back to that of its containing
// work item
static inline ph_job_t *job_from_timer(struct ph_timerwheel_timer *timer)
{
  return ph_container_of(timer, ph_job_t, timer);
}

static bool before_dispatch_timer(
    ph_timerwheel_t *w,
    struct ph_timerwheel_timer *timer,
    struct timeval now,
    void *arg)
{
  ph_job_t *job;
  struct ph_nbio_emitter *emitter = arg;

  ph_unused_parameter(w);
  ph_unused_parameter(now);
  ph_unused_parameter(arg);
  ph_unused_parameter(emitter);

  job = job_from_timer(timer);

  if (job->fd != -1) {
    // Turn off any pending kernel notification
    ph_nbio_emitter_apply_io_mask(emitter, job, 0);
  }

  return true;
}

static void dispatch_timer(
    ph_timerwheel_t *w,
    struct ph_timerwheel_timer *timer,
    struct timeval now,
    void *arg)
{
  ph_job_t *job;
  struct ph_nbio_emitter *emitter = arg;

  ph_unused_parameter(w);
  ph_unused_parameter(now);

  // map the timer address back to that of its containing
  // work item
  job = job_from_timer(timer);

  ph_nbio_emitter_dispatch_immediate(emitter, job, PH_IOMASK_TIME);
}

void ph_nbio_emitter_timer_tick(struct ph_nbio_emitter *emitter)
{
  struct timeval now = ph_time_now();
  while (timercmp(&emitter->wheel.next_run, &now, <)) {
    ph_counter_block_add(emitter->cblock, SLOT_TIMER_TICK, 1);
    ph_timerwheel_tick(&emitter->wheel, now,
        before_dispatch_timer, dispatch_timer, emitter);
  }
}

void ph_nbio_process_affine_jobs(struct ph_nbio_emitter *emitter)
{
  struct ph_nbio_affine_job *ajob, *tmp;

  if (ph_pingfd_consume_all(&emitter->affine_ping)) {
    ph_nbio_affine_job_stailq_t list;

    PH_STAILQ_INIT(&list);
    ck_rwlock_write_lock(&emitter->wheel.lock);
    PH_STAILQ_SWAP(&list, &emitter->affine_jobs, ph_nbio_affine_job);
    ck_rwlock_write_unlock(&emitter->wheel.lock);

    PH_STAILQ_FOREACH_SAFE(ajob, &list, ent, tmp) {
      PH_STAILQ_REMOVE(&list, ajob, ph_nbio_affine_job, ent);

      ajob->func(ajob->code, ajob->arg);
      ph_mem_free(mt_ajob, ajob);
    }
  }
}

ph_result_t ph_nbio_queue_affine_func(uint32_t emitter_affinity,
    ph_nbio_affine_func func, intptr_t code, void *arg)
{
  struct ph_nbio_affine_job *ajob;
  struct ph_nbio_emitter *emitter;
  bool need_ping;

  ajob = ph_mem_alloc(mt_ajob);
  if (!ajob) {
    return PH_NOMEM;
  }

  ajob->func = func;
  ajob->code = code;
  ajob->arg = arg;

  emitter = emitter_for_affinity(emitter_affinity);
  ck_rwlock_write_lock(&emitter->wheel.lock);
  need_ping = PH_STAILQ_EMPTY(&emitter->affine_jobs);
  PH_STAILQ_INSERT_TAIL(&emitter->affine_jobs, ajob, ent);
  ck_rwlock_write_unlock(&emitter->wheel.lock);

  if (need_ping) {
    ph_pingfd_ping(&emitter->affine_ping);
  }
  return PH_OK;
}
{'fraction_non_alphanumeric': 0.10299194476409666, 'fraction_numerical': 0.004602991944764097, 'mean_word_length': 4.727073036792971, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 58, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4366971', 'n_tokens_mistral': 4008, 'n_tokens_neox': 3812, 'n_words': 864}