parser.cc File Reference

#include <sys/types.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>
#include <netinet/tcp.h>
#include <csignal>
#include <fstream>
#include <netdb.h>
#include <cerrno>
#include <cstdio>
#include <cassert>
#include "parse_example.h"
#include "cache.h"
#include "unique_sort.h"
#include "constant.h"
#include "vw.h"
#include "interactions.h"
#include "vw_exception.h"
#include "parse_example_json.h"
#include "parse_dispatch_loop.h"
#include "parse_args.h"

Go to the source code of this file.

Namespaces
	VW

Macros
#define	MAP_ANONYMOUS   MAP_ANON

Functions
void	handle_sigterm (int)
bool	is_test_only (uint32_t counter, uint32_t period, uint32_t after, bool holdout_off, uint32_t target_modulus)
void	set_compressed (parser *par)
uint32_t	cache_numbits (io_buf *buf, int filepointer)
void	reset_source (vw &all, size_t numbits)
void	finalize_source (parser *p)
void	make_write_cache (vw &all, std::string &newname, bool quiet)
void	parse_cache (vw &all, std::vector< std::string > cache_files, bool kill_cache, bool quiet)
void	enable_sources (vw &all, bool quiet, size_t passes, input_options &input_options)
void	lock_done (parser &p)
void	set_done (vw &all)
void	addgrams (vw &all, size_t ngram, size_t skip_gram, features &fs, size_t initial_length, v_array< size_t > &gram_mask, size_t skips)
void	generateGrams (vw &all, example *&ex)
void	end_pass_example (vw &all, example *ae)
void	feature_limit (vw &all, example *ex)
example &	VW::get_unused_example (vw *all)
void	VW::setup_examples (vw &all, v_array< example *> &examples)
void	VW::setup_example (vw &all, example *ae)
example *	VW::new_unused_example (vw &all)
example *	VW::read_example (vw &all, char *example_line)
example *	VW::read_example (vw &all, std::string example_line)
void	VW::add_constant_feature (vw &vw, example *ec)
void	VW::add_label (example *ec, float label, float weight, float base)
example *	VW::import_example (vw &all, const std::string &label, primitive_feature_space *features, size_t len)
primitive_feature_space *	VW::export_example (vw &all, example *ec, size_t &len)
void	VW::releaseFeatureSpace (primitive_feature_space *features, size_t len)
void	VW::parse_example_label (vw &all, example &ec, std::string label)
void	VW::empty_example (vw &, example &ec)
void	VW::clean_example (vw &, example &, bool rewind)
void	VW::finish_example (vw &, example &)
void	thread_dispatch (vw &all, v_array< example *> examples)
void	main_parse_loop (vw *all)
example *	VW::get_example (parser *p)
float	VW::get_topic_prediction (example *ec, size_t i)
float	VW::get_label (example *ec)
float	VW::get_importance (example *ec)
float	VW::get_initial (example *ec)
float	VW::get_prediction (example *ec)
float	VW::get_cost_sensitive_prediction (example *ec)
v_array< float > &	VW::get_cost_sensitive_prediction_confidence_scores (example *ec)
uint32_t *	VW::get_multilabel_predictions (example *ec, size_t &len)
float	VW::get_action_score (example *ec, size_t i)
size_t	VW::get_action_score_length (example *ec)
size_t	VW::get_tag_length (example *ec)
const char *	VW::get_tag (example *ec)
size_t	VW::get_feature_number (example *ec)
float	VW::get_confidence (example *ec)
void	adjust_used_index (vw &)
void	VW::start_parser (vw &all)
void	free_parser (vw &all)
void	VW::end_parser (vw &all)
bool	VW::is_ring_example (vw &all, example *ae)

Variables
bool	got_sigterm

Macro Definition Documentation

MAP_ANONYMOUS

#define MAP_ANONYMOUS   MAP_ANON

Definition at line 309 of file parser.cc.

Referenced by enable_sources(), and dense_parameters::share().

Function Documentation

addgrams()

void addgrams	(	vw &	all,
		size_t	ngram,
		size_t	skip_gram,
		features &	fs,
		size_t	initial_length,
		v_array< size_t > &	gram_mask,
		size_t	skips
	)

Definition at line 584 of file parser.cc.

