The Gaudi Framework  v41r0 (bdbe7744)
Loading...
Searching...
No Matches
GaudiMP.GMPBase.Coord Class Reference

Public Member Functions

 __init__ (self, nWorkers, config, log)
 getSyncEvents (self, nodeID)
 getQueues (self, nodeID)
 Go (self)
 Terminate (self)
 Stop (self)
 SetupQueues (self)

Public Attributes

 log = log
 config = config
 nWorkers = cpu_count()
 qs = self.SetupQueues()
 hq = JoinableQueue()
 fq = JoinableQueue()
 sInit
 sRun
 sFin
 histSyncEvent = Event()
list subworkers = []
 reader
list workers = []
 writer
list system = []

Detailed Description

Definition at line 1319 of file GMPBase.py.

Constructor & Destructor Documentation

◆ __init__()

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

Definition at line 1320 of file GMPBase.py.

1320 def __init__(self, nWorkers, config, log):
1321 self.log = log
1322 self.config = config
1323 # set up Logging
1324 self.log.name = "GaudiPython-Parallel-Logger"
1325 self.log.info("GaudiPython Parallel Process Co-ordinator beginning")
1326
1327 if nWorkers == -1:
1328 # The user has requested all available cpus in the machine
1329 self.nWorkers = cpu_count()
1330 else:
1331 self.nWorkers = nWorkers
1332
1333 self.qs = self.SetupQueues() # a dictionary of queues (for Events)
1334 self.hq = JoinableQueue() # for Histogram data
1335 self.fq = JoinableQueue() # for FileRecords data
1336
1337 # Make a Syncer for Initalise, Run, and Finalise
1338 self.sInit = Syncer(
1339 self.nWorkers, self.log, limit=WAIT_INITIALISE, step=STEP_INITIALISE
1340 )
1341 self.sRun = Syncer(
1342 self.nWorkers,
1343 self.log,
1344 manyEvents=True,
1345 limit=WAIT_SINGLE_EVENT,
1346 step=STEP_EVENT,
1347 firstEvent=WAIT_FIRST_EVENT,
1348 )
1349 self.sFin = Syncer(
1350 self.nWorkers, self.log, limit=WAIT_FINALISE, step=STEP_FINALISE
1351 )
1352 # and one final one for Histogram Transfer
1353 self.histSyncEvent = Event()
1354
1355 # params are common to al subprocesses
1356 params = (self.nWorkers, self.histSyncEvent, self.config, self.log)
1357
1358 self.subworkers = []
1359 # Declare SubProcesses!
1360 for i in range(1, self.nWorkers):
1361 sub = Subworker(
1362 i, self.getQueues(i), self.getSyncEvents(i), params, self.subworkers
1363 )
1364 self.subworkers.append(sub)
1365 self.reader = Reader(
1366 self.getQueues(-1), self.getSyncEvents(-1), params, self.subworkers
1367 )
1368 self.workers = []
1369 wk = Worker(
1370 0, self.getQueues(0), self.getSyncEvents(0), params, self.subworkers
1371 )
1372 self.writer = Writer(
1373 self.getQueues(-2), self.getSyncEvents(-2), params, self.subworkers
1374 )
1375
1376 self.system = []
1377 self.system.append(self.writer)
1378 self.system.append(wk)
1379 self.system.append(self.reader)
1380

Member Function Documentation

◆ getQueues()

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

Definition at line 1387 of file GMPBase.py.

1387 def getQueues(self, nodeID):
1388 eventQ = self.qs[nodeID]
1389 histQ = self.hq
1390 fsrQ = self.fq
1391 return (eventQ, histQ, fsrQ)
1392

◆ getSyncEvents()

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

Definition at line 1381 of file GMPBase.py.

1381 def getSyncEvents(self, nodeID):
1382 init = self.sInit.d[nodeID].event
1383 run = (self.sRun.d[nodeID].event, self.sRun.d[nodeID].lastEvent)
1384 fin = self.sFin.d[nodeID].event
1385 return (init, run, fin)
1386

◆ Go()

GaudiMP.GMPBase.Coord.Go ( self)

Definition at line 1393 of file GMPBase.py.

1393 def Go(self):
1394 # Initialise
1395 self.log.name = "GaudiPython-Parallel-Logger"
1396 self.log.info("INITIALISING SYSTEM")
1397
1398 # Start reader, writer and main worker
1399 for p in self.system:
1400 p.Start()
1401
1402 sc = self.sInit.syncAll(step="Initialise")
1403 if sc == SUCCESS:
1404 pass
1405 else:
1406 self.Terminate()
1407 return FAILURE
1408
1409 # Run
1410 self.log.name = "GaudiPython-Parallel-Logger"
1411 self.log.info("RUNNING SYSTEM")
1412 sc = self.sRun.syncAll(step="Run")
1413 if sc == SUCCESS:
1414 pass
1415 else:
1416 self.Terminate()
1417 return FAILURE
1418
1419 # Finalise
1420 self.log.name = "GaudiPython-Parallel-Logger"
1421 self.log.info("FINALISING SYSTEM")
1422 sc = self.sFin.syncAll(step="Finalise")
1423 if sc == SUCCESS:
1424 pass
1425 else:
1426 self.Terminate()
1427 return FAILURE
1428
1429 # if we've got this far, finally report SUCCESS
1430 self.log.info("Cleanly join all Processes")
1431 self.Stop()
1432 self.log.info("Report Total Success to Main.py")
1433 return SUCCESS
1434

