C समानांतर नेटवर्क कम्प्यूटिंग

शुभ दोपहर, ह्रज़ितली! हाल ही में मुझे एक निश्चित अभिन्न के समानांतर गणना के लिए एक कार्यक्रम लिखना था। स्वाभाविक रूप से, काम के अंत के बाद, कुछ अनुभव प्राप्त हुआ था, और मैं इस अनुभव को आपके साथ साझा करना चाहूंगा। मुझे तुरंत यह कहना चाहिए कि लेख के ढांचे के भीतर मैं एक पूर्णांक फ़ंक्शन के पार्सिंग पर विचार नहीं करूंगा, लेकिन मैं एक दूसरे के साथ थ्रेड्स और कंप्यूटरों की बातचीत पर ध्यान केंद्रित करूंगा।

समस्या का बयान

बेशक, इससे पहले कि आप कोड लिखना शुरू करें, आपको अपने आप को सही कार्य निर्धारित करने की आवश्यकता है, कार्यक्रम की आवश्यकताएं इस प्रकार थीं:

• गणना समानांतर में कई मशीनों पर की जानी चाहिए (स्थानीय नेटवर्क के भीतर)
• अभिकलन बहु-थ्रेडेड होना चाहिए
• कंप्यूटर तक पहुंच जिस पर गणना की जाती है, रुक-रुक कर होती है

यही है, एक प्रोग्राम लिखना आवश्यक है जो कई कंप्यूटरों (जाहिरा तौर पर वर्कस्टेशन पर) पर चलेगा और अभिन्न गणना करने के निर्देशों की प्रतीक्षा करेगा, फिर इसे पढ़ें और परिणाम को वापस उसी को भेजें जिसने गिनती करने के लिए कहा था।

चूंकि हमें नेटवर्क से निपटना है, इसलिए हम अभिन्न की गणना के लिए संभाव्य विधि का चयन करते हैं, हम मोंटे कार्लो पद्धति का उपयोग करेंगे। समानांतर करना आसान है, दो तरीके स्पष्ट हैं:

• सेगमेंट द्वारा, अर्थात प्रत्येक कैलकुलेटर एक निश्चित सेगमेंट पर n अंक उत्पन्न करता है
• उत्पन्न अंकों की संख्या से, अर्थात्, प्रत्येक कैलकुलेटर पूरे एकीकरण अंतराल पर n / (कैलकुलेटरों की संख्या) अंक उत्पन्न करता है

मैं दूसरी विधि चुनूंगा, विशेष रूप से कुछ भी प्रेरित किए बिना, मुझे गणनाकर्ताओं को अपने खंडों की शुरुआत और अंत पास करने की ज़रूरत नहीं है।
इसलिए हमें एक सर्वर (कैलकुलेटर) की आवश्यकता है जो नौकरी के लिए इंतजार करेगा, परिणाम पढ़ेगा और वापस करेगा। और हमें एक क्लाइंट की भी आवश्यकता है जिसके साथ उपयोगकर्ता बातचीत करेगा।
कंप्यूटिंग के लिए सर्वर कैसे खोजें? सभी आईपीओं को दिल से जानना या उन्हें कहीं लिख देना सबसे अच्छा विकल्प नहीं है, यह देखते हुए कि सर्वर के आईपी पते गतिशील हैं, साथ ही वे इस समय किसी अन्य कार्य पर विचार कर सकते हैं या बस ऑफ़लाइन हो सकते हैं। समाधान सरल है, हम वर्तमान में उपलब्ध सर्वरों को खोजने के लिए प्रसारण अनुरोध का उपयोग करेंगे।

दूसरी समस्या, सूचना के आदान-प्रदान के लिए किस परिवहन का उपयोग करना है? एक प्रसारण अनुरोध के लिए, हम स्पष्ट रूप से यूडीपी का उपयोग करेंगे। लेकिन क्लाइंट और सर्वर के इंटरैक्शन के लिए, आप यूडीपी और टीसीपी दोनों का उपयोग कर सकते हैं। लेकिन टीसीपी के साथ कम समस्याएं होंगी, क्योंकि हमें कनेक्शन की स्थिति की जांच करने की आवश्यकता नहीं होगी। यदि दूसरी तरफ सॉकेट बंद हो जाता है, तो ओएस इसका पता लगाएगा और हमें बताएगा।