References features::indicies, v_array< T >::last(), v_array< T >::pop(), v_array< T >::push_back(), features::push_back(), quadratic_constant, v_array< T >::size(), and features::space_names.

Referenced by generateGrams().

{
  if (ngram == 0 && gram_mask.last() < initial_length)
  {
    size_t last = initial_length - gram_mask.last();
    for (size_t i = 0; i < last; i++)
    {
      uint64_t new_index = fs.indicies[i];
      for (size_t n = 1; n < gram_mask.size(); n++)
        new_index = new_index * quadratic_constant + fs.indicies[i + gram_mask[n]];

      fs.push_back(1., new_index);
      if (fs.space_names.size() > 0)
      {
        std::string feature_name(fs.space_names[i].get()->second);
        for (size_t n = 1; n < gram_mask.size(); n++)
        {
          feature_name += std::string("^");
          feature_name += std::string(fs.space_names[i + gram_mask[n]].get()->second);
        }
        fs.space_names.push_back(audit_strings_ptr(new audit_strings(fs.space_names[i].get()->first, feature_name)));
      }
    }
  }
  if (ngram > 0)
  {
    gram_mask.push_back(gram_mask.last() + 1 + skips);
    addgrams(all, ngram - 1, skip_gram, fs, initial_length, gram_mask, 0);
    gram_mask.pop();
  }
  if (skip_gram > 0 && ngram > 0)
    addgrams(all, ngram, skip_gram - 1, fs, initial_length, gram_mask, skips + 1);
}

adjust_used_index()

void adjust_used_index

(

vw &

)

Definition at line 968 of file parser.cc.

Referenced by VW_Finish_Passes().

{ /* no longer used */
}

cache_numbits()

uint32_t cache_numbits	(	io_buf *	buf,
		int	filepointer
	)

Definition at line 91 of file parser.cc.

References io_buf::read_file(), THROW, and VW::version().

Referenced by parse_cache(), and reset_source().

{
  size_t v_length;
  buf->read_file(filepointer, (char*)&v_length, sizeof(v_length));
  if (v_length > 61)
    THROW("cache version too long, cache file is probably invalid");

  if (v_length == 0)
    THROW("cache version too short, cache file is probably invalid");

  std::vector<char> t(v_length);
  buf->read_file(filepointer, t.data(), v_length);
  VW::version_struct v_tmp(t.data());
  if (v_tmp != VW::version)
  {
    //      cout << "cache has possibly incompatible version, rebuilding" << endl;
    return 0;
  }

  char temp;
  if (buf->read_file(filepointer, &temp, 1) < 1)
    THROW("failed to read");

  if (temp != 'c')
    THROW("data file is not a cache file");

  uint32_t cache_numbits;
  if (buf->read_file(filepointer, &cache_numbits, sizeof(cache_numbits)) < (int)sizeof(cache_numbits))
  {
    return true;
  }

  return cache_numbits;
}

enable_sources()

void enable_sources	(	vw &	all,
		bool	quiet,
		size_t	passes,
		input_options &	input_options
	)

Definition at line 312 of file parser.cc.

References vw::active, parser::audit, vw::audit, binary_print_result(), parser::bound_sock, input_options::cache_files, children(), io_buf::count, io_buf::current, vw::daemon, vw::data_filename, parser::decision_service_json, input_options::dsjson, v_array< T >::empty(), f, io_buf::files, vw::final_prediction_sink, VW::finish(), input_options::foreground, handle_sigterm(), vw::hash_inv, parser::input, isbinary(), input_options::json, input_options::kill_cache, parser::label_sock, vw::length(), MAP_ANONYMOUS, parser::max_fd, vw::num_children, io_buf::open_file(), vw::options, vw::p, parse_cache(), input_options::pid_file, input_options::port, input_options::port_file, vw::print, print_result(), v_array< T >::push_back(), vw::quiet, io_buf::READ, read_cached_features(), read_features_string(), VW::read_lines(), parser::reader, parser::resettable, v_array< T >::resize(), vw::sd, parameters::share(), v_array< T >::size(), parser::sorted_cache, vw::stdin_off, parser::text_reader, THROW, THROWERRNO, vw::trace_message, VW::config::options_i::was_supplied(), vw::weights, and parser::write_cache.

Referenced by parse_sources().

