- From: Torbjörn Andersson <torbjorn.andersson AT kau.se>
 
- To: <discussion AT sctp.de>
 
- Subject: RE: [SCTP-Discussion] An error in the SCTPlib/SocketAPI
 
- Date: Thu Oct  9 09:02:01 2003
 
- Importance: Normal
 
- List-id: SCTP Discussions <discussion.sctp.de>
 
 
I'm not getting my programs working correctly. During some time it halts
and never resumes. I append the source code so that you can test it your
self. Run them a few times because it halts every now and then,
annoying:)!
I'm using sctplib-1-rc2 and socket-api1.3.1
A short summary of the programs:
The server initiates a UDP-style socket and waits for a new association.
The client initiates a UDP-style socket and connects to the server. The
server sends messages of a given size, defined as an argument to the
server (must be >100bytes) to the client.
The client logs each packet and when x packets are retrieved the
association is closed by the server.
-----Ursprungligt meddelande-----
Från: Andreas Jungmaier 
[mailto:ajung AT exp-math.uni-essen.de]
 
Skickat: den 18 september 2003 15:19
Till: Torbjörn Andersson
Ämne: Re: SV: [SCTP-Discussion] An error in the SCTPlib
-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1
Okay,
in line 499 of the file "flowcontrol.c" of the sctplib, try changing
"ERROR_FATAL" to "ERROR_MAJOR". Then the sctplib will continue to
work.
This was an alert I put in to make sure that when something unexpected
happens, I would not forget about it. 
However, the application should continue fine, when you simply change
the
statement as mentioned above.
I will try to look into the matter, and put it on my test list.
Anyway, this may just be one of these "strange interactions" 
between socketapi and sctplib (we had a number of those :-)
Hope that helps,
Best regards,
Andreas
- -- 
Dipl.-Ing. Andreas Jungmaier              
Computer Networking Technology Group     
University of Duisburg-Essen                       
http://www.exp-math.uni-essen.de/~ajung   
My PGP Key-ID: D3824AC0             
My PGP Key-Fingerprint: 228C 7C2C 3381 2DD4  9998 2812 5C0B 0B04  D382
4AC0
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.2.2 (GNU/Linux)
iD8DBQE/abC9XAsLBNOCSsARAlu0AKCln33yJkpZKMxkm74JgtjT2J6AlwCfawz6
2LYQZZDmYYaUBFb3Lrlce/Y=
=B/GQ
-----END PGP SIGNATURE-----
#include <stdio.h>
#include <sys/time.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h> 
#include <stdlib.h>
#include <unistd.h> 
#include <sys/uio.h> 
#include <ext_socket.h>
#include <netdb.h>
#include <iostream>
#include <list>
using std::list;
#ifdef __APPLE__
  #include "inet_ntop.c"