इंटरैक्शन का अंतिम संस्करण निम्नानुसार है: सर्वर टीसीपी सॉकेट पर कनेक्शन के लिए इंतजार कर रहा है और उसी समय प्रसारण अनुरोधों का जवाब देता है, जो ग्राहकों को अपनी उपस्थिति का संकेत देता है। जैसे ही क्लाइंट ने कनेक्ट किया और कार्य दिया, हम प्रसारण अनुरोधों का जवाब देते हैं, गणना करते हैं, क्लाइंट को जवाब देते हैं, और फिर से चक्र शुरू करते हैं। क्लाइंट: एक अनुरोध भेजता है, सर्वरों की एक सूची बनाता है, उनके बीच कार्य साझा करता है, परिणाम प्राप्त करता है और उन्हें उपयोगकर्ता को प्रदर्शित करता है।

सर्वर

पहले एक सर्वर लिखते हैं। सबसे पहले, हम इस बात पर सहमत होंगे कि हम क्लाइंट से अनुरोधों के लिए किस पोर्ट का इंतजार करेंगे, इसे पोर्ट नंबर 38199 होने दें। फिर हम क्लाइंट को सर्वर पर नौकरी भेजने के लिए संरचना की घोषणा करेंगे।

#define RCVPORT 38199
#define FUNC(x) x*x // x^2
//
typedef struct {
int limits;
int numoftry;
} task_data_t;

* This source code was highlighted with Source Code Highlighter .

जैसा कि ऊपर दिए गए कोड से देखा जा सकता है, क्लाइंट सर्वर को एकीकरण की ऊपरी सीमा और उसके द्वारा किए जाने वाले प्रयासों की संख्या भेज देगा।

यह स्पष्ट है कि हम गणना कई थ्रेड्स में करेंगे, इसलिए हम एक तर्क संरचना और कैलकुलेटर थ्रेड्स के लिए एक फ़ंक्शन बनाएंगे:

//
typedef struct {
int limits; //
long numoftry; //
long double *results; //
} thread_args_t;

//
void *calculate( void *arg) {
// thread_args_t
thread_args_t *tinfo = (thread_args_t*) arg;
long double result = 0;
int xlim = tinfo->limits;
int trys = tinfo->numoftry;
unsigned a = xlim;
for ( int i = 0; i < trys; ++i) {
int div = rand_r(&a);
int div2 = rand_r(&a);
double x = div % xlim + (div2/(div2*1.0 + div*1.0)) ;
result += FUNC(x);
}
*(tinfo->results) = result;
return NULL;
}

* This source code was highlighted with Source Code Highlighter .

मैं इस तथ्य पर ध्यान आकर्षित करता हूं कि रैंड_आर फ़ंक्शन का उपयोग किया जाता है (अहस्ताक्षरित इंट * सीडैप) । क्योंकि यह फ़ंक्शन कॉल के बीच एक मध्यवर्ती मूल्य को संग्रहीत करने के लिए एक स्थानीय चर का उपयोग करता है। हम सभी थ्रेड्स के लिए वैश्विक वैरिएबल के उपयोग की अनुमति नहीं दे सकते हैं, क्योंकि यह रैंड () फ़ंक्शन का उपयोग करने के मामले में होगा, क्योंकि यह उनके आपसी अवरोध का कारण होगा।

सभी कम्प्यूटेशनल थ्रेड शुरू करने के बाद, मुख्य धागा pthread_join () फ़ंक्शन पर लटकाएगा और कुछ भी नहीं कर सकता है, यदि क्लाइंट खाली गणना से बचने के लिए गणना के दौरान मर जाता है, तो एक और थ्रेड चलाएँ जो क्लाइंट की स्थिति की जांच करेगा।

