docs: def write_abstract_dag(self):
      """
      Write all the nodes in the workflow to the DAX file.
      """
  
      # keep track of if we are using stampede at TACC
      using_stampede = False
  
      if not self.__dax_file_path:
        # this workflow is not dax-compatible, so don't write a dax
        return
  
      import Pegasus.DAX3
      # create the workflow object
      dax_name = os.path.split(self.__dax_file_path)[-1]
      dax_basename = '.'.join(dax_name.split('.')[0:-1])
      workflow = Pegasus.DAX3.ADAG( dax_basename )
  
      # we save the ID number to DAG node name mapping so that
      # we can easily write out the child/parent relationship
      # later
      node_job_object_dict = {}
  
      # FIXME disctionary of executables and pfns in the workflow
      # Pegasus should take care of this so we don't have to
      workflow_executable_dict = {}
      workflow_pfn_dict = {}
  
      # Add PFN caches for this workflow
      for pfn_tuple in self.get_pfn_cache():
          workflow_pfn_dict[pfn_tuple[0]] = pfn_tuple
  
      if self.get_pegasus_worker():
        # write the executable into the dax
        worker_package = Pegasus.DAX3.Executable(
          namespace="pegasus", name="worker",
          os="linux", arch="x86_64", installed=False)
        worker_package.addPFN(Pegasus.DAX3.PFN(self.get_pegasus_worker(),"local"))
        workflow_executable_dict['pegasus-pegasus_worker'] = worker_package
  
      # check for the pegasus-cluster package
      for path in os.environ["PATH"].split(":"):
        cluster_path = os.path.join(path,"pegasus-cluster")
        if os.path.exists(cluster_path):
          # and add to the dax if it exists
          seqexec_package = Pegasus.DAX3.Executable(
            namespace="pegasus", name="seqexec",
            os="linux", arch="x86_64", installed=True)
          seqexec_package.addPFN(Pegasus.DAX3.PFN(cluster_path,"local"))
          workflow_executable_dict['pegasus-pegasus_seqexec'] = seqexec_package
  
      id = 0
      for node in self.__nodes:
        if self.is_dax() and isinstance(node, LSCDataFindNode):
          pass
  
        elif isinstance(node.job(), CondorDAGManJob):
          id += 1
          id_tag = "ID%06d" % id
          node_name = node._CondorDAGNode__name
  
          if node.job().get_dax() is None:
            # write this node as a sub-dag
            subdag_name = os.path.split(node.job().get_dag())[-1]
            try:
              subdag_exec_path = os.path.join(
                os.getcwd(),node.job().get_dag_directory())
            except AttributeError:
              subdag_exec_path = os.getcwd()
  
            subdag = Pegasus.DAX3.DAG(subdag_name,id=id_tag)
            subdag.addProfile(Pegasus.DAX3.Profile("dagman","DIR",subdag_exec_path))
  
            subdag_file = Pegasus.DAX3.File(subdag_name)
            subdag_file.addPFN(Pegasus.DAX3.PFN(os.path.join(subdag_exec_path,subdag_name),"local"))
            workflow.addFile(subdag_file)
            workflow.addDAG(subdag)
            node_job_object_dict[node_name] = subdag
  
          else:
            # write this node as a sub-dax
            subdax_name = os.path.split(node.job().get_dax())[-1]
            dax_subdir = node.job().get_dag_directory()
            if dax_subdir:
              subdax_path = os.path.join(
                os.getcwd(),node.job().get_dag_directory(),subdax_name)
            else:
              subdax_path = os.path.join(os.getcwd(),subdax_name)
              dax_subdir = '.'
  
            subdax = Pegasus.DAX3.DAX(subdax_name,id=id_tag)
  
            # FIXME pegasus should ensure these are unique
            for pfn_tuple in node.job().get_pfn_cache():
              workflow_pfn_dict[pfn_tuple[0]] = pfn_tuple
  
            # set the storage, execute, and output directory locations
            pegasus_args = """--dir %s """ % dax_subdir
            pegasus_args += """--output-dir %s """ % dax_subdir
  
            # set the maxjobs categories for the subdax
            # FIXME pegasus should expose this in the dax, so it can
            # be handled like the MAXJOBS keyword in dag files
            for maxjobcat in node.get_maxjobs_categories():
              pegasus_args += "-Dpegasus.dagman." + maxjobcat[0] + ".maxjobs=" + maxjobcat[1] + " "
  
            if not self.is_dax():
              pegasus_args += "--nocleanup "
  
            if node.get_cluster_jobs():
              pegasus_args += "--cluster " + node.get_cluster_jobs() + " "
  
            if node.get_reduce_dax() is False:
              pegasus_args += " --force "
  
            if node.get_static_pfn_cache():
              pegasus_args += " --cache " + node.get_static_pfn_cache() + " "
  
            pegasus_args += "--output-site local -vvvvvv"
            subdax.addArguments(pegasus_args)
  
            subdax_file = Pegasus.DAX3.File(subdax_name)
            subdax_file.addPFN(Pegasus.DAX3.PFN(subdax_path,"local"))
            workflow.addFile(subdax_file)
            workflow.addDAX(subdax)
            node_job_object_dict[node_name] = subdax
  
        else:
          # write this job as a regular node
          executable = node.job()._CondorJob__executable
          node_name = node._CondorDAGNode__name
  
          id += 1
          id_tag = "ID%06d" % id
          node_job_object_dict[node_name] = id_tag
  
          # get the name of the executable
          executable_namespace = 'ligo-' + str(node.job().__class__.__name__).lower()
          executable_base = os.path.basename(executable)
  
          workflow_job = Pegasus.DAX3.Job( namespace=executable_namespace,
            name=executable_base, version="1.0", id=id_tag)
  
          cmd_line = node.get_cmd_tuple_list()
  
          # loop through all filenames looking for them in the command
          # line so that they can be replaced appropriately by xml tags
          input_node_file_dict = {}
          for f in node.get_input_files():
            input_node_file_dict[f] = 1
  
          for f in input_node_file_dict.keys():
            workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.INPUT,register=False,transfer=True)
  
          output_node_file_dict = {}
          for f in node.get_output_files():
            output_node_file_dict[f] = 1
  
          checkpoint_node_file_dict = {}
          for f in node.get_checkpoint_files():
              checkpoint_node_file_dict[f] = 1
  
          for f in output_node_file_dict.keys():
            workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.OUTPUT,register=False,transfer=True)
  
          for f in checkpoint_node_file_dict.keys():
            workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.CHECKPOINT,register=False,transfer=True)
  
          node_file_dict = dict( input_node_file_dict.items() + output_node_file_dict.items() + checkpoint_node_file_dict.items() )
  
          for job_arg in cmd_line:
            try:
              if node_file_dict.has_key(job_arg[0]):
                workflow_job.addArguments(Pegasus.DAX3.File(os.path.basename(job_arg[0])))
              elif node_file_dict.has_key(job_arg[1]):
                workflow_job.addArguments(job_arg[0], Pegasus.DAX3.File(os.path.basename(job_arg[1])))
              elif len(job_arg[1].split(' ')) != 1:
                  args = [job_arg[0]]
                  for arg in job_arg[1].split(' '):
                      if node_file_dict.has_key(arg):
                          args.append(Pegasus.DAX3.File(os.path.basename(arg)))
                      else:
                          args.append(arg)
                  workflow_job.addArguments(*args)
              else:
                workflow_job.addArguments(job_arg[0], job_arg[1])
            except IndexError:
              pass
  
          # Check for desired grid site
          if node.job().get_grid_site():
              this_grid_site = node.job().get_grid_site()
              workflow_job.addProfile(Pegasus.DAX3.Profile('hints','execution.site',this_grid_site))
              if this_grid_site == 'stampede-dev' or this_grid_site=='stampede':
                using_stampede = True
  
          # write the executable into the dax
          job_executable = Pegasus.DAX3.Executable(
            namespace=executable_namespace,
            name=executable_base, version="1.0",
            os="linux", arch="x86_64",
            installed=node.job().get_executable_installed())
  
          executable_path = os.path.join(os.getcwd(),executable)
          job_executable.addPFN(Pegasus.DAX3.PFN(executable_path,"local"))
  
          workflow_executable_dict[executable_namespace + executable_base] = job_executable
  
          # write the mpi cluster parameter for the job
          if node.job().get_dax_mpi_cluster():
            workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","job.aggregator","mpiexec"))
            workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","clusters.size",str(node.job().get_dax_mpi_cluster())))
  
          # write the grid start parameter for this node
          # if the grid start is not None
          if node.get_grid_start():
            workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","gridstart",node.get_grid_start()))
  
          # write the bundle parameter if this node has one
          if node.get_dax_collapse():
            workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","clusters.size",str(node.get_dax_collapse())))
  
          # write number of times the node should be retried
          if node.get_retry():
            workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","retry",str(node.get_retry())))
  
          # write the post script for this node
          if node.get_post_script():
            post_script_base = os.path.basename(node.get_post_script())
            post_script_path = os.path.join(os.getcwd(),node.get_post_script())
            workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post",post_script_base))
            workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post.path." + post_script_base,post_script_path))
  
          # write the post script for this node
          if node.get_post_script_arg():
            workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post.arguments",' '.join(node.get_post_script_arg())))
  
          # write the dag node category if this node has one
          if node.get_category():
            workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","category",str(node.get_category())))
  
          # write the dag node priority if this node has one
          if node.get_priority():
            workflow_job.addProfile(Pegasus.DAX3.Profile("condor","priority",str(node.get_priority())))
  
          # write the universe that this job should run in to the dax
          if node.get_dax_collapse():
            # collapsed jobs must run in the vanilla universe
            workflow_job.addProfile(Pegasus.DAX3.Profile("condor","universe","vanilla"))
          else:
            workflow_job.addProfile(Pegasus.DAX3.Profile("condor","universe",node.job().get_universe()))
  
          # add any other user specified condor commands or classads
          for p in node.get_pegasus_profile():
              workflow_job.addProfile(Pegasus.DAX3.Profile(p[0],p[1],p[2]))
  
          # finally add this job to the workflow
          workflow.addJob(workflow_job)
          node_job_object_dict[node_name] = workflow_job
  
  
      # print parent-child relationships to DAX
      for node in self.__nodes:
        if self.is_dax() and isinstance(node, LSCDataFindNode):
          pass
        elif self.is_dax() and ( len(node._CondorDAGNode__parents) == 1 ) and isinstance(node._CondorDAGNode__parents[0], LSCDataFindNode):
          pass
        else:
          child_job_object = node_job_object_dict[str(node)]
          if node._CondorDAGNode__parents:
            for parent in node._CondorDAGNode__parents:
              if self.is_dax() and isinstance(parent, LSCDataFindNode):
                pass
              else:
                parent_job_object = node_job_object_dict[str(parent)]
                workflow.addDependency(Pegasus.DAX3.Dependency(parent=parent_job_object, child=child_job_object))
  
      # FIXME put all the executables in the workflow
      for exec_key in workflow_executable_dict.keys():
        workflow.addExecutable(workflow_executable_dict[exec_key])
  
      # FIXME if we are running on stampede, add the mpi wrapper job
      if using_stampede:
        prod_mpiexec = Pegasus.DAX3.Executable(namespace="pegasus",
          name="mpiexec", os="linux", arch="x86_64", installed="true")
        prod_mpiexec.addPFN(Pegasus.DAX3.PFN("file:///home1/02796/dabrown/bin/mpi-cluster-wrapper-impi.sh","stampede"))
        workflow.addExecutable(prod_mpiexec)
  
        dev_mpiexec = Pegasus.DAX3.Executable(namespace="pegasus",
          name="mpiexec", os="linux", arch="x86_64", installed="true")
        dev_mpiexec.addPFN(Pegasus.DAX3.PFN("file:///home1/02796/dabrown/bin/mpi-cluster-wrapper-impi.sh","stampede-dev"))
        workflow.addExecutable(dev_mpiexec)
  
      # FIXME put all the pfns in the workflow
      for pfn_key in workflow_pfn_dict.keys():
        f = Pegasus.DAX3.File(workflow_pfn_dict[pfn_key][0])
        f.addPFN(Pegasus.DAX3.PFN(workflow_pfn_dict[pfn_key][1],workflow_pfn_dict[pfn_key][2]))
        workflow.addFile(f)
  
      f = open(self.__dax_file_path,"w")
      workflow.writeXML(f)
      f.close()