parser.h File Reference

#include "io_buf.h"
#include "parse_primitives.h"
#include "example.h"
#include "future_compat.h"
#include <mutex>
#include <condition_variable>
#include <memory>
#include "queue.h"
#include "object_pool.h"

Go to the source code of this file.

Classes
struct	example_initializer
struct	parser

Functions
void	enable_sources (vw &all, bool quiet, size_t passes, input_options &input_options)
void	adjust_used_index (vw &all)
void	lock_done (parser &p)
void	set_done (vw &all)
void	reset_source (vw &all, size_t numbits)
void	finalize_source (parser *source)
void	set_compressed (parser *par)
void	free_parser (vw &all)

Function Documentation

adjust_used_index()

void adjust_used_index

(

vw &

all

)

Definition at line 968 of file parser.cc.

Referenced by VW_Finish_Passes().

{ /* no longer used */
}

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;
}

finalize_source()

void finalize_source

(

parser *

source

)

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();
}

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();
}

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);
}