//
typedef struct {
int sock; //
pthread_t *calcthreads; //
int threadnum; //
} checker_args_t;

// SIGUSR1
void thread_cancel( int signo) {
pthread_exit(PTHREAD_CANCELED);
}

//
void *client_check( void *arg) {
// checker_args_t
checker_args_t *args = (checker_args_t*) arg;
char a[10];
recv(args->sock, &a, 10, 0); // TCP,
// , recv -1
int st;
for ( int i = 0; i < args->threadnum; ++i)
st = pthread_kill(args->calcthreads[i], SIGUSR1); // SIGUSR1
return NULL;
}

* This source code was highlighted with Source Code Highlighter .

कंप्यूटिंग थ्रेड्स की जबरन समाप्ति के लिए, मैंने संकेतों का उपयोग करने का फैसला किया, क्योंकि कोई स्मृति उनके काम के परिणामस्वरूप आवंटित नहीं की जाती है, डरने की कोई बात नहीं है। हालाँकि pthread_cancel () फ़ंक्शन और pthread_cleanup_push () और pthread_cleanup_pop () मैक्रोज़ का उपयोग करना अधिक सही होगा। बेशक, थ्रेड_कैंसेल () फ़ंक्शन लिखा जाता है जिसे सिग्नल मिलने पर निष्पादित किया जाएगा। और याद रखें कि प्रोग्राम की शुरुआत में, थ्रेड्स शुरू करने से पहले, आपको सिग्नल प्रोसेसिंग के लिए सही मास्क सेट करने की आवश्यकता होती है, अन्यथा हम प्रोग्राम को छोड़ने के लिए जोखिम लेते हैं।

अब एक धागा लिखते हैं जो प्रसारण अनुरोधों का जवाब देगा। ताकि मुख्य धागा शांति से लटका रह सके, ग्राहक का इंतजार कर सके और हमारा अतिरिक्त धागा, उस पल के अनुरोधों का जवाब दे सके।

void *listen_broadcast( void *arg) {
int *isbusy = arg;
// broadcast
int sockbrcast = socket(PF_INET, SOCK_DGRAM, 0);
if (sockbrcast == -1) {
perror( "Create broadcast socket failed" );
exit(EXIT_FAILURE);
}

// broadcast
int port_rcv = RCVPORT;
struct sockaddr_in addrbrcast_rcv;
bzero(&addrbrcast_rcv, sizeof (addrbrcast_rcv));
addrbrcast_rcv.sin_family = AF_INET;
addrbrcast_rcv.sin_addr.s_addr = htonl(INADDR_ANY);
addrbrcast_rcv.sin_port = htons(port_rcv);
//
if (bind(sockbrcast, ( struct sockaddr *) &addrbrcast_rcv,
sizeof (addrbrcast_rcv)) < 0) {
perror( "Bind broadcast socket failed" );
close(sockbrcast);
exit(EXIT_FAILURE);
}

int msgsize = sizeof ( char ) * 18;
char hellomesg[18];
bzero(hellomesg, msgsize);
// broadcast'
fcntl(sockbrcast, F_SETFL, O_NONBLOCK);

//
fd_set readset;
FD_ZERO(&readset);
FD_SET(sockbrcast, &readset);

//
struct timeval timeout;
timeout.tv_sec = 3;
timeout.tv_usec = 0;

struct sockaddr_in client;;
bzero(&client, sizeof (client));
socklen_t servaddrlen = sizeof ( struct sockaddr_in);
char helloanswer[18];
bzero(helloanswer, msgsize);
strcpy(helloanswer, "Hello Client" );
int sockst = 1;
while (sockst > 0) {
sockst = select (sockbrcast + 1, &readset, NULL, &readset, NULL);
if (sockst == -1) {
perror( "Broblems on broadcast socket" );
exit(EXIT_FAILURE);
}
int rdbyte = recvfrom(sockbrcast, ( void *) hellomesg, msgsize,MSG_TRUNC,
( struct sockaddr*) &client,
&servaddrlen);
if (rdbyte == msgsize && strcmp(hellomesg, "Hello Integral" ) == 0 &&
*isbusy == 0) {
if (sendto(sockbrcast, helloanswer, msgsize, 0,
( struct sockaddr*) &client, sizeof ( struct sockaddr_in)) < 0) {
perror( "Sending answer" );
close(sockbrcast);
exit(EXIT_FAILURE);
}
}
FD_ZERO(&readset);
FD_SET(sockbrcast, &readset);
}
return NULL;
}

