| The Gaudi Framework v32r1 (f65d50dc)

Public Member Functions | Public Attributes | List of all members
GaudiMP.GMPBase.Coord Class Reference
Inheritance diagram for GaudiMP.GMPBase.Coord:
[legend]
Collaboration diagram for GaudiMP.GMPBase.Coord:
[legend]

## Public Member Functions
def | __init__ (self, nWorkers, config, log)
def | getSyncEvents (self, nodeID)
def | getQueues (self, nodeID)
def | Go (self)
def | Terminate (self)
def | Stop (self)
def | SetupQueues (self)

## Public Attributes
| log
| config
| nWorkers
| qs
| hq
| fq
| sInit
| sRun
| sFin
| histSyncEvent
| subworkers
| reader
| workers
| writer
| system

## Detailed Description

Definition at line 1337 of file GMPBase.py.

## Constructor & Destructor Documentation

## ◆ __init__()

def GaudiMP.GMPBase.Coord.__init__ | ( |  | self,
|  |  | nWorkers,
|  |  | config,
|  |  | log
| ) |  |

Definition at line 1338 of file GMPBase.py.

1338 def __init__(self, nWorkers, config, log):
1339
1340 self.log = log
1341 self.config = config
1342 # set up Logging
1343 self.log.name = 'GaudiPython-Parallel-Logger'
1344 self.log.info('GaudiPython Parallel Process Co-ordinator beginning')
1345
1346 if nWorkers == -1:
1347 # The user has requested all available cpus in the machine
1348 self.nWorkers = cpu_count()
1349 else:
1350 self.nWorkers = nWorkers
1351
1352 self.qs = self.SetupQueues() # a dictionary of queues (for Events)
1353 self.hq = JoinableQueue() # for Histogram data
1354 self.fq = JoinableQueue() # for FileRecords data
1355
1356 # Make a Syncer for Initalise, Run, and Finalise
1357 self.sInit = Syncer(
1358 self.nWorkers,
1359 self.log,
1360 limit=WAIT_INITIALISE,
1361 step=STEP_INITIALISE)
1362 self.sRun = Syncer(
1363 self.nWorkers,
1364 self.log,
1365 manyEvents=True,
1366 limit=WAIT_SINGLE_EVENT,
1367 step=STEP_EVENT,
1368 firstEvent=WAIT_FIRST_EVENT)
1369 self.sFin = Syncer(
1370 self.nWorkers, self.log, limit=WAIT_FINALISE, step=STEP_FINALISE)
1371 # and one final one for Histogram Transfer
1372 self.histSyncEvent = Event()
1373
1374 # params are common to al subprocesses
1375 params = (self.nWorkers, self.histSyncEvent, self.config, self.log)
1376
1377 self.subworkers = []
1378 # Declare SubProcesses!
1379 for i in range(1, self.nWorkers):
1380 sub = Subworker(i, self.getQueues(i), self.getSyncEvents(i),
1381 params, self.subworkers)
1382 self.subworkers.append(sub)
1383 self.reader = Reader(
1384 self.getQueues(-1), self.getSyncEvents(-1), params,
1385 self.subworkers)
1386 self.workers = []
1387 wk = Worker(0, self.getQueues(0), self.getSyncEvents(0), params,
1388 self.subworkers)
1389 self.writer = Writer(
1390 self.getQueues(-2), self.getSyncEvents(-2), params,
1391 self.subworkers)
1392
1393 self.system = []
1394 self.system.append(self.writer)
1395 self.system.append(wk)
1396 self.system.append(self.reader)
1397
Gaudi::Functional::details::zip::range
decltype(auto) range(Args &&... args)
Zips multiple containers together to form a single range.
Definition: FunctionalDetails.h:71
compareOutputFiles.Reader
def Reader(readerType, filename, qacross, qToEngine)
Definition: compareOutputFiles.py:126

## Member Function Documentation

## ◆ getQueues()

def GaudiMP.GMPBase.Coord.getQueues | ( |  | self,
|  |  | nodeID
| ) |  |

Definition at line 1404 of file GMPBase.py.

1404 def getQueues(self, nodeID):
1405 eventQ = self.qs[nodeID]
1406 histQ = self.hq
1407 fsrQ = self.fq
1408 return (eventQ, histQ, fsrQ)
1409

## ◆ getSyncEvents()

def GaudiMP.GMPBase.Coord.getSyncEvents | ( |  | self,
|  |  | nodeID
| ) |  |

Definition at line 1398 of file GMPBase.py.

1398 def getSyncEvents(self, nodeID):
1399 init = self.sInit.d[nodeID].event
1400 run = (self.sRun.d[nodeID].event, self.sRun.d[nodeID].lastEvent)
1401 fin = self.sFin.d[nodeID].event
1402 return (init, run, fin)
1403

## ◆ Go()

def GaudiMP.GMPBase.Coord.Go | ( |  | self | ) |

Definition at line 1410 of file GMPBase.py.

