pathrate_snd.c

pathrate是pathload的改进型

    fflush(stdout);
    
    
    
    
    /*
      loop: 
        1) Get control messages for next phase (until SEND command)
        2) Send packets for that phase (until complete, or CONTINUE command)
    */
    
    reset_flag=0; 
    train_id=0;
    trains_lost=0;
    done=0;
    
    while(!done) {
      do {
        /* Wait until a control message arrives (unless if reset_flag=1) */
        /* If reset_flag=1, a control message is already here */
        if (!reset_flag) {
          ctr_code=recv_ctr_msg(ctr_strm, ctr_buff);
        }
        reset_flag=0;
    
        /* Get the command and the data fields from the control message */
        ctr_code_cmnd = ctr_code & 0x000000ff;
        ctr_code_data = ctr_code >> 8;
    
        switch(ctr_code_cmnd) { 
          /* Get maximum packet size from receiver */
          case MAX_PCK_LEN:
            max_pack_sz=ctr_code_data;
            if(Verbose){
              printf("--> Maximum packet size: %d bytes \n", max_pack_sz+28);
              if (file)
              fprintf(pathrate_fp,"--> Maximum packet size: %d bytes \n", max_pack_sz+28);
            }
            break;
      
          /* Get packet size from receiver */
          case PCK_LEN:
            pack_sz=ctr_code_data;
            if (Verbose){
              printf("--> Packet size: %d bytes \n", pack_sz+28);
              if (file)
                fprintf(pathrate_fp,"--> Packet size: %d bytes \n", pack_sz+28);
            }
            break;
      
          /* Get train length from receiver */
          case TRAIN_LEN:
            train_len=ctr_code_data;
            if (Verbose){
              printf("--> New train length: %d\n", train_len);
              if (file) 
                 fprintf(pathrate_fp,"--> New train length: %d\n", train_len);
            }
            break;
      
          /* Get number of trains from receiver */
          case NO_TRAINS:
            //no_trains=ctr_code_data;
            no_trains=1;
            if (Verbose){
              printf("--> New number of trains: %d\n", no_trains);
              if (file) 
                fprintf(pathrate_fp,"--> New number of trains: %d\n", no_trains);
            }
            break;
      
          /* End of measurements */
          case GAME_OVER:
            localtm = time(NULL); 
            if (file) fprintf(pathrate_fp,"Receiver terminates measurements on %s", ctime(&localtm));
            if (verbose) printf("Receiver terminates measurements on %s", ctime(&localtm));
            close(ctr_strm);
            sleep(1);
            done=1;
            break;
      
          /* Skip sending trains and go to next input phase of control messages */ 
          case CONTINUE:
            if (Verbose) {
              printf("--> Continue with next round of measurements \n");
              if (file)
                fprintf(pathrate_fp,"--> Continue with next round of measurements \n");
            }
            break;
      
          /* An ACK for a packet train; ignore at this point  */
          case NEG_ACK_TRAIN:
            if (Verbose) {
              printf("--> Redundant NEG_ACK \n");
              if (file) 
                fprintf(pathrate_fp,"--> Redundant NEG_ACK \n");
            }
            break;
      
          case ACK_TRAIN:
            if (Verbose) {
              printf("--> Redundant ACK \n");
              if (file)
                 fprintf(pathrate_fp,"--> Redundant ACK \n");
            }
            break;
      
          /* Train spacing between successive packet pairs/trains */
          case TRAIN_SPACING:
            train_spacing=ctr_code_data; /* msec */
            if (sleep_usecs) {
              sleep_time.tv_sec = train_spacing/1000;
              sleep_time.tv_usec = train_spacing*1000 - sleep_time.tv_sec*1000000;
              if (Verbose) {
                printf("Time period between packet pairs/trains: %d msec\n",train_spacing); 
                if (file) 
                  fprintf(pathrate_fp,"Time period between packet pairs/trains: %d msec\n",train_spacing); 
              }
            }
            else {
              sleep_secs = (int)floor(train_spacing/1000.)+1; 
              if (Verbose) {
                printf("Time period between packet pairs/trains: %d msec\n",sleep_secs*1000);
                if (file)
                   fprintf(pathrate_fp,"Time period between packet pairs/trains: %d msec\n",sleep_secs*1000);
              }
            }
            break;
      
          /* Start sending the packet trains with the specified
            packet size, train length, and number of trains */ 
          case SEND:
            round_id=ctr_code_data;
            if (Verbose) {
              printf("--> New round number: %d\n", round_id);
              if (file) 
                 fprintf(pathrate_fp,"--> New round number: %d\n", round_id);
            }
            break;
      
          default:
            printf("Unknown control message.. aborting\n");
            done=1;
        }
      } while(ctr_code_cmnd!=SEND && !done);
    
      if (ctr_code_cmnd==SEND) {
         /* Send <no_trains> of length <train_len> with packets of size <pack_sz>.
         * NOTE: We always send one more packet in the train. The first packet
         * (and the corresponding spacing) is ignored, because the processing 
         * of that packet takes longer (due to cache misses). 
         * That first packet has pack_id=0.  */

         train_no=0; trains_ackd=0; reset_flag=0;
         do {
           /* Send train of <train_len> packets. 
            * Each packet carries a packet id (unique in each train),
            * a train id (unique in each round), and a round id (unique
            * in the entire execution).  */
           train_id++; 
           train_id_n = htonl(train_id);
           memcpy(pack_buf+sizeof(long),&train_id_n,sizeof(long));
           round_id_n = htonl(round_id);
           memcpy(pack_buf+2*sizeof(long),&round_id_n,sizeof(long));
           for (pack_id=0; pack_id <= train_len; pack_id++) {
             pack_id_n = htonl(pack_id);
             memcpy(pack_buf, &pack_id_n, sizeof(long));
             sendto(sock_udp, pack_buf, pack_sz, 0, (struct sockaddr*)&rcv_udp_addr,
                sizeof(rcv_udp_addr));
           }
           train_no++;
           if (Verbose) {
              printf("\tTrain-%d:", train_no);
              if (file)
                 fprintf(pathrate_fp,"\tTrain-%d:", train_no);
           }
           
           /* Introduce some delay between consecutive packet trains */
           if (sleep_usecs){
              select(1, NULL, NULL, NULL, &sleep_time);
              sleep_time.tv_sec = train_spacing/1000;
              sleep_time.tv_usec = train_spacing*1000 - sleep_time.tv_sec*1000000;
           }
           else 
             sleep(sleep_secs);
    
           /* Send SENT_TRAIN ctr msgs and wait for ACK/NEG_ACK message */
           ctr_code = SENT_TRAIN;
           send_ctr_msg(ctr_strm,ctr_code);

           ctr_code=recv_ctr_msg(ctr_strm,ctr_buff);
           if ((ctr_code & 0x000000ff)==ACK_TRAIN) {   /* the train has been ACKed */
             trains_ackd++;
             trains_lost=0;
             if (Verbose) {
               printf("  => ACKed\n");
               if (file)
                  fprintf(pathrate_fp,"  => ACKed\n");
             }
           }
           else if ((ctr_code & 0x000000ff)==NEG_ACK_TRAIN) {
             /* Received negative ack, Retransmit */
             if (Verbose){
               printf("  => Retransmit\n");
               if (file)
                  fprintf(pathrate_fp,"  => Retransmit\n");
             }
             /* Primitive congestion avoidance */
             if (++trains_lost > MAX_CONSEC_LOSSES) {
               if (file){
                 fprintf(pathrate_fp,"Too many losses.."); 
                 fprintf(pathrate_fp,"Aborting measurements to avoid congestion \n\n");
               }
               if (verbose){
                 printf("Too many losses.."); 
                 printf("Aborting measurements to avoid congestion \n\n");
               }
               /* Send GAME_OVER signal to rcvr on TCP, then  quit*/
               ctr_code = GAME_OVER;
               send_ctr_msg(ctr_strm,ctr_code);
               sleep(1);
               close(ctr_strm);
               done=1;
             }
           }
           else {   
             /* The receiver has already started a new phase, 
              * and sends different control messages. 
              * Abort this phase, and start the next. */
             printf("DEBUG:: Enter here with code %x\n", (ctr_code & 0x000000FF));
             reset_flag=1;
             trains_lost=0;
           }
         } while (trains_ackd < no_trains && !reset_flag && !done);
      } /* if (ctr_command==SEND) */
    } /* while (!done)  */
  } while(iterative);  /* Repeat forever if in iterative mode */
  
  
  exit(0); 
}