* This source code was highlighted with Source Code Highlighter .

यहां सब कुछ सरल है, एक सॉकेट बनाएं और अनुरोधों की प्रतीक्षा करें। प्राप्त अनुरोध के रूप में - हम इसका जवाब देते हैं। और एक जटिलता, अनुरोध का उत्तर देना है या नहीं, इसबाय चर के मूल्य से तय किया जाता है।

अंत में मुख्य हो गया:

int main( int argc, char ** argv) {
// - -
if (argc > 2) {
fprintf(stderr, "Usage: %s [numofcpus]\n" , argv[0]);
exit(EXIT_FAILURE);
}

int numofthread;

if (argc == 2) {
numofthread = atoi(argv[1]);
if (numofthread < 1) {
fprintf(stderr, "Incorrect num of threads!\n" );
exit(EXIT_FAILURE);
}
fprintf(stdout, "Num of threads forced to %d\n" , numofthread);
} else {
// , -
numofthread = sysconf(_SC_NPROCESSORS_ONLN);
if (numofthread < 1) {
fprintf(stderr, "Can't detect num of processors\n"
"Continue in two threads\n" );
numofthread = 2;
}
fprintf(stdout, "Num of threads detected automatically it's %d\n\n" ,
numofthread);
}

* This source code was highlighted with Source Code Highlighter .

मुझे लगता है कि तर्कों के सत्यापन की व्याख्या नहीं की जा सकती ...

संकेतों के लिए एक मुखौटा सेट करें, और अनुरोधों को सुनने के लिए एक धागा चलाएं:

struct sigaction cancel_act;
memset(&cancel_act, 0, sizeof (cancel_act));
cancel_act.sa_handler = thread_cancel;
sigfillset(&cancel_act.sa_mask);
sigaction(SIGUSR1, &cancel_act, NULL);

// broadcast'
pthread_t broadcast_thread;
int isbusy = 1; //(int*) malloc(sizeof(int));
// broadcast
// 0 - , 1-
isbusy = 1;
if (pthread_create(&broadcast_thread, NULL, listen_broadcast, &isbusy)) {
fprintf(stderr, "Can't create broadcast listen thread" );
perror( "Detail:" );
exit(EXIT_FAILURE);
}

* This source code was highlighted with Source Code Highlighter .

अब हम एक सॉकेट बनाते हैं जिसके साथ ग्राहक एक कनेक्शन स्थापित करेंगे:

int listener;
struct sockaddr_in addr;
listener = socket(PF_INET, SOCK_STREAM, 0);
if (listener < 0) {
perror( "Can't create listen socket" );
exit(EXIT_FAILURE);
}

addr.sin_family = AF_INET;
addr.sin_port = htons(RCVPORT);
addr.sin_addr.s_addr = INADDR_ANY;
int a = 1;
// SO_REUSEADDR
if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &a, sizeof (a)) < 0) {
perror( "Set listener socket options" );
exit(EXIT_FAILURE);
}

//
if (bind(listener, ( struct sockaddr*) &addr, sizeof (addr)) < 0) {
perror( "Can't bind listen socket" );
exit(EXIT_FAILURE);
}

//
if (listen(listener, 1) < 0) {
perror( "Eror listen socket" );
exit(EXIT_FAILURE);
}

* This source code was highlighted with Source Code Highlighter .

हम कम्प्यूटेशनल चक्र शुरू करते हैं:

//
int needexit = 0;
while (needexit == 0) {
fprintf(stdout, "\nWait new connection...\n\n" );
int client;
isbusy = 0; //
struct sockaddr_in addrclient;
socklen_t addrclientsize = sizeof (addrclient);
client = accept(listener, ( struct sockaddr*)&addrclient,
&addrclientsize);
if (client < 0) {
perror( "Client accepting" );
}

* This source code was highlighted with Source Code Highlighter .

प्रसारण अनुरोध के साथ एक अलग थ्रेड सौदों के बाद से हम चुपचाप स्वीकार करते हैं।
क्लाइंट हमारे से कनेक्ट होने के बाद, हम उससे डेटा की जांच करते हैं, और गणना करना शुरू करते हैं:

isbusy = 1; //
task_data_t data;
int read_bytes = recv(client, &data, sizeof (data), 0);
if (read_bytes != sizeof (data) || data.limits < 1 || data.numoftry < 1) {
fprintf(stderr, "Invalid data from %s on port %d, reset peer\n" ,
inet_ntoa(addrclient.sin_addr), ntohs(addrclient.sin_port));
close(client);
isbusy = 0;
} else {
fprintf(stdout, "New task from %s on port %d\nlimits: %d\n"
"numoftrys: %d\n" , inet_ntoa(addrclient.sin_addr),
ntohs(addrclient.sin_port), data.limits, data.numoftry);
thread_args_t *tinfo;
pthread_t *calc_threads =
(pthread_t*) malloc( sizeof (pthread_t) * numofthread);
int threads_trys = data.numoftry % numofthread;
long double *results =
( long double *) malloc( sizeof ( long double ) * numofthread);
tinfo = (thread_args_t*) malloc( sizeof (thread_args_t) *
numofthread);
//
int numofthreadtry = data.numoftry / numofthread + 1;
for ( int i = 0; i < numofthread; ++i) {
tinfo[i].limits = data.limits;
tinfo[i].numoftry = numofthreadtry;
tinfo[i].results = &results[i];
if (pthread_create(&calc_threads[i], NULL, calculate, &tinfo[i])
!= 0) {
fprintf(stderr, "Can't create thread by num %d" , i);
perror( "Detail:" );
exit(EXIT_FAILURE);
}
}

//
checker_args_t checker_arg;
checker_arg.calcthreads = calc_threads;
checker_arg.threadnum = numofthread;
checker_arg.sock = client;
pthread_t checker_thread;
if (pthread_create(&checker_thread, NULL, client_check,
&checker_arg) != 0) {
fprintf(stderr, "Can't create checker thread" );
perror( "Detail:" );
exit(EXIT_FAILURE);
}
int iscanceled = 0; // ?
int *exitstat;
for ( int i = 0; i < numofthread; ++i) {
pthread_join(calc_threads[i], ( void *) &exitstat);
if (exitstat == PTHREAD_CANCELED)
iscanceled = 1; //
}
if (iscanceled != 1) {
long double *res = ( long double *) malloc( sizeof ( long double ));
bzero(res, sizeof ( long double ));
*res = 0.0;
for ( int i = 0; i < numofthread; ++i)
*res += results[i];
pthread_kill(checker_thread, SIGUSR1);
if (send(client, res, sizeof ( long double ), 0) < 0) {
perror( "Sending error" );
}
close(client);
free(res);
//free(checker_arg);
free(results);
free(calc_threads);
free(tinfo);
isbusy = 0;
fprintf(stdout, "Calculate and send finish!\n" );
} else {
fprintf(stderr, "Client die!\n" );
close(client);
//free(checker_arg);
free(results);
free(calc_threads);
free(tinfo);
}


}

}

return (EXIT_SUCCESS);
}

* This source code was highlighted with Source Code Highlighter .

बाकी कोड सरल है:

1. हम धागे की गणना शुरू करते हैं
2. थ्रेड चेकिंग स्थिति चलाएँ
3. हम तब तक इंतजार करते हैं जब तक कि सब कुछ निपट नहीं जाता।
4. हम क्लाइंट को जवाब देते हैं
5. कंप्यूटिंग चक्र नए सिरे से शुरू करना

