Chronological Thread 

< Chronological >      < Thread >

---

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
  • <Possible follow-up(s)>
  • Re: [SCTP-Discussion] An error in the SCTPlib/SocketAPI, Andreas Jungmaier, 10/09/2003
  • Re: [SCTP-Discussion] An error in the SCTPlib/SocketAPI, Andreas Jungmaier, 10/12/2003
    • SV: [SCTP-Discussion] An error in the SCTPlib/SocketAPI, Torbjörn Andersson, 10/13/2003
      • Re: SV: [SCTP-Discussion] An error in the SCTPlib/SocketAPI, Andreas Jungmaier, 10/13/2003

---

Archive powered by MHonArc 2.6.19+.