sender.cc File Reference

#include <vector>
#include <netdb.h>
#include "io_buf.h"
#include "cache.h"
#include "network.h"
#include "reductions.h"

Go to the source code of this file.

Classes
struct	sender

Functions
void	open_sockets (sender &s, std::string host)
void	send_features (io_buf *b, example &ec, uint32_t mask)
void	receive_result (sender &s)
void	learn (sender &s, LEARNER::single_learner &, example &ec)
void	finish_example (vw &, sender &, example &)
void	end_examples (sender &s)
LEARNER::base_learner *	sender_setup (options_i &options, vw &all)

Function Documentation

end_examples()

void end_examples

(

sender &

s

)

Definition at line 93 of file sender.cc.

References sender::buf, io_buf::files, receive_result(), sender::received_index, and sender::sent_index.

Referenced by sender_setup().

{
  // close our outputs to signal finishing.
  while (s.received_index != s.sent_index) receive_result(s);
  shutdown(s.buf->files[0], SHUT_WR);
}

finish_example()

void finish_example	(	vw &	,
		sender &	,
		example &
	)

Definition at line 91 of file sender.cc.

{}

learn()

void learn	(	sender &	s,
		LEARNER::single_learner &	,
		example &	ec
	)

Definition at line 79 of file sender.cc.

References sender::all, sender::buf, label_parser::cache_label, cache_tag(), sender::delay_ring, example::l, label_data::label, parser::lp, vw::p, vw::parse_mask, receive_result(), sender::received_index, parser::ring_size, vw::sd, send_features(), sender::sent_index, vw::set_minmax, polylabel::simple, and example::tag.

Referenced by sender_setup().

{
  if (s.received_index + s.all->p->ring_size / 2 - 1 == s.sent_index)
    receive_result(s);

  s.all->set_minmax(s.all->sd, ec.l.simple.label);
  s.all->p->lp.cache_label(&ec.l, *s.buf);  // send label information.
  cache_tag(*s.buf, ec.tag);
  send_features(s.buf, ec, (uint32_t)s.all->parse_mask);
  s.delay_ring[s.sent_index++ % s.all->p->ring_size] = &ec;
}

open_sockets()

void open_sockets	(	sender &	s,
		std::string	host
	)

Definition at line 44 of file sender.cc.

References sender::buf, io_buf::files, open_socket(), v_array< T >::push_back(), and sender::sd.

Referenced by sender_setup().

{
  s.sd = open_socket(host.c_str());
  s.buf = new io_buf();
  s.buf->files.push_back(s.sd);
}

receive_result()

void receive_result

(

sender &

s

)

Definition at line 65 of file sender.cc.

References sender::all, sender::delay_ring, VW::get_prediction(), loss_function::getLoss(), example::l, label_data::label, example::loss, vw::loss, vw::p, example::pred, sender::received_index, return_simple_example(), parser::ring_size, polyprediction::scalar, sender::sd, vw::sd, polylabel::simple, and example::weight.

Referenced by end_examples(), and learn().

{
  float res, weight;

  get_prediction(s.sd, res, weight);
  example& ec = *s.delay_ring[s.received_index++ % s.all->p->ring_size];
  ec.pred.scalar = res;

  label_data& ld = ec.l.simple;
  ec.loss = s.all->loss->getLoss(s.all->sd, ec.pred.scalar, ld.label) * ec.weight;

  return_simple_example(*(s.all), nullptr, ec);
}

send_features()

void send_features	(	io_buf *	b,
		example &	ec,
		uint32_t	mask
	)

Definition at line 51 of file sender.cc.

References constant_namespace, example_predict::feature_space, io_buf::flush(), example_predict::indices, output_byte(), output_features(), and v_array< T >::size().

Referenced by learn().

{
  // note: subtracting 1 b/c not sending constant
  output_byte(*b, (unsigned char)(ec.indices.size() - 1));

  for (namespace_index ns : ec.indices)
  {
    if (ns == constant_namespace)
      continue;
    output_features(*b, ns, ec.feature_space[ns], mask);
  }
  b->flush();
}

sender_setup()

LEARNER::base_learner* sender_setup	(	options_i &	options,
		vw &	all
	)

Definition at line 100 of file sender.cc.

References VW::config::option_group_definition::add(), VW::config::options_i::add_and_parse(), end_examples(), VW::finish_example(), LEARNER::init_learner(), learn(), LEARNER::make_base(), VW::config::make_option(), open_sockets(), vw::p, parser::ring_size, and VW::config::options_i::was_supplied().

Referenced by parse_reductions().

{
  std::string host;

  option_group_definition sender_options("Network sending");
  sender_options.add(make_option("sendto", host).keep().help("send examples to <host>"));
  options.add_and_parse(sender_options);

  if (!options.was_supplied("sendto"))
  {
    return nullptr;
  }

  auto s = scoped_calloc_or_throw<sender>();
  s->sd = -1;
  open_sockets(*s.get(), host);

  s->all = &all;
  s->delay_ring = calloc_or_throw<example*>(all.p->ring_size);

  LEARNER::learner<sender, example>& l = init_learner(s, learn, learn, 1);
  l.set_finish_example(finish_example);
  l.set_end_examples(end_examples);
  return make_base(l);
}