मैं इस तथ्य पर आपका ध्यान आकर्षित करता हूं कि आपको चेकिंग थ्रेड को पूरा करने की आवश्यकता नहीं है, क्योंकि यह तब समाप्त होता है जब क्लाइंट के साथ सॉकेट बंद हो जाता है।

सभी सर्वर तैयार है।

ग्राहक

ग्राहक को निम्नानुसार व्यवस्थित किया जाएगा:

1. हम सर्वर की सूची बनाते हैं
2. प्रत्येक सर्वर के साथ काम करने के लिए, हम एक थ्रेड बनाते हैं
3. हम नतीजों और बाहर निकलने का इंतजार कर रहे हैं

प्रत्येक सर्वर के साथ काम करने के लिए, एक अलग धागा बनाना सुविधाजनक है, यह आपको उनके साथ अतुल्यकालिक रूप से काम करने की अनुमति देगा।

सर्वर की तरह, हम डेटा विनिमय के लिए एक संरचना की घोषणा करेंगे। साथ ही सर्वर के साथ काम करने वाले थ्रेड्स के लिए तर्क संरचना।

//
typedef struct {
int limits;
int numoftry;
} task_data_t;

//
typedef struct {
int limits; //
int numoftry; //
struct sockaddr_in *server; //
long double *results; //
} thread_args_t;

* This source code was highlighted with Source Code Highlighter .

क्लाइंट में, इसे सुनने के लिए पोर्ट सेट करने की आवश्यकता नहीं है, क्योंकि सर्वर जवाब देता है कि पैकेट कहाँ से आया है, इसलिए चलो वह ले जो ओएस हमें असाइन करेगा। तो, सर्वर के साथ काम करने वाला फ़ंक्शन (थ्रेड):

//
void *send_thread( void *arg) {
thread_args_t *task_data = (thread_args_t*) arg;
int servsock = socket(PF_INET, SOCK_STREAM, 0);
if (servsock < 0) {
perror( "Create new socket to server" );
exit(EXIT_FAILURE);
}
struct sockaddr_in listenaddr;
listenaddr.sin_family = AF_INET;
listenaddr.sin_addr.s_addr = INADDR_ANY;
listenaddr.sin_port = 0;

if (bind(servsock, ( struct sockaddr*) &listenaddr, sizeof (listenaddr)) < 0) {
perror( "Can't create listen socket" );
exit(EXIT_FAILURE);
}
socklen_t servaddrlen = sizeof ( struct sockaddr_in);
if (connect(servsock, ( struct sockaddr*)task_data->server,
servaddrlen) < 0) {
perror( "Connect to server failed!" );
exit(EXIT_FAILURE);
}
task_data_t senddata;
senddata.limits = task_data->limits;
senddata.numoftry = task_data->numoftry;

if (send(servsock, &senddata, sizeof (senddata), 0) < 0) {
perror( "Sending data to server failed" );
exit(EXIT_FAILURE);
}

int recv_byte = recv(servsock, task_data->results, sizeof ( long double ), 0);
if (recv_byte == 0) {
fprintf(stderr, "Server %s on port %d die!\nCancel calculate, on all" ,
inet_ntoa(task_data->server->sin_addr),
ntohs(task_data->server->sin_port));
exit(EXIT_FAILURE);
}
fprintf(stdout, "Server %s on port %d finish!\n" ,
inet_ntoa(task_data->server->sin_addr),
ntohs(task_data->server->sin_port));
return NULL;
}

* This source code was highlighted with Source Code Highlighter .

मुख्य, मुख्य सर्वर के समान है, साथ ही इस पर कुछ विस्तार से टिप्पणी की गई है, मैं इस पर ज्यादा चर्चा नहीं करूंगा।

