Annotation of /trunk/paramin-beta/netcomminterface.cc

Parent Directory | Revision Log

---

Revision 1 - (view) (download)

1 :	agomez	1	#include "netinterface.h"
2 :
3 :			// ********************************************************
4 :			// Functions for sending/receiving
5 :			// ********************************************************
6 :			void NetInterface::sendOne() {
7 :			int cansend, tid, dataid;
8 :			//tid = pManager->getNextTidToSend(net);
9 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
10 :			while (tid == waitForBetterProcesses())
11 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
12 :
13 :			if (tid < 0) {
14 :			cerr << "Error in netinterface - no processes available\n";
15 :			stopNetComm();
16 :			exit(EXIT_FAILURE);
17 :			}
18 :
19 :			assert(!dataSet->isEmpty());
20 :			dataid = dataSet->get();
21 :			cansend = sendOne(tid, dataid);
22 :			if (cansend != 1) {
23 :			// could not send data with id = dataid sendDataso update queue
24 :			dataSet->putFirst(dataid);
25 :			cerr << "Error in netinterface - failed to send data\n";
26 :			stopNetComm();
27 :			exit(EXIT_FAILURE);
28 :			}
29 :			}
30 :
31 :			int NetInterface::sendOne(int processID, int x_id) {
32 :			// Function returns ERROR if error occures while trying to send
33 :			// Function returns SUCCESS if successfully sent data
34 :			// Function return 0 if process with process identity = processID can not be used.
35 :			int i, cansend;
36 :			// Vector vec;
37 :			DoubleVector vecSend;
38 :
39 :			if (dctrl == NULL) {
40 :			cerr << "Error in netinterface - no valid datagroup\n";
41 :			exit(EXIT_FAILURE);
42 :			}
43 :			if (processID < 0 || (processID >= net->getNumProcesses())) {
44 :			cerr << "Error in netinterface - invalid process id\n";
45 :			stopNetComm();
46 :			return net->netError();
47 :			}
48 :			if (x_id < 0 || (x_id >= dctrl->getTotalSet())) {
49 :			cerr << "Error in netinterface - invalid vector id\n";
50 :			stopNetComm();
51 :			return net->netError();
52 :			}
53 :
54 :			// prepare data to be sent
55 :			assert(numVarToSend > 0);
56 :			NetDataVariables* sp = new NetDataVariables(numVarToSend);
57 :			// vec = dctrl->getX(x_id);
58 :			vecSend = prepareVectorToSend(dctrl->getX(x_id));
59 :			for (i = 0; i < numVarToSend; i++) {
60 :			sp->x[i] = vecSend[i];
61 :			}
62 :
63 :			sp->x_id = x_id;
64 :			sp->tag = dctrl->getTag();
65 :			cansend = net->sendData(sp, processID);
66 :			delete sp;
67 :			if (cansend == 1) {
68 :			//successfully sent sp using processID
69 :			dctrl->sentOne(x_id);
70 :			pManager->sent(processID);
71 :			return net->netSuccess();
72 :
73 :			} else if (cansend == -1) {
74 :			cerr << "Error in netinterface - failed to send data\n";
75 :			return net->netError();
76 :			} else if (cansend == 0) {
77 :			// process with id = processID is down
78 :			pManager->processDown(processID);
79 :			return cansend;
80 :			} else {
81 :			cerr << "Error in netinterface - unrecognised return value\n";
82 :			stopNetComm();
83 :			return net->netError();
84 :			}
85 :			}
86 :
87 :			int NetInterface::resend() {
88 :			int i, canresend;
89 :			int tid = -1;
90 :			int sendID;
91 :			// Vector vec;
92 :			DoubleVector vecSend;
93 :
94 :			if (dctrl == NULL) {
95 :			cerr << "Error in netinterface - no valid datagroup\n";
96 :			exit(EXIT_FAILURE);
97 :			}
98 :			if (!pManager->isFreeProc()) {
99 :			cerr << "Error in netinterface - no free processes\n";
100 :			stopNetComm();
101 :			return net->netError();
102 :			}
103 :			if (dctrl->allReceived()) {
104 :			cerr << "Error in netinterface - no data to resend\n";
105 :			stopNetComm();
106 :			return net->netError();
107 :			}
108 :
109 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
110 :			if (tid < 0)
111 :			tid = pManager->getTidToSend(net);
112 :
113 :			if (tid < 0) {
114 :			// there are no free processes
115 :			cerr << "Error in netinterface - no free processes\n";
116 :			stopNetComm();
117 :			return net->netError();
118 :			}
119 :
120 :			sendID = dctrl->getNextXToResend();
121 :			// vec = dctrl->getX(sendID);
122 :			vecSend = prepareVectorToSend(dctrl->getX(sendID));
123 :			assert(numVarToSend > 0);
124 :			NetDataVariables* sp = new NetDataVariables(numVarToSend);
125 :			for (i = 0; i < numVarToSend; i++)
126 :			sp->x[i] = vecSend[i];
127 :
128 :			sp->x_id = sendID;
129 :			sp->tag = dctrl->getTag();
130 :			canresend = net->sendData(sp, tid);
131 :			while (canresend == 0 && pManager->isFreeProc()) {
132 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
133 :			if (tid < 0)
134 :			tid = pManager->getTidToSend(net);
135 :			canresend = net->sendData(sp, tid);
136 :			}
137 :
138 :			delete sp;
139 :			if (canresend == 1) {
140 :			dctrl->resentOne(sendID);
141 :			pManager->sent(tid);
142 :			return net->netSuccess();
143 :
144 :			} else if (canresend == -1) {
145 :			cerr << "Error in netinterface - failed to resend data\n";
146 :			return net->netError();
147 :			} else if (canresend == 0 && !pManager->isFreeProc()) {
148 :			cerr << "Error in netinterface - failed to resend data\n";
149 :			stopNetComm();
150 :			return net->netError();
151 :			} else {
152 :			cerr << "Error in netinterface - unrecognised return value\n";
153 :			stopNetComm();
154 :			return net->netError();
155 :			}
156 :			}
157 :
158 :			int NetInterface::receiveOne() {
159 :			// function returns SUCCESS if successfully could receive data
160 :			// function return ERROR if netcommunication is down or error occurs
161 :			// functions returns 0 if cannot receive data but netcommunication is OK
162 :
163 :			int received;
164 :			double res;
165 :			if (dctrl == NULL) {
166 :			cerr << "Error in netinterface - no valid datagroup\n";
167 :			exit(EXIT_FAILURE);
168 :			}
169 :
170 :			NetDataResult* dp = new NetDataResult;
171 :			received = net->receiveData(dp);
172 :			if (received == 1 && (dp->who >= 0)) {
173 :			// received data successfully
174 :			pManager->setFreeProc(dp->who);
175 :			if (dctrl->getTag() == dp->tag) {
176 :			// received data belonging to datagroup
177 :			if (scaler != NULL) {
178 :			// Vector tempV;
179 :			// tempV = makeVector(dctrl->getX(dp->x_id));
180 :			// res = scaler->scaleResult(dp->result, dp->x_id, tempV);
181 :			res = scaler->scaleResult(dp->result, dp->x_id, makeVector(dctrl->getX(dp->x_id)));
182 :			} else {
183 :			res = dp->result;
184 :			}
185 :
186 :			if (dctrl->hasAnswer(dp->x_id)) {
187 :			receiveID = -1;
188 :			} else {
189 :			receiveID = dp->x_id;
190 :			dctrl->setY(dp->x_id, res);
191 :			}
192 :
193 :			} else {
194 :			// received data which does not belong to current datagroup
195 :			receiveID = -1;
196 :			}
197 :
198 :			delete dp;
199 :			return received;
200 :
201 :			} else if (received == 1 && dp->who == -1) {
202 :			// Was not able to receive but can continue netcommunication,
203 :			cerr << "Error in netinterface - could not receive data\n";
204 :			receiveID = -1;
205 :			delete dp;
206 :			// probably timeout occured so should check for health of processes.
207 :			net->getHealthOfProcesses(pManager->getStatus());
208 :			return !received;
209 :
210 :			} else if (received == -1) {
211 :			cerr << "Error in netinterface - failed to receive data\n";
212 :			receiveID = -1;
213 :			delete dp;
214 :			return net->netError();
215 :
216 :			} else {
217 :			cerr << "Error in netinterface - unrecognised return value\n";
218 :			receiveID = -1;
219 :			stopNetComm();
220 :			delete dp;
221 :			return net->netError();
222 :			}
223 :			}
224 :
225 :			int NetInterface::sendToAllIdleHosts() {
226 :			// functions returns ERROR if error occures in netcommunication
227 :			// function return SUCCESS if successfully sent all available data
228 :
229 :			int tid = 0;
230 :			int dataID;
231 :			int cansend = 1;
232 :
233 :			if (dctrl == NULL) {
234 :			cerr << "Error in netinterface - no valid datagroup\n";
235 :			exit(EXIT_FAILURE);
236 :			}
237 :
238 :			if (!dctrl->allSent()) {
239 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
240 :			};
241 :			while ((tid != this->noAvailableProcesses()) && !dctrl->allSent()
242 :			&& (cansend == 1 || cansend == 0)) {
243 :
244 :			// have not sent all data belonging to dctrl
245 :			if (tid == this->waitForBetterProcesses())
246 :			tid = pManager->getTidToSend(net);
247 :
248 :			if (tid != this->noAvailableProcesses()) {
249 :			assert(tid >= 0);
250 :			assert(!dataSet->isEmpty());
251 :			dataID = dataSet->get();
252 :			cansend = sendOne(tid, dataID);
253 :
254 :			if (cansend != 1)
255 :			dataSet->putFirst(dataID);
256 :
257 :			if (!dctrl->allSent() && (cansend == 1 || cansend == 0))
258 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
259 :			}
260 :			}
261 :
262 :			if (cansend > 1 || cansend < -1) {
263 :			cerr << "Error in netinterface - unrecognised return value\n";
264 :			stopNetComm();
265 :			return net->netError();
266 :			} else if (cansend == -1) {
267 :			return net->netError();
268 :			} else {
269 :			return net->netSuccess();
270 :			}
271 :			}
272 :
273 :			int NetInterface::sendToIdleHosts() {
274 :			// function returns ERROR if error occures while trying to send data,
275 :			// function returns SUCCESS if successfully sent available data
276 :			int tid = 0;
277 :			int dataID;
278 :			int cansend = 1;
279 :			if (dctrl == NULL) {
280 :			cerr << "Error in netinterface - no valid datagroup\n";
281 :			exit(EXIT_FAILURE);
282 :			}
283 :			if (!dctrl->allSent()) {
284 :			// dctrl->getNumLeftToSend() er í lagi, vesen í getNextTidToSend
285 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
286 :			//tid = 0;
287 :			}
288 :			while ((tid != this->noAvailableProcesses()) && (tid != this->waitForBetterProcesses())
289 :			&& (!dctrl->allSent()) && (cansend == 1 || cansend == 0)) {
290 :			// Have not sent all data and there is a suitable process available
291 :			assert(tid >= 0);
292 :			assert(!dataSet->isEmpty());
293 :			dataID = dataSet->get();
294 :			cansend = sendOne(tid, dataID);
295 :			if (cansend != 1)
296 :			dataSet->putFirst(dataID);
297 :			if ((cansend == 1 || cansend == 0) && !dctrl->allSent()) {
298 :			tid = pManager->getNextTidToSend(dctrl->getNumLeftToSend(), net);
299 :			};
300 :			}
301 :
302 :			if (cansend > 1 || cansend < -1) {
303 :			cerr << "Error in netinterface - unrecognised return value\n";
304 :			stopNetComm();
305 :			return net->netError();
306 :			} else if (cansend == -1) {
307 :			return net->netError();
308 :			} else {
309 :			return net->netSuccess();
310 :			}
311 :			}
312 :
313 :			int NetInterface::sendAll() {
314 :			assert(dctrl != NULL);
315 :			int OK = sendToIdleHosts();
316 :			while (!dctrl->allSent() && OK == 1) {
317 :			// try to receive data from a host as no more suitable processes
318 :			// available and still have data to send
319 :			OK = receiveAndSend();
320 :			}
321 :			return OK;
322 :			}
323 :
324 :			int NetInterface::receiveAndSend() {
325 :			// This function tries to keep all processes busy by first trying to
326 :			// receive and then sending available data to all suitable processes
327 :			// It is assumed that there is an outstanding message to be received.
328 :			int counter = 0;
329 :			int cansend = 1;
330 :			int canreceive = 0;
331 :
332 :			while (counter < numTries && canreceive == 0) {
333 :			// Try to receive again from a process, netcommunication is OK but
334 :			// could not receive probably because there are no messages coming in
335 :			canreceive = receiveOne();
336 :			counter++;
337 :			}
338 :
339 :			if (canreceive == 1) {
340 :			// Received successfully, at least one process is free
341 :			cansend = sendToIdleHosts();
342 :			return cansend;
343 :
344 :			} else if (canreceive == -1) {
345 :			// Error occured while trying to receive, netcommunication is down
346 :			return net->netError();
347 :
348 :			} else if (canreceive == 0) {
349 :			// Did not receive any data
350 :			if (!pManager->isFreeProc()) {
351 :			// no free process after trying to receive
352 :			cerr << "Error in netinterface - not receiving from busy process\n";
353 :			return net->netError();
354 :			} else {
355 :			// There is a free processs that can be used, will try to send
356 :			cansend = sendToAllIdleHosts();
357 :			return cansend;
358 :			}
359 :
360 :			} else {
361 :			cerr << "Error in netinterface - unrecognised return value\n";
362 :			stopNetComm();
363 :			return net->netError();
364 :			}
365 :			}
366 :
367 :			int NetInterface::receiveOnCondition(Condition* con) {
368 :			int counter = 0;
369 :			int cond = 0;
370 :			int receiveID;
371 :			int canreceive = 1;
372 :
373 :			while (!dctrl->allReceived() && !cond&& canreceive == 1) {
374 :			canreceive = receiveOne();
375 :			while (counter < (numTries - 1) && canreceive == 0) {
376 :			canreceive = receiveOne();
377 :			counter++;
378 :			}
379 :
380 :			if (canreceive == 1) {
381 :			receiveID = getReceiveID();
382 :			if (receiveID >= 0)
383 :			cond = con->computeCondition();
384 :			}
385 :			counter = 0;
386 :			}
387 :
388 :			if (canreceive == -1) {
389 :			return net->netError();
390 :
391 :			} else if (canreceive == 0) {
392 :			cerr << "Error in netinterface - no received data\n";
393 :			stopNetComm();
394 :			return net->netError();
395 :
396 :			} else if (cond == 1) {
397 :			// condition has been met
398 :			return cond;
399 :
400 :			} else if (dctrl->allReceived() && cond == 0) {
401 :			// condition has not been met and nothing to receive
402 :			return cond;
403 :
404 :			} else {
405 :			cerr << "Error in netinterface - unrecognised return value\n";
406 :			stopNetComm();
407 :			return net->netError();
408 :			}
409 :			}
410 :
411 :			int NetInterface::receiveAll() {
412 :			int counter = 0;
413 :			int canreceive = 1;
414 :			int cansend = 1;
415 :
416 :			while (!(dctrl->allReceived()) && (canreceive == 0 || canreceive == 1) && cansend == 1) {
417 :			canreceive = receiveOne();
418 :			while (counter < (numTries - 1) && canreceive == 0) {
419 :			canreceive = receiveOne();
420 :			counter++;
421 :			}
422 :
423 :			if (canreceive == 0) {
424 :			// Did not receive anything - try to resend data
425 :			if (!pManager->isFreeProc()) {
426 :			cerr << "Error in netinterface - no received data\n";
427 :			canreceive = -1;
428 :			} else if (!dctrl->allReceived() && pManager->isFreeProc()) {
429 :			// have not received all so will resend data
430 :			cansend = resend();
431 :			}
432 :			}
433 :			counter = 0;
434 :			}
435 :
436 :			if (canreceive == -1 || cansend == -1) {
437 :			return net->netError();
438 :			} else if (canreceive > 1 || canreceive < -1) {
439 :			stopNetComm();
440 :			return net->netError();
441 :			} else if (cansend > 1 || cansend < -1) {
442 :			stopNetComm();
443 :			return net->netError();
444 :			} else if (dctrl->allReceived()) {
445 :			return net->netSuccess();
446 :			} else {
447 :			cerr << "Error in netinterface - unrecognised return value\n";
448 :			stopNetComm();
449 :			return net->netError();
450 :			}
451 :			}
452 :
453 :			int NetInterface::sendAndReceiveAllData() {
454 :			int cansend_receive;
455 :			cansend_receive = sendAll();
456 :			if (cansend_receive == 1)
457 :			cansend_receive = receiveAll();
458 :
459 :			if (cansend_receive == 1)
460 :			return net->netSuccess();
461 :			else
462 :			return net->netError();
463 :			}
464 :
465 :			int NetInterface::sendAllOnCondition(Condition* con) {
466 :			int cond = 0;
467 :			int receiveID;
468 :			int OK = 1;
469 :
470 :			// start by sending data to all free suitable hosts.
471 :			OK = sendToIdleHosts();
472 :			// receive and send until received all or condition is true or error
473 :			while (!dctrl->allSent() && cond == 0&& OK == 1) {
474 :			// try to receive data from a host
475 :			OK = receiveAndSend();
476 :			// if received from dctrl then check if condition is true
477 :			if (OK == 1) {
478 :			receiveID = getReceiveID();
479 :			if (receiveID >= 0)
480 :			cond = con->computeCondition();
481 :			}
482 :			}
483 :
484 :			if (OK != 1)
485 :			return net->netError();
486 :			else
487 :			return cond;
488 :			}
489 :
490 :			int NetInterface::sendAndReceiveSetData(Condition* con) {
491 :			int cond = 0;
492 :			int sendreceive = 1;
493 :			int numTries = 0;
494 :			int counter, newreturns, numLeftToReceive, totalNumHosts;
495 :
496 :			while ((sendreceive != -1) && cond == 0) {
497 :			counter = getNumDataItemsSet();
498 :			newreturns = getNumDataItemsAnswered();
499 :			numLeftToReceive = counter - newreturns;
500 :			totalNumHosts = getTotalNumProc();
501 :
502 :			sendreceive = sendToAllIdleHosts();
503 :			if (sendreceive == 1) {
504 :			numTries = 0;
505 :			if ((numLeftToReceive < totalNumHosts) && dctrl->allSent()) {
506 :			// can start resending, waiting for last data to come in
507 :			sendreceive = receiveOne();
508 :			while (numTries < (numTries - 1) && sendreceive == 0) {
509 :			sendreceive = receiveOne();
510 :			numTries++;
511 :			}
512 :
513 :			if (sendreceive == 0) {
514 :			if (!pManager->isFreeProc()) {
515 :			cerr << "Error in netinterface - received no data\n";
516 :			sendreceive = -1;
517 :			stopNetComm();
518 :			} else if (!dctrl->allReceived() && pManager->isFreeProc()) {
519 :			sendreceive = resend();
520 :			}
521 :			}
522 :			numTries = 0;
523 :
524 :			} else {
525 :			sendreceive = receiveOne();
526 :			while (numTries < (numTries - 1) && sendreceive == 0) {
527 :			sendreceive = receiveOne();
528 :			numTries++;
529 :			}
530 :
531 :			if (sendreceive == 0) {
532 :			if (!pManager->isFreeProc()) {
533 :			cerr << "Error in netinterface - received no data\n";
534 :			sendreceive = -1;
535 :			stopNetComm();
536 :			}
537 :			}
538 :			}
539 :
540 :			if (sendreceive == 1)
541 :			cond = con->computeCondition();
542 :			}
543 :			}
544 :
545 :			if (sendreceive == -1)
546 :			return net->netError();
547 :			else if (cond == 1)
548 :			return cond;
549 :			else
550 :			return net->netError();
551 :			}
552 :
553 :			int NetInterface::sendAndReceiveTillCondition(Condition* con) {
554 :			int cond;
555 :			cond = sendAllOnCondition(con);
556 :			if (cond == 0)
557 :			cond = receiveOnCondition(con);
558 :			return cond;
559 :			}
560 :
561 :			// ********************************************************
562 :			// Functions concerning netcommunication
563 :			// ********************************************************
564 :			int NetInterface::startNetComm() {
565 :			return net->startNetCommunication();
566 :			}
567 :
568 :			void NetInterface::sendStringValue() {
569 :			assert(numVarToSend > 0);
570 :			assert(switches.Size() == numVarToSend);
571 :			int SEND = net->sendData(switches);
572 :			if (!(SEND == netSuccess())) {
573 :			cerr << "Error in netinterface - failed to send switches\n";
574 :			exit(EXIT_FAILURE);
575 :			}
576 :			}
577 :
578 :			void NetInterface::sendStringValue(int processID) {
579 :			assert(numVarToSend > 0);
580 :			assert(switches.Size() == numVarToSend);
581 :			int SEND = net->sendData(switches,processID);
582 :			if (!(SEND == netSuccess())) {
583 :			cerr << "Error in netinterface - failed to send switches\n";
584 :			exit(EXIT_FAILURE);
585 :			}
586 :			}
587 :
588 :			void NetInterface::sendBoundValues() {
589 :			assert(numVarToSend > 0);
590 :			assert(lowerBound.Size() == numVarToSend);
591 :			int SEND = net->sendBoundData(lowerBound);
592 :			if (!(SEND == netSuccess())) {
593 :			cerr << "Error in netinterface - failed to send lowerbounds\n";
594 :			exit(EXIT_FAILURE);
595 :			}
596 :
597 :			assert(upperBound.Size() == numVarToSend);
598 :			SEND = net->sendBoundData(upperBound);
599 :			if (!(SEND == netSuccess())) {
600 :			cerr << "Error in netinterface - failed to send upperbounds\n";
601 :			exit(EXIT_FAILURE);
602 :			}
603 :			}
604 :
605 :			void NetInterface::sendBoundValues(int processID) {
606 :			assert(numVarToSend > 0);
607 :			assert(lowerBound.Size() == numVarToSend);
608 :			int SEND = net->sendBoundData(lowerBound, processID);
609 :			if (!(SEND == netSuccess())) {
610 :			cerr << "Error in netinterface - failed to send lowerbounds\n";
611 :			exit(EXIT_FAILURE);
612 :			}
613 :			assert(upperBound.Size() == numVarToSend);
614 :			SEND = net->sendBoundData(upperBound, processID);
615 :			if (!(SEND == netSuccess())) {
616 :			cerr << "Error in netinterface - failed to send upperbounds\n";
617 :			exit(EXIT_FAILURE);
618 :			}
619 :			}
620 :
621 :			void NetInterface::stopNetComm() {
622 :			net->stopNetCommunication();
623 :			pManager->noProcessesRunning();
624 :			}
625 :
626 :			int NetInterface::getNumFreeProcesses() {
627 :			return pManager->getNumFreeProc();
628 :			}
629 :
630 :			int NetInterface::getTotalNumProc() {
631 :			return net->getNumProcesses();
632 :			}
633 :
634 :			int NetInterface::getNumTags() {
635 :			return numberOfTags;
636 :			}
637 :
638 :			int NetInterface::getNextMsgTag() {
639 :			int tempTag = numberOfTags;
640 :			numberOfTags++;
641 :			return tempTag;
642 :			}
643 :
644 :			int NetInterface::isUpAndRunning() {
645 :			return net->netCommStarted();
646 :			}
647 :
648 :			int NetInterface::netError() {
649 :			return net->netError();
650 :			}
651 :
652 :			int NetInterface::netSuccess() {
653 :			return net->netSuccess();
654 :			}
655 :
656 :			int NetInterface::noAvailableProcesses() {
657 :			return pManager->noAvailableProcesses();
658 :			}
659 :
660 :			int NetInterface::waitForBetterProcesses() {
661 :			return pManager->waitForBetterProcesses();
662 :			}

---

root@forge.cesga.es	ViewVC Help
Powered by ViewVC 1.0.0