Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def as_graph(processes, settings):
"""
Return a graph from a list of process objects.
"""
logger.info('Building graph ...')
from altgraph.Graph import Graph
graph = Graph()
##############################################################################
# TODO: CRITICAL to avoid cycles in the graph and ensure this is DAG
# directed acyclic graph: http://en.wikipedia.org/wiki/Directed_acyclic_graph
# we should take timing into consideration:
# the first read to a given path can only happen after the last write has been
# completed, IFF there are writes that precede a given read
# therefore a path may appear several times in the graph based on the
# read/writes
# this applies to actual files, but be may not to pipes or sockets
##############################################################################
for proc in processes:
if proc.is_empty():
continue
# FIXME: handle forks
# if not settings.forks and proc.is_pure_forker():
# continue
def __init__(self, graph=None, debug=0):
if graph is None:
graph = Graph()
self.graphident = self
self.graph = graph
self.debug = debug
self.indent = 0
graph.add_node(self, None)
def __init__(self, graph=None, debug=0):
if graph is None:
graph = Graph()
self.graphident = self
self.graph = graph
self.debug = debug
self.indent = 0
graph.add_node(self, None)