int main( int argc, char ** argv) {
if (argc < 3) {
fprintf(stderr, "Usage: %s limits numoftry [maxserv]\n" , argv[0]);
exit(EXIT_FAILURE);
}

int numoftry = atoi(argv[2]);
if (numoftry == 0) {
fprintf(stderr, "Num of try is invalid\n" );
exit(EXIT_FAILURE);
}
int maxservu = 1000000;
if (argc == 4) {
maxservu = atoi(argv[3]);
if (maxservu < 1) {
fprintf(stderr, "Error number of max servers\n" );
exit(EXIT_FAILURE);
}
}
int limits = atoi(argv[1]);
if (limits == 0) {
fprintf(stderr, "Limits is invalid\n" );
exit(EXIT_FAILURE);
}

// broadcast
int sockbrcast = socket(PF_INET, SOCK_DGRAM, 0);
if (sockbrcast == -1) {
perror( "Create broadcast socket failed" );
exit(EXIT_FAILURE);
}


// broadcast
int port_rcv = 0;
struct sockaddr_in addrbrcast_rcv;
bzero(&addrbrcast_rcv, sizeof (addrbrcast_rcv));
addrbrcast_rcv.sin_family = AF_INET;
addrbrcast_rcv.sin_addr.s_addr = htonl(INADDR_ANY);
addrbrcast_rcv.sin_port = 0; //htons(port_rcv);
//
if (bind(sockbrcast, ( struct sockaddr *) &addrbrcast_rcv,
sizeof (addrbrcast_rcv)) < 0) {
perror( "Bind broadcast socket failed" );
close(sockbrcast);
exit(EXIT_FAILURE);
}

// broadcast
int port_snd = 38199;
struct sockaddr_in addrbrcast_snd;
bzero(&addrbrcast_snd, sizeof (addrbrcast_snd));
addrbrcast_snd.sin_family = AF_INET;
addrbrcast_snd.sin_port = htons(port_snd);
addrbrcast_snd.sin_addr.s_addr = htonl(0xffffffff);

// broadcast
int access = 1;
if (setsockopt(sockbrcast, SOL_SOCKET, SO_BROADCAST,
( const void *) &access, sizeof (access)) < 0) {
perror( "Can't accept broadcast option at socket to send" );
close(sockbrcast);
exit(EXIT_FAILURE);
}
int msgsize = sizeof ( char ) * 18;
void *hellomesg = malloc(msgsize);
bzero(hellomesg, msgsize);
strcpy(hellomesg, "Hello Integral" );
// broadcast
if (sendto(sockbrcast, hellomesg, msgsize, 0,
( struct sockaddr*) &addrbrcast_snd, sizeof (addrbrcast_snd)) < 0) {
perror( "Sending broadcast" );
close(sockbrcast);
exit(EXIT_FAILURE);
}

// broadcast'
fcntl(sockbrcast, F_SETFL, O_NONBLOCK);

//
fd_set readset;
FD_ZERO(&readset);
FD_SET(sockbrcast, &readset);

//
struct timeval timeout;
timeout.tv_sec = 3;
timeout.tv_usec = 0;

struct sockaddr_in *servers =
( struct sockaddr_in*) malloc( sizeof ( struct sockaddr_in));
bzero(servers, sizeof ( struct sockaddr_in));
int servcount = 0;
int maxserv = 1;
socklen_t servaddrlen = sizeof ( struct sockaddr_in);
// servers
while ( select (sockbrcast + 1, &readset, NULL, &readset, &timeout) > 0) {
int rdbyte = recvfrom(sockbrcast, ( void *) hellomesg, msgsize,MSG_TRUNC,
( struct sockaddr*) &servers[servcount],
&servaddrlen);
if (rdbyte == msgsize && strcmp(hellomesg, "Hello Client" ) == 0) {
servcount++;

if (servcount >= maxserv) {
servers = realloc(servers,
sizeof ( struct sockaddr_in) * (maxserv + 1));
if (servers == NULL) {
perror( "Realloc failed" );
close(sockbrcast);
exit(EXIT_FAILURE);
}
bzero(&servers[servcount], servaddrlen);
maxserv++;
}
FD_ZERO(&readset);
FD_SET(sockbrcast, &readset);
}
}
int i;
if (servcount < 1) {
fprintf(stderr, "No servers found!\n" );
exit(EXIT_FAILURE);
}
if (argc > 3 && maxservu <= servcount)
servcount = maxservu;
for (i = 0; i < servcount; ++i) {
printf( "Server answer from %s on port %d\n" ,
inet_ntoa(servers[i].sin_addr), ntohs(servers[i].sin_port));
}
printf( "\n" );
free(hellomesg);

long double *results =
( long double *) malloc( sizeof ( long double ) * servcount);
//
pthread_t *tid = (pthread_t*) malloc( sizeof (pthread_t) * servcount);
for (i = 0; i < servcount; ++i) {
thread_args_t *args = (thread_args_t*) malloc ( sizeof (thread_args_t));
args->limits = limits;
args->numoftry = numoftry / servcount + 1;
args->results = &results[i];
args->server = &servers[i];

if (pthread_create(&tid[i], NULL, send_thread, args) != 0) {
perror( "Create send thread failed" );
exit(EXIT_FAILURE);
}
}
long double res = 0;
//
for (i = 0; i < servcount; ++i)
pthread_join(tid[i], NULL);

//
for (i = 0; i < servcount; ++i)
res += results[i];
res /= numoftry;
res *= limits;


free(servers);
printf( "\nResult: %Lf\n" , res);
return (EXIT_SUCCESS);
}

