Deal correctly with infix type constructors in GADT decls
[ghc-hetmet.git] / rts / parallel / ParInit.c
1 /* --------------------------------------------------------------------------
2    Time-stamp: <Wed Mar 21 2001 16:37:16 Stardate: [-30]6363.46 hwloidl>
3
4    Initialising the parallel RTS
5
6    An extension based on Kevin Hammond's GRAPH for PVM version
7    P. Trinder, January 17th 1995.
8    Adapted for the new RTS
9    P. Trinder, July 1997.
10    H-W. Loidl, November 1999.
11
12    ------------------------------------------------------------------------ */
13
14 #ifdef PAR /* whole file */
15
16 //@menu
17 //* Includes::                  
18 //* Global variables::          
19 //* Initialisation Routines::   
20 //@end menu
21
22 //@node Includes, Global variables
23 //@subsection Includes
24
25 /* Evidently not Posix */
26 /* #include "PosixSource.h" */
27
28 #include <setjmp.h>
29 #include "Rts.h"
30 #include "RtsFlags.h"
31 #include "RtsUtils.h"
32 #include "ParallelRts.h"
33 #include "Sparks.h"
34 #include "LLC.h"
35 #include "HLC.h"
36
37 //@node Global variables, Initialisation Routines, Includes
38 //@subsection Global variables
39
40 /* Global conditions defined here. */
41
42 rtsBool IAmMainThread = rtsFalse;       /* Set for the main thread      */
43
44 /* Task identifiers for various interesting global tasks. */
45
46 GlobalTaskId IOTask = 0,                /* The IO Task Id               */
47              SysManTask = 0,            /* The System Manager Task Id   */
48              mytid = 0;                 /* This PE's Task Id            */
49
50 rtsTime         main_start_time;        /* When the program started     */
51 rtsTime         main_stop_time;         /* When the program finished    */
52 jmp_buf         exit_parallel_system;   /* How to abort from the RTS    */
53
54
55 //rtsBool fishing = rtsFalse;             /* We have no fish out in the stream */
56 rtsTime last_fish_arrived_at = 0;       /* Time of arrival of most recent fish*/
57 nat     outstandingFishes = 0;          /* Number of active fishes */ 
58
59 //@cindex spark queue
60 /* GranSim: a globally visible array of spark queues */
61 rtsSpark *pending_sparks_hd[SPARK_POOLS],  /* ptr to start of a spark pool */ 
62          *pending_sparks_tl[SPARK_POOLS],  /* ptr to end of a spark pool */ 
63          *pending_sparks_lim[SPARK_POOLS],
64          *pending_sparks_base[SPARK_POOLS]; 
65
66 //@cindex spark_limit
67 /* max number of sparks permitted on the PE; 
68    see RtsFlags.ParFlags.maxLocalSparks */
69 nat spark_limit[SPARK_POOLS];
70
71 //@cindex PendingFetches
72 /* A list of fetch reply messages not yet processed; this list is filled
73    by awaken_blocked_queue and processed by processFetches */
74 StgBlockedFetch *PendingFetches = END_BF_QUEUE;
75
76 //@cindex allPEs
77 GlobalTaskId *allPEs;
78
79 //@cindex nPEs
80 nat nPEs = 0;
81
82 //@cindex sparksIgnored
83 nat sparksIgnored = 0, sparksCreated = 0, 
84     threadsIgnored = 0, threadsCreated = 0;
85
86 //@cindex advisory_thread_count
87 nat advisory_thread_count = 0;
88
89 globalAddr theGlobalFromGA;
90
91 /* For flag handling see RtsFlags.h */
92
93 //@node Prototypes
94 //@subsection Prototypes
95
96 /* Needed for FISH messages (initialisation of random number generator) */
97 void srand48 (long);
98 time_t time (time_t *);
99
100 //@node Initialisation Routines,  , Global variables
101 //@subsection Initialisation Routines
102
103 /*
104   par_exit defines how to terminate the program.  If the exit code is
105   non-zero (i.e. an error has occurred), the PE should not halt until
106   outstanding error messages have been processed.  Otherwise, messages
107   might be sent to non-existent Task Ids.  The infinite loop will actually
108   terminate, since STG_Exception will call myexit\tr{(0)} when
109   it received a PP_FINISH from the system manager task.
110 */
111 //@cindex shutdownParallelSystem
112 void
113 shutdownParallelSystem(StgInt n)
114 {
115   /* use the file specified via -S */ 
116   FILE *sf = RtsFlags.GcFlags.statsFile;
117
118   IF_PAR_DEBUG(verbose,
119                if (n==0)
120                  belch("==== entered shutdownParallelSystem ...");
121                else
122                  belch("==== entered shutdownParallelSystem (ERROR %d)...", n);
123                );
124   
125   stopPEComms(n);
126
127 #if 0
128   if (sf!=(FILE*)NULL) 
129     fprintf(sf, "PE %x: %u sparks created, %u sparks Ignored, %u threads created, %u threads Ignored", 
130             (W_) mytid, sparksCreated, sparksIgnored,
131             threadsCreated, threadsIgnored);
132 #endif
133
134   ShutdownEachPEHook();
135 }
136
137 //@cindex initParallelSystem
138 void
139 initParallelSystem(void)
140 {
141   /* Don't buffer standard channels... */
142   setbuf(stdout,NULL);
143   setbuf(stderr,NULL);
144   
145   srand48(time(NULL) * getpid()); /* Initialise Random-number generator seed*/
146                                   /* used to select target of FISH message*/
147   if (!InitPackBuffer())
148     barf("InitPackBuffer");
149
150   if (!initMoreBuffers())
151     barf("initMoreBuffers");
152
153   if (!initSparkPools())
154     barf("initSparkPools");
155 }
156
157 /* 
158  * SynchroniseSystem synchronises the reduction task with the system
159  * manager, and initialises the Global address tables (LAGA & GALA)
160  */
161
162 //@cindex synchroniseSystem
163 void
164 synchroniseSystem(void)
165 {
166   /* Only in debug mode? */
167   fprintf(stderr, "==== Starting parallel execution on %d processors ...\n", nPEs);
168
169   InitEachPEHook();                  /* HWL: hook to be execed on each PE */
170
171   /* Initialize global address tables */
172   initGAtables();
173
174   initParallelSystem();
175   
176   startPEComms();
177 }
178
179 /* 
180   Do the startup stuff (this is PVM specific!).
181   Determines global vars: mytid, IAmMainThread, SysManTask, nPEs
182   Called at the beginning of RtsStartup.startupHaskell
183 */
184 void 
185 startupParallelSystem(char *argv[]) { 
186  mytid = pvm_mytid();           /* Connect to PVM */
187
188  if (*argv[0] == '-') {         /* Look to see whether we're the Main Thread */
189   IAmMainThread = rtsTrue;
190   sscanf(argv[0],"-%0X",&SysManTask);  /* extract SysMan task ID*/      
191   argv++;                              /* Strip off flag argument */
192  } else {
193   SysManTask = pvm_parent();
194  }
195
196  IF_PAR_DEBUG(verbose,
197                fprintf(stderr, "==== [%x] %s PE located SysMan at %x\n",
198                        mytid, IAmMainThread?"Main":"Remote", SysManTask));
199
200  nPEs = atoi(argv[1]);
201 }
202
203 /* 
204    Exception handler during startup.
205 */
206 void *
207 processUnexpectedMessageDuringStartup(rtsPacket p) {
208   OpCode opCode;
209   GlobalTaskId sender_id;
210
211   getOpcodeAndSender(p, &opCode, &sender_id);
212
213   switch(opCode) { 
214       case PP_FISH:
215         bounceFish();
216         break;
217 #if defined(DIST)
218       case PP_REVAL:
219         bounceReval();
220         break;
221 #endif
222       case PP_FINISH:
223         stg_exit(EXIT_SUCCESS); 
224         break;
225       default:
226         fprintf(stderr,"== Task %x: Unexpected OpCode %x (%s) from %x in startPEComms\n",
227                 mytid, opCode, getOpName(opCode), sender_id);
228     }
229 }
230
231 void 
232 startPEComms(void){ 
233
234   startUpPE(); 
235   allPEs = (GlobalTaskId *) stgMallocBytes(sizeof(GlobalTaskId) * MAX_PES,
236                                            "(PEs)");
237   
238   /* Send our tid and IAmMainThread flag back to SysMan */
239   sendOp1(PP_READY, SysManTask, (StgWord)IAmMainThread);  
240   /* Wait until we get the PE-Id table from Sysman */    
241   waitForPEOp(PP_PETIDS, SysManTask, processUnexpectedMessageDuringStartup); 
242
243   IF_PAR_DEBUG(verbose,
244                belch("==-- startPEComms: methinks we just received a PP_PETIDS message"));
245
246   /* Digest the PE table we received */
247   processPEtids();
248 }
249
250 void
251 processPEtids(void) { 
252   long newPE;
253   nat i, sentPEs, currentPEs;
254
255   nPEs=0;
256           
257   currentPEs = nPEs;
258
259   IF_PAR_DEBUG(verbose,
260                 belch("==-- processPEtids: starting to iterate over a PVM buffer"));
261   /* ToDo: this has to go into LLComms !!! */
262   GetArgs(&sentPEs,1);
263
264   ASSERT(sentPEs > currentPEs);
265   ASSERT(sentPEs < MAX_PES); /* enforced by SysMan too*/  
266   
267   for (i = 0; i < sentPEs; i++) { 
268     GetArgs(&newPE,1);
269     if (i<currentPEs) { 
270       ASSERT(newPE == allPEs[i]);
271     } else { 
272 #if defined(DIST)
273       // breaks with PAR && !DEBUG
274       IF_PAR_DEBUG(verbose,
275         fprintf(stderr, "[%x] registering %d'th %x\n", mytid, i, newPE)); 
276       if(!looks_like_tid(newPE))
277           barf("unacceptable taskID %x\n",newPE);
278 #endif
279       allPEs[i] = newPE;      
280       nPEs++;
281       registerTask(newPE); 
282     }
283   }
284
285   IF_PAR_DEBUG(verbose,
286                /* debugging */
287                belch("++++ [%x] PE table as I see it:", mytid);
288                for (i = 0; i < sentPEs; i++) { 
289                  belch("++++ allPEs[%d] = %x", i, allPEs[i]);
290                });
291 }
292
293 void 
294 stopPEComms(StgInt n) { 
295   if (n != 0) { 
296     /* In case sysman doesn't know about us yet...
297     pvm_initsend(PvmDataDefault);
298     PutArgs(&IAmMainThread,1);
299     pvm_send(SysManTask, PP_READY);
300      */
301     sendOp(PP_READY, SysManTask);  
302   } 
303   
304   sendOp2(PP_FINISH, SysManTask, n, n);  
305   waitForPEOp(PP_FINISH, SysManTask, NULL);
306   fflush(gr_file);
307   shutDownPE();
308 }
309
310 #endif /* PAR -- whole file */
311
312 //@index
313 //* PendingFetches::  @cindex\s-+PendingFetches
314 //* SynchroniseSystem::  @cindex\s-+SynchroniseSystem
315 //* allPEs::  @cindex\s-+allPEs
316 //* initParallelSystem::  @cindex\s-+initParallelSystem
317 //* nPEs::  @cindex\s-+nPEs
318 //* par_exit::  @cindex\s-+par_exit
319 //* spark queue::  @cindex\s-+spark queue
320 //* sparksIgnored::  @cindex\s-+sparksIgnored
321 //@end index
322