#endif
#include <string.h>
#define BUFLEN  100
#define PACKETSIZE 1400
#define S2MS 1000
//#define DEBUG 1
typedef struct{
  u_long seq_num;
  struct timeval queued;
  struct timeval recved;
}message_t;
static void
handle_event(void *buf)
{
  struct sctp_assoc_change *sac;
  struct sctp_send_failed *ssf;
  struct sctp_paddr_change *spc;
  struct sctp_remote_error *sre;
  union sctp_notification *snp;
  char addrbuf[INET6_ADDRSTRLEN];
  const char *ap;
  struct sockaddr_in *sin;
  struct sockaddr_in6 *sin6;
  snp = (sctp_notification *)buf;
  switch ((snp->sn_header).sn_type) {
    case SCTP_ASSOC_CHANGE:
      sac = &snp->sn_assoc_change;
      printf("^^^ assoc_change: state=%hu, error=%hu, instr=%hu, 
outstr=%hu\n",
              sac->sac_state, sac->sac_error, sac->sac_inbound_streams, 
sac->sac_outbound_streams);
      break;
    case SCTP_SEND_FAILED:
      ssf = &snp->sn_send_failed;
      printf("^^^ sendfailed: len=%hu err=%d\n", ssf->ssf_length, 
ssf->ssf_error);
      break;
    case SCTP_PEER_ADDR_CHANGE:
      spc = &snp->sn_paddr_change; 
      if (((struct sockaddr_in *)&spc->spc_aaddr)->sin_family == AF_INET) {
        sin = (struct sockaddr_in *)&spc->spc_aaddr; 
        ap = inet_ntop(AF_INET, &sin->sin_addr, addrbuf, INET6_ADDRSTRLEN);
      } else {
        sin6 = (struct sockaddr_in6 *)&spc->spc_aaddr; 
        ap = inet_ntop(AF_INET6, &sin6->sin6_addr, addrbuf, INET6_ADDRSTRLEN);
      }
      printf("^^^ intf_change: %s state=%d, error=%d\n", ap, spc->spc_state, 
spc->spc_error);
      break;
    case SCTP_REMOTE_ERROR:
      sre = &snp->sn_remote_error;
      printf("^^^ remote_error: err=%hu len=%hu\n", ntohs(sre->sre_error), 
ntohs(sre->sre_length));
      break;
    case SCTP_SHUTDOWN_EVENT:
      printf("^^^ shutdown event\n");
      break;
    default:
      printf("unknown type: %hu\n", snp->sn_header.sn_type);
      break;
  }
}
void prnMessage(message_t msg)
{
  struct timeval curr_time;
  gettimeofday(&curr_time,NULL);
  printf("%d\t", msg.seq_num);
  printf("%lf\t", msg.queued.tv_sec + msg.queued.tv_usec/1000000.0);
  printf("%lf\n", msg.recved.tv_sec + msg.recved.tv_usec/1000000.0);
}
unsigned long receiveData(int fd, list<message_t>& mesgs)
{
  unsigned long total = 0;
  struct msghdr msg[1];
  struct cmsghdr *cmsg;
  struct iovec iov[1];
  char cbuf[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
  char *buf;
  ssize_t buflen;
  ssize_t cmsglen = CMSG_SPACE(sizeof(struct sctp_sndrcvinfo));
  
  
  /* Allocate the initial data buffer */
  buflen = PACKETSIZE;
  if (!(buf = (char*)malloc(sizeof(char)*PACKETSIZE))) {
    fprintf(stderr, "out of memory\n");
    exit(1);
  }
  iov[0].iov_base = buf;
  iov[0].iov_len = sizeof(char)*PACKETSIZE;
  /* Set up the msghdr structure for receiving */
  memset(msg, 0, sizeof (*msg));
  msg->msg_name = NULL;
  msg->msg_namelen = 0;
  msg->msg_control = cbuf;
  msg->msg_controllen = cmsglen;
  msg->msg_flags = 0;
  msg->msg_iov = iov;
  msg->msg_iovlen = 1;
  cmsg = (struct cmsghdr *)cbuf;
  /* Wait for something to echo */
  while (1){
    u_int nr = 0;
    nr = ext_recvmsg(fd, msg, 0);
    if( (msg->msg_flags & MSG_EOR) == 0) //Everything is not read
      printf("Everything is not read yet");
    /* Intercept notifications here */
    if (msg->msg_flags & MSG_NOTIFICATION) {
#ifdef DEBUG
      handle_event(buf);
#endif
      union sctp_notification *snp = (union sctp_notification *)buf;
      if(snp->sn_header.sn_type == SCTP_SHUTDOWN_EVENT){
#ifdef DEBUG
        printf("Shutdown notification!\n");
        fflush(stdout);
#endif
        return total;
      }
    }
    else if(nr>0){
      total += nr;
#ifdef DEBUG
      struct sctp_sndrcvinfo *sri = 
        (struct sctp_sndrcvinfo *)CMSG_FIRSTHDR(msg);
      printf("%d bytes received at stream %d\n",nr,sri->sinfo_stream);
#endif
      message_t msg;
      bcopy(buf,&msg,sizeof(message_t));
      gettimeofday(&(msg.recved),0);
      mesgs.insert(mesgs.end(),msg);
    }
  }
  return total;
}
int main(int argc, char** argv)
{
  int fd;
  /* Create destination address */
  struct hostent *phe;
  struct in_addr addr[1];
  char * localIP, * remoteIP;
  int port;
  if(argc != 4){
    printf("Usage: sctpClient <local ip> <remote ip> <port>\n");
    perror("Got:");
    for(int i = 1; i < argc; i++){
      fprintf(stderr,"%s ",argv[1]);
    }
    perror("\n");
    exit(-1);
  }
  localIP = argv[1];
  remoteIP = argv[2];
  port = atoi(argv[3]);
  struct sockaddr_in sin_src[1];
  sin_src->sin_family = AF_INET;
  sin_src->sin_port = htons(0); //Client port is picked by SCTP
  sin_src->sin_addr.s_addr = inet_addr(localIP); 
  if(remoteIP[0]> 47 || remoteIP[0]< 58){
    phe = gethostbyname(remoteIP);
    if(phe == 0){
      perror("Bad host lookup.\n");
    }
    memcpy(addr, phe->h_addr_list[0], sizeof(struct in_addr));
  }
  else{
    if(inet_aton(remoteIP,addr) < 0){
      perror ("Unalbe to create internet address!\n");
      exit(-1);
    }
  }
  struct sockaddr_in sin_dest[1];
  sin_dest->sin_family = AF_INET;
  sin_dest->sin_port = htons(port);
  sin_dest->sin_addr.s_addr = addr[0].s_addr;
  if ((fd = ext_socket(PF_INET, SOCK_SEQPACKET, IPPROTO_SCTP)) == -1) {
    perror("socket");
    exit(1);
  }
  
  
  /* Enable notifications */
  struct sctp_event_subscribe subscription;
  subscription.sctp_data_io_event = 1;
  subscription.sctp_association_event = 0;
  subscription.sctp_address_event = 0;
  subscription.sctp_send_failure_event = 0;
  subscription.sctp_peer_error_event = 0;
  subscription.sctp_partial_delivery_event = 0;
  subscription.sctp_adaption_layer_event = 0;
  subscription.sctp_shutdown_event = 1;
  if(ext_setsockopt(fd,IPPROTO_SCTP,SCTP_EVENTS, &subscription, 
sizeof(subscription)) != 0){
    perror("setevent failed\n");
    exit(1);
  }
  struct sctp_initmsg initmsg;
  initmsg.sinit_num_ostreams = 1;
  initmsg.sinit_max_instreams = 1;
  initmsg.sinit_max_attempts = 0;
  initmsg.sinit_max_init_timeo  = 0;
  unsigned long size = sizeof(initmsg);
  if (ext_setsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, size) < 0) {
    perror("setsockopt SCTP_INITMSG");
    exit(1);
  }
  if (ext_bind(fd, (struct sockaddr *)sin_src, sizeof (*sin_src)) == -1) {
    perror("bind");
    exit(1);
  }
  ext_connect(fd,(struct sockaddr*)sin_dest,sizeof(struct sockaddr));
  class list<message_t> results;
  receiveData(fd,results);
  close(fd);
  class list<message_t>::iterator iter = results.begin();
  while(iter != results.end()){
    prnMessage(*iter);
    iter++;
  }
  printf("%d messages were retrieved\n",results.size());
}
#include <ext_socket.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <sys/time.h>
#include <string.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#define MS2USEC 1000
#define BUFLEN  2000
#define INTERVALL_CORRECTION 10000
//#define VERBOSE_1
//#define VERBOSE_2
#include <stdio.h>
#include <stdlib.h>
struct message{
  u_long seq_num;
  struct timeval queued;
  struct timeval recved;
};
void shutdownAssoc(int fd, sctp_assoc_t aId){
  struct sctp_sndrcvinfo *sri;
  struct msghdr msg[1];
  struct cmsghdr *cmsg;
  struct iovec iov[1];
  char cbuf[CMSG_SPACE(sizeof(*sri))];
  msg->msg_name = 0; //peer;
  msg->msg_namelen = 0;//addr_len; //addrLen;
  iov->iov_base = 0;
  iov->iov_len = 0;
  msg->msg_iov = iov;
  msg->msg_iovlen = 1;
  msg->msg_flags = 0;
  msg->msg_control = cbuf;
  msg->msg_controllen = CMSG_SPACE(sizeof(*sri));
  cmsg = CMSG_FIRSTHDR(msg);
  cmsg->cmsg_len = CMSG_LEN(sizeof(*sri));
  cmsg->cmsg_level = IPPROTO_SCTP;
  cmsg->cmsg_type = SCTP_SNDRCV;
  sri = (struct sctp_sndrcvinfo*)CMSG_DATA(cmsg);
  sri->sinfo_stream = 0;
  sri->sinfo_ssn = 0; //Ignored when sending
  sri->sinfo_flags =  MSG_EOF; //Unordered delivery
  sri->sinfo_ppid = 0;
  sri->sinfo_assoc_id = aId; //0 assoc_id;
  if(ext_sendmsg(fd,msg,0) < 0){
    perror("something went wrong during sending");
    exit(-1);
  }
}
void 
send(int socket, sctp_assoc_t aId, struct message msgStamp, unsigned int 
msgSize)
{
  struct sctp_sndrcvinfo *sri;
  struct msghdr msg[1];
  struct cmsghdr *cmsg;
  char cbuf[CMSG_SPACE(sizeof(*sri))];
  char *buf;
  struct iovec iov[1];
 
  /* Allocate the initial data buffer */
  if (!(buf = (char*)malloc(sizeof(char)*msgSize))) {
    fprintf(stderr, "out of memory\n");
    exit(1);
  }
  memset(buf,'1',msgSize);
 
 /* Insert messagestamp */
  bcopy(&msgStamp,buf,sizeof(struct message));
  /* Set up the msghdr structure for receiving */
  msg->msg_name = 0; //peer;
  msg->msg_namelen = 0;//addr_len; //addrLen;
  msg->msg_iov = iov;
  msg->msg_iovlen = 1;
  iov->iov_base = buf;
  iov->iov_len = msgSize;
  msg->msg_control = cbuf;
  msg->msg_controllen = CMSG_SPACE(sizeof(*sri));
  msg->msg_flags = 0;
  cmsg = CMSG_FIRSTHDR(msg);
  cmsg->cmsg_len = CMSG_LEN(sizeof(*sri));
  cmsg->cmsg_level = IPPROTO_SCTP;
  cmsg->cmsg_type = SCTP_SNDRCV;
  sri = (struct sctp_sndrcvinfo*)CMSG_DATA(cmsg);
  sri->sinfo_stream = 0;
  sri->sinfo_ssn = 0; //Ignored when sending
  sri->sinfo_flags = 0;// MSG_UNORDERED;// | last == true ? MSG_EOF : 0; 
//Unordered delivery
  sri->sinfo_ppid = getpid();
  sri->sinfo_assoc_id = aId; //0 assoc_id;
  if(ext_sendmsg(socket,msg,0) < 0){
    perror("something went wrong during sending");
    exit(-1);
  }
  free(buf);
}
static void
waitForShutdown(int fd)
{
  struct iovec iov[1];
  struct msghdr msg[1];
  ssize_t len;
  char *buf;
  ssize_t buflen;
  /* Allocate the initial data buffer */
  buflen = BUFLEN;
  if (!(buf = (char*)malloc(BUFLEN))) {
    fprintf(stderr, "out of memory\n");
    exit(1);
  }
  iov[0].iov_base = buf;
  iov[0].iov_len = sizeof(char)*BUFLEN;
  /* Set up the msghdr structure for receiving */
  memset(msg, 0, sizeof (*msg));
  msg->msg_iov = &iov[0];
  msg->msg_iovlen = 1;
  /* Wait for a notification */
  while (1){
    len = ext_recvmsg(fd, msg, 0);
    /* Intercept notifications here */
    if(len > 0){
#ifdef VERBOSE_2
      printf("Received message!\n");
//      handle_event(buf);
#endif
      if (msg->msg_flags & MSG_NOTIFICATION) {
        union sctp_notification *snp = (union sctp_notification *)buf;
        if(snp->sn_header.sn_type == SCTP_ASSOC_CHANGE &&  
           snp->sn_assoc_change.sac_state == SCTP_SHUTDOWN_COMP){
          return;
        }
        else{
#ifdef VERBOSE_2
          printf("Some other state!\n"); 
#endif
        }
      }
      else{
#ifdef VERBOSE_2
        printf("Some other notification!\n");
#endif
      }
    }
    else{
#ifdef VERBOSE_2
      printf("Unalbe to read from fd!\n");
#endif
      return;
    }
  } 
}
sctp_assoc_t getNewConnection(int fd)
{
  struct msghdr msg[1];
  char *buf;
  struct iovec iov[1];
  ssize_t buflen;
  ssize_t len;
  /* Allocate the initial data buffer */
  buflen = BUFLEN;
  if (!(buf = (char*)malloc(BUFLEN))) {
    fprintf(stderr, "out of memory\n");
    exit(1);
  }
  iov[0].iov_base = buf;
  iov[0].iov_len = sizeof(char)*BUFLEN;
  /* Set up the msghdr structure for receiving */
  memset(msg, 0, sizeof (*msg));
  msg->msg_iov = &iov[0];
  msg->msg_iovlen = 1;
#ifdef VERBOSE_1
      printf("Waiting for new connection\n");
      fflush(stdout);
#endif
  while (1){
    len = ext_recvmsg(fd, msg, 0);
    /* Intercept notifications here */
#ifdef VERBOSE_2
     printf("Received message!\n");
     fflush(stdout);
#endif
    if (msg->msg_flags & MSG_NOTIFICATION) {
      union sctp_notification *snp = (union sctp_notification *)buf;
      if(snp->sn_header.sn_type == SCTP_ASSOC_CHANGE){
        if(snp->sn_assoc_change.sac_state == SCTP_COMM_UP){
          return snp->sn_assoc_change.sac_assoc_id;
        }
      }
    }
  }
}
int setSocketOptions(int fd)
{
  /* Enable notifications */
  struct sctp_event_subscribe subscription;
  subscription.sctp_data_io_event = 1;
  subscription.sctp_association_event = 1;
  subscription.sctp_address_event = 0;
  subscription.sctp_send_failure_event = 1;
  subscription.sctp_peer_error_event = 0;
  subscription.sctp_partial_delivery_event = 0;
  subscription.sctp_adaption_layer_event = 0;
  subscription.sctp_shutdown_event = 1;
  
  if(ext_setsockopt(fd,IPPROTO_SCTP,SCTP_EVENTS, &subscription, 
sizeof(subscription)) != 0){
    perror("setevent failed\n");
    return -1;
  }
  return 0;
}
struct message createMsg(struct timeval stamp){
  static int id = 0;
  struct message msg;
  msg.seq_num = id++;
  msg.queued = stamp;
  memset(&(msg.recved),'A',sizeof(struct timeval));
  return msg;
}
void prnMessage(struct message msg)
{
  struct timeval curr_time;
  gettimeofday(&curr_time,NULL);
  printf("%d\t", msg.seq_num);
  printf("%lf\t", msg.queued.tv_sec + msg.queued.tv_usec/1000000.0);
  printf("%lf\n", msg.recved.tv_sec + msg.recved.tv_usec/1000000.0);
}
void waitFor(long time){
  if(time > 10000){
#ifdef VERBOSE_2
      printf("Sleeping\n");
      fflush(stdout);
#endif
    usleep(time - INTERVALL_CORRECTION);
  }
  else{
    struct timeval start, now;
    long currTime, startTime;
#ifdef VERBOSE_2
      printf("busy-wait sleep: %ld ms\n", time/1000);
      fflush(stdout);
#endif
    gettimeofday(&start,0);
    startTime = start.tv_sec*1000000 + start.tv_usec;
    do{
      gettimeofday(&now,0);
      currTime = now.tv_sec*1000000 + now.tv_usec;
    }while( (currTime - startTime) < time );
  }
}
void sendData(int newFd, sctp_assoc_t aId, unsigned int interval, unsigned 
int msgSize, unsigned int msgCnt){
  unsigned int sent = 0;
  struct timeval start[1], now[1], tmp[1];
  gettimeofday(start,0);
  while(sent < msgCnt){
    gettimeofday(now,0);
    tmp->tv_sec = start->tv_sec + (interval/1000)*sent;
    tmp->tv_usec = start->tv_usec + (interval%1000)*MS2USEC*sent;
    if(tmp->tv_usec/1000000 > 0){
      tmp->tv_sec += tmp->tv_usec/1000000;
      tmp->tv_usec = tmp->tv_usec%1000000;
    }
    if( timercmp(now, tmp, >=) ){
      struct message msg = createMsg(*tmp);
      
#ifdef VERBOSE_1
      printf("Sending msg\n");
#endif
#ifdef VERBOSE_2
      printf("stamp %lf\t", tmp->tv_sec + tmp->tv_usec/1000000.0);
      printf("now   %lf\t\n", now->tv_sec + now->tv_usec/1000000.0);
      fflush(stdout);
#endif
      send(newFd,aId,msg,msgSize);
      sent++;
    }
    else{
      long sleepTime = (tmp->tv_sec*1000000 + tmp->tv_usec) - 
        (now->tv_sec*1000000 + now->tv_usec);
#ifdef VERBOSE_1
      printf("Waiting %ld ms\n",sleepTime/1000);
      fflush(stdout);
#endif     
      waitFor(sleepTime);
    }
  }
}
int main(int argc, char** argv)
{
  int port, interval, msgsize, msgCnt;
  int fd;
  char* localIP;
  struct sockaddr_in sin_src[1];
  struct sockaddr_in sin_remote[1];
  socklen_t addr_len;
  if(argc != 6){
    perror("Usage: sctpServer <local ip> <port> <interval between msgs(ms)> 
<message size>(bytes) <messages>\n");
    perror("Got:");
    for(int i = 1; i < argc; i++){
      fprintf(stderr,"%s ",argv[1]);
    }
    perror("\n");
    exit(-1);
  }
  
  if ((fd = ext_socket(PF_INET, SOCK_SEQPACKET, IPPROTO_SCTP)) == -1) {
    perror("socket");
    exit(1);
  }
  setSocketOptions(fd);
  localIP  = argv[1];
  port     = atoi(argv[2]);
  interval = atoi(argv[3]);
  msgsize  = atoi(argv[4]);
  msgCnt   = atoi(argv[5]);
  sin_src->sin_family = AF_INET;
  sin_src->sin_port = htons(port);
  sin_src->sin_addr.s_addr = inet_addr(localIP);
  if (ext_bind(fd, (struct sockaddr *)sin_src, sizeof (*sin_src)) == -1) {
    perror("Unable to bind socket\n");
    exit(1);
  }
  
  struct sctp_initmsg initmsg;
  initmsg.sinit_num_ostreams = 1;
  initmsg.sinit_max_instreams = 1;
  initmsg.sinit_max_attempts = 0;
  initmsg.sinit_max_init_timeo  = 0;
  unsigned long size = sizeof(initmsg);
  if (ext_setsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, size) < 0) {
    perror("setsockopt SCTP_INITMSG");
    exit(1);
  }
  /* Allow new associations to be accepted */
  if (ext_listen(fd, 10) < 0) {
    perror("listen");
    exit(1);
  }
  sctp_assoc_t aId = getNewConnection(fd);
#ifdef VERBOSE_1
  printf("New connection recieved!\n");
  fflush(stdout);
#endif
  sendData(fd,aId,interval,msgsize,msgCnt);
  shutdownAssoc(fd,aId);
//  ext_close(fd);
  waitForShutdown(fd);
/*  while(1){
    sleep(2);
  }
*/
}
- RE: [SCTP-Discussion] An error in the SCTPlib/SocketAPI, Torbjörn Andersson, 10/09/2003
 
Archive powered by MHonArc 2.6.19+.