Chronological Thread 

< Chronological >      < Thread >

---

Hi all

I'm writing to you again now because I still have problem with the SCTP
layer. The 1.0.0 version did improve things but there is still a problem
when one sends messages without intermediate delays.

You can try it yourself if you compile my program and send 1000 messages
with an interval of 0ms, each message containing 500 bytes. SCTP does
not fail every time but it will eventually. I guess within 20 retries.
The simplest network setup will do fine(2 hosts). What happens is that
SCTP stalls somewhere as

I hope you find this information valuable and find the problem, what
ever it is.

Another problem:
When I initiate a shutdown on an association in a UDP socket I should be
able to retrieve the SCTP_SHUT_COMP notification but I'm not. I'm
sending a message with MSG_EOF set and the association is closed
correctly except that the initiator does not retrieve the
SCTP_SHUT_COMP.

Can you look into this? I find it difficult to debug your code so I
leave it to you.


Regards, Torbjörn Andersson
#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 VERBOSE_1
//#define VERBOSE_0

typedef struct{
u_long seq_num;
struct timeval queued;
struct timeval recved;
}message_t;

#ifdef VERBOSE_1
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;
}
}
#endif

void prnMessage(message_t msg)
{
struct timeval curr_time;
gettimeofday(&curr_time,NULL);

printf("%ld\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;
msg->msg_flags = 0;
iov[0].iov_len = sizeof(char)*PACKETSIZE;
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 VERBOSE_2
handle_event(buf);
#endif
union sctp_notification *snp = (union sctp_notification *)buf;
if(snp->sn_header.sn_type == SCTP_SHUTDOWN_EVENT){
#ifdef VERBOSE_0
printf("Shutdown notification!\n");
fflush(stdout);
#endif
}

if(snp->sn_header.sn_type == SCTP_ASSOC_CHANGE){
#ifdef VERBOSE_0
printf("SCTP ASSOC CHANGE!\n");
fflush(stdout);
#endif
if(snp->sn_assoc_change.sac_state == SCTP_SHUTDOWN_COMP){
#ifdef VERBOSE_0
printf("SCTP SHUTDOWN COMPLETE!\n");
fflush(stdout);
#endif
return total;
}
}
}
else if(nr != 20){
total += nr;
#ifdef VERBOSE_0
struct sctp_sndrcvinfo *sri =
(struct sctp_sndrcvinfo *)CMSG_FIRSTHDR(msg);
printf("Msg[%d]%d bytes received at stream
%d\n",mesgs.size(),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;
struct hostent *addr;
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];
if (atoi(localIP) > 0 ) {
/* Numeric */
sin_src->sin_family = AF_INET;
sin_src->sin_addr.s_addr = inet_addr(localIP);
} else {
if ((addr=gethostbyname(localIP)) == NULL){
perror("bad hostname");
exit(-1);
}
sin_src->sin_family = addr->h_addrtype;

struct in_addr addr_tmp[1];
bcopy(addr->h_addr_list[0],(char*)&addr_tmp, sizeof(struct in_addr));
sin_src->sin_addr.s_addr = addr_tmp->s_addr;
}
sin_src->sin_port = htons(0);

struct sockaddr_in sin_dest[1];
if (atoi(remoteIP) > 0 ) {
/* Numeric */
sin_dest->sin_family = AF_INET;
sin_dest->sin_addr.s_addr = inet_addr(remoteIP);
} else {
if ((addr=gethostbyname(remoteIP)) == NULL){
perror("bad hostname");
exit(-1);
}
sin_dest->sin_family = addr->h_addrtype;

struct in_addr addr_tmp[1];
bcopy(addr->h_addr_list[0],(char*)&addr_tmp, sizeof(struct in_addr));
sin_dest->sin_addr.s_addr = addr_tmp->s_addr;
}
sin_dest->sin_port = htons(port);

/*
if ((fd = ext_socket(PF_INET, SOCK_SEQPACKET, IPPROTO_SCTP)) == -1) {
perror("socket");
exit(1);
}
*/
if ((fd = ext_socket(PF_INET, SOCK_STREAM, IPPROTO_SCTP)) == -1) {
perror("socket");
exit(1);
}

/* 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 = 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 = 10;
initmsg.sinit_max_init_timeo = 60;

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);
}

if(ext_connect(fd,(struct sockaddr*)sin_dest,sizeof(struct sockaddr)) != 0){
printf("0 messages were retrieved\n");
printf("Unable to connect to server after 10 retries\n");
exit(-1);
}
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++;
}
}

#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_0
//#define VERBOSE_1
//#define VERBOSE_2
#define VERBOSE_SHUT

#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 = 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){
#ifdef VERBOSE_SHUT
printf("Waiting for shutdown_comp event!\n");
#endif
len = ext_recvmsg(fd, msg, 0);
/* Intercept notifications here */
if(len > 0){
#ifdef VERBOSE_SHUT
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_SHUT
printf("Some other state!\n");
#endif
}
}
else{
#ifdef VERBOSE_SHUT
printf("Some other notification!\n");
#endif
}
}
else{
#ifdef VERBOSE_SHUT
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("%ld\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){
long sleepTime;
// if(time > 10000){
#ifdef VERBOSE_2
printf("Sleeping\n");
fflush(stdout);
#endif
sleepTime = time - INTERVALL_CORRECTION;
usleep( (sleepTime < 0) ? 0: sleepTime);
/* }
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: %d...",sent+1);
fflush(stdout);
#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);
}
}
#ifdef VERBOSE_0
printf("Everything sent!\n");
fflush(stdout);
#endif
}



int main(int argc, char** argv)
{
int port, interval, msgsize, msgCnt;
int fd;
char* localIP;
struct sockaddr_in sin_src[1];
struct hostent *addr;

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]);

if (atoi(localIP) > 0 ) {
/* Numeric */
sin_src->sin_family = AF_INET;
sin_src->sin_addr.s_addr = inet_addr(localIP);
} else {
if ((addr=gethostbyname(localIP)) == NULL){
perror("bad hostname");
exit(-1);
}
sin_src->sin_family = addr->h_addrtype;

struct in_addr addr_tmp[1];
bcopy(addr->h_addr_list[0],(char*)&addr_tmp, sizeof(struct in_addr));
sin_src->sin_addr.s_addr = addr_tmp->s_addr;
}
sin_src->sin_port = htons(port);

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 = 10;
initmsg.sinit_max_init_timeo = 600;

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);
}
*/
}

---

• [SCTP-Discussion] RETRY: An error in the SCTPlib/SocketAPI, Torbjörn Andersson, 10/23/2003
  • Re: [SCTP-Discussion] RETRY: An error in the SCTPlib/SocketAPI, Andreas Jungmaier, 10/28/2003
  • <Possible follow-up(s)>
  • [SCTP-Discussion] RETRY: An error in the SCTPlib/SocketAPI, Andreas Jungmaier, 10/23/2003

---

Archive powered by MHonArc 2.6.19+.