<commit_before>// Copyright 2015 <NAME>. All Rights Reserved.
// Author: <EMAIL> (<NAME>)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package dag

import (
	"errors"
	"fmt"
	"log"
	"sync"

	"github.com/jacobsa/syncutil"

	"golang.org/x/net/context"
)

// Call the visitor once for each unique node in the union of startNodes and
// all of its transitive dependencies, with bounded parallelism.
//
// Guarantees:
//
//  *  If the graph contains a cycle, this function will not succeed.
//
//  *  If a node N depends on a node M, v.Visit(N) will be called only after
//     v.Visit(M) returns successfully.
//
//  *  For each unique node N, dr.FindDependencies(N) and v.Visit(N) will each
//     be called at most once. Moreover, v.Visit(N) will be called only after
//     dr.FindDependencies(N) returns successfully.
//
func Visit(
	ctx context.Context,
	startNodes []Node,
	dr DependencyResolver,
	v Visitor,
	parallelism int) (err error) {
	err = errors.New("TODO")
	return
}

// Given
//
// *   a channel whose contents are a topologically-sorted list of the unique
//     nodes in a DAG (i.e. a node appears only after all of its predecessors
//     have) and
//
// *   a successor finder that agrees with the topological sort about the
//     structure of the graph,
//
// invoke the supplied visitor once for each node in the graph with, bounded
// parallelism. The visitor will be called for a node N only after it has
// returned success for all of N's predecessors.
//
// The successor finder may be called multiple times per node, and must be
// idempotent.
func TraverseDAG(
	ctx context.Context,
	nodes <-chan Node,
	sf SuccessorFinder,
	v Visitor,
	parallelism int) (err error) {
	b := syncutil.NewBundle(ctx)

	// Set up a state struct.
	state := &traverseDAGState{
		notReadyToVisit: make(map[Node]traverseDAGNodeState),

		// One, for processIncomingNodes, to prevent visitNodes from returning
		// immediately. processIncomingNodes will mark itself as no longer busy
		// when it returns.
		busyWorkers: 1,
	}

	state.mu = syncutil.NewInvariantMutex(state.checkInvariants)
	state.cond.L = &state.mu

	// Process incoming nodes from the user.
	b.Add(func(ctx context.Context) (err error) {
		err = processIncomingNodes(ctx, nodes, sf, state)
		if err != nil {
			err = fmt.Errorf("processIncomingNodes: %v", err)
			return
		}

		return
	})

	// Run multiple workers to deal with the nodes that are ready to visit.
	for i := 0; i < parallelism; i++ {
		b.Add(func(ctx context.Context) (err error) {
			err = visitNodes(ctx, sf, v, state)
			if err != nil {
				err = fmt.Errorf("visitNodes: %v", err)
				return
			}

			return
		})
	}

	// Join the bundle, but use the explicitly tracked first worker error in
	// order to circumvent the following race:
	//
	//  *  Worker A encounters an error, sets firstErr, and returns.
	//
	//  *  Worker B wakes up, sees firstErr, and returns with a junk follow-on
	//     error.
	//
	//  *  The bundle observes worker B's error before worker A's.
	//
	b.Join()
	state.mu.Lock()
	err = state.firstErr
	state.mu.Unlock()

	return
}

////////////////////////////////////////////////////////////////////////
// nodeInfo
////////////////////////////////////////////////////////////////////////

type nodeState int

// States in which a node involved in a call to Visit may be.
const (
	// The dependency resolver has not yet been called for this node, or a call
	// is currently in progress.
	state_DependenciesUnresolved nodeState = iota

	// The dependencies of this node have been resolved, but they have not yet
	// all been visited.
	state_DependenciesUnsatisfied

	// This node is eligible to be visited, but the visitor has not yet been
	// called or a call is currently in progress.
	state_Unvisited

	// The visitor has been called for this node and returned successfully.
	state_Visited
)

type nodeInfo struct {
	node  Node
	state nodeState

	// The number of unsatisfied dependencies remaining for this node.
	//
	// INVARIANT: unsatisfied >= 0
	// INVARIANT: unsatisfied > 0 iff state == state_DependenciesUnsatisfied
	unsatisfied int64

	// The set of unsatisfied nodes for which this node is a blocker.
	//
	// INVARIANT: len(dependants) > 0 implies state < state_Unvisited.
	// INVARIANT: For each n, n.state == state_DependenciesUnsatisfied
	dependants []*nodeInfo
}

////////////////////////////////////////////////////////////////////////
// visitState
////////////////////////////////////////////////////////////////////////