{
  all.p->input->current = 0;
  parse_cache(all, input_options.cache_files, input_options.kill_cache, quiet);

  if (all.daemon || all.active)
  {
#ifdef _WIN32
    WSAData wsaData;
    int lastError = WSAStartup(MAKEWORD(2, 2), &wsaData);
    if (lastError != 0)
      THROWERRNO("WSAStartup() returned error:" << lastError);
#endif
    all.p->bound_sock = (int)socket(PF_INET, SOCK_STREAM, 0);
    if (all.p->bound_sock < 0)
    {
      std::stringstream msg;
      msg << "socket: " << strerror(errno);
      all.trace_message << msg.str() << endl;
      THROW(msg.str().c_str());
    }

    int on = 1;
    if (setsockopt(all.p->bound_sock, SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof(on)) < 0)
      all.trace_message << "setsockopt SO_REUSEADDR: " << strerror(errno) << endl;

    // Enable TCP Keep Alive to prevent socket leaks
    int enableTKA = 1;
    if (setsockopt(all.p->bound_sock, SOL_SOCKET, SO_KEEPALIVE, (char*)&enableTKA, sizeof(enableTKA)) < 0)
      all.trace_message << "setsockopt SO_KEEPALIVE: " << strerror(errno) << endl;

    sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    short unsigned int port = 26542;
    if (all.options->was_supplied("port"))
      port = (uint16_t)input_options.port;
    address.sin_port = htons(port);

    // attempt to bind to socket
    if (::bind(all.p->bound_sock, (sockaddr*)&address, sizeof(address)) < 0)
      THROWERRNO("bind");

    // listen on socket
    if (listen(all.p->bound_sock, 1) < 0)
      THROWERRNO("listen");

    // write port file
    if (all.options->was_supplied("port_file"))
    {
      socklen_t address_size = sizeof(address);
      if (getsockname(all.p->bound_sock, (sockaddr*)&address, &address_size) < 0)
      {
        all.trace_message << "getsockname: " << strerror(errno) << endl;
      }
      std::ofstream port_file;
      port_file.open(input_options.port_file.c_str());
      if (!port_file.is_open())
        THROW("error writing port file: " << input_options.port_file);

      port_file << ntohs(address.sin_port) << endl;
      port_file.close();
    }

    // background process (if foreground is not set)
    if (!input_options.foreground)
    {
      // FIXME switch to posix_spawn
      if (!all.active && daemon(1, 1))
        THROWERRNO("daemon");
    }

    // write pid file
    if (all.options->was_supplied("pid_file"))
    {
      std::ofstream pid_file;
      pid_file.open(input_options.pid_file.c_str());
      if (!pid_file.is_open())
        THROW("error writing pid file");

      pid_file << getpid() << endl;
      pid_file.close();
    }

    if (all.daemon && !all.active)
    {
#ifdef _WIN32
      THROW("not supported on windows");
#else
      fclose(stdin);
      // weights will be shared across processes, accessible to children
      all.weights.share(all.length());

      // learning state to be shared across children
      shared_data* sd =
          (shared_data*)mmap(0, sizeof(shared_data), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0);
      memcpy(sd, all.sd, sizeof(shared_data));
      free(all.sd);
      all.sd = sd;

      // create children
      size_t num_children = all.num_children;
      v_array<int> children = v_init<int>();
      children.resize(num_children);
      for (size_t i = 0; i < num_children; i++)
      {
        // fork() returns pid if parent, 0 if child
        // store fork value and run child process if child
        if ((children[i] = fork()) == 0)
        {
          all.quiet |= (i > 0);
          goto child;
        }
      }

      // install signal handler so we can kill children when killed
      {
        struct sigaction sa;
        // specifically don't set SA_RESTART in sa.sa_flags, so that
        // waitid will be interrupted by SIGTERM with handler installed
        memset(&sa, 0, sizeof(sa));
        sa.sa_handler = handle_sigterm;
        sigaction(SIGTERM, &sa, nullptr);
      }

      while (true)
      {
        // wait for child to change state; if finished, then respawn
        int status;
        pid_t pid = wait(&status);
        if (got_sigterm)
        {
          for (size_t i = 0; i < num_children; i++) kill(children[i], SIGTERM);
          VW::finish(all);
          exit(0);
        }
        if (pid < 0)
          continue;
        for (size_t i = 0; i < num_children; i++)
          if (pid == children[i])
          {
            if ((children[i] = fork()) == 0)
            {
              all.quiet |= (i > 0);
              goto child;
            }
            break;
          }
      }

#endif
    }

#ifndef _WIN32
  child:
#endif
    sockaddr_in client_address;
    socklen_t size = sizeof(client_address);
    all.p->max_fd = 0;
    if (!all.quiet)
      all.trace_message << "calling accept" << endl;
    int f = (int)accept(all.p->bound_sock, (sockaddr*)&client_address, &size);
    if (f < 0)
      THROWERRNO("accept");

    all.p->label_sock = f;
    all.print = print_result;

    all.final_prediction_sink.push_back((size_t)f);

    all.p->input->files.push_back(f);
    all.p->max_fd = std::max(f, all.p->max_fd);
    if (!all.quiet)
      all.trace_message << "reading data from port " << port << endl;

    all.p->max_fd++;
    if (all.active)
      all.p->reader = read_features_string;
    else
    {
      if (isbinary(*(all.p->input)))
      {
        all.p->reader = read_cached_features;
        all.print = binary_print_result;
      }
      else
      {
        all.p->reader = read_features_string;
      }
      all.p->sorted_cache = true;
    }
    all.p->resettable = all.p->write_cache || all.daemon;
  }
  else
  {
    if (!all.p->input->files.empty())
    {
      if (!quiet)
        all.trace_message << "ignoring text input in favor of cache input" << endl;
    }
    else
    {
      std::string temp = all.data_filename;
      if (!quiet)
        all.trace_message << "Reading datafile = " << temp << endl;
      try
      {
        all.p->input->open_file(temp.c_str(), all.stdin_off, io_buf::READ);
      }
      catch (std::exception const&)
      {
        // when trying to fix this exception, consider that an empty temp is valid if all.stdin_off is false
        if (!temp.empty())
        {
          all.trace_message << "can't open '" << temp << "', sailing on!" << endl;
        }
        else
        {
          throw;
        }
      }

      if (input_options.json || input_options.dsjson)
      {
        // TODO: change to class with virtual method
        // --invert_hash requires the audit parser version to save the extra information.
        if (all.audit || all.hash_inv)
        {
          all.p->reader = &read_features_json<true>;
          all.p->text_reader = &line_to_examples_json<true>;
          all.p->audit = true;
        }
        else
        {
          all.p->reader = &read_features_json<false>;
          all.p->text_reader = &line_to_examples_json<false>;
          all.p->audit = false;
        }

        all.p->decision_service_json = input_options.dsjson;
      }
      else
      {
        all.p->reader = read_features_string;
        all.p->text_reader = VW::read_lines;
      }

      all.p->resettable = all.p->write_cache;
    }
  }

  if (passes > 1 && !all.p->resettable)
    THROW("need a cache file for multiple passes : try using --cache_file");

  all.p->input->count = all.p->input->files.size();
  if (!quiet && !all.daemon)
    all.trace_message << "num sources = " << all.p->input->files.size() << endl;
}