◆ SetupQueues()

GaudiMP.GMPBase.Coord.SetupQueues ( self)

Definition at line 1452 of file GMPBase.py.

1452 def SetupQueues(self):
1453 # This method will set up the network of Queues needed
1454 # N Queues = nWorkers + 1
1455 # Each Worker has a Queue in, and a Queue out
1456 # Reader has Queue out only
1457 # Writer has nWorkers Queues in
1458
1459 # one queue from Reader-Workers
1460 rwk = JoinableQueue()
1461 # one queue from each worker to writer
1462 workersWriter = [JoinableQueue() for i in range(self.nWorkers)]
1463 d = {}
1464 d[-1] = (None, rwk) # Reader
1465 d[-2] = (workersWriter, None) # Writer
1466 for i in range(self.nWorkers):
1467 d[i] = (rwk, workersWriter[i])
1468 return d
1469
1470
1471# ============================= EOF ===========================================

◆ Stop()

GaudiMP.GMPBase.Coord.Stop ( self)

Definition at line 1445 of file GMPBase.py.

1445 def Stop(self):
1446 # procs should be joined in reverse order to launch
1447 self.system.reverse()
1448 for s in self.system:
1449 s.proc.join()
1450 return SUCCESS
1451
reverse
::details::reverse_wrapper< T > reverse(T &&iterable)
Definition reverse.h:58

◆ Terminate()

GaudiMP.GMPBase.Coord.Terminate ( self)

Definition at line 1435 of file GMPBase.py.

1435 def Terminate(self):
1436 # Brutally kill sub-processes
1437 children = multiprocessing.active_children()
1438 for i in children:
1439 i.terminate()
1440
1441 # self.writer.proc.terminate()
1442 # [ w.proc.terminate() for w in self.workers]
1443 # self.reader.proc.terminate()
1444

Member Data Documentation

◆ config

GaudiMP.GMPBase.Coord.config = config

Definition at line 1322 of file GMPBase.py.

◆ fq

GaudiMP.GMPBase.Coord.fq = JoinableQueue()

Definition at line 1335 of file GMPBase.py.

◆ histSyncEvent

GaudiMP.GMPBase.Coord.histSyncEvent = Event()

Definition at line 1353 of file GMPBase.py.

◆ hq

GaudiMP.GMPBase.Coord.hq = JoinableQueue()

Definition at line 1334 of file GMPBase.py.

◆ log

GaudiMP.GMPBase.Coord.log = log

Definition at line 1321 of file GMPBase.py.

◆ nWorkers

GaudiMP.GMPBase.Coord.nWorkers = cpu_count()

Definition at line 1329 of file GMPBase.py.

◆ qs

GaudiMP.GMPBase.Coord.qs = self.SetupQueues()

Definition at line 1333 of file GMPBase.py.

◆ reader

GaudiMP.GMPBase.Coord.reader
Initial value:
= Reader(
self.getQueues(-1), self.getSyncEvents(-1), params, self.subworkers
)

Definition at line 1365 of file GMPBase.py.

◆ sFin

GaudiMP.GMPBase.Coord.sFin
Initial value:
= Syncer(
self.nWorkers, self.log, limit=WAIT_FINALISE, step=STEP_FINALISE
)

Definition at line 1349 of file GMPBase.py.

◆ sInit

GaudiMP.GMPBase.Coord.sInit
Initial value:
= Syncer(
self.nWorkers, self.log, limit=WAIT_INITIALISE, step=STEP_INITIALISE
)

Definition at line 1338 of file GMPBase.py.

◆ sRun

GaudiMP.GMPBase.Coord.sRun
Initial value:
= Syncer(
self.nWorkers,
self.log,
manyEvents=True,
limit=WAIT_SINGLE_EVENT,
step=STEP_EVENT,
firstEvent=WAIT_FIRST_EVENT,
)

Definition at line 1341 of file GMPBase.py.

◆ subworkers

list GaudiMP.GMPBase.Coord.subworkers = []

Definition at line 1358 of file GMPBase.py.

◆ system

list GaudiMP.GMPBase.Coord.system = []

Definition at line 1376 of file GMPBase.py.

◆ workers

list GaudiMP.GMPBase.Coord.workers = []

Definition at line 1368 of file GMPBase.py.

◆ writer

GaudiMP.GMPBase.Coord.writer
Initial value:
= Writer(
self.getQueues(-2), self.getSyncEvents(-2), params, self.subworkers
)

Definition at line 1372 of file GMPBase.py.

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