type visitState struct {
	mu syncutil.InvariantMutex

	// All of the nodes we've yet encountered and their current state.
	//
	// INVARIANT: For each k, v, v.node == k
	// INVARIANT: For each v, v.checkInvariants() doesn't panic
	//
	// GUARDED_BY(mu)
	nodes map[Node]*nodeInfo

	// The set of nodes in state_DependenciesUnresolved for which we haven't yet
	// started a call to the dependency resolver.
	//
	// INVARIANT: For each v, v.state == state_DependenciesUnresolved
	// INVARIANT: For each v, v.node is a key in nodes
	//
	// GUARDED_BY(mu)<commit_msg>Wrote the skeleton of dag.Visit.<commit_after>// Copyright 2015 <NAME>. All Rights Reserved.
// Author: <EMAIL> (<NAME>)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package dag

import (
	"errors"
	"fmt"
	"log"
	"sync"

	"github.com/jacobsa/syncutil"

	"golang.org/x/net/context"
)

// Call the visitor once for each unique node in the union of startNodes and
// all of its transitive dependencies, with bounded parallelism.
//
// Guarantees:
//
//  *  If the graph contains a cycle, this function will not succeed.
//
//  *  If a node N depends on a node M, v.Visit(N) will be called only after
//     v.Visit(M) returns successfully.
//
//  *  For each unique node N, dr.FindDependencies(N) and v.Visit(N) will each
//     be called at most once. Moreover, v.Visit(N) will be called only after
//     dr.FindDependencies(N) returns successfully.
//
func Visit(
	ctx context.Context,
	startNodes []Node,
	dr DependencyResolver,
	v Visitor,
	parallelism int) (err error) {
	b := syncutil.NewBundle(ctx)

	// Set up a state struct.
	state := &visitState{
		dr:          dr,
		visitor:     v,
		nodes:       make(map[Node]*nodeInfo),
		unsatisfied: make(map[*nodeInfo]struct{}),
	}

	state.mu = syncutil.NewInvariantMutex(state.checkInvariants)
	state.cond.L = &state.mu

	// Add each of the start nodes.
	state.mu.Lock()
	state.addNodes(startNodes)
	state.mu.Unlock()

	// Run multiple workers.
	for i := 0; i < parallelism; i++ {
		b.Add(func(ctx context.Context) (err error) {
			err = state.processNodes(ctx)
			if err != nil {
				err = fmt.Errorf("processNodes: %v", err)
				return
			}

			return
		})
	}

	// Join the bundle, but use the explicitly tracked first worker error in
	// order to circumvent the following race:
	//
	//  *  Worker A encounters an error, sets firstErr, and returns.
	//
	//  *  Worker B wakes up, sees firstErr, and returns with a follow-on
	//     "cancelled" error.
	//
	//  *  The bundle observes worker B's error before worker A's.
	//
	b.Join()
	state.mu.Lock()
	err = state.firstErr
	state.mu.Unlock()

	return
}

////////////////////////////////////////////////////////////////////////
// nodeInfo
////////////////////////////////////////////////////////////////////////

type nodeState int

// States in which a node involved in a call to Visit may be.
const (
	// The dependency resolver has not yet been called for this node, or a call
	// is currently in progress.
	state_DependenciesUnresolved nodeState = iota

	// The dependencies of this node have been resolved, but they have not yet
	// all been visited.
	state_DependenciesUnsatisfied

	// This node is eligible to be visited, but the visitor has not yet been
	// called or a call is currently in progress.
	state_Unvisited

	// The visitor has been called for this node and returned successfully.
	state_Visited
)

type nodeInfo struct {
	node  Node
	state nodeState

	// The number of unsatisfied dependencies remaining for this node.
	//
	// INVARIANT: unsatisfied >= 0
	// INVARIANT: unsatisfied > 0 iff state == state_DependenciesUnsatisfied
	unsatisfied int64

	// The set of unsatisfied nodes for which this node is a blocker.
	//
	// INVARIANT: len(dependants) > 0 implies state < state_Unvisited.
	// INVARIANT: For each n, n.state == state_DependenciesUnsatisfied
	dependants []*nodeInfo
}

////////////////////////////////////////////////////////////////////////
// visitState
////////////////////////////////////////////////////////////////////////

type visitState struct {
	dr      DependencyResolver
	visitor Visitor

	mu syncutil.InvariantMutex

	// All of the nodes we've yet encountered and their current state.
	//
	// INVARIANT: For each k, v, v.node == k
	// INVARIANT: For each v, v.checkInvariants() doesn't panic
	//
	// GUARDED_BY(mu)
	nodes map[Node]*nodeInfo

	// The set of nodes in state_DependenciesUnresolved for which we haven't yet
	// started a call to the dependency resolver.
	//
	// INVARIANT: For each v, v.state == state_DependenciesUnresolved
	// INVARIANT: For each v, v.node is a key in nodes
	//
	// GUARDED_BY(mu)
{'fraction_non_alphanumeric': 0.15106424911312574, 'fraction_numerical': 0.002857705951911707, 'mean_word_length': 2.4876288659793815, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1318358', 'n_tokens_mistral': 3145, 'n_tokens_neox': 2945, 'n_words': 1180}