macedon-conv.c

这是一个著名的应用层组播中间件的源码

    fprintf(hfile, "  char* create_%s(int &bufsize, int toaddr", tempm->simple_name);
    field = tempm->fields;
    while (field)
    {
      if (!strcmp(field->field_size,"1")) {
        fprintf(hfile,", %s %s", field->field_type, field->field_nickname);
      }
      else {
        fprintf(hfile,", %s* %s", field->field_type, field->field_nickname);
      }
      field = field->next;
    }
    fprintf(hfile,", char* msg, int size);\n");
    tempm=tempm->next;
  }
  fprintf(hfile, "  virtual int command(int argc, const char*const *argv);\n");
  fprintf(hfile, "  virtual void check_neighbors(double curtime);\n");
  fprintf(hfile, "  virtual void exit_agent();\n");
  fprintf(hfile, "  void dump_packet(Packet *p, char *descrip, int recv_flag);\n");
  fprintf(hfile, "  void dump_state();\n");
  fprintf(hfile, "  void dump_timings();\n");
  fprintf(hfile, "  virtual void recv(Packet *p, Handler *h);\n");
  if (!base_name)
    fprintf(hfile, "  void recv_from_pipe(int from, unsigned char * hdata, int hsize, unsigned char * data, int size);\n");
  fprintf(hfile, "  int forward(Packet *p);\n");
  fprintf(hfile, "  virtual int macedon_upcall_ext(int operation, void* arg);\n");
  fprintf(hfile, "  virtual int macedon_downcall_ext(int operation, void* arg);\n");
  fprintf(hfile, "  virtual void sendmsg(int nbytes, AppData *data, const char *flags=0);\n");
  fprintf(hfile, "  virtual void macedon_register_handlers(macedon_forward_handler, macedon_deliver_handler, macedon_notify_handler, macedon_upcall_handler=0);\n");

  for(i=0; i<API_num_types; i++) {
    fprintf(hfile, "  virtual %s %s;\n", API_returns[i], API_prototype[i]);
  }

  fprintf(hfile, "  hdr_%s* gethdr%s( Packet* );\n", protocol_name, protocol_name);
  fprintf(hfile, "  struct macedon_fields *getmacedonfields(Packet *);\n");

  fprintf(hfile, "\n");

  fprintf(hfile, "  void multi_send(Packet *);\n");
  fprintf(hfile, "\n");
  fprintf(hfile, "  Packet* prep_delivery(char *msg, int size, int type, int nextkey, macedon_key nextkeyhash, int from_pipe=0);\n");
  fprintf(hfile, "\n");

  fprintf(hfile, "  int fsm;\n");
  fprintf(hfile, "  int fsm_hint;\n");

  if (!base_name) {
    transport *temptr = my_transports;
    while (temptr) {
      if (temptr->type)
        fprintf(hfile, "  %s_transport* %s;\n", temptr->type, temptr->over);
      temptr = temptr->next;
    }
  }
  fprintf(hfile, "  macedon_forward_handler forward_handler;\n");
  fprintf(hfile, "  macedon_deliver_handler deliver_handler;\n");
  fprintf(hfile, "  macedon_notify_handler notify_handler;\n");
  fprintf(hfile, "  macedon_upcall_handler upcall_handler;\n");
  fprintf(hfile, "  int node_type;\n");
  fprintf(hfile, "  long long bytes_sent;");

  fprintf(hfile, "  MACEDON_Timer utility_timer_;\n", protocol_name);
  temp = my_periphs;
  while (temp)
  {
    switch(temp->array)
    {
      case 0:
        fprintf(hfile, "  %s %s;\n", temp->type, temp->name);
        break;
      case 1:
        fprintf(hfile, "  %s %s[%s];\n", temp->type, temp->name, temp->size);
        break;
      case 2:
        fprintf(hfile, "  %s %s[%s][%s];\n", temp->type, temp->name,
            temp->size, temp->size2);
        break;
      case 3:
        fprintf(hfile, "  %s %s[%s][%s][%s];\n", temp->type, temp->name,
            temp->size, temp->size2, temp->size3);
        break;
      default:
        break;
    }
    temp = temp->next;
  }

  fprintf(hfile, "  int bytes_recv_;\n");
  fprintf(hfile, "  int prev_send_seqno_;\n");
  fprintf(hfile, "  int prev_recv_seqno_;	\n");

  fprintf(hfile, "\n");

  fprintf(hfile, "  int utility_expire(int);\n");

  tempt = my_timers;
  while (tempt)
  {
    fprintf(hfile, "  MACEDON_Timer *%s;\n", tempt->timer_name);
    fprintf(hfile, "  int expire_%s(int);\n", tempt->timer_name);
    tempt = tempt->next;
  }

  cfile_stuff();

}

