<commit_before>/*
 * This file is distributed under the terms in the attached LICENSE file.
 * If you do not find this file, copies can be found by writing to:
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300,
 * Berkeley, CA, 94704.  Attention:  Intel License Inquiry.
 * Or
 * UC Berkeley EECS Computer Science Division, 387 Soda Hall #1776, 
 * Berkeley, CA,  94707. Attention: P2 Group.
 * 
 * DESCRIPTION: Element which simply prints any tuple that passes
 * through it.
 */

#include "p2.h"
#include "loop.h"
#include "udp.h"
#include "dDuplicateConservative.h"
#include "overlogCompiler.h"
#include "dataflowInstaller.h"
#include "timedPullPush.h"
#include "queue.h"
#include "ol_lexer.h"
#include "ol_context.h"
#include "plmb_confgen.h"
#include "val_str.h"
#include <iostream>

P2::P2(string hostname, string port, uint tc) 
  : _plumber(new Plumber()), _id(hostname+":"+port), _transport_conf(tc)
{
  eventLoopInitialize();
  Py_Initialize();
  boost::python::object 
    parser(boost::python::handle<> 
           (boost::python::borrowed(PyImport_ImportModule("dfparser"))));
  _parser = parser;
  install("dataflow", stub(hostname, port));
  _tupleSourceInterface = dynamic_cast<TupleSourceInterface*>( 
    _plumber->dataflow("P2")->find("tupleSourceInterface")->element().get());
  _plumber->toDot("p2.dot");

  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr dDemux      = edit->find("dDemux");
  ElementSpecPtr dRoundRobin = edit->find("dRoundRobin");
  ElementSpecPtr qinput      = edit->addElement(
    ElementPtr(new Queue("demuxQueue_overlogInstall", 100)));
  ElementSpecPtr qoutput     = edit->addElement(
    ElementPtr(new Queue("output_queue", 100)));
  ElementSpecPtr pullPush    = edit->addElement(
    ElementPtr(new TimedPullPush("demuxTimedPullPush_overlogInstall", 0)));
  ElementSpecPtr duplicator  = edit->addElement(
    ElementPtr(new DDuplicateConservative("dc_overlogInstall", 1)));
  ElementSpecPtr overlogCompiler = edit->addElement(
    ElementPtr(new OverlogCompiler("overlogCompiler", _plumber, _id, "P2")));
  ElementSpecPtr dataflowInstaller = edit->addElement(
    ElementPtr(new DataflowInstaller("dataflowInstaller", _plumber, _parser)));

  edit->hookUp(dDemux, dDemux->add_output(Val_Str::mk("overlogInstall")), qinput, 0);
  edit->hookUp(qinput, 0, pullPush, 0);
  edit->hookUp(pullPush, 0, duplicator, 0);
  edit->hookUp(duplicator, 0, overlogCompiler, 0);
  edit->hookUp(overlogCompiler, 0, dataflowInstaller, 0);
  edit->hookUp(dataflowInstaller, 0, qoutput, 0);
  edit->hookUp(qoutput, 0, dRoundRobin, dRoundRobin->add_input());
  if (_plumber->install(edit) < 0) {
    std::cerr << "OVERLOG AND DATAFLOW INSTALLER FAILURE" << std::endl;
    exit(1);
  }
#ifdef DEBUG
  _plumber->toDot("p2.dot");
#endif
}