* This source code was highlighted with Source Code Highlighter .

प्रदर्शन परीक्षण

थ्रेड्स की संख्या के आधार पर प्रदर्शन

मैंने एक ही कंप्यूटर पर, एक ही इनपुट डेटा के साथ, विभिन्न थ्रेड्स के साथ प्रोग्राम चलाया।
हम क्लाइंट को मापदंडों 2 1000000000 से शुरू करते हैं; 1, 2, 4, 8 के मापदंडों वाला सर्वर।
परिणाम, क्रमशः: 0m37.063s, 0m20.576s, 0m20.329s, 0m21.029s। यह ध्यान में रखना चाहिए कि जो परिणाम समय ने दिखाया, उससे आपको 4 सेकंड घटाना होगा जो हम सर्वर के लिए प्रतीक्षा करते हैं।
एक कोर (टीएम) 2 डुओ सीपीयू T5470 पर प्रोसेसर। जैसा कि अपेक्षित था, प्रोसेसर में दो कोर हैं, और तदनुसार 2 से अधिक थ्रेड लॉन्च करना व्यर्थ है, और इसलिए: थ्रेड्स की संख्या को दोगुना करना
दोहरा त्वरण।

हमने Asus 1215p पर भी परीक्षण किया, परिणाम में उपरोक्त धागे की संख्या पर समान निर्भरता है

कारों की संख्या के आधार पर प्रदर्शन

अब मैंने 1, 2, 4, 8, 16 कंप्यूटरों पर सर्वर शुरू किया, थ्रेड्स की संख्या कोर की संख्या के बराबर है, यानी दो थ्रेड्स।
यहाँ परिणाम हैं: 0m10.268s, 0m5.122s, 0m2.487s, 0m1.265, 0m0.766s।
फिर से हमें अपेक्षित परिणाम मिलता है, प्रत्येक नए कंप्यूटर के साथ, गणना 2 बार तेज हो जाएगी।

कुल मिलाकर

नतीजतन, हमें एक कार्यक्रम मिलता है जो एक साथ अभिन्न की गणना करता है और सभी प्रकार की परेशानियों के लिए प्रतिरोधी होता है जो ऑपरेशन के दौरान हो सकता है। दुर्भाग्य से, लेख के बड़े आकार के कारण, अधिक विस्तार से सब कुछ पर विचार करना संभव नहीं था। लेकिन मुझे उम्मीद है कि मेरा उदाहरण नए लोगों को मेरे साथ हुई कुछ त्रुटियों से बचने में मदद करेगा। मेक फाइल्स वाले कार्यक्रमों के स्रोत यहां देखे जा सकते हैं ।

आपका ध्यान के लिए धन्यवाद!