cfile_stuff()
{
  fprintf(cfile, "int hdr_%s::offset_;\n",protocol_name);

  fprintf(cfile, "\n");
  fprintf(cfile, "static class %sHeaderClass : public PacketHeaderClass {\n", protocol_name);
  fprintf(cfile, "public:\n");
  fprintf(cfile, "  %sHeaderClass() : PacketHeaderClass(\"PacketHeader/%s\",\n", protocol_name, protocol_name); 
  fprintf(cfile, "                                          sizeof(hdr_%s)) {}\n", protocol_name);
  fprintf(cfile, "} class_%shdr;\n", protocol_name);
  fprintf(cfile, "\n");

  fprintf(cfile, "static class %sClass : public TclClass {\n", protocol_name);
  fprintf(cfile, "public:\n");
  fprintf(cfile, "  %sClass() : TclClass(\"Agent/%s\") {}\n", protocol_name, protocol_name); 
  fprintf(cfile, "  TclObject* create(int, const char*const*) {\n");
  fprintf(cfile, "    return (new %s_Agent());\n", protocol_name);
  fprintf(cfile, "  }\n");
  fprintf(cfile, "} class_%s;\n", protocol_name);
  fprintf(cfile, "\n");

  fprintf(cfile, "hdr_%s* %s_Agent::gethdr%s( Packet* pkt )\n", protocol_name, protocol_name, protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "  return ( hdr_%s::access(pkt) );\n", protocol_name);
  fprintf(cfile, "}\n");
  fprintf(cfile, "\n");

  fprintf(cfile, "struct macedon_fields* %s_Agent::getmacedonfields( Packet* pkt )\n", protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "  return (struct macedon_fields*)gethdr%s(pkt);\n", protocol_name);
  fprintf(cfile, "}\n");
  fprintf(cfile, "\n");

  /*   fprintf(cfile, "void %s_Timer::expire(Event *E)\n", protocol_name); */
  /*   fprintf(cfile, "{\n"); */
  /*   fprintf(cfile, "\n"); */
  /*   fprintf(cfile, "(a_->*handler_) (e);\n"); */
  /*   fprintf(cfile, "}\n"); */
  /*   fprintf(cfile, "\n"); */

  fprintf(cfile, "int %s_Agent::utility_expire(int yo)\n", protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "  double curtime = Scheduler::instance().clock();\n");
  /*  fprintf(cfile, "  debug_trace(\"Timer:  Utility timer expired. Rechedding in %%f time.\\n \", MACEDON_UTIL_TIMER_VALUE);\n"); */
//  fprintf(cfile, "  retrans_check();\n");
  fprintf(cfile, "  check_neighbors(curtime);\n");
  fprintf(cfile, "  utility_timer_.start(MACEDON_UTIL_TIMER_VALUE); \n");
  fprintf(cfile, "  return 0;\n");
  fprintf(cfile, "\n");
  fprintf(cfile, "}\n");
  fprintf(cfile, "\n");

  init_function();
  multicast_functions();
  command_function();
  ext_functions();
  register_function();
  create_function();
  deliver_functions();
  check_function();
  dumppkt_function();
  dumpstate_function();

  recv_functions();  // must be last

}