int P2::install(string type, string program)
{
  if (type == string("overlog")) {
    ostringstream script;
    compileOverlog(program, script);
    program = script.str();
    type = "dataflow";
  }

  if (type == string("dataflow")) {
    _parser.attr("clear")();
    boost::python::tuple result =
      boost::python::extract<boost::python::tuple>(
        _parser.attr("compile")(_plumber, program));
    boost::python::dict dataflows = 
      boost::python::extract<boost::python::dict>(result[0]);
    boost::python::list edits     = 
      boost::python::extract<boost::python::list>(result[1]);

    int ndataflows = boost::python::extract<int>(dataflows.attr("__len__")());
    int nedits     = boost::python::extract<int>(edits.attr("__len__")());

    if (ndataflows) {
      boost::python::tuple t = dataflows.popitem();
      t[1].attr("eval_dataflow")();
      Plumber::DataflowPtr d = 
        boost::python::extract<Plumber::DataflowPtr>(t[1].attr("conf"));
      if (_plumber->install(d) < 0) {
        std::cerr << "DATAFLOW INSTALLATION FAILURE" << std::endl;
        return -1;
      }
    }
    if (nedits) {
      edits[0].attr("eval_dataflow")();
      Plumber::DataflowEditPtr e = 
        boost::python::extract<Plumber::DataflowEditPtr>(edits[0].attr("conf"));
      if (_plumber->install(e) < 0) {
        std::cerr << "EDIT INSTALLATION FAILURE" << std::endl;
        return -1;
      }
    }
#ifdef DEBUG
    _plumber->toDot("p2.dot");
#endif
  }
  return 0;
}

void P2::run()
{
  eventLoop();
}

P2::CallbackHandlePtr P2::subscribe(string tupleName, cb_tp callback)
{
  int port = -1;	// The port that the listener sits on
  // Create a new edit 
  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr listener = 
    edit->addElement(ElementPtr(new TupleListener("listener_"+tupleName, callback)));

  // The planner creates the duplicator name, see if one already exists 
  ElementSpecPtr duplicator = edit->find("dc_"+tupleName);
  if (!duplicator) {
    /** We want to listen for something that has no rule strand
     *  We first need to register it with the demux then hang a duplicator off of the demux
     *  A rule strand that is installed later, and assumes the tuple name is an event,
     *  will find this duplicator and install the strand properly. At this point this
     *  event can no longer be materialized */
    ElementSpecPtr ddemux   = edit->find("dDemux");
    ElementSpecPtr queue    = edit->addElement(
      ElementPtr(new Queue("demuxQueue_"+tupleName, 100)));
    ElementSpecPtr pullPush = edit->addElement(
      ElementPtr(new TimedPullPush("demuxTimedPullPush_"+tupleName, 0)));
    duplicator = edit->addElement(
      ElementPtr(new DDuplicateConservative("dc_"+tupleName, 0)));

    // Hook everything up except duplicator output
    edit->hookUp(ddemux, ddemux->add_output(Val_Str::mk(tupleName)), queue, 0);
    edit->hookUp(queue, 0, pullPush, 0);
    edit->hookUp(pullPush, 0, duplicator, 0);
  }
  // Not hookup the listener to the duplicator output
  port = duplicator->add_output();
  edit->hookUp(duplicator, port, listener, 0);
  if (_plumber->install(edit) < 0) {
    std::cerr << "ERROR: Couldn't install listener" << std::endl;
    return P2::CallbackHandlePtr();
  }
#ifdef DEBUG
  _plumber->toDot("p2.dot");
#endif
  // Return a handle to the caller for unsubscribing to the listener.
  return P2::CallbackHandlePtr(new P2::CallbackHandle(tupleName, port));	
}

void P2::unsubscribe(P2::CallbackHandlePtr handle)
{
  if (!handle || handle->_valid == false) {
    std::cerr << "ERROR: invalid callback handle." << std::endl;
  } 
  else {
    // Invalidate the handle
    handle->_valid = false;
  }
  unsigned port    = handle->_port;
  string tupleName = handle->_name;

  /* All this routine does it take out an edit, uses it to locate the
   * duplicator, and removes the port. */
  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr     duplicator = edit->find("dc_"+tupleName);
  if (!duplicator) {
    std::cerr << "ERROR: no listener registered for " << tupleName << std::endl;
    return;
  }
  edit->disconnect_output(duplicator, port);
  if (_plumber->install(edit) < 0) {
    std::cerr << "ERROR: uninstall listener" << std::endl;
    return;
  }

}

int P2::tuple(TuplePtr tp)
{
  /* Using the tupleSourceInterface, attached right before the demux, to 
   * inject a tuple into the system. */
  return _tupleSourceInterface->tuple(tp);
}