end_pass_example()

void end_pass_example	(	vw &	all,
		example *	ae
	)

Definition at line 644 of file parser.cc.

References label_parser::default_label, example::end_pass, parser::in_pass_counter, example::l, parser::lp, and vw::p.

{
  all.p->lp.default_label(&ae->l);
  ae->end_pass = true;
  all.p->in_pass_counter = 0;
}

feature_limit()

void feature_limit	(	vw &	all,
		example *	ex
	)

Definition at line 651 of file parser.cc.

References example_predict::feature_space, example_predict::indices, vw::limit, vw::parse_mask, features::sort(), and unique_features().

Referenced by VW::setup_example().

{
  for (namespace_index index : ex->indices)
    if (all.limit[index] < ex->feature_space[index].size())
    {
      features& fs = ex->feature_space[index];
      fs.sort(all.parse_mask);
      unique_features(fs, all.limit[index]);
    }
}

finalize_source()

void finalize_source

(

parser *

p

)

Definition at line 206 of file parser.cc.

References io_buf::close_files(), v_array< T >::empty(), f, io_buf::files, parser::input, v_array< T >::last(), parser::output, and v_array< T >::pop().

Referenced by VW::finish(), and set_compressed().

{
#ifdef _WIN32
  int f = _fileno(stdin);
#else
  int f = fileno(stdin);
#endif
  while (!p->input->files.empty() && p->input->files.last() == f) p->input->files.pop();
  p->input->close_files();

  delete p->input;
  p->input = nullptr;
  p->output->close_files();
  delete p->output;
  p->output = nullptr;
}

