forked from lanl/coNCePTuaL
-
Notifications
You must be signed in to change notification settings - Fork 1
/
codegen_interpret.py
2924 lines (2675 loc) · 125 KB
/
codegen_interpret.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#! /usr/bin/env python
########################################################################
#
# Code generation module for the coNCePTuaL language:
# Interpreter of coNCePTuaL programs
#
# By Scott Pakin <pakin@lanl.gov>
#
# ----------------------------------------------------------------------
#
#
# Copyright (C) 2015, Los Alamos National Security, LLC
# All rights reserved.
#
# Copyright (2015). Los Alamos National Security, LLC. This software
# was produced under U.S. Government contract DE-AC52-06NA25396
# for Los Alamos National Laboratory (LANL), which is operated by
# Los Alamos National Security, LLC (LANS) for the U.S. Department
# of Energy. The U.S. Government has rights to use, reproduce,
# and distribute this software. NEITHER THE GOVERNMENT NOR LANS
# MAKES ANY WARRANTY, EXPRESS OR IMPLIED, OR ASSUMES ANY LIABILITY
# FOR THE USE OF THIS SOFTWARE. If software is modified to produce
# derivative works, such modified software should be clearly marked,
# so as not to confuse it with the version available from LANL.
#
# Additionally, redistribution and use in source and binary forms,
# with or without modification, are permitted provided that the
# following conditions are met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#
# * Neither the name of Los Alamos National Security, LLC, Los Alamos
# National Laboratory, the U.S. Government, nor the names of its
# contributors may be used to endorse or promote products derived
# from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY LANS AND CONTRIBUTORS "AS IS" AND ANY
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL LANS OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
# OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
# OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
# BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
# WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
# OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
# EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
#
########################################################################
import sys
import os
import string
import re
import math
import types
import random
import copy
from ncptl_ast import AST
from ncptl_error import NCPTL_Error
from ncptl_variables import Variables
try:
import resource
except ImportError:
# Not every Python installation provides the resource module.
# Because we need it only to acquire the OS page size, it's not
# critical; we can safely utilize a default page size.
pass
# To support the coNCePTuaL GUI (built using Jython) we need to make
# some packages optional.
try:
# If we can import the java module then we must be running Jython
# (either the interpreter or the compiler).
import java
from gui_patches import *
except ImportError:
# We're running from ordinary C-based Python.
from pyncptl import *
# Define "safe" versions of ncptl_virtual_to_physical.
def ncptl_func_processor_of(procmap, vtask, num_tasks):
if vtask < 0 or vtask >= num_tasks:
return -1L
else:
return ncptl_virtual_to_physical(procmap, vtask)
def ncptl_dfunc_processor_of(procmap, vtask, num_tasks):
return ncptl_func_processor_of(procmap, int(vtask), num_tasks)
# Define "safe" versions of ncptl_physical_to_virtual.
def ncptl_func_task_of(procmap, ptask, num_tasks):
if ptask < 0 or ptask >= num_tasks:
return -1L
else:
return ncptl_physical_to_virtual(procmap, ptask)
def ncptl_dfunc_task_of(procmap, ptask, num_tasks):
return ncptl_func_task_of(procmap, int(ptask), num_tasks)
class NCPTL_CodeGen:
thisfile = globals()["__file__"]
bytes_per_int = long(1.0+math.log(sys.maxint+1.0)/math.log(2.0)) / 8L
try:
bytes_per_page = resource.getpagesize()
except NameError:
# It shouldn't affect anything too seriously if we simply
# assume 4KB pages.
bytes_per_page = 4096
trivial_nodes = [
"top_level_stmt_list",
"header_decl_list",
"header_decl",
"version_decl",
"simple_stmt_list",
"simple_stmt",
"let_binding_list",
"source_task",
"target_tasks",
"rel_expr",
"rel_primary_expr",
"expr",
"for_each_expr",
"primary_expr",
"item_count"]
sub_statement_nodes = \
dict([(ntype, 1) for ntype in [
"rel_disj_expr",
"rel_conj_expr",
"eq_expr",
"ifelse_expr",
"add_expr",
"mult_expr",
"unary_expr",
"power_expr",
"integer",
"ident",
"my_task",
"real",
"dimension",
"dimension_list",
"func_call",
"item_size",
"data_type",
"byte_count",
"data_multiplier",
"time_unit",
"aggregate_func",
"an",
"message_alignment",
"touch_repeat_count",
"expr_list",
"task_expr",
"restricted_ident",
"string_or_expr_list",
"string_or_log_comment",
"string",
"such_that",
"comma",
"stride",
"range_list",
"range",
"aggregate_expr",
"log_expr_list",
"log_expr_list_elt",
"message_spec",
"touching_type",
"buffer_offset",
"buffer_number",
"recv_buffer_number",
"send_attrs",
"receive_attrs"]])
#---------------------#
# Helper class that #
# represents an event #
#---------------------#
class Event:
def __init__(self, operation, task, srclines, peers=None, msgsize=None,
tag=None, blocking=1, attributes=None, collective_id=None):
"Define a new event, initially with no timing information."
self.operation = operation
self.task = task
self.srclines = srclines
if peers == None:
self.peers = []
else:
self.peers = peers
self.msgsize = msgsize
self.tag = tag
self.blocking = blocking
if attributes == None:
self.attributes = []
else:
self.attributes = attributes
self.collective_id = collective_id
self.posttime = None # We don't know when we were posted.
self.completetime = None # We don't know when we completed.
self.found_match = 0 # We haven't processed a matching event
def contents(self):
"Return a tuple representing our internal state."
return [self.operation, self.task, self.peers, self.msgsize,
self.tag, self.blocking, self.attributes, self.posttime,
self.completetime, self.found_match]
#---------------------#
# Helper class that #
# represents a list #
# of events #
#---------------------#
class EventList:
def __init__(self, parent):
"Encapsulate all of the variables that relate to event lists."
self.parent = parent # Parent object (of type NCPTL_CodeGen)
self.errmsg = parent.errmsg # Error-message object
self.events = [] # The list of events proper
self.first_incomplete = 0 # Index of first incomplete event
self.first_unposted = 0 # Index of first unposted event
self.length = 0 # Number of entries in events[]
def all_complete(self):
"Return 1 if all events have completed, 0 otherwise."
return self.first_incomplete >= self.length
def push(self, event):
"""
Modify an event to use physical instead of virtual
ranks then push the event onto the end of the event
list.
"""
self.events.append(event)
self.length = self.length + 1
def get_first_incomplete(self):
"Return the first incomplete event in the event list."
return self.events[self.first_incomplete]
def try_posting_all(self):
"Post as many events as possible."
while self.first_unposted < self.length:
event = self.events[self.first_unposted]
if self.first_unposted == 0:
# First event posts immediately.
event.posttime = 0
self.first_unposted = self.first_unposted + 1
elif self.events[self.first_unposted-1].completetime != None:
# Post only after the previous event completes.
prev_event = self.events[self.first_unposted-1]
event.posttime = prev_event.completetime + self.complete_post_overhead(prev_event, event)
self.first_unposted = self.first_unposted + 1
else:
# We can't complete anything else.
break
def complete(self, peerlist=None, nolat_peerlist=None):
'''
Mark the first incomplete event as completed and
return the completion time. If a list of peer events
is given, set the completion time to the maximum of
the event\'s completion time and the post time +
message latency of each of the peer events. If a
list of "no latency" peer events is specified, do the
same thing but using the maximum of the completion
times.
'''
self.try_posting_all()
this_ev = self.events[self.first_incomplete]
if this_ev.posttime == None:
self.errmsg.error_internal("Task %d, event %d completed before being posted" %
(this_ev.task, self.first_incomplete))
newtime = this_ev.posttime + self.post_complete_overhead(this_ev)
if peerlist != None:
# PEERLIST represents senders.
for peer_ev in peerlist:
if peer_ev.posttime == None:
self.errmsg.error_internal("An event on task %d completed before being posted" % peer_ev.task)
newtime = max(newtime, peer_ev.posttime + self.message_latency(peer_ev))
if nolat_peerlist != None:
# NOLAT_PEERLIST represents equals.
for peer_ev in nolat_peerlist:
if peer_ev.posttime == None:
self.errmsg.error_internal("An event on task %d completed before being posted" % peer_ev.task)
newtime = max(newtime, peer_ev.posttime + self.post_complete_overhead(peer_ev))
this_ev.completetime = newtime
self.first_incomplete = self.first_incomplete + 1
return newtime
def post_complete_overhead(self, event):
"Return the overhead between posting and completing an event."
return 0
def complete_post_overhead(self, prev_ev, this_ev):
"Return the overhead between completing an event and posting the next event."
# Receiving a blocking message takes no time; everything else does.
if prev_ev.operation == "RECEIVE" and prev_ev.blocking:
return 0
elif prev_ev.operation == "NEWSTMT":
return 0
else:
return 1
def message_latency(self, event):
"Return the message latency for a given event."
# Determine the set of source tasks and the set of target tasks.
if event.operation == "REDUCE":
# REDUCE events -- peer list contains the source and
# target lists
source_tasks, target_tasks = event.peers
else:
# Other communication events -- peer list contains
# only the targets.
source_tasks = [event.task]
target_tasks = event.peers
# Compute the maximum latency from any source to any
# target and return that.
latency = -1
latency_list = self.parent.latency_list
for source in source_tasks:
for target in target_tasks:
if source == target:
latency = max(latency, latency_list[0][1])
else:
for taskcount, newlatency in latency_list:
if source/taskcount == target/taskcount:
# We're guaranteed to reach this point
# because we added (numtasks, _) to the
# end of latency_list.
latency = max(latency, newlatency)
break
return latency
def find_unmatched(self):
"Return a list of events with no matching event."
return filter(lambda ev: not ev.found_match, self.events)
def delete_unposted(self):
"""
Delete from the first unposted event onwards (invoked
when a task is blocked on account of deadlock).
"""
del self.events[self.first_unposted:]
self.length = len(self.events)
#----------------------#
# Helper class that #
# represents a message #
# queue #
#----------------------#
class MessageQueue:
def __init__(self, errmsg):
"Initialize a message queue."
self.errmsg = errmsg # Error-message object
self.queues = {} # Map from source task to tag to message size to an event queue
def push(self, event):
"Push a new event onto a message queue."
source_task = event.task
tag = event.tag
message_size = event.msgsize
if not self.queues.has_key(source_task):
self.queues[source_task] = {}
if not self.queues[source_task].has_key(tag):
self.queues[source_task][tag] = {}
if not self.queues[source_task][tag].has_key(message_size):
self.queues[source_task][tag][message_size] = []
self.queues[source_task][tag][message_size].append(event)
def pop_match(self, event):
"Pop the first matching event from the queue or return None."
result = self.peek_match(event)
source_task = event.peers[0]
tag = event.tag
message_size = event.msgsize
if result != None:
self.queues[source_task][tag][message_size].pop(0)
result.found_match = 1
return result
def unpop_match(self, event, matched_event):
"Reinsert a previously popped event into the queue."
source_task = event.peers[0]
tag = event.tag
message_size = event.msgsize
self.queues[source_task][tag][message_size].insert(0, matched_event)
def peek_match(self, event):
"Return the first matching event from the queue or None."
source_task = event.peers[0]
tag = event.tag
message_size = event.msgsize
try:
return self.queues[source_task][tag][message_size][0]
except KeyError:
return None
except IndexError:
return None
#---------------------#
# Exported functions #
# (called from the #
# compiler front end) #
#---------------------#
def __init__(self, options=[], numtasks=1L):
"Initialize the coNCePTuaL interpreter."
self.errmsg = NCPTL_Error() # Placeholder until generate is called
self.scopes = [] # Variable scopes (innermost first)
self.numtasks = numtasks # Number of tasks to simulate
self.cmdline = [] # Command-line passed to the backend
self.options = [] # Supported command-line options
self.context = "int" # Evaluation context (int or float)
self.next_byte = {} # Map from touch node to next byte to touch
self.logstate = {} # Map from a physical rank to log state
self.suppress_output = 0 # 1=temporarily ignore OUTPUTS and LOGS
self.program_uses_log_file = 0 # 1=LOGS or COMPUTES AGGREGATES was used
self.program_can_use_log_file = 1 # 0=supress logging; 1=allow it
self.logcolumn = 0L # Current "column" in the log file
self.for_time_reps = 3L # Number of repetitions to use for FOR <time>
self.random_seed = ncptl_seed_random_task(0L, 0L) # Seed for the RNG
self.mcastsync = 0L # 1=synchronize after a multicast
self.latency_list = [(1,1)] # Hierarchy of message latencies
self.kill_reps = 0L # 1=FOR...REPETITIONS limited to one iteration
self.timing_flag = ncptl_allocate_timing_flag() # Used by FOR <time>
self.type2method = {} # Map from a node type to a method that can handle it
self.stuck_tasks = {} # Set of deadlocked tasks
self.applicable_tasks = {} # Set of tasks that should execute the current statement
self.unique_id = 0L # Unique identifier for a collective operation
self.backend_name = "interpret"
self.backend_desc = "coNCePTuaL interpreter"
# Process the "--tasks" argument but store all others. Note
# that we do this "manually" without relying on
# ncptl_parse_command_line() because we need the number of
# tasks before we begin interpreting.
arg = 0
while arg < len(options):
# Search for "-T#", "-T #" and "--tasks=#".
taskstr = ""
if options[arg] == "-T" and arg+1 < len(options):
taskstr = options[arg+1]
argname = "-T"
arg = arg + 1
else:
arg_match = re.match(r'(--tasks=)(.*)', options[arg]) or re.match(r'(-T)(.*)', options[arg])
if arg_match:
argname = arg_match.group(1)
taskstr = arg_match.group(2)
# Verify that the task count is numeric.
if taskstr:
try:
self.numtasks = long(taskstr)
except ValueError:
self.errmsg.error_fatal('%s expected a number but received "%s"' %
(argname, taskstr),
filename=self.backend_name)
taskstr = ""
else:
self.cmdline.append(options[arg])
arg = arg + 1
# Point each method name in the trivial_nodes list to the
# n_trivial_node method.
for mname in self.trivial_nodes:
setattr(self, "n_" + mname, self.n_trivial_node)
# By default, all tasks perform every statement.
for task in range(0, self.numtasks):
self.applicable_tasks[task] = 1
def clear_events(self):
"""Restart the interpreter by clearing all of top-level state
(needed by the coNCePTuaL GUI)."""
self.errmsg = NCPTL_Error("internal")
self.eventlist = map(lambda self: self.EventList(self), [self] * int(self.numtasks))
self.msgqueue = map(lambda self: self.MessageQueue(self.errmsg), [self] * int(self.numtasks))
self.pendingevents = map(lambda h: [], [None] * int(self.numtasks))
self.timer_start = [0] * int(self.numtasks)
self.counters = []
self.counter_stack = map(lambda t: [], range(0, self.numtasks))
def fake_semantic_analysis(self, node):
"Pretend we ran the semantic analyzer (needed by the coNCePTuaL GUI)."
node.sem = {"is_constant": False}
for kid in node.kids:
self.fake_semantic_analysis(kid)
def generate(self, ast, filesource='<stdin>', filetarget="-", sourcecode=None):
"Interpret an AST."
self.generate_initialize(ast, filesource, filetarget, sourcecode)
self.process_node(ast) # Prefix traversal (roughly)
self.generate_finalize(ast, filesource, sourcecode)
return []
def compile_only(self, progfilename, codelines, outfilename, verbose=0, keepints=0):
"""
Do nothing unless an output file was specified, in which
case we dump event state to that file.
"""
if outfilename != "-":
if verbose:
sys.stderr.write("# Dumping final event state to %s ...\n" % outfilename)
self.dump_event_lists(outfilename)
else:
if verbose:
sys.stderr.write("# [Nothing to do here]\n")
return outfilename
def compile_and_link(self, progfilename, codelines, outfilename, verbose=0, keepints=0):
"Pass control to compile_only, as linking is not meaningful here."
self.compile_only(progfilename, codelines, outfilename, verbose, keepints)
return outfilename
#------------------#
# Internal utility #
# functions #
#------------------#
def filter_task_list(self, tasklist):
"""
Return only those tasks that are applicable to the current
statement. As a side effect, convert all tasks to int type.
"""
return map(int, filter(lambda t: self.applicable_tasks.has_key(t), tasklist))
def get_unique_id(self):
"Return the next available ID number."
unique_id = self.unique_id
self.unique_id += 1
return unique_id
def collectives_match(self, event1, event2):
"Return true if and only if two collective events match each other."
if event1.collective_id == None or event2.collective_id == None:
# At least one event is not a collective operation.
return False
return event1.collective_id == event2.collective_id
def set_log_file_status(self, enable):
"Force log-file usage on or off (intended to be called by derived classes)."
if enable:
self.program_uses_log_file = 1
self.program_can_use_log_file = 1
else:
self.program_can_use_log_file = 0
self.options = filter(lambda ev: ev[2] != "logfile", self.options)
def dump_event_lists(self, outfilename):
"Write an easy-to-parse list of events to a file."
try:
outfile = open(outfilename, "w")
for task in range(0, self.numtasks):
for event in self.eventlist[task].events:
if event.completetime == None:
outfile.write(self.format_plurals("Task %d posted %C %s at time %d but never completed it\n",
1, (task, event.operation, event.posttime)))
else:
outfile.write(self.format_plurals("Task %d posted %C %s at time %d and completed it at time %d\n",
1, (task, event.operation, event.posttime, event.completetime)))
outfile.close()
except IOError, (errno, strerror):
self.errmsg.error_fatal("Unable to produce %s (%s)" % (outfilename, strerror),
filename=self.backend_name)
def parse_latency_hierarchy(self, taskstr):
"""Parse a hierarchy of task counts and latencies into a list
of {tasks, latency} tuples."""
tasks_cost_list = []
prev_tasks = 1
prev_latency = 0
found_ellipsis = 0
numtasks = self.numtasks
taskspec_re = re.compile(r'^(\d+)(:\d+)?$')
# Process each comma-separated task:latency pair in turn.
for taskspec in string.split(re.sub(r'\s+', "",
string.replace(taskstr, "tasks", str(numtasks))),
","):
# Handle a trailing ellipsis.
if found_ellipsis:
self.errmsg.error_fatal('"..." may appear only at the end of a task hierarchy',
filename=self.backend_name)
if taskspec == "...":
# Repeat the previous task:latency pair until we cover all
# numtasks tasks.
found_ellipsis = 1
if tasks_cost_list == []:
self.errmsg.error_fatal('"..." may not appear at the beginning of a task hierarchy',
filename=self.backend_name)
while prev_tasks*task_factor < numtasks:
prev_tasks = prev_tasks * task_factor;
prev_latency = prev_latency + latency_delta
tasks_cost_list.append((prev_tasks, prev_latency))
continue
# Handle the common case, a task factor followed by an
# optional latency delta.
taskspec_match = taskspec_re.search(taskspec)
if not taskspec_match:
self.errmsg.error_fatal('Unable to parse "%s" (in hierarchy "%s")' % (taskspec, taskstr),
filename=self.backend_name)
task_factor = int(taskspec_match.group(1))
try:
latency_delta = int(taskspec_match.group(2)[1:])
except TypeError:
latency_delta = 1
if task_factor < 1:
self.errmsg.error_fatal('Task factor must be positive in "%s"' % taskspec,
filename=self.backend_name)
prev_tasks = prev_tasks * task_factor;
prev_latency = prev_latency + latency_delta
tasks_cost_list.append((prev_tasks, prev_latency))
# Append a catch-all case if necessary then return the final list.
if tasks_cost_list[-1][0] < numtasks:
tasks_cost_list.append((numtasks, tasks_cost_list[-1][1]+1))
return tasks_cost_list
def generate_initialize(self, ast, filesource='<stdin>', filetarget="-", sourcecode=None):
"Perform all of the initialization needed by the generate method."
# Define various parameters that depend upon filesource,
# sourcecode, and/or numtasks.
self.filesource = filesource # Input file
self.sourcecode = sourcecode # coNCePTuaL source code
self.errmsg = NCPTL_Error(filesource) # Error-handling methods
self.eventlist = map(lambda self: self.EventList(self), # Map from task to event list
[self] * int(self.numtasks))
self.msgqueue = map(lambda self: self.MessageQueue(self.errmsg), # Map from source task to message size to event list
[self] * int(self.numtasks))
self.pendingevents = map(lambda h: [], # Asynchronous events not yet waited for
[None] * int(self.numtasks))
self.timer_start = [0] * int(self.numtasks) # Logical time at which the timer started
self.counters = [] # Per-task counters exposed to programs
for task in range(0, self.numtasks):
self.counters.append({})
for varname in Variables.variables.keys():
self.counters[task][varname] = 0L
self.counters[task]["num_tasks"] = self.numtasks
self.counter_stack = map(lambda t: [], range(0, self.numtasks)) # Stack of counter values
if self.filesource == "<command line>":
self.filename = "a.out"
else:
self.filename = self.filesource
if filetarget == "-":
# If no target filename was specified, derive the log
# filename template from the source filename.
filebase = re.sub(r'\.ncptl$', "", self.filename)
else:
# If a target filename was specified, derive the log
# filename template from that.
filebase = os.path.splitext(filetarget)[0]
self.logfiletemplate = "%s-%%p.log" % os.path.basename(filebase)
self.options.extend([
["numtasks", "Number of tasks to use", "tasks", "T", 1L],
["mcastsync",
"Perform an implicit synchronization after a multicast (0=no; 1=yes)",
"mcastsync", "M", 0L],
["latency_list",
"Latency hierarchy as a comma-separated list of task_factor:latency_delta pairs",
"hierarchy", "H", "tasks:1"],
["random_seed", "Seed for the random-number generator",
"seed", "S", self.random_seed],
["kill_reps",
"If nonzero, perform FOR...REPETITIONS loop bodies exactly once",
"kill-reps", "K", 0L],
["logfiletmpl", "Log-file template", "logfile",
"L", self.logfiletemplate]])
def generate_finalize(self, ast, filesource='<stdin>', sourcecode=None):
"Perform all of the finalization needed by the generate method."
# Process all events in the event lists.
self.process_all_events()
# Cleanly shut down coNCePTuaL.
if self.program_uses_log_file:
for logstate in self.logstate.values():
ncptl_log_commit_data(logstate)
ncptl_log_write_epilogue(logstate)
ncptl_log_close(logstate)
ncptl_finalize()
def invoke_hook(self, hookname, localvars, alternatepy=None, alternate=None):
"""
Invoke a hook method if it exists, passing it a dictionary
of the current scope's local variables. The hook
function's required return type varies from hook to hook.
If the HOOKNAME method does not exist, evaluate ALTERNATEPY
and return the result. If ALTERNATEPY is not defined,
return ALTERNATE.
"""
hookmethod = getattr(self, hookname, None)
if hookmethod:
hookoutput = hookmethod(localvars)
if hookoutput:
return hookoutput
else:
return []
elif alternatepy:
return alternatepy(localvars)
else:
return alternate
def process_node(self, node):
"Given a node, invoke a method that knows how to process it."
# If the node has a constant value and is an expression (as
# opposed to a statement), reuse its previously calculated
# value.
if node.sem["is_constant"] and self.sub_statement_nodes.has_key(node.type):
try:
# Return our previously calculated value.
return node.previous_value
except AttributeError:
# This is the first time we're called. We'll need to
# calculate a value.
pass
# We need to compute a value for this node.
try:
result = self.type2method[node.type](node)
except KeyError:
methodname = "n_" + node.type
methodcode = getattr(self, methodname, self.n_undefined)
self.type2method[node.type] = methodcode
result = methodcode(node)
if node.sem["is_constant"]:
node.previous_value = result
return result
def apply_binary_function(self, ffunc, node, ifunc=None):
"""
Acquire a node's two children as either longs or floats
depending upon the value of self.context. Apply either
IFUNC or FFUNC, as appropriate. If FFUNC fails, return a
tuple consisting of FFUNC and the two operands. IFUNC
defaults to FFUNC.
"""
if len(node.kids) != 2:
self.errmsg.error_internal("Node %s has %d children, not 2" %
(node.type, len(node.kids)))
if self.context == "int":
value1 = long(self.process_node(node.kids[0]))
value2 = long(self.process_node(node.kids[1]))
if ifunc == None:
return ffunc(value1, value2)
else:
return ifunc(value1, value2)
else:
value1 = self.process_node(node.kids[0])
value2 = self.process_node(node.kids[1])
try:
return ffunc(float(value1), float(value2))
except TypeError:
return (ffunc, value1, value2)
def eval_lazy_expr(self, frag, wanttype=types.LongType):
"""Evaluate strings, longs, floats, and futures (really
tuples) in a given type context."""
conversion = {
(types.StringType, types.StringType) : lambda frag: frag,
(types.LongType, types.StringType) : lambda frag: str(frag),
(types.FloatType, types.StringType) : lambda frag: "%.10lg" % frag,
(types.StringType, types.LongType) : lambda frag: long(frag),
(types.LongType, types.LongType) : lambda frag: frag,
(types.FloatType, types.LongType) : lambda frag: long(frag),
(types.StringType, types.FloatType) : lambda frag: float(frag),
(types.LongType, types.FloatType) : lambda frag: float(frag),
(types.FloatType, types.FloatType) : lambda frag: frag}
try:
# Simple types
return conversion[type(frag), wanttype](frag)
except KeyError:
if type(frag) == types.TupleType:
# Future
func = frag[0]
arglist = []
for arg in frag[1:]:
arglist.append(self.eval_lazy_expr(arg, wanttype))
return self.eval_lazy_expr(apply(func, arglist), wanttype)
except:
pass
self.errmsg.error_internal('Unknown expression type %s for expression "%s"' % (str(type(frag)), frag))
def initialize_log_file(self, physrank):
"""Create a set of log files and write a prologue to each of.
Note that repeated calls will have no adverse effect."""
if self.logstate.has_key(physrank):
return
if not self.program_can_use_log_file:
return
ncptl_log_add_comment("Python version", re.sub(r'\s+', " ", sys.version))
for rank in range(self.numtasks):
self.logstate[rank] = ncptl_log_open(self.logfiletemplate, rank)
ncptl_log_write_prologue(self.logstate[rank],
sys.executable, self.logfile_uuid,
self.backend_name, self.backend_desc, self.numtasks,
self.options, len(self.options),
string.split(string.rstrip(self.sourcecode), "\n"))
self.program_uses_log_file = 1
def convert_to_tuple_list(self, messagelist):
"""
Convert all entries in a list of strings, numbers, and
(type, value) pairs to (type, value) pairs.
"""
result = []
for msgfrag in messagelist:
if type(msgfrag) == types.StringType:
result.append(("STRING", msgfrag))
elif type(msgfrag) == types.FloatType:
result.append(("NUMBER", msgfrag))
elif type(msgfrag) == types.TupleType:
result.append(msgfrag)
else:
self.errmsg.error_internal('Unexpected message fragment "%s"' % msgfrag)
return result
def format_plurals(self, format, number, args):
"Wrap the % operator with special cases for plurals."
if number == 1:
format = string.replace(format, "%S", "")
format = string.replace(format, "%W", "was")
format = re.sub(r'%C(?= [AEIOUaeiou])', "an", format)
format = string.replace(format, "%C", "a")
else:
format = string.replace(format, "%S", "s")
format = string.replace(format, "%W", "were")
format = string.replace(format, "%C", str(number))
return format % args
def update_counters(self, event, operation=None):
"Update various counter variables after an event completes."
if event.suppressed:
return
if operation == None:
operation = event.operation
counters = self.counters[event.task]
if operation == "SEND":
counters["msgs_sent"] = counters["msgs_sent"] + 1
counters["bytes_sent"] = counters["bytes_sent"] + event.msgsize
elif operation == "RECEIVE":
counters["msgs_received"] = counters["msgs_received"] + 1
counters["bytes_received"] = counters["bytes_received"] + event.msgsize
else:
self.errmsg.error_internal('Event type "%s" should have been either "SEND" or "RECEIVE"' % operation)
counters["total_bytes"] = counters["bytes_sent"] + counters["bytes_received"]
counters["total_msgs"] = counters["msgs_sent"] + counters["msgs_received"]
def _virtual_to_physical(self, vtask):
"Map a virtual task ID to a physical processor number."
if type(vtask) in (types.IntType, types.LongType):
return int(ncptl_virtual_to_physical(self.procmap, vtask))
elif type(vtask) in (types.ListType, types.TupleType):
ptasks = map(lambda t, self=self: self._virtual_to_physical(t), vtask)
if type(vtask) == types.TupleType:
ptasks = tuple(ptasks)
return ptasks
def push_event(self, event):
"""
Modify an event to use physical instead of virtual
ranks and introduce a suppression flag then push the
event onto the end of the appropriate event list.
"""
physrank = self._virtual_to_physical(event.task)
event.task = physrank
event.peers = self._virtual_to_physical(event.peers)
event.suppressed = self.suppress_output
self.eventlist[physrank].push(event)
def evaluate_for_each(self, for_each_node, expr_node):
"Evaluate an expression for each element in a list of ranges."
resulting_list = []
range_lists = self.process_node(for_each_node.kids[1])
self.scopes.insert(0, {})
for rlist in range_lists:
for var in rlist:
self.scopes[0][for_each_node.kids[0].attr] = var
if len(for_each_node.kids) < 3:
# Base case 1 -- evaluate the expression.
resulting_list.append(self.process_node(expr_node))
elif for_each_node.kids[2].type == "where_expr":
# Base case 2 -- conditionally evaluate the expression
condition = self.process_node(for_each_node.kids[2])
if condition:
resulting_list.append(self.process_node(expr_node))
else:
# Recursive case -- evaluate the next part of the
# list comprehension.
resulting_list.extend(self.evaluate_for_each(for_each_node.kids[2], expr_node))
self.scopes.pop(0)
return resulting_list
#---------------------------------#
# AST interpretation: relational #
# expressions (return true/false) #
#---------------------------------#
def n_rel_disj_expr(self, node):
"Return true if any of our children is true."
if len(node.kids) == 1:
return self.n_trivial_node(node)
else:
value1 = long(self.process_node(node.kids[0]))
if value1:
return 1L
else:
return long(self.process_node(node.kids[1]))
def n_rel_conj_expr(self, node):
"Return true only if all of our children are true."
if len(node.kids) == 1:
return self.n_trivial_node(node)
else:
value1 = long(self.process_node(node.kids[0]))
if value1:
return long(self.process_node(node.kids[1]))
else:
return 0L
def n_eq_expr(self, node):
"Compare our children's values."
attr2func = {
"op_eq": lambda a, b: a==b,
"op_ne": lambda a, b: a!=b,
"op_gt": lambda a, b: a>b,
"op_lt": lambda a, b: a<b,
"op_ge": lambda a, b: a>=b,
"op_le": lambda a, b: a<=b}
try:
return self.apply_binary_function(attr2func[node.attr], node)
except KeyError:
pass
if node.attr == "op_divides":
ifunc = lambda a, b: ncptl_func_modulo(b, a) == 0L
ffunc = lambda a, b: long(ncptl_dfunc_modulo(b, a)) == 0L
return self.apply_binary_function(ffunc, node, ifunc)
elif node.attr == "op_odd":
value = long(self.process_node(node.kids[0]))
return value % 2 != 0
elif node.attr == "op_even":
value = long(self.process_node(node.kids[0]))
return value % 2 == 0
elif node.attr == "op_in_range":
number = self.process_node(node.kids[0])
bounds = [self.process_node(node.kids[1]), self.process_node(node.kids[2])]
bounds.sort()
return bounds[0] <= number <= bounds[1]
elif node.attr == "op_not_in_range":
number = self.process_node(node.kids[0])
bounds = [self.process_node(node.kids[1]), self.process_node(node.kids[2])]
bounds.sort()
return not (bounds[0] <= number <= bounds[1])
elif node.attr == "op_in_range_list":
expression = self.process_node(node.kids[0])
rangelists = self.process_node(node.kids[1])
for rlist in rangelists:
if expression in rlist:
return 1
return 0
elif node.attr == "op_not_in_range_list":
expression = self.process_node(node.kids[0])
rangelists = self.process_node(node.kids[1])
for rlist in rangelists:
if expression in rlist:
return 0
return 1
else:
self.errmsg.error_internal('Unknown eq_expr "%s"' % node.attr)
def n_where_expr(self, node):
"Return our child's value."
return self.process_node(node.kids[0])
#--------------------------------#
# AST interpretation: arithmetic #
# expressions (return numbers) #
#--------------------------------#