« back to all changes in this revision
Viewing changes to src/AIX5SP2/pvmdmimd.c
- browse files at revision 2.1.4
- compare with another revision
- download diff
- download tarball
- view history from revision 2.1.4
- Committer: Bazaar Package Importer
- Author(s): Steinar H. Gunderson
- Date: 2005-10-06 15:25:03 UTC
- mfrom: (1.1.1 upstream)
- mto: This revision was merged to the branch mainline in revision 4.
- Revision ID: james.westby@ubuntu.com-20051006152503-nty60w30a9dja62m
Tags: 3.4.5-5
Use --remove-all on pvm-rsh alternative in prerm instead of --remove.
(Closes: #328366)
(Closes: #328366)
- files added:
- Readme.Beolin
- src/AIX5MP
- src/AIX5SP2
- src/BEOLIN
- files removed:
- Readme.Win32.txt
- WIN32/DevStudio5
- WIN32/DevStudio6
- files modified:
- Makefile
added
removed
src/AIX5SP2/pvmdmimd.c
1
2
static char rcsid[] =
3
"$Id: pvmdmimd.c,v 1.1 2004/01/14 17:34:27 pvmsrc Exp $";
4
5
/*
6
* PVM version 3.4: Parallel Virtual Machine System
7
* University of Tennessee, Knoxville TN.
8
* Oak Ridge National Laboratory, Oak Ridge TN.
9
* Emory University, Atlanta GA.
10
* Authors: J. J. Dongarra, G. E. Fagg, M. Fischer
11
* G. A. Geist, J. A. Kohl, R. J. Manchek, P. Mucci,
12
* P. M. Papadopoulos, S. L. Scott, and V. S. Sunderam
13
* (C) 1997 All Rights Reserved
14
*
15
* NOTICE
16
*
17
* Permission to use, copy, modify, and distribute this software and
18
* its documentation for any purpose and without fee is hereby granted
19
* provided that the above copyright notice appear in all copies and
20
* that both the copyright notice and this permission notice appear in
21
* supporting documentation.
22
*
23
* Neither the Institutions (Emory University, Oak Ridge National
24
* Laboratory, and University of Tennessee) nor the Authors make any
25
* representations about the suitability of this software for any
26
* purpose. This software is provided ``as is'' without express or
27
* implied warranty.
28
*
29
* PVM version 3 was funded in part by the U.S. Department of Energy,
30
* the National Science Foundation and the State of Tennessee.
31
*/
32
33
/*
34
* pvmdmimd.c
35
*
36
* MPP interface.
37
*
38
* void mpp_init(int argc, char **argv):
39
* Initialization. Create a table to keep track of active nodes.
40
* argc, argv: passed from main.
41
*
42
* int mpp_load( struct waitc_spawn *wxp )
43
*
44
* Load executable onto nodes; create new entries in task table,
45
* encode node number and process type into task IDs, etc.
46
*
47
* Construction of Task ID:
48
*
49
* 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0
50
* +-+-+-----------------------+-+-----+--------------------------+
51
* |s|g| host index |n| prt | node # (16384) |
52
* +-+-+-----------------------+-+-----+--------------------------+
53
*
54
* The "n" bit is set for node task but clear for host task.
55
*
56
* flags: exec options;
57
* name: executable to be loaded;
58
* argv: command line argument for executable
59
* count: number of tasks to be created;
60
* tids: array to store new task IDs;
61
* ptid: parent task ID.
62
*
63
* mpp_new(int count, int ptid):
64
* Allocate a set of nodes. (called by mpp_load())
65
* count: number of nodes; ptid: parent task ID.
66
*
67
* int mpp_output():
68
* Send all pending packets to nodes via native send. Node number
69
* and process type are extracted from task ID.
70
*
71
* int mpp_mcast(int src, struct pkt pp, int tids[], int ntask):
72
* Global send.
73
* src: source task ID;
74
* pp: packet;
75
* tids: list of destination task IDs;
76
* ntask: how many.
77
*
78
* int mpp_probe():
79
* Probe for pending packets from nodes (non-blocking). Returns
80
* 1 if packets are dectected, otherwise 0.
81
*
82
* void mpp_input():
83
* Receive pending packets (from nodes) via native recv.
84
*
85
* struct task *mpp_find(int pid):
86
* Find a task in task table by its Unix pid.
87
*
88
* void mpp_free(struct task *tp):
89
* Remove node/process-type from active list.
90
* tp: task pointer.
91
*
92
$Log: pvmdmimd.c,v $
93
Revision 1.1 2004/01/14 17:34:27 pvmsrc
94
New AIX5SP2 arch.
95
(Spanker=kohl)
96
97
Revision 1.4 2002/02/21 23:19:01 pvmsrc
98
Added new (not to be documented!) PVM_MAX_TASKS env var support.
99
- for Mahir Lokvancic <mahir@math.ufl.edu>.
100
- forcefully limits the number of tasks that can attach to a
101
pvmd, required on Solaris in rare circumstances when hard
102
FD_SETSIZE limit is reached, and all hell breaks loose...
103
- check return for task_new() call, can now produce NULL ptr,
104
indicating PvmOutOfRes...
105
(Spanker=kohl)
106
107
Revision 1.3 2001/09/25 21:19:00 pvmsrc
108
Minor TMPNAMFUN()/tmpnam() cleanup.
109
- moved macro def to pvm3.h, renamed PVMTNPMAN().
110
- same for LEN_OF_TMP_NAM -> PVMTMPNAMLEN.
111
- mostly a huge waste of time, since *both* tmpnam() & mktemp()
112
produce the same "dangerous" warning message in Linux/gcc...
113
- damn.
114
(Spanker=kohl)
115
116
Revision 1.2 2000/02/17 21:10:16 pvmsrc
117
Cleaned up comments... mpp_load() args...
118
(Spanker=kohl)
119
120
* Revision 1.1 1998/02/23 21:31:44 pvmsrc
121
* Copied from SP2MPI.
122
*
123
* Revision 1.10 1997/10/01 15:36:06 pvmsrc
124
* Removed unnecessary #include "fromlib.h" header.
125
* - all consts now included in pvm3.h...
126
* - header file eliminated.
127
* (Spanker=kohl)
128
*
129
* Revision 1.9 1997/08/27 20:18:58 pvmsrc
130
* Added blank args (0,0,0) in mpp_load() to make it call forkexec correctly.
131
* Protocol needs to be changed to allow taskers to access IBM poe directly.
132
* (Spanker=fagg)
133
*
134
* Revision 1.8 1997/07/09 13:54:58 pvmsrc
135
* Fixed Author Header.
136
*
137
* Revision 1.7 1997/06/02 13:48:38 pvmsrc
138
* Moved #include host.h above #include waitc.h.
139
* Removed old mesg.h include... gone baby.
140
*
141
* Revision 1.6 1997/05/06 20:14:36 pvmsrc
142
* Catch stdout/stderr correctly and redirect to outtid,tag,ctx
143
*
144
* Revision 1.5 1997/05/05 20:08:56 pvmsrc
145
* Pass outctx and trcctx to mpp tasks.
146
*
147
* Revision 1.4 1997/05/02 13:52:50 pvmsrc
148
* Start up MPI jobs correctly and get them configured.
149
*
150
* Revision 1.3 1997/03/25 15:52:21 pvmsrc
151
* PVM patches from the base 3.3.10 to 3.3.11 versions where applicable.
152
* Originals by Bob Manchek. Altered by Graham Fagg where required.
153
* -IP enabled over switch
154
* -RMPOOL env can be used instead of host list
155
*
156
* Revision 1.2 1997/01/28 19:30:57 pvmsrc
157
* New Copyright Notice & Authors.
158
*
159
* Revision 1.1 1996/09/23 23:15:09 pvmsrc
160
* Initial revision
161
*
162
* Revision 1.3 1996/05/14 14:35:59 manchek
163
* inc'd changes from chulho@kgn.ibm.com
164
*
165
* Revision 1.2 1995/07/25 17:41:27 manchek
166
* mpp_output returns int
167
*
168
* Revision 1.1 1995/05/30 17:23:56 manchek
169
* Initial revision
170
*
171
* Revision 1.3 1994/06/03 20:54:24 manchek
172
* version 3.3.0
173
*
174
* Revision 1.2 1993/12/20 15:39:47 manchek
175
* patch 6 from wcj
176
*
177
* Revision 1.1 1993/08/30 23:35:09 manchek
178
* Initial revision
179
*
180
*/
181
182
#include <stdlib.h>
183
#include <sys/param.h>
184
#include <sys/types.h>
185
#include <sys/time.h>
186
#include <sys/socket.h>
187
#include <netinet/in.h>
188
#include <netinet/tcp.h>
189
#include <sys/stat.h>
190
#include <unistd.h>
191
#include <fcntl.h>
192
#include <errno.h>
193
#include <stdio.h>
194
#include <signal.h>
195
#ifdef SYSVSTR
196
#include <string.h>
197
#define CINDEX(s,c) strchr(s,c)
198
#else
199
#include <strings.h>
200
#define CINDEX(s,c) index(s,c)
201
#endif
202
203
#include <pvm3.h>
204
#include <pvmproto.h>
205
#include "global.h"
206
#include "host.h"
207
#include "waitc.h"
208
#include "pvmalloc.h"
209
#include "pkt.h"
210
#include "task.h"
211
#include "listmac.h"
212
#include "pvmdmp.h"
213
#include "pvmmimd.h"
214
#include "bfunc.h"
215
216
#define MPICOMM "/usr/bin/poe"
217
#define MPIOPT1 "-procs"
218
#define MPIOPT2 "-euilib"
219
#define MPIOPT3 "-hfile"
220
#define MPIOPT4 "-rmpool"
221
#define MPIOPARG2 "us" /* options: -procs # -euilib us */
222
#define MPIOPARG3 "ip" /* Enable IP over switch */
223
#define MPIARGC 7 /* number of command line arguments */
224
225
char *getenv();
226
227
/* Global */
228
229
extern int pvmdebmask; /* from pvmd.c */
230
extern char **epaths; /* from pvmd.c */
231
extern int myhostpart; /* from pvmd.c */
232
extern int tidhmask; /* from pvmd.c */
233
extern int ourudpmtu; /* from pvmd.c */
234
extern struct htab *hosts; /* from pvmd.c */
235
extern struct task *locltasks; /* from task.c */
236
237
int tidtmask = TIDPTYPE; /* mask for ptype field of tids */
238
int tidnmask = TIDNODE; /* mask for node field of tids */
239
240
/* private */
241
242
static int myndf = 0;
243
static struct nodeset *busynodes; /* active nodes; ordered by proc type */
244
static char pvmtxt[512]; /* scratch for error log */
245
static int ptypemask; /* mask; we use these bits of ptype in tids */
246
static char nodefile[PVMTMPNAMLEN]; /* tmp node file */
247
static char **nodelist = 0; /* default poe node list */
248
static int partsize = 0; /* number of nodes allocated */
249
static int hostfileused = TRUE; /* Check if MP_HOSTFILE used */
250
static char defaultpool[64]="1"; /* default MP_POOL if not set */
251
static char mpiadapter[]={MPIOPARG2}; /* default User-Space */
252
253
static int sp2pvminfo[SIZEHINFO];
254
255
256
257
void
258
mpp_init(argc, argv)
259
int *argc;
260
char **argv;
261
{
262
struct stat sb;
263
char *hfn; /* host file name */
264
char nname[128]; /* node name */
265
FILE *hfp;
266
int i;
267
268
if ((hfn = getenv("LOADLBATCH"))) {
269
if (strcmp(hfn, "yes") == 0) {
270
if ((hfn = getenv("LOADL_PROCESSOR_LIST"))) {
271
sprintf(pvmtxt,
272
"LOADL_PROCESSOR_LIST=%s.\n",hfn);
273
pvmlogperror(pvmtxt);
274
hostfileused=FALSE;
275
for (hfn;*hfn!='\0';hfn++) {
276
if (*hfn == ' ')
277
partsize++;
278
}
279
} else {
280
sprintf(pvmtxt,
281
"LOADL_PROCESSOR_LIST=%s - is not set\n",hfn);
282
pvmlogperror(pvmtxt);
283
exit(PvmHostFail);
284
}
285
} else {
286
sprintf(pvmtxt,"LOADLBATCH=%s - not set to yes\n",hfn);
287
pvmlogperror(pvmtxt);
288
exit(PvmHostFail);
289
}
290
} else if ((hfn = getenv("MP_PROCS"))) {
291
if ((partsize = atoi(hfn)) < 1) {
292
sprintf(pvmtxt,"MP_PROCS=%d must be >= to 1\n",partsize);
293
pvmlogperror(pvmtxt);
294
exit(PvmHostFail);
295
}
296
hostfileused=FALSE;
297
if ((hfn = getenv("MP_RMPOOL"))) {
298
if (atoi(hfn) < 0) {
299
sprintf(pvmtxt,
300
"MP_RMPOOL=%d must be >= to 0\n",defaultpool);
301
pvmlogperror(pvmtxt);
302
exit(PvmHostFail);
303
}
304
strcpy(defaultpool,hfn);
305
}
306
} else if ((hfn = getenv("MP_HOSTFILE")))
307
{
308
if (!(hfp = fopen(hfn, "r"))) {
309
sprintf(pvmtxt, "sp2hostfile() fopen %s\n", hfn);
310
pvmlogperror(pvmtxt);
311
exit(PvmHostFail);
312
}
313
hostfileused=TRUE;
314
while (fscanf(hfp, "%s", nname) != EOF)
315
partsize++;
316
317
nodelist = TALLOC(partsize, char*, "nname");
318
rewind(hfp);
319
for (i = 0; i < partsize; i++) {
320
fscanf(hfp, "%s", nname);
321
nodelist[i] = STRALLOC(nname);
322
}
323
fclose(hfp);
324
325
} else {
326
pvmlogerror("mpp_init() no POE host file.\n");
327
pvmlogerror("mpp_init() MP_PROCS, MP_RMPOOL or MP_HOSTFILE must be set.\n");
328
exit(PvmHostFail);
329
}
330
if ((hfn = getenv("MP_EUILIB"))) {
331
if (strcmp(hfn, "ip") == 0) {
332
strcpy(mpiadapter,MPIOPARG3); /* IP over switch */
333
}
334
}
335
sprintf(pvmtxt, "%d nodes allocated.\n", partsize);
336
pvmlogerror(pvmtxt);
337
338
busynodes = TALLOC(1, struct nodeset, "nsets");
339
BZERO((char*)busynodes, sizeof(struct nodeset));
340
busynodes->n_link = busynodes;
341
busynodes->n_rlink = busynodes;
342
343
ptypemask = tidtmask >> (ffs(tidtmask) - 1);
344
}
345
346
347
/* create tmp poe host file from default */
348
char *
349
sp2hostfile(first, count)
350
int first; /* first node in the set */
351
int count; /* number of nodes requested */
352
{
353
int i;
354
FILE *tmpfp;
355
356
if (partsize < count) {
357
sprintf(pvmtxt, "sp2hostfile() need at least %d nodes\n", count+1);
358
pvmlogperror(pvmtxt);
359
return (char *)0;
360
}
361
(void)PVMTMPNAMFUN(nodefile);
362
if (!(tmpfp = fopen(nodefile, "w"))) {
363
sprintf(pvmtxt, "sp2hostfile() fopen %s", nodefile);
364
pvmlogperror(pvmtxt);
365
return (char *)0;
366
}
367
if (pvmdebmask & PDMNODE) {
368
sprintf(pvmtxt, "sp2hostfile() POE host file: %s\n", nodefile);
369
pvmlogerror(pvmtxt);
370
}
371
for (i = first; i < count + first; i++)
372
fprintf(tmpfp, "%s\n", nodelist[i]);
373
fclose(tmpfp);
374
return nodefile;
375
}
376
377
378
/*
379
* find a set of free nodes from nodelist; assign ptype sequentially,
380
* only tasks spawned together get the same ptype
381
*/
382
struct nodeset *
383
mpp_new(count, ptid)
384
int count; /* number of nodes requested */
385
int ptid; /* parent's tid */
386
{
387
struct nodeset *sp, *newp, *sp2;
388
int last = -1;
389
int ptype = 0;
390
391
if (!(newp = TALLOC(1, struct nodeset, "nsets"))) {
392
pvmlogerror("mpp_new() can't get memory\n");
393
pvmbailout(0);
394
}
395
BZERO((char*)newp, sizeof(struct nodeset));
396
397
newp->n_size = count;
398
for (sp = busynodes->n_link; sp != busynodes; sp = sp->n_link) {
399
if (sp->n_first - last > count)
400
break;
401
last = sp->n_first + sp->n_size - 1;
402
/*
403
if (sp->n_link == busynodes && partsize - last > count)
404
break;
405
*/
406
if (ptype <= sp->n_ptype)
407
ptype = sp->n_ptype + 1;
408
}
409
if (sp == busynodes && partsize - last <= count) {
410
pvmlogerror("mpp_new() not enough nodes in partition\n");
411
PVM_FREE(newp);
412
return (struct nodeset *)0;
413
}
414
for (sp2 = busynodes->n_link; sp2 != busynodes; sp2 = sp2->n_link)
415
if ((sp2->n_ptype & ptypemask) == (ptype & ptypemask))
416
break;
417
if (sp2 != busynodes || ptype == NPARTITIONS) {
418
for (ptype = 0; ptype < NPARTITIONS; ptype++) {
419
for (sp2 = busynodes->n_link; sp2 != busynodes; sp2 = sp2->n_link)
420
if ((sp2->n_ptype & ptypemask) == (ptype & ptypemask))
421
break;
422
if (sp2 == busynodes)
423
break;
424
}
425
if (ptype == NPARTITIONS) {
426
pvmlogerror("mpp_new() out of ptypes: too many spawns\n");
427
return (struct nodeset *)0;
428
}
429
}
430
431
done:
432
if (pvmdebmask & PDMNODE) {
433
sprintf(pvmtxt, "mpp_new() %d nodes %d ... ptype=%d ptid=%x\n",
434
count, last+1, ptype, ptid);
435
pvmlogerror(pvmtxt);
436
}
437
newp->n_first = last + 1;
438
if (hostfileused) {
439
if (!sp2hostfile(newp->n_first, count)) {
440
PVM_FREE(newp);
441
return (struct nodeset *)0;
442
}
443
}
444
newp->n_ptype = ptype;
445
newp->n_ptid = ptid;
446
newp->n_alive = count - 1;
447
LISTPUTBEFORE(sp, newp, n_link, n_rlink);
448
449
return newp;
450
}
451
452
453
/*
454
* remove node/ptype from active list; if tid is the last to go, shutdown
455
* pvmhost's socket, but do not destroy the node set because pvmhost may
456
* not exit immediately. To avoid a race condition, let mpp_output()
457
* do the cleanup.
458
*/
459
void
460
mpp_free(tp)
461
struct task *tp;
462
{
463
struct nodeset *sp;
464
int ptype;
465
struct task *tp2;
466
int tid = tp->t_tid;
467
468
if (!TIDISNODE(tid))
469
return;
470
471
ptype = TIDTOTYPE(tid);
472
tp->t_txq = 0; /* don't free pvmhost's txq */
473
for (sp = busynodes->n_link; sp != busynodes; sp = sp->n_link) {
474
if ((sp->n_ptype & ptypemask) == ptype) {
475
476
if (pvmdebmask & PDMNODE) {
477
sprintf(pvmtxt, "mpp_free() t%x type=%ld alive=%d\n",
478
tid, sp->n_ptype, sp->n_alive);
479
pvmlogerror(pvmtxt);
480
}
481
if (--sp->n_alive == 0) {
482
if (tp2 = task_find(sp->n_ptid)) {
483
tp2->t_flag |= TF_CLOSE;
484
if (tp2->t_sock != -1) {
485
/*
486
wrk_fds_delete(tp2->t_sock, 3);
487
(void)close(tp2->t_sock);
488
tp2->t_sock = -1;
489
*/
490
shutdown(tp2->t_sock, 1);
491
/* close stdout after pvmhost dies */
492
tp2->t_out = tp->t_out;
493
}
494
}
495
/*
496
LISTDELETE(sp, n_link, n_rlink);
497
PVM_FREE(sp);
498
*/
499
}
500
tp->t_out = -1; /* don't free shared stdout if alive > 0 */
501
return;
502
}
503
}
504
sprintf(pvmtxt, "mpp_free() t%x not active\n", tid);
505
pvmlogerror(pvmtxt);
506
return;
507
}
508
509
510
/* load executable onto the given set of nodes */
511
int
512
mpp_load( wxp )
513
struct waitc_spawn *wxp;
514
{
515
int flags = 0; /* exec options */
516
char *name; /* executable */
517
char **argv; /* arg list (argv[-1] must be there) */
518
int count; /* how many */
519
int *tids; /* array to store new tids */
520
int ptid; /* parent task ID */
521
int nenv; /* length of environment */
522
char **envp; /* environment strings */
523
int ptypepart; /* type field */
524
int j;
525
struct task *tp;
526
struct pkt *hosttxq; /* out-going queue of pvmhost */
527
int err = 0;
528
struct nodeset *sp;
529
char c[128]; /* buffer to store count, name.host */
530
char **ar = 0;
531
int nargs;
532
char **ep, **eplist;
533
char path[MAXPATHLEN];
534
struct stat sb;
535
char **av;
536
int hostout; /* stdout of pvmhost */
537
struct hostd *hp = hosts->ht_hosts[hosts->ht_local];
538
int hostpid; /* Unix pid of pvmhost */
539
char htid[128]; /* buffer to store pvmhost tid */
540
char *p;
541
542
static char *nullep[] = { "", 0 };
543
544
/* -- initialize some variables from the waitc_spawn struct -- */
545
546
name = wxp->w_argv[0];
547
argv = wxp->w_argv;
548
count = wxp->w_veclen;
549
tids = wxp->w_vec;
550
ptid = wxp->w_ptid;
551
nenv = wxp->w_nenv;
552
envp = wxp->w_env;
553
554
555
556
eplist = CINDEX(name, '/') ? nullep : epaths;
557
558
for (ep = eplist; *ep; ep++) {
559
/* search for file */
560
(void)strcpy(path, *ep);
561
if (path[0])
562
(void)strcat(path, "/");
563
(void)strncat(path, name, sizeof(path) - strlen(path) - 1);
564
565
if (stat(path, &sb) == -1
566
|| ((sb.st_mode & S_IFMT) != S_IFREG)
567
|| !(sb.st_mode & S_IEXEC)) {
568
if (pvmdebmask & PDMTASK) {
569
sprintf(pvmtxt, "mpp_load() stat failed <%s>\n", path);
570
pvmlogerror(pvmtxt);
571
}
572
continue;
573
}
574
575
if (!(sp = mpp_new(count+1, ptid))) {
576
err = PvmOutOfRes;
577
goto done;
578
}
579
ptypepart = (sp->n_ptype << (ffs(tidtmask) - 1)) | TIDONNODE;
580
581
if (argv)
582
for (nargs = 0; argv[nargs]; nargs++);
583
else
584
nargs = 0;
585
/* ar[-1], poe, -procs, #, -euilib, us, -hfile fname */
586
nargs += MPIARGC + 1;
587
av = TALLOC(nargs + 1, char*, "argv");
588
av++; /* reserve room for debugger */
589
BZERO((char*)av, nargs * sizeof(char*));
590
av[0] = MPICOMM;
591
av[1] = path;
592
av[--nargs] = 0;
593
if (hostfileused) {
594
av[--nargs] = nodefile;
595
av[--nargs] = MPIOPT3;
596
} else {
597
av[--nargs] = defaultpool;
598
av[--nargs] = MPIOPT4;
599
}
600
av[--nargs] = mpiadapter;
601
av[--nargs] = MPIOPT2;
602
sprintf(c, "%d", count+1);
603
av[--nargs] = c;
604
av[--nargs] = MPIOPT1;
605
for (j = 2; j < nargs; j++)
606
av[j] = argv[j - 1]; /* poe name argv -procs # -euilib us */
607
/*
608
if ((sock = mksock()) == -1) {
609
err = PvmSysErr;
610
goto done;
611
}
612
*/
613
if (flags & PvmTaskDebug)
614
av++; /* pdbx name -procs # -euilib us */
615
/* if (err = forkexec(flags, av[0], av, 0, (char **)0, &tp))
616
*/
617
if (err = forkexec(flags, av[0], av, 0, (char **)0, 0,
618
0, 0, &tp))
619
goto done;
620
tp->t_ptid = ptid;
621
PVM_FREE(tp->t_a_out);
622
sprintf(c, "%s.host", name);
623
tp->t_a_out = STRALLOC(c);
624
sp->n_ptid = tp->t_tid; /* pvmhost's tid */
625
hosttxq = tp->t_txq;
626
hostout = tp->t_out;
627
hostpid = tp->t_pid;
628
tp->t_out = -1;
629
sprintf(htid, "PVMHTID=%d", tp->t_tid);
630
p = STRALLOC(htid);
631
pvmputenv(p);
632
sp2pvminfo[0] = TDPROTOCOL;
633
sp2pvminfo[1] = myhostpart + ptypepart;
634
sp2pvminfo[2] = ptid;
635
sp2pvminfo[3] = MAXFRAGSIZE;
636
sp2pvminfo[4] = myndf;
637
sp2pvminfo[5] = partsize;
638
sp2pvminfo[6] = wxp->w_outtid;
639
sp2pvminfo[7] = wxp->w_outtag;
640
sp2pvminfo[8] = wxp->w_outctx;
641
sp2pvminfo[9] = wxp->w_trctid;
642
sp2pvminfo[10] = wxp->w_trctag;
643
sp2pvminfo[11] = wxp->w_trcctx;
644
645
646
/*
647
if (sockconn(sock, tp, pvminfo) == -1) {
648
err = PvmSysErr;
649
task_free(tp);
650
goto done;
651
}
652
*/
653
/* XXX task may not be on same host; can't do auth with tmp file */
654
tp->t_flag |= TF_CONN;
655
if (pvmdebmask & PDMTASK) {
656
sprintf(pvmtxt, "mpp_load() %d type=%d ptid=%x t%x...\n",
657
count, sp->n_ptype, ptid, myhostpart + ptypepart);
658
pvmlogerror(pvmtxt);
659
}
660
661
/* create new task structs */
662
663
for (j = 0; j < count; j++) {
664
if ((tp = task_new(myhostpart + ptypepart + j)) == NULL) {
665
err = PvmOutOfRes;
666
goto done;
667
}
668
tp->t_a_out = STRALLOC(name);
669
tp->t_ptid = ptid;
670
tp->t_flag |= TF_CONN; /* no need for the auth crap */
671
tids[j] = tp->t_tid;
672
PVM_FREE(tp->t_txq);
673
tp->t_txq = hosttxq; /* node tasks share pvmhost's txq */
674
tp->t_out = hostout; /* and stdout */
675
tp->t_pid = hostpid; /* pvm_kill should kill pvmhost */
676
tp->t_outtid = wxp->w_outtid; /* catch stdout/stderr */
677
tp->t_outtag = wxp->w_outtag;
678
tp->t_outctx = wxp->w_outctx;
679
}
680
return 0;
681
}
682
if (pvmdebmask & PDMTASK) {
683
sprintf(pvmtxt, "mpp_load() didn't find <%s>\n", name);
684
pvmlogerror(pvmtxt);
685
}
686
err = PvmNoFile;
687
688
done:
689
for (j = 0; j < count; j++)
690
tids[j] = err;
691
return err;
692
}
693
694
695
#if 0
696
697
/* kill poe process */
698
void
699
sp2kill(a_out, node)
700
char *a_out;
701
int node;
702
{
703
FILE *fp;
704
int i;
705
char nname[128]; /* node name */
706
char comm[512]; /* command to issue */
707
char *hfn; /* host file name */
708
struct stat sb;
709
char *av[8]; /* for rsh args */
710
int ac;
711
int pid = -1; /* pid of rsh */
712
713
node++;
714
if ((hfn = getenv("MP_HOSTFILE")) || stat(hfn = "host.list", &sb) != -1) {
715
if (fp = fopen(hfn, "r")) {
716
for (i = 0; i < node; i++)
717
fscanf(fp, "%s", nname);
718
fclose(hfn);
719
if ((pid = fork()) == -1) {
720
pvmlogperror("sp2kill() fork");
721
return;
722
}
723
if (!pid) {
724
ac = 0;
725
av[ac++] = "/usr/bin/rsh";
726
av[ac++] = nname;
727
av[ac++] = "poekill";
728
av[ac++] = a_out;
729
av[ac++] = 0;
730
for (i = getdtablesize(); --i > 2; )
731
(void)close(i);
732
execvp(av[0], av);
733
_exit(1);
734
}
735
} else
736
pvmlogperror("sp2kill() fopen");
737
return;
738
}
739
pvmlogerror("sp2kill() no host file");
740
}
741
742
#endif /*0*/
743
744
745
void
746
mpp_kill(tp, signum)
747
struct task *tp;
748
int signum;
749
{
750
751
if (TIDISNODE(tp->t_tid)) {
752
if (signum == SIGTERM || signum == SIGKILL) {
753
/* sp2kill(tp->t_a_out, tp->t_tid & tidnmask); */
754
(void)kill(tp->t_pid, signum);
755
mpp_free(tp);
756
task_cleanup(tp);
757
task_free(tp);
758
} else {
759
sprintf(pvmtxt,"mpp_kill() signal %d to node t%x ignored\n",
760
signum, tp->t_tid);
761
pvmlogerror(pvmtxt);
762
}
763
} else
764
(void)kill(tp->t_pid, signum);
765
}
766
767
768
/*
769
* Add pvmhost's socket to wfds if there are packets waiting to
770
* be sent to a related node task. Node tasks have no sockets;
771
* they share pvmhost's packet queue (txq). Pvmhost simply
772
* forwards any packets it receives to the appropriate node.
773
*/
774
775
int
776
mpp_output(dummy1, dummy2)
777
struct task *dummy1;
778
struct pkt *dummy2;
779
{
780
struct nodeset *sp, *sp2;
781
struct task *tp;
782
int ptype;
783
784
for (sp = busynodes->n_link; sp != busynodes; sp = sp->n_link)
785
if ((tp = task_find(sp->n_ptid))) {
786
if (tp->t_txq->pk_link->pk_buf && tp->t_sock != -1)
787
wrk_fds_add(tp->t_sock, 2);
788
} else {
789
if (sp->n_alive) {
790
sprintf(pvmtxt, "mpp_output() pvmhost %d died!\n", sp->n_ptype);
791
pvmlogerror(pvmtxt);
792
/* clean up tasks it serves */
793
ptype = sp->n_ptype & ptypemask;
794
for (tp = locltasks->t_link; tp != locltasks; tp = tp->t_link)
795
if (TIDISNODE(tp->t_tid) && TIDTOTYPE(tp->t_tid) == ptype) {
796
tp->t_txq = 0;
797
tp = tp->t_rlink;
798
task_cleanup(tp->t_link);
799
task_free(tp->t_link);
800
}
801
}
802
/* pvmhost has died, destroy the node set */
803
sp2 = sp;
804
sp = sp->n_rlink;
805
LISTDELETE(sp2, n_link, n_rlink);
806
PVM_FREE(sp2);
807
}
808
return 0;
809
}
810
811
812
/* replace tm_connect and tm_conn2 */
813
int
814
mpp_conn(tp, tp2)
815
struct task *tp;
816
struct task *tp2;
817
{
818
struct nodeset *sp;
819
int pvminfo[SIZEHINFO]; /* host info */
820
int ptypepart; /* type field */
821
822
if (pvmdebmask & PDMNODE) {
823
sprintf(pvmtxt, "mpp_conn() pvmhost %x", tp2->t_tid);
824
pvmlogerror(pvmtxt);
825
}
826
tp2->t_sock = tp->t_sock;
827
tp2->t_sad = tp->t_sad;
828
tp2->t_salen = tp->t_salen;
829
tp2->t_flag |= TF_CONN;
830
tp->t_sock = -1;
831
for (sp = busynodes->n_link; sp != busynodes; sp = sp->n_link)
832
if (sp->n_ptid == tp2->t_tid)
833
break;
834
if (sp == busynodes) {
835
pvmlogerror("mpp_conn() task is not pvmhost\n");
836
return -1;
837
}
838
ptypepart = (sp->n_ptype << (ffs(tidtmask) - 1)) | TIDONNODE;
839
if (write(tp2->t_sock, sp2pvminfo, sizeof(sp2pvminfo))
840
!= sizeof(sp2pvminfo)) {
841
pvmlogperror("mpp_conn() write");
842
return -1;
843
}
844
return 0;
845
}
846
847
848
#if 0
849
850
/*
851
* Create socket to talk to pvmhost.
852
* Return socket descriptor if successful, -1 otherwise.
853
*/
854
int
855
mksock()
856
{
857
struct hostd *hp = hosts->ht_hosts[hosts->ht_local];
858
struct sockaddr_in sin;
859
char buf[128];
860
char *p;
861
int cc;
862
int sock;
863
864
if ((sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
865
pvmlogperror("mksock() socket");
866
return -1;
867
}
868
sin = hp->hd_sad;
869
if (bind(sock, (struct sockaddr*)&sin, sizeof(sin)) == -1) {
870
pvmlogperror("mksock() bind");
871
(void)close(sock);
872
return -1;
873
}
874
cc = sizeof(sin);
875
if (getsockname(sock, (struct sockaddr*)&sin, &cc) == -1) {
876
pvmlogperror("mksock() getsockname");
877
(void)close(sock);
878
return -1;
879
}
880
if (listen(sock, 1) == -1) {
881
pvmlogperror("mksock() listen");
882
(void)close(sock);
883
return -1;
884
}
885
886
p = inadport_hex(&sin);
887
sprintf(buf, "PVMSOCK=%s", p);
888
p = STRALLOC(buf);
889
pvmputenv(p);
890
891
return sock;
892
}
893
894
895
/*
896
* Wait for connect request from pvmhost and establish connection.
897
* Return 0 if successful, -1 otherwise.
898
* Close listening socket.
899
*/
900
int
901
sockconn(sock, tp, hinfo)
902
int sock; /* listening post */
903
struct task *tp; /* pvm host */
904
int hinfo[]; /* host info to pass along */
905
{
906
int i;
907
908
if ((tp->t_sock = accept(sock, (struct sockaddr*)&tp->t_sad,
909
&tp->t_salen)) == -1) {
910
pvmlogperror("sockconn() accept");
911
return -1;
912
} else {
913
if (pvmdebmask & (PDMPACKET|PDMTASK)) {
914
sprintf(pvmtxt, "sockconn() accept from %s sock %d\n",
915
inadport_decimal(&tp->t_sad), tp->t_sock);
916
pvmlogerror(pvmtxt);
917
}
918
close(sock);
919
#ifndef NOSOCKOPT
920
i = 1;
921
if (setsockopt(tp->t_sock, IPPROTO_TCP, TCP_NODELAY,
922
(char*)&i, sizeof(int)) == -1)
923
pvmlogperror("sockconn() setsockopt");
924
#endif
925
}
926
if (write(tp->t_sock, hinfo, SIZEHINFO*sizeof(int))
927
!= SIZEHINFO*sizeof(int)) {
928
pvmlogperror("sockconn: write");
929
return -1;
930
}
931
if ((i = fcntl(tp->t_sock, F_GETFL, 0)) == -1)
932
pvmlogperror("sockconn: fcntl");
933
else {
934
i |= O_NDELAY;
935
(void)fcntl(tp->t_sock, F_SETFL, i);
936
}
937
wrk_fds_add(tp->t_sock, 1);
938
939
return 0;
940
}
941
942
#endif /*0*/
Loggerhead is a web-based interface for Breezy
Version: 2.0.1