void P2::compileOverlog(string overlog, std::ostringstream& script) {
  std::istringstream overlog_iss(overlog, std::istringstream::in);
  boost::shared_ptr< OL_Context > ctxt(new OL_Context());
  ctxt->parse_stream(&overlog_iss);

  Plumber::DataflowEditPtr conf = _plumber->new_dataflow_edit("P2");
  Plmb_ConfGen *gen = new Plmb_ConfGen(ctxt.get(), conf, false, false, false, 
                                       string("overlogCompiler"), false, script, true);
  gen->createTables(_id);
  gen->configurePlumber(boost::shared_ptr<Udp>(), _id);
}

string P2::stub(string hostname, string port)
{
  ostringstream stub;
  stub << "dataflow P2 { \
           let udp = Udp2(\"udp\"," << port << "); \
           let wrapAroundDemux = Demux(\"wrapAroundSendDemux\", \{Val_Str(\""
                                                                << hostname << ":" << port
                                                                << "\")}, 0); \
           let inputRR = RoundRobin(\"inputRR\", 3); \
           let header  = PelTransform(\"source\", \"$0 pop \\\""
                                      << hostname <<":" << port 
                                      << "\\\" pop swallow unbox drop "
                                      << "0 pop 0 pop 0 pop 0 pop 0 pop 0 pop popall\"); ";


  if (_transport_conf & (RELIABLE | RCC | CC | ORDERED)) {
    stub << "let rtt      = RoundTripTimer(\"rtt_timer\"); ";
    stub << "let ackDemux = Demux(\"ackDemux\", {Val_Str(\"ACK\")}, 1); ";
    stub << "let netoutRR = RoundRobin(\"netoutRR\", 2); ";
    stub << "let ack = BasicAck(\"acknowledge\"); ";

    if (_transport_conf & CC) {
      stub << "let cct = CCT(\"cct\"); ";
    }
    else if (_transport_conf & RCC) {
      stub << "let rcct = RateCCT(\"rcct\"); ";
    }

    if (_transport_conf & RELIABLE)  {
      stub << "let rdelv = RDelivery(\"reliable_delivery\"); ";
    }
  }

  stub << "udp-> UnmarshalField(\"unmarshal\", 1) -> \
           PelTransform(\"unRoute\", \"$1 unboxPop\") ->";

  stub << "Defrag(\"defrag\") -> TimedPullPush(\"defrag_pull\", 0) ->";
  if (_transport_conf & (RCC | CC | RELIABLE | ORDERED)) {
    stub << "ackDemux[1] -> ack ->";
  }

  if (_transport_conf & ORDERED) {
    stub << "ODelivery(\"order\") ->";
  }
  else if (_transport_conf & RELIABLE) {
    stub << "DupRemove(\"dupremove\") ->";
  }
  stub << "PelTransform(\"unPackage\", \"$8 unboxPop\")->";

  stub << "Queue -> inputRR -> \
           TimedPullPush(\"pullDriver\", 0)     -> \
           PrintWatch(\"printWatch\", {str}) -> \
           DDemux(\"dDemux\", {value}, 0) ->  \
           Discard(\"discard\"); \
           DRoundRobin(\"dRoundRobin\", 0) -> \
           TimedPullPush(\"rrout_pullPush\", 0) -> \
           wrapAroundDemux -> \
           UnboxField(\"unboxWrapAround\", 1) -> \
           Queue(\"wrapAroundQueue\", 1000) -> \
           [1]inputRR; \
           wrapAroundDemux[1] -> Queue -> \
           header -> \
           Sequence(\"terminal_sequence\", 1) ->";


  /** Send side reliable delivery and congestion control */
  if (_transport_conf & (RELIABLE | RCC | CC | ORDERED)) {
    stub << "rtt ->";<commit_msg>*** empty log message ***<commit_after>/*
 * This file is distributed under the terms in the attached LICENSE file.
 * If you do not find this file, copies can be found by writing to:
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300,
 * Berkeley, CA, 94704.  Attention:  Intel License Inquiry.
 * Or
 * UC Berkeley EECS Computer Science Division, 387 Soda Hall #1776, 
 * Berkeley, CA,  94707. Attention: P2 Group.
 * 
 * DESCRIPTION: Element which simply prints any tuple that passes
 * through it.
 */

#include "p2.h"
#include "loop.h"
#include "udp.h"
#include "dDuplicateConservative.h"
#include "overlogCompiler.h"
#include "dataflowInstaller.h"
#include "timedPullPush.h"
#include "queue.h"
#include "ol_lexer.h"
#include "ol_context.h"
#include "plmb_confgen.h"
#include "val_str.h"
#include <iostream>

P2::P2(string hostname, string port, uint tc) 
  : _plumber(new Plumber()), _id(hostname+":"+port), _transport_conf(tc)
{
  eventLoopInitialize();
  Py_Initialize();
  boost::python::object 
    parser(boost::python::handle<> 
           (boost::python::borrowed(PyImport_ImportModule("dfparser"))));
  _parser = parser;
  install("dataflow", stub(hostname, port));
  _tupleSourceInterface = dynamic_cast<TupleSourceInterface*>( 
    _plumber->dataflow("P2")->find("tupleSourceInterface")->element().get());
  _plumber->toDot("p2.dot");

  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr dDemux      = edit->find("dDemux");
  ElementSpecPtr dRoundRobin = edit->find("dRoundRobin");
  ElementSpecPtr qinput      = edit->addElement(
    ElementPtr(new Queue("demuxQueue_overlogInstall", 100)));
  ElementSpecPtr qoutput     = edit->addElement(
    ElementPtr(new Queue("output_queue", 100)));
  ElementSpecPtr pullPush    = edit->addElement(
    ElementPtr(new TimedPullPush("demuxTimedPullPush_overlogInstall", 0)));
  ElementSpecPtr duplicator  = edit->addElement(
    ElementPtr(new DDuplicateConservative("dc_overlogInstall", 1)));
  ElementSpecPtr overlogCompiler = edit->addElement(
    ElementPtr(new OverlogCompiler("overlogCompiler", _plumber, _id, "P2")));
  ElementSpecPtr dataflowInstaller = edit->addElement(
    ElementPtr(new DataflowInstaller("dataflowInstaller", _plumber, _parser)));

  edit->hookUp(dDemux, dDemux->add_output(Val_Str::mk("overlogInstall")), qinput, 0);
  edit->hookUp(qinput, 0, pullPush, 0);
  edit->hookUp(pullPush, 0, duplicator, 0);
  edit->hookUp(duplicator, 0, overlogCompiler, 0);
  edit->hookUp(overlogCompiler, 0, dataflowInstaller, 0);
  edit->hookUp(dataflowInstaller, 0, qoutput, 0);
  edit->hookUp(qoutput, 0, dRoundRobin, dRoundRobin->add_input());
  if (_plumber->install(edit) < 0) {
    std::cerr << "OVERLOG AND DATAFLOW INSTALLER FAILURE" << std::endl;
    exit(1);
  }
#ifdef DEBUG
  _plumber->toDot("p2.dot");
#endif
}


int P2::install(string type, string program)
{
  if (type == string("overlog")) {
    ostringstream script;
    compileOverlog(program, script);
    program = script.str();
    type = "dataflow";
  }

  if (type == string("dataflow")) {
    _parser.attr("clear")();
    boost::python::tuple result =
      boost::python::extract<boost::python::tuple>(
        _parser.attr("compile")(_plumber, program));
    boost::python::dict dataflows = 
      boost::python::extract<boost::python::dict>(result[0]);
    boost::python::list edits     = 
      boost::python::extract<boost::python::list>(result[1]);

    int ndataflows = boost::python::extract<int>(dataflows.attr("__len__")());
    int nedits     = boost::python::extract<int>(edits.attr("__len__")());

    if (ndataflows) {
      boost::python::tuple t = dataflows.popitem();
      t[1].attr("eval_dataflow")();
      Plumber::DataflowPtr d = 
        boost::python::extract<Plumber::DataflowPtr>(t[1].attr("conf"));
      if (_plumber->install(d) < 0) {
        std::cerr << "DATAFLOW INSTALLATION FAILURE" << std::endl;
        return -1;
      }
    }
    if (nedits) {
      edits[0].attr("eval_dataflow")();
      Plumber::DataflowEditPtr e = 
        boost::python::extract<Plumber::DataflowEditPtr>(edits[0].attr("conf"));
      if (_plumber->install(e) < 0) {
        std::cerr << "EDIT INSTALLATION FAILURE" << std::endl;
        return -1;
      }
    }
#ifdef DEBUG
    _plumber->toDot("p2.dot");
#endif
  }
  return 0;
}

void P2::run()
{
  eventLoop();
}

P2::CallbackHandlePtr P2::subscribe(string tupleName, cb_tp callback)
{
  int port = -1;	// The port that the listener sits on
  // Create a new edit 
  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr listener = 
    edit->addElement(ElementPtr(new TupleListener("listener_"+tupleName, callback)));

  // The planner creates the duplicator name, see if one already exists 
  ElementSpecPtr duplicator = edit->find("dc_"+tupleName);
  if (!duplicator) {
    /** We want to listen for something that has no rule strand
     *  We first need to register it with the demux then hang a duplicator off of the demux
     *  A rule strand that is installed later, and assumes the tuple name is an event,
     *  will find this duplicator and install the strand properly. At this point this
     *  event can no longer be materialized */
    ElementSpecPtr ddemux   = edit->find("dDemux");
    ElementSpecPtr queue    = edit->addElement(
      ElementPtr(new Queue("demuxQueue_"+tupleName, 100)));
    ElementSpecPtr pullPush = edit->addElement(
      ElementPtr(new TimedPullPush("demuxTimedPullPush_"+tupleName, 0)));
    duplicator = edit->addElement(
      ElementPtr(new DDuplicateConservative("dc_"+tupleName, 0)));

    // Hook everything up except duplicator output
    edit->hookUp(ddemux, ddemux->add_output(Val_Str::mk(tupleName)), queue, 0);
    edit->hookUp(queue, 0, pullPush, 0);
    edit->hookUp(pullPush, 0, duplicator, 0);
  }
  // Not hookup the listener to the duplicator output
  port = duplicator->add_output();
  edit->hookUp(duplicator, port, listener, 0);
  if (_plumber->install(edit) < 0) {
    std::cerr << "ERROR: Couldn't install listener" << std::endl;
    return P2::CallbackHandlePtr();
  }
#ifdef DEBUG
  _plumber->toDot("p2.dot");
#endif
  // Return a handle to the caller for unsubscribing to the listener.
  return P2::CallbackHandlePtr(new P2::CallbackHandle(tupleName, port));	
}

void P2::unsubscribe(P2::CallbackHandlePtr handle)
{
  if (!handle || handle->_valid == false) {
    std::cerr << "ERROR: invalid callback handle." << std::endl;
  } 
  else {
    // Invalidate the handle
    handle->_valid = false;
  }
  unsigned port    = handle->_port;
  string tupleName = handle->_name;

  /* All this routine does it take out an edit, uses it to locate the
   * duplicator, and removes the port. */
  Plumber::DataflowEditPtr edit = _plumber->new_dataflow_edit("P2");
  ElementSpecPtr     duplicator = edit->find("dc_"+tupleName);
  if (!duplicator) {
    std::cerr << "ERROR: no listener registered for " << tupleName << std::endl;
    return;
  }
  edit->disconnect_output(duplicator, port);
  if (_plumber->install(edit) < 0) {
    std::cerr << "ERROR: uninstall listener" << std::endl;
    return;
  }

}

int P2::tuple(TuplePtr tp)
{
  /* Using the tupleSourceInterface, attached right before the demux, to 
   * inject a tuple into the system. */
  return _tupleSourceInterface->tuple(tp);
}

void P2::compileOverlog(string overlog, std::ostringstream& script) {
  std::istringstream overlog_iss(overlog, std::istringstream::in);
  boost::shared_ptr< OL_Context > ctxt(new OL_Context());
  ctxt->parse_stream(&overlog_iss);

  Plumber::DataflowEditPtr conf = _plumber->new_dataflow_edit("P2");
  Plmb_ConfGen *gen = new Plmb_ConfGen(ctxt.get(), conf, false, false, false, 
                                       string("overlogCompiler"), false, script, true);
  gen->createTables(_id);
  gen->configurePlumber(boost::shared_ptr<Udp>(), _id);
}

string P2::stub(string hostname, string port)
{
  ostringstream stub;
  stub << "dataflow P2 { \
           let udp = Udp2(\"udp\"," << port << "); \
           let wrapAroundDemux = Demux(\"wrapAroundSendDemux\", \{Val_Str(\""
                                                                << hostname << ":" << port
                                                                << "\")}, 0); \
           let inputRR = RoundRobin(\"inputRR\", 3); \
           let header  = PelTransform(\"source\", \"$0 pop \\\""
                                      << hostname <<":" << port 
                                      << "\\\" pop swallow unbox drop "
                                      << "0 pop 0 pop 0 pop 0 pop 0 pop 0 pop popall\"); ";


  if (_transport_conf & (RELIABLE | RCC | CC | ORDERED)) {
    stub << "let rtt      = RoundTripTimer(\"rtt_timer\"); ";
    stub << "let ackDemux = Demux(\"ackDemux\", {Val_Str(\"ACK\")}, 1); ";
    stub << "let netoutRR = RoundRobin(\"netoutRR\", 2); ";
    stub << "let ack = BasicAck(\"acknowledge\"); ";

    if (_transport_conf & CC) {
      stub << "let cct = CCT(\"cct\"); ";
    }
    else if (_transport_conf & RCC) {
      stub << "let rcct = RateCCT(\"rcct\"); ";
    }

    if (_transport_conf & RELIABLE)  {
      stub << "let rdelv = RDelivery(\"reliable_delivery\"); ";
    }
  }

  stub << "udp-> UnmarshalField(\"unmarshal\", 1) -> \
           PelTransform(\"unRoute\", \"$1 unboxPop\") ->";

  stub << "Defrag(\"defrag\") -> TimedPullPush(\"defrag_pull\", 0) ->";
  if (_transport_conf & (RCC | CC | RELIABLE | ORDERED)) {
    stub << "ackDemux[1] -> ack ->";
  }

  if (_transport_conf & ORDERED) {
    stub << "ODelivery(\"order\") ->";
  }
  else if (_transport_conf & RELIABLE) {
    // stub << "DupRemove(\"dupremove\") ->";
  }
  stub << "PelTransform(\"unPackage\", \"$8 unboxPop\")->";

  stub << "Queue -> inputRR -> \
           TimedPullPush(\"pullDriver\", 0)     -> \
           PrintWatch(\"printWatch\", {str}) -> \
           DDemux(\"dDemux\", {value}, 0) ->  \
           Discard(\"discard\"); \
           DRoundRobin(\"dRoundRobin\", 0) -> \
           TimedPullPush(\"rrout_pullPush\", 0) -> \
           wrapAroundDemux -> \
           UnboxField(\"unboxWrapAround\", 1) -> \
           Queue(\"wrapAroundQueue\", 1000) -> \
           [1]inputRR; \
           wrapAroundDemux[1] -> Queue -> \
           header -> \
           Sequence(\"terminal_sequence\", 1) ->";


  /** Send side reliable delivery and congestion control */
  if (_transport_conf & (RELIABLE | RCC | CC | ORDERED)) {
    stub << "rtt ->";
{'fraction_non_alphanumeric': 0.15727441904852793, 'fraction_numerical': 0.012215488858138092, 'mean_word_length': 3.227107704719645, 'pattern_counts': {'":': 6, '<': 205, '<?xml version=': 0, '>': 187, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2695663', 'n_tokens_mistral': 7094, 'n_tokens_neox': 6778, 'n_words': 1552}