<commit_before>/*
 *  Copyright (c) 2010 <NAME>
 *  Copyright (c) 2010 <NAME>
 *
 *  This is free software; see gpl-3.0.txt
 */
#include <sys/select.h>
#include <sys/time.h>
#include <time.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>

#include <net_helper.h>
#include <topmanager.h>
#include <msg_types.h>

#include "streaming.h"
#include "loop.h"
#include "dbg.h"

#define BUFFSIZE 64 * 1024
static struct timeval period = {0, 500000};
static struct timeval tnext;

void tout_init(struct timeval *tv)
{
  struct timeval tnow;

  if (tnext.tv_sec == 0) {
    gettimeofday(&tnext, NULL);
  }
  gettimeofday(&tnow, NULL);
  if(timercmp(&tnow, &tnext, <)) {
    timersub(&tnext, &tnow, tv);
  } else {
    *tv = (struct timeval){0, 0};
  }
}

void loop(struct nodeID *s, int csize, int buff_size)
{
  int done = 0;
  static uint8_t buff[BUFFSIZE];
  int cnt = 0;
  
  period.tv_sec = csize / 1000000;
  period.tv_usec = csize % 1000000;
  
  topParseData(NULL, 0);
  stream_init(buff_size, s);
  while (!done) {
    int len, res;
    struct timeval tv;

    tout_init(&tv);
    res = wait4data(s, tv);
    if (res > 0) {
      struct nodeID *remote;

      len = recv_from_peer(s, &remote, buff, BUFFSIZE);
      switch (buff[0] /* Message Type */) {
        case MSG_TYPE_TOPOLOGY:
          topParseData(buff, len);
          break;
        case MSG_TYPE_CHUNK:
          received_chunk(buff, len);
          break;
        default:
          fprintf(stderr, "Unknown Message Type %x\n", buff[0]);
      }
      nodeid_free(remote);
    } else {
      const struct nodeID **neighbours;
      int n;
      struct timeval tmp;

      neighbours = topGetNeighbourhood(&n);
      send_chunk(neighbours, n);
      if (cnt++ % 10 == 0) {
        topParseData(NULL, 0);
      }
      timeradd(&tnext, &period, &tmp);
      tnext = tmp;
    }
  }
}

void source_loop(const char *fname, struct nodeID *s, int csize, int chunks, bool loop)
{
  int done = 0;
  static uint8_t buff[BUFFSIZE];
  int cnt = 0;

  source_init(fname, s, loop);
  while (!done) {
    int len, res;
    struct timeval tv;

    tout_init(&tv);
    res = wait4data(s, tv);<commit_msg>follow net_helper.h changes to wait4data<commit_after>/*
 *  Copyright (c) 2010 <NAME>
 *  Copyright (c) 2010 <NAME>
 *
 *  This is free software; see gpl-3.0.txt
 */
#include <sys/select.h>
#include <sys/time.h>
#include <time.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>

#include <net_helper.h>
#include <topmanager.h>
#include <msg_types.h>

#include "streaming.h"
#include "loop.h"
#include "dbg.h"

#define BUFFSIZE 64 * 1024
static struct timeval period = {0, 500000};
static struct timeval tnext;

void tout_init(struct timeval *tv)
{
  struct timeval tnow;

  if (tnext.tv_sec == 0) {
    gettimeofday(&tnext, NULL);
  }
  gettimeofday(&tnow, NULL);
  if(timercmp(&tnow, &tnext, <)) {
    timersub(&tnext, &tnow, tv);
  } else {
    *tv = (struct timeval){0, 0};
  }
}

void loop(struct nodeID *s, int csize, int buff_size)
{
  int done = 0;
  static uint8_t buff[BUFFSIZE];
  int cnt = 0;
  
  period.tv_sec = csize / 1000000;
  period.tv_usec = csize % 1000000;
  
  topParseData(NULL, 0);
  stream_init(buff_size, s);
  while (!done) {
    int len, res;
    struct timeval tv;

    tout_init(&tv);
    res = wait4data(s, &tv);
    if (res > 0) {
      struct nodeID *remote;

      len = recv_from_peer(s, &remote, buff, BUFFSIZE);
      switch (buff[0] /* Message Type */) {
        case MSG_TYPE_TOPOLOGY:
          topParseData(buff, len);
          break;
        case MSG_TYPE_CHUNK:
          received_chunk(buff, len);
          break;
        default:
          fprintf(stderr, "Unknown Message Type %x\n", buff[0]);
      }
      nodeid_free(remote);
    } else {
      const struct nodeID **neighbours;
      int n;
      struct timeval tmp;

      neighbours = topGetNeighbourhood(&n);
      send_chunk(neighbours, n);
      if (cnt++ % 10 == 0) {
        topParseData(NULL, 0);
      }
      timeradd(&tnext, &period, &tmp);
      tnext = tmp;
    }
  }
}

void source_loop(const char *fname, struct nodeID *s, int csize, int chunks, bool loop)
{
  int done = 0;
  static uint8_t buff[BUFFSIZE];
  int cnt = 0;

  source_init(fname, s, loop);
  while (!done) {
    int len, res;
    struct timeval tv;

    tout_init(&tv);
    res = wait4data(s, &tv);
{'fraction_non_alphanumeric': 0.14214236979758926, 'fraction_numerical': 0.02569934045940414, 'mean_word_length': 2.7783505154639174, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, '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': '1278913', 'n_tokens_mistral': 1775, 'n_tokens_neox': 1676, 'n_words': 429}