Vector in = (Vector)RegisterStreams.getNodeInStreams(node.contents); Vector out_v = (Vector)RegisterStreams.getNodeOutStreams(node.contents); NetStream out = (NetStream)out_v.elementAt(0);
List<NetStream> in = RegisterStreams.getNodeInStreams(node.contents); List<NetStream> out_v = RegisterStreams.getNodeOutStreams(node.contents); NetStream out = out_v.get(0);
    @Deprecated public static void generateJoiner2(FlatNode node) {                SIRJoiner joiner = (SIRJoiner)node.contents;        // The joiner is not doing any work        if (joiner.getSumOfWeights() == 0) return;         int init_counts, steady_counts;        Integer init_int = (Integer)ClusterBackend.initExecutionCounts.get(node);        if (init_int==null) {            init_counts = 0;        } else {            init_counts = init_int.intValue();        }        steady_counts = ((Integer)ClusterBackend.steadyExecutionCounts.get(node)).intValue();        CType baseType = CommonUtils.getBaseType(CommonUtils.getJoinerType(node));        int id = NodeEnumerator.getSIROperatorId(node.contents);        Vector in = (Vector)RegisterStreams.getNodeInStreams(node.contents);        Vector out_v = (Vector)RegisterStreams.getNodeOutStreams(node.contents);        NetStream out = (NetStream)out_v.elementAt(0);        int sum_of_weights = joiner.getSumOfWeights();            CodegenPrintWriter p = null;        try {            p = new CodegenPrintWriter(new FileWriter("thread_"+id+".cpp"));        } catch (IOException e) {            e.printStackTrace();            System.exit(1);        }        p.println("#include <stream_node.h>\n");        p.println("class thread"+id+" : public stream_node<"+baseType+","+baseType+"> {");        p.println("public:");        p.println("  thread"+id+"() : stream_node<"+baseType+","+baseType+">("+id+","+NodeEnumerator.getNumberOfNodes()+","+sum_of_weights+","+sum_of_weights+","+sum_of_weights+") {");        for (int i = 0; i < in.size(); i++) {            NetStream s = (NetStream)in.elementAt(i);                   int num = joiner.getWeight(i);              p.println("    add_input_rate("+s.getSource()+","+num+");");        }        p.println("    add_output("+out.getDest()+");");            p.println("  }");        p.newLine();        p.println("  "+baseType+" buf["+sum_of_weights+"];");        p.newLine();        p.println("  int state_size() { return 0; }");        p.println("  void save_state(object_write_buffer *buf) {}");        p.println("  void load_state(object_write_buffer *buf) {}");        p.println("  void init_state() {}");        p.println("  void send_credits() {}");        p.println("  void exec_message(message *msg) {}");        p.newLine();            p.println("  void work() {");        int offs = 0;        for (int i = 0; i < in.size(); i++) {            NetStream s = (NetStream)in.elementAt(i);                   int num = joiner.getWeight(i);              p.println("    consumer_array["+s.getSource()+"]->pop_items(buf+"+offs+", "+num+");");            offs += num;        }        p.println("    producer_array["+out.getDest()+"]->push_items(buf, "+sum_of_weights+");");        p.println("  }");        p.newLine();        p.println("  void work_n(int __n) {");        p.println("    for (int y = 0; y < __n; y++) {");        offs = 0;        for (int i = 0; i < in.size(); i++) {            NetStream s = (NetStream)in.elementAt(i);                   int num = joiner.getWeight(i);              p.println("      consumer_array["+s.getSource()+"]->pop_items(buf+"+offs+", "+num+");");            offs += num;        }        p.println("      producer_array["+out.getDest()+"]->push_items(buf, "+sum_of_weights+");");        p.println("    }");        p.println("  }");        p.newLine();        p.println("};");        p.newLine();        p.println("thread"+id+" *instance_"+id+" = NULL;");        p.println("thread"+id+" *get_instance_"+id+"() {");        p.println("  if (instance_"+id+" == NULL) { instance_"+id+" = new thread"+id+"();");        p.println("    instance_"+id+"->init_stream(); }");        p.println("  return instance_"+id+";");        p.println("}");        p.println("void __get_thread_info_"+id+"() { get_instance_"+id+"()->get_thread_info(); }");        p.println("void __declare_sockets_"+id+"() { get_instance_"+id+"()->declare_sockets(); }");        p.println("extern int __max_iteration;");        p.println("void run_"+id+"() { get_instance_"+id+"()->run_simple("+init_counts+"+("+steady_counts+"*__max_iteration)); }");        try {            p.close();        }        catch (Exception e) {            System.err.println("Unable to write code to file thread_"+id+".cpp");        }            if (ClusterBackend.debugPrint)            System.out.println("Code for " + node.contents.getName() +                               " written to thread_"+id+".cpp");    }