Chronological Thread 

< Chronological >      < Thread >

---

Hi Michael!
I don't undestand... where (or when) should I do locking?
I attach the sample app. It's based en echo_tool.c
I don't do anything complicated. Just launch 4 threads and start sending packages to another pc, which have the same app that launch 4 threads.
should I use the sockeapi to be thread safe ?
Regards,
Julio




--
--
Stewie: [After Lois tries to feed Stewie his broccoli "airplane style"] Damn you, Damn the Broccoli, and Damn the Wright Brothers.
----------------------------
Julio Kriger
mailto:juliokriger AT gmail.com
#include "sctp_wrapper.h"
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif


#define MAXIMUM_NUMBER_OF_LOCAL_ADDRESSES 2
#define MAXIMUM_NUMBER_OF_ASSOCIATIONS 8
#define MAXIMUM_PAYLOAD_LENGTH 32768

#ifndef min
#define min(x,y) (x)<(y)?(x):(y)
#endif

#define PORT_BASE 12000

#define MESSAGE_START "MSG_START"
#define MESSAGE_STOP "MSG_STOP"

#define SHORTLIVE_CLIENT 0
#define LONGLIVE_CLIENT 1
#define SHORTLIVE_SERVER 2
#define LONGLIVE_SERVER 3

struct thread_params {
unsigned long nrOfReceivedBytes;
unsigned int assocID;
int threadType;
int sourcePort;
int destPort;
int stop;
};

struct thread_params threadParams[4];

unsigned char
localAddressList[MAXIMUM_NUMBER_OF_LOCAL_ADDRESSES][SCTP_MAX_IP_LEN];
unsigned char destinationAddress[SCTP_MAX_IP_LEN] = "";
unsigned short noOfLocalAddresses = 0;
unsigned int measureTime = 10000;
int unknownCommand = 0;
int inputStreams = 1;
int outputStreams = 1;
int verbose = 0;
FILE * hFile = NULL;
char filename[256] = "";
int runAsServer = 1;


void MeasurementTimerRunOffFunction(unsigned int timerID, void *
threadParamIndex, void * dummy);

/////////////////////////////////////////////////////////////////////////////////////////
int randomMessageLength() {
int r;

r = 1024 + (int) ((8192.0 - 1024.0) * rand() / (RAND_MAX + 1.0));

if (!(1024 <= r && r <= 8192)) {
printf("ERROR en randomMessageLength!!!");
exit(1);
}

return r;
}

/////////////////////////////////////////////////////////////////////////////////////////