free_parser()

void free_parser

(

vw &

all

)

Definition at line 976 of file parser.cc.

References parser::counts, io_buf::currentname, VW::dealloc_example(), label_parser::delete_label, vw::delete_prediction, v_array< T >::delete_v(), VW::object_pool< T, TInitializer, TCleanup >::empty(), parser::example_pool, io_buf::finalname, VW::object_pool< T, TInitializer, TCleanup >::get_object(), parser::gram_mask, parser::lp, parser::name, vw::ngram_strings, parser::output, vw::p, VW::ptr_queue< T >::pop(), parser::ready_parsed_examples, VW::ptr_queue< T >::size(), and parser::words.

Referenced by VW::finish().

{
  all.p->words.delete_v();
  all.p->name.delete_v();

  if (!all.ngram_strings.empty())
    all.p->gram_mask.delete_v();

  io_buf* output = all.p->output;
  if (output != nullptr)
  {
    output->finalname.delete_v();
    output->currentname.delete_v();
  }

  while (!all.p->example_pool.empty())
  {
    example* temp = all.p->example_pool.get_object();
    VW::dealloc_example(all.p->lp.delete_label, *temp, all.delete_prediction);
  }

  while (all.p->ready_parsed_examples.size() != 0)
  {
    example* temp = all.p->ready_parsed_examples.pop();
    VW::dealloc_example(all.p->lp.delete_label, *temp, all.delete_prediction);
  }
  all.p->counts.delete_v();
}

generateGrams()

void generateGrams	(	vw &	all,
		example *&	ex
	)

This function adds k-skip-n-grams to the feature vector. Definition of k-skip-n-grams: Consider a feature vector - a, b, c, d, e, f 2-skip-2-grams would be - ab, ac, ad, bc, bd, be, cd, ce, cf, de, df, ef 1-skip-3-grams would be - abc, abd, acd, ace, bcd, bce, bde, bdf, cde, cdf, cef, def Note that for a n-gram, (n-1)-grams, (n-2)-grams... 2-grams are also appended The k-skip-n-grams are appended to the feature vector. Hash is evaluated using the principle h(a, b) = h(a)*X + h(b), where X is a random no. 32 random nos. are maintained in an array and are used in the hashing.

Definition at line 630 of file parser.cc.

References addgrams(), v_array< T >::clear(), example_predict::feature_space, parser::gram_mask, example_predict::indices, vw::ngram, vw::p, v_array< T >::push_back(), and vw::skips.

Referenced by VW::setup_example().

{
  for (namespace_index index : ex->indices)
  {
    size_t length = ex->feature_space[index].size();
    for (size_t n = 1; n < all.ngram[index]; n++)
    {
      all.p->gram_mask.clear();
      all.p->gram_mask.push_back((size_t)0);
      addgrams(all, n, all.skips[index], ex->feature_space[index], length, all.p->gram_mask, 0);
    }
  }
}

handle_sigterm()

void handle_sigterm

(

int

)

Definition at line 68 of file parser.cc.

Referenced by enable_sources().

{ got_sigterm = true; }

is_test_only()

bool is_test_only	(	uint32_t	counter,
		uint32_t	period,
		uint32_t	after,
		bool	holdout_off,
		uint32_t	target_modulus
	)

Definition at line 70 of file parser.cc.

Referenced by VW::setup_example().

{
  if (holdout_off)
    return false;
  if (after == 0)  // hold out by period
    return (counter % period == target_modulus);
  else  // hold out by position
    return (counter > after);
}

lock_done()

void lock_done

(

parser &

p

)

Definition at line 571 of file parser.cc.

References parser::done, parser::ready_parsed_examples, and VW::ptr_queue< T >::set_done().

Referenced by parse_dispatch(), and set_done().

