<commit_before>#include "strm.h"
#include <pthread.h>

struct strm_thread {
  pthread_t th;
  strm_queue *queue;
} *threads;

static int thread_max;
static int pipeline_count = 0;
static pthread_mutex_t pipeline_mutex;
static pthread_cond_t pipeline_cond;

#include <assert.h>

static void task_init();

static int
task_tid(strm_stream *s, int tid)
{
  int i;

  if (s->tid < 0) {
    if (tid >= 0) {
      s->tid = tid % thread_max;
    }
    else {
      int n = 0;
      int max = 0;

      for (i=0; i<thread_max; i++) {
        int size = strm_queue_size(threads[i].queue);
        if (size == 0) break;
        if (size > max) {
          max = size;
          n = i;
        }
      }
      if (i == thread_max) {
        s->tid = n;
      }
      else {
        s->tid = i;
      }
    }
  }
  return s->tid;
}

static void
task_push(int tid, struct strm_queue_task *t)
{
  strm_stream *s = t->strm;

  assert(threads != NULL);
  task_tid(s, tid);
  strm_queue_push(threads[s->tid].queue, t);
}

void
strm_task_push(struct strm_queue_task *t)
{
  task_push(-1, t);
}

void
strm_emit(strm_stream *strm, strm_value data, strm_func func)
{
  strm_stream *d = strm->dst;

  while (d) {
    task_push(strm->tid+1, strm_queue_task(d, d->start_func, data));
    d = d->nextd;
  }
  if (func) {
    strm_task_push(strm_queue_task(strm, func, strm_null_value()));
  }
}

int
strm_connect(strm_stream *src, strm_stream *dst)
{
  strm_stream *s;

  assert(dst->mode != strm_task_prod);
  s = src->dst;
  if (s) {
    while (s->nextd) {
      s = s->nextd;
    }
    s->nextd = dst;
  }
  else {
    src->dst = dst;
  }

  if (src->mode == strm_task_prod) {
    task_init();<commit_msg>run first emit target task on the same thread<commit_after>#include "strm.h"
#include <pthread.h>

struct strm_thread {
  pthread_t th;
  strm_queue *queue;
} *threads;

static int thread_max;
static int pipeline_count = 0;
static pthread_mutex_t pipeline_mutex;
static pthread_cond_t pipeline_cond;

#include <assert.h>

static void task_init();

static int
task_tid(strm_stream *s, int tid)
{
  int i;

  if (s->tid < 0) {
    if (tid >= 0) {
      s->tid = tid % thread_max;
    }
    else {
      int n = 0;
      int max = 0;

      for (i=0; i<thread_max; i++) {
        int size = strm_queue_size(threads[i].queue);
        if (size == 0) break;
        if (size > max) {
          max = size;
          n = i;
        }
      }
      if (i == thread_max) {
        s->tid = n;
      }
      else {
        s->tid = i;
      }
    }
  }
  return s->tid;
}

static void
task_push(int tid, struct strm_queue_task *t)
{
  strm_stream *s = t->strm;

  assert(threads != NULL);
  task_tid(s, tid);
  strm_queue_push(threads[s->tid].queue, t);
}

void
strm_task_push(struct strm_queue_task *t)
{
  task_push(-1, t);
}

void
strm_emit(strm_stream *strm, strm_value data, strm_func func)
{
  strm_stream *d = strm->dst;
  int tid = strm->tid;

  while (d) {
    task_push(tid, strm_queue_task(d, d->start_func, data));
    d = d->nextd;
    tid++;
  }
  if (func) {
    strm_task_push(strm_queue_task(strm, func, strm_null_value()));
  }
}

int
strm_connect(strm_stream *src, strm_stream *dst)
{
  strm_stream *s;

  assert(dst->mode != strm_task_prod);
  s = src->dst;
  if (s) {
    while (s->nextd) {
      s = s->nextd;
    }
    s->nextd = dst;
  }
  else {
    src->dst = dst;
  }

  if (src->mode == strm_task_prod) {
    task_init();
{'fraction_non_alphanumeric': 0.146697837521917, 'fraction_numerical': 0.004967855055523086, 'mean_word_length': 2.528865979381443, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 47, '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': '4753534', 'n_tokens_mistral': 1436, 'n_tokens_neox': 1411, 'n_words': 315}