deliver_functions()
{
  message *tempm;
  message_field *field;
  fprintf(cfile, "Packet * %s_Agent::prep_delivery(char *msg, int size, int type, int nextkey, macedon_key nextkeyhash, int from_pipe) { return %s_prep_delivery(msg, size, type, nextkey, nextkeyhash, from_pipe); }\n", protocol_name, protocol_name);
  fprintf(cfile, "Packet * %s_prep_delivery(char *msg, int size, int type, int nextkey, macedon_key nextkeyhash, int from_pipe)\n", protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "   Packet *pkt;\n");
  fprintf(cfile, "   pkt = new Packet();\n");
  fprintf(cfile, "   extern int big_hdr_size;\n");
#ifndef VARIABLE_HEADERS  
  fprintf(cfile, "pkt->bits_ = (unsigned char *)malloc(size+sizeof(hdr_ip)+sizeof(hdr_cmn)+100);\n");
  fprintf(cfile, "if (pkt->bits_ == 0) {\n");
  fprintf(cfile, "  printf(\"Malloc exception 1!\\n\"); \n");
  fprintf(cfile, "  exit(72);\n");
  fprintf(cfile, "}\n");
  fprintf(cfile, "bcopy( msg, pkt->bits_+sizeof(hdr_ip)+sizeof(hdr_cmn), size );\n");
  fprintf(cfile, " hdr_%s* recv_hdr%s = global%s->gethdr%s( pkt );\n", protocol_name, protocol_name, protocol_name, protocol_name);
#else  
  fprintf(cfile, "   ASSERT(size >= sizeof(macedon_fields)+from_pipe*(sizeof(hdr_ip)+sizeof(hdr_cmn)));\n");
  fprintf(cfile, "   int toalloc = ((macedon_fields*)(msg+from_pipe*(sizeof(hdr_ip)+sizeof(hdr_cmn))))->mh_expanded_size_+sizeof(hdr_ip)+sizeof(hdr_cmn);\n");
  fprintf(cfile, "   pkt->bits_ = (unsigned char *)malloc(toalloc+100);\n");
  fprintf(cfile, "   if (pkt->bits_ == 0) {\n");
  fprintf(cfile, "     printf(\"Malloc exception 1!\\n\"); \n");
  fprintf(cfile, "     exit(72);\n");
  fprintf(cfile, "   }\n");
  fprintf(cfile, "   //bcopy( msg, pkt->bits_+sizeof(hdr_ip)+sizeof(hdr_cmn), size );\n");
  fprintf(cfile, "   if(from_pipe) {\n");
  fprintf(cfile, "     bcopy(msg, pkt->bits_, sizeof(hdr_ip)+sizeof(hdr_cmn));\n");
  fprintf(cfile, "   }\n");

  fprintf(cfile, "   int pktPosition=sizeof(hdr_ip)+sizeof(hdr_cmn);\n");
  fprintf(cfile, "   int msgPosition=(from_pipe)?sizeof(hdr_ip)+sizeof(hdr_cmn):0;\n");
  fprintf(cfile, "   ASSERT(msgPosition + sizeof(hdr_%s) <= size);\n",protocol_name);
  fprintf(cfile, "   ASSERT(pktPosition + sizeof(hdr_%s) <= toalloc);\n",protocol_name);
  fprintf(cfile, "   bcopy( msg+msgPosition, pkt->bits_+pktPosition, sizeof(hdr_%s) );\n",protocol_name);
  fprintf(cfile, "   pktPosition+=sizeof(hdr_%s);\n",protocol_name);
  fprintf(cfile, "   msgPosition+=sizeof(hdr_%s);\n",protocol_name);
  fprintf(cfile, "   hdr_%s* recv_hdr%s = global%s->gethdr%s( pkt );\n", protocol_name, protocol_name, protocol_name, protocol_name);
#endif  
  tempm = my_messages;
  fprintf(cfile, "   ASSERT(recv_hdr%s->%s_mf_.mh_type_ >= PT_%s * 100 && recv_hdr%s->%s_mf_.mh_type_ < (PT_%s + 1)*100);\n",protocol_name,protocol_name,protocol_name,protocol_name,protocol_name,protocol_name);
  while (tempm){
    fprintf(cfile, "   if ( recv_hdr%s->%s_mf_.mh_type_ ==  %s_message_%s)\n", protocol_name, protocol_name, protocol_name, tempm->simple_name);
    fprintf(cfile, "   {\n");
#ifndef VARIABLE_HEADERS    
    fprintf(cfile, "     pkt->hdrlen_ = sizeof(hdr_%s_%s) + sizeof(hdr_ip) + sizeof(hdr_cmn);\n", protocol_name, tempm->simple_name);
    fprintf(cfile, "     if(size > sizeof(hdr_%s_%s))\n", protocol_name, tempm->simple_name);
    fprintf(cfile, "     {\n");
    fprintf(cfile, "       pkt->size_ = size - sizeof(hdr_%s_%s);\n", protocol_name, tempm->simple_name);
    fprintf(cfile, "       pkt->data_ = pkt->bits_+pkt->hdrlen_;\n");
    fprintf(cfile, "     }\n");
#else
    fprintf(cfile, "     int contiguous=0;\n");
    field = tempm->fields;
    while(field) {
      if(field->neighbor) {
        if (!strcmp(field->field_size,"1"))
        {
          fprintf(cfile, "     msgPosition += ((%s*)(pkt->bits_+pktPosition))->deserialize(msg+msgPosition);\n",field->field_type);
          fprintf(cfile, "     pktPosition += sizeof(%s); // field(%s)\n",field->field_type,field->field_nickname);
          fprintf(cfile, "     ASSERT(msgPosition <= size);\n",protocol_name);
          fprintf(cfile, "     ASSERT(pktPosition <= toalloc);\n",protocol_name);
        }
        else
        {
          fprintf(cfile, "     for (int k=0; k<%s; k++) \n", field->field_size);
          fprintf(cfile, "     { \n");
          fprintf(cfile, "       msgPosition += ((%s*)(pkt->bits_+pktPosition))->deserialize(msg+msgPosition); // field(%s)[k]\n",field->field_type,field->field_nickname);
          fprintf(cfile, "       pktPosition += sizeof(%s);\n",field->field_type);
          fprintf(cfile, "       ASSERT(msgPosition <= size);\n",protocol_name);
          fprintf(cfile, "       ASSERT(pktPosition <= toalloc);\n",protocol_name);
          fprintf(cfile, "     } \n");
        }
      } else {
        if (!strcmp(field->field_size,"1"))
        {
          fprintf(cfile, "     contiguous+=sizeof(%s); // field(%s)\n",field->field_type,field->field_nickname);
        }
        else
        {
          fprintf(cfile, "     contiguous+=(sizeof(%s)*%s); //field(%s[])\n",field->field_type,field->field_size,field->field_nickname);
        }
        if(field->next == NULL || field->next->neighbor != NULL) {
          fprintf(cfile, "     ASSERT(msgPosition+contiguous <= size);\n",protocol_name);
          fprintf(cfile, "     ASSERT(pktPosition+contiguous <= toalloc);\n",protocol_name);
          fprintf(cfile, "     bcopy( msg+msgPosition, pkt->bits_+pktPosition, contiguous );\n");
          fprintf(cfile, "     pktPosition+=contiguous;\n");
          fprintf(cfile, "     msgPosition+=contiguous;\n");
          fprintf(cfile, "     contiguous=0;\n");
        }
      }
      field = field->next;
    }
    fprintf(cfile, "     //pkt->hdrlen_ = sizeof(hdr_%s_%s) + sizeof(hdr_ip) + sizeof(hdr_cmn);\n", protocol_name, tempm->simple_name);
    fprintf(cfile, "     pkt->hdrlen_ = pktPosition;\n");
    fprintf(cfile, "     if(size > msgPosition)\n");
    fprintf(cfile, "     {\n");
    fprintf(cfile, "       pkt->size_ = size - msgPosition;\n");
    fprintf(cfile, "       pkt->data_ = pkt->bits_+pktPosition;\n");
    fprintf(cfile, "       ASSERT(msgPosition+pkt->size_ <= size);\n",protocol_name);
    fprintf(cfile, "       ASSERT(pktPosition+pkt->size_ <= toalloc);\n",protocol_name);
    fprintf(cfile, "       bcopy(msg+msgPosition,pkt->data_,pkt->size_);\n");
    fprintf(cfile, "     }\n");
#endif
    fprintf(cfile, "   }\n");
    tempm = tempm->next;
  }
  fprintf(cfile, "  return pkt;\n");
  fprintf(cfile, "}\n");

  fprintf(cfile, "int %s_my_forward(char *msg, int size, int type, int nextkey, macedon_key nextkeyhash)\n", protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "  Packet *pkt;\n");
  //  fprintf(cfile, "  printf(\"%s_my_forward called\\n\");\n", protocol_name);

  fprintf(cfile, "  pkt = %s_prep_delivery(msg, size, type, nextkey, nextkeyhash);\n", protocol_name);
  fprintf(cfile, "  return global%s->forward(pkt);\n", protocol_name);
  fprintf(cfile, "}\n");

  fprintf(cfile, "void %s_my_deliver(char *msg, int size, int type)\n", protocol_name);
  fprintf(cfile, "{\n");
  fprintf(cfile, "  Packet *pkt;\n");
  //  fprintf(cfile, "  printf(\"%s_my_deliver called\\n\");\n", protocol_name);
  fprintf(cfile, "  pkt = %s_prep_delivery(msg, size, type, 0, 0);\n", protocol_name);
  fprintf(cfile, "  global%s->recv(pkt, 0);\n", protocol_name);
  fprintf(cfile, "  return;\n", protocol_name);
  fprintf(cfile, "}\n");

  fprintf(cfile, "void %s_my_notify(int type, int size, int *neighbors)\n", protocol_name);
  fprintf(cfile, "{\n");
  //  fprintf(cfile, "  printf(\"%s_my_notify called\\n\");\n", protocol_name);
  fprintf(cfile, "  global%s->macedon_notify(type, size, neighbors);\n", protocol_name);
  fprintf(cfile, "  return;\n", protocol_name);
  fprintf(cfile, "}\n");