{
  p.done = true;
  // in case get_example() is waiting for a fresh example, wake so it can realize there are no more.
  p.ready_parsed_examples.set_done();
}

main_parse_loop()

void main_parse_loop

(

vw *

all

)

Definition at line 905 of file parser.cc.

References parse_dispatch(), and thread_dispatch().

Referenced by VW::start_parser().

{ parse_dispatch(*all, thread_dispatch); }

make_write_cache()

void make_write_cache	(	vw &	all,
		std::string &	newname,
		bool	quiet
	)

Definition at line 223 of file parser.cc.

References io_buf::currentname, f, io_buf::files, io_buf::finalname, vw::num_bits, io_buf::open_file(), parser::output, vw::p, push_many(), v_array< T >::size(), vw::stdin_off, prediction_type::to_string(), vw::trace_message, VW::version(), io_buf::WRITE, parser::write_cache, and io_buf::write_file().

Referenced by parse_cache().

{
  io_buf* output = all.p->output;
  if (output->files.size() != 0)
  {
    all.trace_message << "Warning: you tried to make two write caches.  Only the first one will be made." << endl;
    return;
  }

  std::string temp = newname + std::string(".writing");
  push_many(output->currentname, temp.c_str(), temp.length() + 1);

  int f = output->open_file(temp.c_str(), all.stdin_off, io_buf::WRITE);
  if (f == -1)
  {
    all.trace_message << "can't create cache file !" << endl;
    return;
  }

  size_t v_length = (uint64_t)VW::version.to_string().length() + 1;

  output->write_file(f, &v_length, sizeof(v_length));
  output->write_file(f, VW::version.to_string().c_str(), v_length);
  output->write_file(f, "c", 1);
  output->write_file(f, &all.num_bits, sizeof(all.num_bits));

  push_many(output->finalname, newname.c_str(), newname.length() + 1);
  all.p->write_cache = true;
  if (!quiet)
    all.trace_message << "creating cache_file = " << newname << endl;
}

parse_cache()

void parse_cache	(	vw &	all,
		std::vector< std::string >	cache_files,
		bool	kill_cache,
		bool	quiet
	)

Definition at line 255 of file parser.cc.

References c, cache_numbits(), io_buf::close_file(), v_array< T >::delete_v(), f, parser::input, make_write_cache(), vw::num_bits, io_buf::open_file(), parser::output, vw::p, vw::parse_mask, io_buf::READ, read_cached_features(), parser::reader, parser::resettable, parser::sorted_cache, io_buf::space, vw::stdin_off, vw::trace_message, and parser::write_cache.

Referenced by enable_sources().

{
  all.p->write_cache = false;

  for (auto& file : cache_files)
  {
    int f = -1;
    if (!kill_cache)
      try
      {
        f = all.p->input->open_file(file.c_str(), all.stdin_off, io_buf::READ);
      }
      catch (const std::exception&)
      {
        f = -1;
      }
    if (f == -1)
      make_write_cache(all, file, quiet);
    else
    {
      uint64_t c = cache_numbits(all.p->input, f);
      if (c < all.num_bits)
      {
        if (!quiet)
          all.trace_message << "WARNING: cache file is ignored as it's made with less bit precision than required!"
                            << endl;
        all.p->input->close_file();
        make_write_cache(all, file, quiet);
      }
      else
      {
        if (!quiet)
          all.trace_message << "using cache_file = " << file.c_str() << endl;
        all.p->reader = read_cached_features;
        if (c == all.num_bits)
          all.p->sorted_cache = true;
        else
          all.p->sorted_cache = false;
        all.p->resettable = true;
      }
    }
  }

  all.parse_mask = ((uint64_t)1 << all.num_bits) - 1;
  if (cache_files.size() == 0)
  {
    if (!quiet)
      all.trace_message << "using no cache" << endl;
    all.p->output->space.delete_v();
  }
}

reset_source()

void reset_source	(	vw &	all,
		size_t	numbits
	)

Definition at line 126 of file parser.cc.