1410 def Go(self):
1411
1412 # Initialise
1413 self.log.name = 'GaudiPython-Parallel-Logger'
1414 self.log.info('INITIALISING SYSTEM')
1415
1416 # Start reader, writer and main worker
1417 for p in self.system:
1418 p.Start()
1419
1420 sc = self.sInit.syncAll(step="Initialise")
1421 if sc == SUCCESS:
1422 pass
1423 else:
1424 self.Terminate()
1425 return FAILURE
1426
1427 # Run
1428 self.log.name = 'GaudiPython-Parallel-Logger'
1429 self.log.info('RUNNING SYSTEM')
1430 sc = self.sRun.syncAll(step="Run")
1431 if sc == SUCCESS:
1432 pass
1433 else:
1434 self.Terminate()
1435 return FAILURE
1436
1437 # Finalise
1438 self.log.name = 'GaudiPython-Parallel-Logger'
1439 self.log.info('FINALISING SYSTEM')
1440 sc = self.sFin.syncAll(step="Finalise")
1441 if sc == SUCCESS:
1442 pass
1443 else:
1444 self.Terminate()
1445 return FAILURE
1446
1447 # if we've got this far, finally report SUCCESS
1448 self.log.info("Cleanly join all Processes")
1449 self.Stop()
1450 self.log.info("Report Total Success to Main.py")
1451 return SUCCESS
1452

## ◆ SetupQueues()

def GaudiMP.GMPBase.Coord.SetupQueues | ( |  | self | ) |

Definition at line 1470 of file GMPBase.py.

1470 def SetupQueues(self):
1471 # This method will set up the network of Queues needed
1472 # N Queues = nWorkers + 1
1473 # Each Worker has a Queue in, and a Queue out
1474 # Reader has Queue out only
1475 # Writer has nWorkers Queues in
1476
1477 # one queue from Reader-Workers
1478 rwk = JoinableQueue()
1479 # one queue from each worker to writer
1480 workersWriter = [JoinableQueue() for i in xrange(self.nWorkers)]
1481 d = {}
1482 d[-1] = (None, rwk) # Reader
1483 d[-2] = (workersWriter, None) # Writer
1484 for i in xrange(self.nWorkers):
1485 d[i] = (rwk, workersWriter[i])
1486 return d
1487
1488
1489 # ============================= EOF ===========================================

## ◆ Stop()

def GaudiMP.GMPBase.Coord.Stop | ( |  | self | ) |

Definition at line 1463 of file GMPBase.py.

1463 def Stop(self):
1464 # procs should be joined in reverse order to launch
1465 self.system.reverse()
1466 for s in self.system:
1467 s.proc.join()
1468 return SUCCESS
1469
reverse
::details::reverse_wrapper< T > reverse(T &&iterable)
Definition: reverse.h:49

## ◆ Terminate()

def GaudiMP.GMPBase.Coord.Terminate | ( |  | self | ) |

Definition at line 1453 of file GMPBase.py.

1453 def Terminate(self):
1454 # Brutally kill sub-processes
1455 children = multiprocessing.active_children()
1456 for i in children:
1457 i.terminate()
1458
1459 # self.writer.proc.terminate()
1460 #[ w.proc.terminate() for w in self.workers]
1461 # self.reader.proc.terminate()
1462

## Member Data Documentation

## ◆ config

GaudiMP.GMPBase.Coord.config

Definition at line 1341 of file GMPBase.py.

## ◆ fq

GaudiMP.GMPBase.Coord.fq

Definition at line 1354 of file GMPBase.py.

## ◆ histSyncEvent

GaudiMP.GMPBase.Coord.histSyncEvent

Definition at line 1372 of file GMPBase.py.

## ◆ hq

GaudiMP.GMPBase.Coord.hq

Definition at line 1353 of file GMPBase.py.

## ◆ log

GaudiMP.GMPBase.Coord.log

Definition at line 1340 of file GMPBase.py.

## ◆ nWorkers

GaudiMP.GMPBase.Coord.nWorkers

Definition at line 1348 of file GMPBase.py.

## ◆ qs

GaudiMP.GMPBase.Coord.qs

Definition at line 1352 of file GMPBase.py.

## ◆ reader

GaudiMP.GMPBase.Coord.reader

Definition at line 1383 of file GMPBase.py.

## ◆ sFin

GaudiMP.GMPBase.Coord.sFin

Definition at line 1369 of file GMPBase.py.

## ◆ sInit

GaudiMP.GMPBase.Coord.sInit

Definition at line 1357 of file GMPBase.py.

## ◆ sRun

GaudiMP.GMPBase.Coord.sRun

Definition at line 1362 of file GMPBase.py.

## ◆ subworkers

GaudiMP.GMPBase.Coord.subworkers

Definition at line 1377 of file GMPBase.py.

## ◆ system

GaudiMP.GMPBase.Coord.system

Definition at line 1393 of file GMPBase.py.

## ◆ workers

GaudiMP.GMPBase.Coord.workers

Definition at line 1386 of file GMPBase.py.

## ◆ writer

GaudiMP.GMPBase.Coord.writer

Definition at line 1389 of file GMPBase.py.

The documentation for this class was generated from the following file:

- GaudiMP/python/GaudiMP/GMPBase.py