<commit_before>/// \file buffer.cpp
/// Contains the main code for the Buffer.

#include <fcntl.h>
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sstream>
#include <sys/time.h>
#include <mist/config.h>
#include <mist/timing.h>
#include "buffer_stream.h"
#include <mist/stream.h>

/// Holds all code unique to the Buffer.
namespace Buffer {

  volatile bool buffer_running = true; ///< Set to false when shutting down.
  Stream * thisStream = 0;
  Socket::Server SS; ///< The server socket.

  ///\brief A function running in a thread to send all statistics.
  ///\param empty A null pointer.
  void handleStats(void * empty){
#if defined(_TTHREAD_POSIX_) && defined(WITH_THREADNAMES) && !(defined(__FreeBSD__) || defined(__APPLE__) || defined(__MACH__) || defined(_WIN32) || defined(__CYGWIN__))
    pthread_setname_np(pthread_self(), "StatsHandler");
#endif
    if (empty != 0){
      return;
    }
    std::string double_newline = "\n\n";
    Socket::Connection StatsSocket = Socket::Connection(Util::getTmpFolder() + "statistics", true);
    while (buffer_running){
      Util::sleep(1000); //sleep one second
      if ( !StatsSocket.connected()){
        StatsSocket = Socket::Connection(Util::getTmpFolder() + "statistics", true);
      }
      if (StatsSocket.connected()){
        StatsSocket.SendNow(Stream::get()->getStats());
        StatsSocket.SendNow(double_newline);
        if (StatsSocket.spool()){
          //Got a response.
          buffer_running = false;
        }
      }
    }
    StatsSocket.close();
  }

  ///\brief A function to handle input data.
  ///\param conn A socket reference.
  void handlePushIn(Socket::Connection & conn){
    #if defined(_TTHREAD_POSIX_) && defined(WITH_THREADNAMES) && !(defined(__FreeBSD__) || defined(__APPLE__) || defined(__MACH__) || defined(_WIN32) || defined(__CYGWIN__))
    pthread_setname_np(pthread_self(), "Push Input");
    #endif
    conn.setBlocking(true);
    int sockNo = 0;
    while (buffer_running && conn.connected()){
      while (thisStream->parsePacket(conn)){
        //do nothing while parsing
      }
      Util::sleep(10);//sleep to prevent high CPU usage
    }
    if (buffer_running){
      thisStream->endStream();
    }
    long long int wait_time = Util::getMS();
    while (Util::getMS() - wait_time < thisStream->metadata.bufferWindow){<commit_msg>Fix MistBuffer 100% CPU time after stream end.<commit_after>/// \file buffer.cpp
/// Contains the main code for the Buffer.

#include <fcntl.h>
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sstream>
#include <sys/time.h>
#include <mist/config.h>
#include <mist/timing.h>
#include "buffer_stream.h"
#include <mist/stream.h>

/// Holds all code unique to the Buffer.
namespace Buffer {

  volatile bool buffer_running = true; ///< Set to false when shutting down.
  Stream * thisStream = 0;
  Socket::Server SS; ///< The server socket.

  ///\brief A function running in a thread to send all statistics.
  ///\param empty A null pointer.
  void handleStats(void * empty){
#if defined(_TTHREAD_POSIX_) && defined(WITH_THREADNAMES) && !(defined(__FreeBSD__) || defined(__APPLE__) || defined(__MACH__) || defined(_WIN32) || defined(__CYGWIN__))
    pthread_setname_np(pthread_self(), "StatsHandler");
#endif
    if (empty != 0){
      return;
    }
    std::string double_newline = "\n\n";
    Socket::Connection StatsSocket = Socket::Connection(Util::getTmpFolder() + "statistics", true);
    while (buffer_running){
      Util::sleep(1000); //sleep one second
      if ( !StatsSocket.connected()){
        StatsSocket = Socket::Connection(Util::getTmpFolder() + "statistics", true);
      }
      if (StatsSocket.connected()){
        StatsSocket.SendNow(Stream::get()->getStats());
        StatsSocket.SendNow(double_newline);
        if (StatsSocket.spool()){
          //Got a response.
          buffer_running = false;
        }
      }
    }
    StatsSocket.close();
  }

  ///\brief A function to handle input data.
  ///\param conn A socket reference.
  void handlePushIn(Socket::Connection & conn){
    #if defined(_TTHREAD_POSIX_) && defined(WITH_THREADNAMES) && !(defined(__FreeBSD__) || defined(__APPLE__) || defined(__MACH__) || defined(_WIN32) || defined(__CYGWIN__))
    pthread_setname_np(pthread_self(), "Push Input");
    #endif
    conn.setBlocking(true);
    int sockNo = 0;
    while (buffer_running && conn.connected()){
      while (thisStream->parsePacket(conn)){
        //do nothing while parsing
      }
      Util::sleep(10);//sleep to prevent high CPU usage
    }
    conn.close();
    if (buffer_running){
      thisStream->endStream();
    }
    long long int wait_time = Util::getMS();
    while (Util::getMS() - wait_time < thisStream->metadata.bufferWindow){
{'fraction_non_alphanumeric': 0.1479540415238863, 'fraction_numerical': 0.005845595646039105, 'mean_word_length': 3.5150136487716104, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 39, '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': '474409', 'n_tokens_mistral': 1595, 'n_tokens_neox': 1529, 'n_words': 389}