References v_array< T >::begin(), binary_print_result(), parser::bound_sock, cache_numbits(), v_array< T >::clear(), io_buf::close_file(), io_buf::close_file_or_socket(), io_buf::compressed(), io_buf::current, io_buf::currentname, vw::daemon, f, io_buf::files, vw::final_prediction_sink, io_buf::finalname, recall_tree_ns::find(), io_buf::flush(), parser::input, isbinary(), io_buf::num_files(), io_buf::open_file(), parser::output, parser::output_done, parser::output_lock, vw::p, v_array< T >::pop(), vw::print, print_result(), v_array< T >::push_back(), io_buf::READ, read_cached_features(), read_features_string(), parser::reader, parser::ready_parsed_examples, io_buf::reset_file(), parser::resettable, v_array< T >::size(), VW::ptr_queue< T >::size(), vw::stdin_off, THROW, and parser::write_cache.

Referenced by parse_dispatch().

{
  io_buf* input = all.p->input;
  input->current = 0;
  if (all.p->write_cache)
  {
    all.p->output->flush();
    all.p->write_cache = false;
    all.p->output->close_file();
    remove(all.p->output->finalname.begin());

    if (0 != rename(all.p->output->currentname.begin(), all.p->output->finalname.begin()))
      THROW("WARN: reset_source(vw& all, size_t numbits) cannot rename: " << all.p->output->currentname << " to "
                                                                          << all.p->output->finalname);

    while (input->num_files() > 0)
      if (input->compressed())
        input->close_file();
      else
      {
        int fd = input->files.pop();
        const auto& fps = all.final_prediction_sink;

        // If the current popped file is not in the list of final predictions sinks, close it.
        if (std::find(fps.cbegin(), fps.cend(), fd) == fps.cend())
          io_buf::close_file_or_socket(fd);
      }
    input->open_file(all.p->output->finalname.begin(), all.stdin_off, io_buf::READ);  // pushing is merged into
                                                                                      // open_file
    all.p->reader = read_cached_features;
  }
  if (all.p->resettable == true)
  {
    if (all.daemon)
    {
      // wait for all predictions to be sent back to client
      {
        std::unique_lock<std::mutex> lock(all.p->output_lock);
        all.p->output_done.wait(lock, [&] { return all.p->ready_parsed_examples.size() == 0; });
      }

      // close socket, erase final prediction sink and socket
      io_buf::close_file_or_socket(all.p->input->files[0]);
      all.final_prediction_sink.clear();
      all.p->input->files.clear();

      sockaddr_in client_address;
      socklen_t size = sizeof(client_address);
      int f = (int)accept(all.p->bound_sock, (sockaddr*)&client_address, &size);
      if (f < 0)
        THROW("accept: " << strerror(errno));

      // note: breaking cluster parallel online learning by dropping support for id

      all.final_prediction_sink.push_back((size_t)f);
      all.p->input->files.push_back(f);

      if (isbinary(*(all.p->input)))
      {
        all.p->reader = read_cached_features;
        all.print = binary_print_result;
      }
      else
      {
        all.p->reader = read_features_string;
        all.print = print_result;
      }
    }
    else
    {
      for (size_t i = 0; i < input->files.size(); i++)
      {
        input->reset_file(input->files[i]);
        if (cache_numbits(input, input->files[i]) < numbits)
          THROW("argh, a bug in caching of some sort!");
      }
    }
  }
}

set_compressed()

void set_compressed

(

parser *

par

)

Definition at line 82 of file parser.cc.

References finalize_source(), parser::input, and parser::output.

Referenced by parse_source().

{
  finalize_source(par);
  delete par->input;
  par->input = new comp_io_buf;
  delete par->output;
  par->output = new comp_io_buf;
}

set_done()

void set_done

(

vw &

all

)

Definition at line 578 of file parser.cc.

References vw::early_terminate, lock_done(), and vw::p.

Referenced by GD::end_pass(), end_pass(), and finish_example().

{
  all.early_terminate = true;
  lock_done(*all.p);
}

thread_dispatch()

void thread_dispatch	(	vw &	all,
		v_array< example *>	examples
	)

Definition at line 896 of file parser.cc.

References parser::end_parsed_examples, vw::p, VW::ptr_queue< T >::push(), parser::ready_parsed_examples, and v_array< T >::size().

Referenced by main_parse_loop().

{
  all.p->end_parsed_examples += examples.size();
  for (auto example : examples)
  {
    all.p->ready_parsed_examples.push(example);
  }
}

Variable Documentation

got_sigterm

bool got_sigterm

Definition at line 66 of file parser.cc.