r.add("producer2<"+TypeToC(out.getType())+"> "+out.producer_name()+";\n"); r.add("extern void "+out.push_name()+"("+TypeToC(out.getType())+" data);\n");
r.add("producer2<"+ClusterUtils.CTypeToString(out.getType())+"> "+out.producer_name()+";\n"); r.add("extern void "+out.push_name()+"("+ClusterUtils.CTypeToString(out.getType())+" data);\n");
    public Vector generatePreamble() {	Vector r = new Vector();	Iterator i;	r.add("// ClusterFusion isEliminated: "+isEliminated+"\n");	Iterator iter = fusedWith.iterator();	while (iter.hasNext()) {	    r.add("// Fused with: "+iter.next()+"\n");	}	r.add("\n");	r.add("#include <stdlib.h>\n");	r.add("#include <unistd.h>\n");	r.add("#include <math.h>\n");		r.add("\n");	r.add("#include <init_instance.h>\n");        r.add("#include <mysocket.h>\n");	r.add("#include <object_write_buffer.h>\n");	r.add("#include <save_state.h>\n");	r.add("#include <sdep.h>\n");	r.add("#include <message.h>\n");	r.add("#include <timer.h>\n");	r.add("#include <thread_info.h>\n");	r.add("#include <consumer2.h>\n");	r.add("#include <producer2.h>\n");	r.add("#include \"cluster.h\"\n");	r.add("#include \"fusion.h\"\n");	//r.add("#include \"structs.h\"\n");	r.add("#include \"global.h\"\n");	r.add("\n");	//r.add("#include <peek_stream.h>\n");	//r.add("#include <data_consumer.h>\n");	//r.add("#include <data_producer.h>\n");		r.add("extern int __max_iteration;\n");	r.add("extern int __init_iter;\n");	r.add("extern int __timer_enabled;\n");	r.add("extern int __frequency_of_chkpts;\n");	r.add("extern volatile int __vol;\n");	r.add("message *__msg_stack_"+id+";\n");	r.add("int __number_of_iterations_"+id+";\n");	r.add("int __counter_"+id+" = 0;\n");	r.add("int __steady_"+id+" = 0;\n");	r.add("int __tmp_"+id+" = 0;\n");	r.add("int __tmp2_"+id+" = 0;\n");	r.add("int *__state_flag_"+id+" = NULL;\n");	r.add("thread_info *__thread_"+id+" = NULL;\n");	if (restrictedExecution) {	    r.add("int __credit_"+id+" = "+initCredit+";\n");	}		i = msg_to.iterator();	while (i.hasNext()) {	    SIRStream str = (SIRStream)i.next();	    r.add("sdep *sdep_"+id+"_"+NodeEnumerator.getSIROperatorId(str)+";\n");	}	r.add("\n");		//  +=============================+	//  | Communication Variables     |	//  +=============================+	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    r.add("consumer2<"+TypeToC(in.getType())+"> "+in.consumer_name()+";\n");	    r.add("extern "+TypeToC(in.getType())+" "+in.pop_name()+"();\n");	    /*	    if (oper instanceof SIRFilter) {		String type = ((SIRFilter)oper).getInputType().toString();		r.add("peek_stream<"+type+"> "+in.name()+"in(&"+in.consumer_name()+");\n");	    }	    */	}		i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    r.add("producer2<"+TypeToC(out.getType())+"> "+out.producer_name()+";\n");	    r.add("extern void "+out.push_name()+"("+TypeToC(out.getType())+" data);\n");	    r.add("    // this-part:"+ClusterFusion.getPartition(NodeEnumerator.getFlatNode(id))+" dst-part:"+ClusterFusion.getPartition(NodeEnumerator.getFlatNode(out.getDest()))+"\n");	}		i = msg_from.iterator();	while (i.hasNext()) {	    int src = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("netsocket *__msg_sock_"+src+"_"+id+"in;\n");		}	i = msg_to.iterator();	while (i.hasNext()) {	    int dst = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("netsocket *__msg_sock_"+id+"_"+dst+"out;\n");	}		r.add("\n");	//  +=============================+	//  | Fields                      |	//  +=============================+	for (int f = 0; f < fields.length; f++) {	    CType type = fields[f].getType();	    if (type.toString().endsWith("Portal")) continue;	    JExpression init_val = fields[f].getVariable().getValue();	    String ident = fields[f].getVariable().getIdent();	    r.add(TypeToC(type)+" "+ident+"__"+id);	    if (init_val == null) {		if (type.isOrdinal()) r.add(" = 0");		if (type.isFloatingPoint()) r.add(" = 0.0f");	    }	    if (init_val != null && init_val instanceof JIntLiteral) {		r.add(" = "+((JIntLiteral)init_val).intValue());	    }	    if (init_val != null && init_val instanceof JFloatLiteral) {		r.add(" = "+((JFloatLiteral)init_val).floatValue());	    }	    r.add(";\n");	}	r.add("\n");	//  +=============================+	//  | Read / Write Thread         |	//  +=============================+	r.add("void save_peek_buffer__"+id+"(object_write_buffer *buf);\n");	r.add("void load_peek_buffer__"+id+"(object_write_buffer *buf);\n");	r.add("\n");	r.add("void __write_thread__"+id+"(object_write_buffer *buf) {\n");	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    r.add("  "+in.consumer_name()+".write_object(buf);\n");	    	    /*	    if (oper instanceof SIRFilter) {	       r.add("  "+in.name()+"in.write_object(buf);\n");	    }	    */	}	r.add("  save_peek_buffer__"+id+"(buf);\n");	for (int f = 0; f < fields.length; f++) {	    CType type = fields[f].getType();	    String ident = fields[f].getVariable().getIdent();	    DetectConst dc = DetectConst.getInstance((SIRFilter)oper);	    if (dc != null && dc.isConst(ident)) continue;	    if (type.isArrayType()) {		int size = 0;		String dims[] = ArrayDim.findDim((SIRFilter)oper, ident);		CType base = ((CArrayType)type).getBaseType();		try {		    size = Integer.valueOf(dims[0]).intValue();		} catch (NumberFormatException ex) {		    System.out.println("Warning! Could not estimate size of an array: "+ident);		}		r.add("  buf->write("+ident+"__"+id+", "+size+" * sizeof("+TypeToC(base)+"));\n");	    } else {		r.add("  buf->write(&"+ident+"__"+id+", sizeof("+TypeToC(type)+"));\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    r.add("  "+out.producer_name()+".write_object(buf);\n");	}	r.add("}\n");	r.add("\n");		r.add("void __read_thread__"+id+"(object_write_buffer *buf) {\n");	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    r.add("  "+in.consumer_name()+".read_object(buf);\n");	    /*	    if (oper instanceof SIRFilter) {		r.add("  "+in.name()+"in.read_object(buf);\n");	    }	    */	}	r.add("  load_peek_buffer__"+id+"(buf);\n");	for (int f = 0; f < fields.length; f++) {	    CType type = fields[f].getType();	    String ident = fields[f].getVariable().getIdent();	    DetectConst dc = DetectConst.getInstance((SIRFilter)oper);	    if (dc != null && dc.isConst(ident)) continue;	    if (type.isArrayType()) {		int size = 0;		String dims[] = ArrayDim.findDim((SIRFilter)oper, ident);		CType base = ((CArrayType)type).getBaseType();		try {		    size = Integer.valueOf(dims[0]).intValue();		} catch (NumberFormatException ex) {		    System.out.println("Warning! Could not estimate size of an array: "+ident);		}		r.add("  buf->read("+ident+"__"+id+", "+size+" *  sizeof("+TypeToC(base)+"));\n");	    } else {		r.add("  buf->read(&"+ident+"__"+id+", sizeof("+TypeToC(type)+"));\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    r.add("  "+out.producer_name()+".read_object(buf);\n");	}	r.add("}\n");	r.add("\n");		//  +=============================+	//  | Check Thread Status         |	//  +=============================+		r.add("inline void check_status__"+id+"() {\n");	r.add("  check_thread_status(__state_flag_"+id+", __thread_"+id+");\n");	r.add("}\n");	r.add("\n");		r.add("void check_status_during_io__"+id+"() {\n");	r.add("  check_thread_status_during_io(__state_flag_"+id+", __thread_"+id+");\n");	r.add("}\n");	r.add("\n");		//  +=============================+	//  | Fused Methods               |	//  +=============================+	if (!isEliminated) {	    Iterator iter2 = fusedWith.iterator();	    while (iter2.hasNext()) {		FlatNode tmp = (FlatNode)iter2.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("extern void __declare_sockets_"+fid+"();\n");		r.add("extern void __init_sockets_"+fid+"(void (*cs_fptr)());\n");		r.add("extern void __flush_sockets_"+fid+"();\n");		r.add("extern void __peek_sockets_"+fid+"();\n");		r.add("extern void __init_thread_info_"+fid+"(thread_info *);\n");		r.add("\n");	    }	}	//  +=============================+	//  | Thread Info                 |	//  +=============================+	r.add("void __init_thread_info_"+id+"(thread_info *info) {\n");	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    r.add("  info->add_incoming_data_connection(new connection_info("+in.getSource()+","+in.getDest()+",&"+in.consumer_name()+"));\n");	}		i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    r.add("  info->add_outgoing_data_connection(new connection_info("+out.getSource()+","+out.getDest()+",&"+out.producer_name()+"));\n");	}	r.add("  __state_flag_"+id+" = info->get_state_flag();\n");	if (!isEliminated) {	    Iterator _i = fusedWith.iterator();	    while (_i.hasNext()) {		FlatNode tmp = (FlatNode)_i.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("  __init_thread_info_"+fid+"(info);\n");	    }	}	r.add("}\n");	r.add("\n");	r.add("thread_info *__get_thread_info_"+id+"() {\n");	r.add("  if (__thread_"+id+" != NULL) return __thread_"+id+";\n");	r.add("  __thread_"+id+" = new thread_info("+id+", check_status_during_io__"+id+");\n");	r.add("  __init_thread_info_"+id+"(__thread_"+id+");\n");	r.add("  return __thread_"+id+";\n");		r.add("}\n");	r.add("\n");	//  +=============================+	//  | Declare Sockets             |	//  +=============================+	r.add("void __declare_sockets_"+id+"() {\n");	if (!isEliminated) {	    Iterator iter2 = fusedWith.iterator();	    while (iter2.hasNext()) {		FlatNode tmp = (FlatNode)iter2.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("  __declare_sockets_"+fid+"();\n");	    }	}	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(in.getSource());	    if (!fusedWith.contains(tmp)) {			r.add("  init_instance::add_incoming("+in.getSource()+","+in.getDest()+", DATA_SOCKET);\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(out.getDest());	    if (!fusedWith.contains(tmp)) {				r.add("  init_instance::add_outgoing("+out.getSource()+","+out.getDest()+", DATA_SOCKET);\n");	    }	}	i = msg_from.iterator();	while (i.hasNext()) {	    int src = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("  init_instance::add_incoming("+src+","+id+",MESSAGE_SOCKET);\n");	}	i = msg_to.iterator();	while (i.hasNext()) {	    int dst = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("  init_instance::add_outgoing("+id+","+dst+",MESSAGE_SOCKET);\n");	}	r.add("}\n");	r.add("\n");	//  +=============================+	//  | Init Sockets                |	//  +=============================+	r.add("void __init_sockets_"+id+"(void (*cs_fptr)()) {\n");	r.add("  mysocket *sock;\n");	r.add("\n");	if (!isEliminated) {	    Iterator iter2 = fusedWith.iterator();	    while (iter2.hasNext()) {		FlatNode tmp = (FlatNode)iter2.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("  __init_sockets_"+fid+"(cs_fptr);\n");	    }	}	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(in.getSource());	    if (!fusedWith.contains(tmp)) {		r.add("  sock = init_instance::get_incoming_socket("+in.getSource()+","+in.getDest()+",DATA_SOCKET);\n");		r.add("  sock->set_check_thread_status(cs_fptr);\n");		//r.add("  sock->set_item_size(sizeof("+in.getType()+"));\n");		r.add("  "+in.consumer_name()+".set_socket(sock);\n");		r.add("  "+in.consumer_name()+".init();\n");		r.add("\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(out.getDest());	    if (!fusedWith.contains(tmp)) {		r.add("  sock = init_instance::get_outgoing_socket("+out.getSource()+","+out.getDest()+",DATA_SOCKET);\n");		r.add("  sock->set_check_thread_status(cs_fptr);\n");		//r.add("  sock->set_item_size(sizeof("+out.getType()+"));\n");		r.add("  "+out.producer_name()+".set_socket(sock);\n");		r.add("  "+out.producer_name()+".init();\n");		r.add("\n");	    }	}	i = msg_from.iterator();	while (i.hasNext()) {	    int src = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("  __msg_sock_"+src+"_"+id+"in = (netsocket*)init_instance::get_incoming_socket("+src+","+id+",MESSAGE_SOCKET);\n");	    r.add("\n");	}	i = msg_to.iterator();	while (i.hasNext()) {	    int dst = NodeEnumerator.getSIROperatorId((SIRStream)i.next());	    r.add("  __msg_sock_"+id+"_"+dst+"out = (netsocket*)init_instance::get_outgoing_socket("+id+","+dst+",MESSAGE_SOCKET);\n");	    r.add("\n");	}	r.add("}\n");	r.add("\n");	//  +=============================+	//  | Flush Sockets               |	//  +=============================+	r.add("void __flush_sockets_"+id+"() {\n");	if (!isEliminated) {	    Iterator iter2 = fusedWith.iterator();	    while (iter2.hasNext()) {		FlatNode tmp = (FlatNode)iter2.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("  __flush_sockets_"+fid+"();\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(out.getDest());	    if (!fusedWith.contains(tmp)) {		r.add("  "+out.producer_name()+".flush();\n");		r.add("  "+out.producer_name()+".get_socket()->close();\n");	    }	}	i = data_out.iterator();	while (i.hasNext()) {	    NetStream out = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(out.getDest());	    if (!fusedWith.contains(tmp)) {		r.add("  "+out.producer_name()+".delete_socket_obj();\n");	    }	}	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(in.getSource());	    if (!fusedWith.contains(tmp)) {		r.add("  "+in.consumer_name()+".delete_socket_obj();\n");	    }	}	r.add("}\n");	r.add("\n");	//  +=============================+	//  | Peek Sockets                |	//  +=============================+	r.add("void __peek_sockets_"+id+"() {\n");	if (!isEliminated) {	    Iterator iter2 = fusedWith.iterator();	    while (iter2.hasNext()) {		FlatNode tmp = (FlatNode)iter2.next();		int fid = NodeEnumerator.getNodeId(tmp);		r.add("  __peek_sockets_"+fid+"();\n");	    }	}	i = data_in.iterator();	while (i.hasNext()) {	    NetStream in = (NetStream)i.next();	    FlatNode tmp = NodeEnumerator.getFlatNode(in.getSource());	    if (!fusedWith.contains(tmp)) {		r.add("  "+in.consumer_name()+".peek(0);\n");	    }	}	r.add("}\n");	r.add("\n");	return r;    }