char * pathStateName(unsigned int state)
{
switch (state) {
case SCTP_PATH_OK:
return "REACHABLE";
break;
case SCTP_PATH_UNREACHABLE:
return "UNREACHABLE";
break;
case SCTP_PATH_ADDED:
return "ADDED";
break;
case SCTP_PATH_REMOVED:
return "REMOVED";
break;
case SCTP_PATH_CONFIRMED:
return "CONFIRMED";
break;
case SCTP_PATH_UNCONFIRMED:
return "UNCONFIRMED";
break;
default:
return "UNKNOWN";
break;
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void printUsage(void)
{
printf("usage: test_tool2 [options]\n");
printf("options:\n");
printf("-d destination_addr establish a association with the
specified address\n");
printf("-t measure time number of seconds to measure (default
1000)\n");
printf("-f filename filename to dump meditions\n");
printf("-z streams number of input stream (default 1)\n");
printf("-x streams number of output stream (default 1)\n");
printf("-c to run as client, else run as server\n");
printf("-v verbose mode\n");
}

/////////////////////////////////////////////////////////////////////////////////////////

void getArgs(int argc, char * * argv)
{
int i;
char * opt;

for (i=1; i < argc ;i++) {
if (argv[i][0] == '-') {
switch (argv[i][1]) {
case 'd':
if (i+1 >= argc) {
printUsage();
exit(0);
}
opt = argv[++i];
if (strlen(opt) < SCTP_MAX_IP_LEN) {
strcpy((char *)destinationAddress,
opt);
}
break;
case 't':
if (i+1 >= argc) {
printUsage();
exit(0);
}
opt = argv[++i];
measureTime = atoi(opt) * 1000;
break;
case 'f':
if (i+1 >= argc) {
printUsage();
exit(0);
}
opt = argv[++i];
strcpy(filename, opt);
break;
case 'z':
if (i+1 >= argc) {
printUsage();
exit(0);
}
opt = argv[++i];
inputStreams = atoi(opt);
break;
case 'x':
if (i+1 >= argc) {
printUsage();
exit(0);
}
opt = argv[++i];
outputStreams = atoi(opt);
break;
case 'c':
runAsServer = 0;
break;
case 'v':
verbose = 1;
break;
default:
unknownCommand = 1;
break;
}
} else
unknownCommand = 1;
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void checkArgs(void)
{
int abortProgram;
int printUsageInfo;

abortProgram = 0;
printUsageInfo = 0;

if (noOfLocalAddresses == 0) {
strcpy((char *)localAddressList[noOfLocalAddresses],
"0.0.0.0");
noOfLocalAddresses++;
}
if (unknownCommand ==1) {
printf("Error: Unkown options in command.\n");
printUsageInfo = 1;
abortProgram = 1;
}

if (printUsageInfo == 1) {
printUsage();
}
if (abortProgram == 1) {
exit(-1);
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void DataArriveNotif(unsigned int assocID, unsigned short streamID, unsigned
int len,
unsigned short streamSN,unsigned int
TSN, unsigned int protoID,
unsigned int unordered, void *
threadParamIndex)
{
// printf("DataArriveNotif %d\n", assocID);

unsigned char chunk[MAXIMUM_PAYLOAD_LENGTH];
unsigned int length;
unsigned short ssn;
unsigned int the_tsn;
struct thread_params * cp = & threadParams[(int) threadParamIndex];

if (verbose) {
// printf("DataArriveNotif %-8x: Data arrived (%u bytes on
stream %u, %s)\n", assocID, len, streamID,
(unordered==SCTP_ORDERED_DELIVERY)?"ordered":"unordered");
}

/* read it */
length = MAXIMUM_PAYLOAD_LENGTH;
memset(chunk, 0, MAXIMUM_PAYLOAD_LENGTH);
SCTP_receive(cp->assocID, streamID, chunk, &length, &ssn, &the_tsn,
SCTP_MSG_DEFAULT);

/* analize message */
if (strncmp(MESSAGE_START, (char *) chunk, sizeof(MESSAGE_START)) == 0) {
printf("RECEIVED MESSAGE_START %d\n", cp->assocID);
cp->nrOfReceivedBytes = 0;
SCTP_startTimer(measureTime, MeasurementTimerRunOffFunction,
threadParamIndex, NULL);
} else {
if (strncmp(MESSAGE_STOP, (char *) chunk, sizeof(MESSAGE_STOP)) == 0)
{
printf("RECEIVED MESSAGE_STOP %d\n", cp->assocID);
cp->stop = 1;
return;
}
}

if (! cp->stop) {
/* update counter */
cp->nrOfReceivedBytes += length;

/* and send the response */
switch (cp->threadType) {
case SHORTLIVE_CLIENT:
length = randomMessageLength();
break;
case LONGLIVE_CLIENT:
length = MAXIMUM_PAYLOAD_LENGTH;
break;
case SHORTLIVE_SERVER:
length = randomMessageLength();
break;
case LONGLIVE_SERVER:
length = 50;
break;
}

memset(chunk, 0, MAXIMUM_PAYLOAD_LENGTH);
int res = SCTP_send(cp->assocID, streamID, chunk, length, protoID,
SCTP_USE_PRIMARY, SCTP_NO_CONTEXT, SCTP_INFINITE_LIFETIME, unordered,
SCTP_BUNDLING_DISABLED);
if (res != 0) {
printf("1 SCTP_send = %d\n", res);
printf("cp->assocID %d\n", cp->assocID);
}
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void * CommunicationUpNotif(unsigned int assocID, int status, unsigned int
noOfPaths,
unsigned short
noOfInStreams, unsigned short noOfOutStreams,
int
associationSupportsPRSCTP, void * dummy)
{
// printf("CommunicationUpNotif %d\n", assocID);

unsigned char chunk[MAXIMUM_PAYLOAD_LENGTH];
unsigned int length;
SCTP_PathStatus pathStatus;
SCTP_AssociationStatus assocStatus;
unsigned short pathID;
int index;

sctp_getAssocStatus(assocID, &assocStatus);

index = assocStatus.sourcePort - PORT_BASE;

struct thread_params * cp = &threadParams[index];
int res;

cp->assocID = assocID;
//printf("cp->assocID %d\n", cp->assocID);

if (verbose) {
printf("%-8x: Communication up (%u paths:", cp->assocID,
noOfPaths);
for (pathID=0; pathID < noOfPaths; pathID++) {
SCTP_getPathStatus(cp->assocID, pathID, &pathStatus);
printf(" %s", pathStatus.destinationAddress);
}
printf(")\n");
printf("%-8x: %u incoming, %u outgoing streams.\n",
cp->assocID, noOfInStreams, noOfOutStreams);
}

switch (cp->threadType) {
case SHORTLIVE_CLIENT:
length = randomMessageLength();
memset(chunk, 0, MAXIMUM_PAYLOAD_LENGTH);
res = SCTP_send(cp->assocID, 0, chunk, length,
SCTP_GENERIC_PAYLOAD_PROTOCOL_ID, SCTP_USE_PRIMARY, SCTP_NO_CONTEXT,
SCTP_INFINITE_LIFETIME, SCTP_ORDERED_DELIVERY, SCTP_BUNDLING_DISABLED);
if (res != 0) {
printf("2 SCTP_send = %d\n", res);
printf("cp->assocID %d\n", cp->assocID);
}
break;
case LONGLIVE_CLIENT:
length = MAXIMUM_PAYLOAD_LENGTH;
memset(chunk, 0, MAXIMUM_PAYLOAD_LENGTH);
res = SCTP_send(cp->assocID, 0, chunk, length,
SCTP_GENERIC_PAYLOAD_PROTOCOL_ID, SCTP_USE_PRIMARY, SCTP_NO_CONTEXT,
SCTP_INFINITE_LIFETIME, SCTP_ORDERED_DELIVERY, SCTP_BUNDLING_DISABLED);
if (res != 0) {
printf("3 SCTP_send = %d\n", res);
printf("cp->assocID %d\n", cp->assocID);
}
break;
}

return (void *) index;
}

/////////////////////////////////////////////////////////////////////////////////////////

void SendFailureNotif(unsigned int assocID, unsigned char * unsent_data,
unsigned int dataLength, unsigned int * context, void * threadParamIndex)
{
// printf("SendFailureNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

printf("%-8x: Send failure\n", assocID);
}

/////////////////////////////////////////////////////////////////////////////////////////

void NetworkStatusChangeNotif(unsigned int assocID, short affectedPathID,
unsigned short newState, void * threadParamIndex)
{
// printf("NetworkStatusChangeNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
// printf("cp->assocID %d\n", cp->assocID);

SCTP_AssociationStatus assocStatus;
SCTP_PathStatus pathStatus;
unsigned short pathID;

if (verbose) {
SCTP_getPathStatus(cp->assocID, affectedPathID, &pathStatus);
printf("%-8x: Network status change: path %u (towards %s) is
now %s\n", cp->assocID, affectedPathID, pathStatus.destinationAddress,
pathStateName(newState));
}

/* if the primary path has become inactive */
if ((newState == SCTP_PATH_UNREACHABLE) &&
(affectedPathID == SCTP_getPrimary(cp->assocID))) {

/* select a new one */
SCTP_getAssocStatus(cp->assocID, &assocStatus);
for (pathID=0; pathID < assocStatus.numberOfAddresses;
pathID++) {
SCTP_getPathStatus(cp->assocID, pathID, &pathStatus);
if (pathStatus.state == SCTP_PATH_OK) {
break;
}
}
/* and use it */
if (pathID < assocStatus.numberOfAddresses) {
SCTP_setPrimary(cp->assocID, pathID);
}
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void CommunicationLostNotif(unsigned int assocID, unsigned short status, void
* threadParamIndex)
{
// printf("CommunicationLostNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

unsigned char buffer[MAXIMUM_PAYLOAD_LENGTH];
unsigned int bufferLength;
unsigned short streamID, streamSN;
unsigned int protoID;
unsigned int tsn;

if (verbose) {
printf("%-8x: Communication lost (status %u)\n", cp->assocID,
status);
}

/* retrieve data */
bufferLength = sizeof(buffer);
while (SCTP_receiveUnsent(cp->assocID, buffer, &bufferLength, &tsn,
&streamID, &streamSN, &protoID) >= 0) {
/* do something with the retrieved data */
/* after that, reset bufferLength */
bufferLength = sizeof(buffer);
}

bufferLength = sizeof(buffer);
while (SCTP_receiveUnacked(cp->assocID, buffer, &bufferLength, &tsn,
&streamID, &streamSN, &protoID) >= 0) {
/* do something with the retrieved data */
/* after that, reset bufferLength */
bufferLength = sizeof(buffer);
}

/* delete the association */
SCTP_deleteAssociation(cp->assocID);
}

/////////////////////////////////////////////////////////////////////////////////////////

void CommunicationErrorNotif(unsigned int assocID, unsigned short status,
void * threadParamIndex)
{
// printf("CommunicationErrorNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

if (verbose) {
printf("%-8x: Communication error (status %u)\n", assocID,
status);
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void RestartNotif(unsigned int assocID, void * threadParamIndex)
{
// printf("RestartNotif %d\n", assocID);

SCTP_AssociationStatus assocStatus;

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

if (verbose) {
printf("%-8x: Restart\n", cp->assocID);
}

SCTP_getAssocStatus(assocID, &assocStatus);

/* update ULP data */
cp->assocID = assocID;
}

/////////////////////////////////////////////////////////////////////////////////////////

void ShutdownCompleteNotif(unsigned int assocID, void * threadParamIndex)
{
// printf("ShutdownCompleteNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

if (verbose) {
printf("%-8x: Shutdown complete\n", cp->assocID);
}
/* free ULP data */
//((struct ulp_data *) ulpDataPtr)->maximumStreamID = -1;
SCTP_deleteAssociation(cp->assocID);
}

/////////////////////////////////////////////////////////////////////////////////////////

void ShutdownReceivedNotif(unsigned int assocID, void * threadParamIndex)
{
// printf("ShutdownReceivedNotif %d\n", assocID);

struct thread_params * cp = &threadParams[(int) threadParamIndex];
printf("cp->assocID %d\n", cp->assocID);

if (verbose) {
printf("%-8x: Shutdown received\n", cp->assocID);
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void MeasurementTimerRunOffFunction(unsigned int timerID, void *
threadParamIndex, void * dummy)
{
// printf("MeasurementTimerRunOffFunction %d\n", (int) threadParamIndex);

int res;
struct thread_params * cp = &threadParams[(int) threadParamIndex];
// printf("cp->assocID %d\n", cp->assocID);

switch(cp->threadType) {
case SHORTLIVE_CLIENT:
case LONGLIVE_CLIENT:
printf("SEND MESSAGE_START %d\n", cp->assocID);
res = SCTP_send(cp->assocID, 0, (unsigned char *) MESSAGE_START,
sizeof(MESSAGE_START), SCTP_GENERIC_PAYLOAD_PROTOCOL_ID, SCTP_USE_PRIMARY,
SCTP_NO_CONTEXT, SCTP_INFINITE_LIFETIME, SCTP_ORDERED_DELIVERY,
SCTP_BUNDLING_DISABLED);
if (res != 0) {
printf("4 SCTP_send = %d\n", res);
}
break;
case SHORTLIVE_SERVER:
case LONGLIVE_SERVER:
if(hFile != stdout) {
fprintf(hFile, "MEASURE %-8d: %-8lu Bytes received\n",
cp->destPort,cp->nrOfReceivedBytes);
}
printf("MEASURE %-8d: %-8lu Bytes received\n",
cp->destPort,cp->nrOfReceivedBytes);

printf("SEND MESSAGE_STOP %d\n", cp->assocID);
res = SCTP_send(cp->assocID, 0, (unsigned char *) MESSAGE_STOP,
sizeof(MESSAGE_STOP), SCTP_GENERIC_PAYLOAD_PROTOCOL_ID, SCTP_USE_PRIMARY,
SCTP_NO_CONTEXT, SCTP_INFINITE_LIFETIME, SCTP_ORDERED_DELIVERY,
SCTP_BUNDLING_DISABLED);
if (res != 0) {
printf("5 SCTP_send = %d\n", res);
}
cp->stop = 1;
break;
}
}

/////////////////////////////////////////////////////////////////////////////////////////

void * ThreadWorker(void * threadParamIndex)
{
printf("ThreadWorker %d\n", (int)threadParamIndex);

unsigned short sctpInstance;
SCTP_ulpCallbacks LLServerEchoUlp;
struct thread_params * cp = & threadParams[(int) threadParamIndex];

/* initialize the echo_ulp variable */
LLServerEchoUlp.dataArriveNotif = &DataArriveNotif;
LLServerEchoUlp.sendFailureNotif = &SendFailureNotif;
LLServerEchoUlp.networkStatusChangeNotif = &NetworkStatusChangeNotif;
LLServerEchoUlp.communicationUpNotif = &CommunicationUpNotif;
LLServerEchoUlp.communicationLostNotif = &CommunicationLostNotif;
LLServerEchoUlp.communicationErrorNotif = &CommunicationErrorNotif;
LLServerEchoUlp.restartNotif = &RestartNotif;
LLServerEchoUlp.shutdownCompleteNotif = &ShutdownCompleteNotif;
LLServerEchoUlp.peerShutdownReceivedNotif = &ShutdownReceivedNotif;

sctpInstance = SCTP_registerInstance(cp->sourcePort, inputStreams,
outputStreams, noOfLocalAddresses, localAddressList, LLServerEchoUlp);

switch(cp->threadType) {
case SHORTLIVE_CLIENT:
case LONGLIVE_CLIENT:
//sleep(5);

SCTP_associate(sctpInstance, outputStreams, destinationAddress,
cp->destPort, threadParamIndex);

printf("SCTP_startTimer %d\n", threadParamIndex);
SCTP_startTimer(3000, &MeasurementTimerRunOffFunction,
threadParamIndex, NULL);

break;
case SHORTLIVE_SERVER:
case LONGLIVE_SERVER:
break;
}

while (! cp->stop) {
sleep(0.1);
}

printf("EXIT %d\n", cp->assocID);
pthread_exit(NULL);
return NULL;
}

/////////////////////////////////////////////////////////////////////////////////////////

int main(int argc, char **argv)
{
int i;
pthread_t threads[8];
SCTP_LibraryParameters params;

getArgs(argc, argv);
checkArgs();

srand(time(NULL));

SCTP_initLibrary();
SCTP_getLibraryParameters(&params);

params.sendOotbAborts = 0;
/* params.checksumAlgorithm = SCTP_CHECKSUM_ALGORITHM_ADLER32; */
params.checksumAlgorithm = SCTP_CHECKSUM_ALGORITHM_CRC32C;
SCTP_setLibraryParameters(&params);

if (filename[0] != 0) {
hFile = fopen(filename, "w+");
} else {
hFile = stdout;
}

int index = 0;

if (runAsServer) {
// lanzo los thread del short life server
for (i = 0; i < 2; i++) {
threadParams[index].assocID = 0;
threadParams[index].destPort = PORT_BASE + index;
threadParams[index].nrOfReceivedBytes = 0;
threadParams[index].sourcePort = PORT_BASE + index;
threadParams[index].stop = 0;
threadParams[index].threadType = SHORTLIVE_SERVER;
//pthread_create(&threads[index], NULL, ThreadWorker, (void *)
index);
index ++;
}
// lanzo los thread del long life server
for (i = 0; i < 2; i++) {
threadParams[index].assocID = 0;
threadParams[index].destPort = PORT_BASE + index;
threadParams[index].nrOfReceivedBytes = 0;
threadParams[index].sourcePort = PORT_BASE + index;
threadParams[index].stop = 0;
threadParams[index].threadType = LONGLIVE_SERVER;
//pthread_create(&threads[index], NULL, ThreadWorker, (void *)
index);
index ++;
}
} else {
// lanzo los thread del short life client
for (i = 0; i < 2; i++) {
threadParams[index].assocID = 0;
threadParams[index].destPort = PORT_BASE + index;
threadParams[index].nrOfReceivedBytes = 0;
threadParams[index].sourcePort = PORT_BASE + index;
threadParams[index].stop = 0;
threadParams[index].threadType = SHORTLIVE_CLIENT;
pthread_create(&threads[index], NULL, ThreadWorker, (void *)
index);
index ++;
}
// lanzo los thread del long life client
for (i = 0; i < 2; i++) {
threadParams[index].assocID = 0;
threadParams[index].destPort = PORT_BASE + index;
threadParams[index].nrOfReceivedBytes = 0;
threadParams[index].sourcePort = PORT_BASE + index;
threadParams[index].stop = 0;
threadParams[index].threadType = LONGLIVE_CLIENT;
pthread_create(&threads[index], NULL, ThreadWorker, (void *)
index);
index ++;
}
}

/* run the event handler forever */
while (!( (threadParams[0].stop == 1) && (threadParams[1].stop == 1) &&
(threadParams[2].stop == 1) && (threadParams[3].stop == 1))) {
SCTP_eventLoop();
};

for (i = 0; i < 4; i++) {
pthread_join(threads[i], NULL);
}

if (filename[0] != 0) {
fclose(hFile);
}

/* this will never be reached */
exit(0);
}

---

• [SCTP-Discussion] a weird bug (segmentation fault), Julio Kriger, 05/22/2006
  • Re: [SCTP-Discussion] a weird bug (segmentation fault), Michael Tuexen, 05/22/2006
    • Re: [SCTP-Discussion] a weird bug (segmentation fault), Julio Kriger, 05/22/2006
      • Re: [SCTP-Discussion] a weird bug (segmentation fault), Michael Tuexen, 05/22/2006
  • Re: [SCTP-Discussion] a weird bug (segmentation fault), Thomas Dreibholz, 05/23/2006
    • Re: [SCTP-Discussion] a weird bug (segmentation fault), Julio Kriger, 05/23/2006
      • Re: [SCTP-Discussion] a weird bug (segmentation fault), Thomas Dreibholz, 05/23/2006
        • Re: [SCTP-Discussion] a weird bug (segmentation fault), Julio Kriger, 05/23/2006
          • Re: [SCTP-Discussion] a weird bug (segmentation fault), Julio Kriger, 05/28/2006
            • Re: [SCTP-Discussion] a weird bug (segmentation fault), Thomas Dreibholz, 05/31/2006

---

Archive powered by MHonArc 2.6.19+.