<commit_before>// Copyright 2011, <NAME>. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
/*
 *  Filename:    godirs.go
 *  Author:      <NAME> <<EMAIL>>
 *  Created:     Tue Jul  5 22:13:49 PDT 2011
 *  Description: 
 *  Usage:       godirs [options] ARGUMENT ...
 */

//  Package dispatch provides goroutine dispatch and concurrency limiting.
//  It provides an object Dispatch which is a queueing system for concurrent
//  functions. It implements a dynamic limit on the number of routines that
//  it runs simultaneously. It also implements an interface Queue, allowing
//  for alternate queue implementations.
//
//  See github.com/bmatsuo/dispatch/examples for usage examples.
package dispatch
import (
    "sync"
    "github.com/bmatsuo/dispatch/queues"
)

//  A Dispatch is an automated function dispatch queue with a limited
//  number of concurrent gorountines.
type Dispatch struct {
    // The maximum number of goroutines can be changed while the queue is
    // processing.
    MaxGo      int

    // Handle waiting when the limit of concurrent goroutines has been reached.
    waitingToRun bool
    nextWake     chan bool

    // Handle waiting when function queue is empty.
    waitingOnQ   bool
    restart      chan bool

    // Manage the Start()'ing of a Dispatch, avoiding race conditions.
    startLock    *sync.Mutex
    started      bool

    // Handle goroutine-safe queue operations.
    qLock        *sync.Mutex
    queue        Queue

    // Handle goroutine-safe limiting and identifier operations.
    pLock        *sync.Mutex
    processing   int         // Number of QueueTasks running
    idcount      int64       // pid counter

    // Handle stopping of the Start() method.
    kill         chan bool
}

//  Create a new queue object with a specified limit on concurrency.
func New(maxroutines int) *Dispatch {
    return NewCustom(maxroutines, NewFIFO())
}
func NewCustom(maxroutines int, queue Queue) *Dispatch {
    var rl = new(Dispatch)
    rl.startLock = new(sync.Mutex)
    rl.qLock     = new(sync.Mutex)
    rl.pLock     = new(sync.Mutex)
    rl.restart   = make(chan bool)
    rl.kill      = make(chan bool)
    rl.nextWake  = make(chan bool)
    rl.queue     = queue
    rl.MaxGo     = maxroutines
    rl.idcount   = 0
    return rl
}

//  Goroutines called from a Dispatch are given an int identifier unique
//  to that routine.
type StdTask struct {
    F func(id int64)
}
func (dt StdTask) Func() func(id int64) {
    return dt.F
}
type dispatchTaskWrapper struct {
    id int64
    t  Task
}
func (dtw dispatchTaskWrapper) Func() func(id int64) {
    return dtw.t.Func()
}
func (dtw dispatchTaskWrapper) Id() int64 {
    return dtw.id
}
func (dtw dispatchTaskWrapper) Task() Task {
    return dtw.t
}

//  Enqueue a task for execution as a goroutine.
func (gq *Dispatch) Enqueue(t Task) int64 {
    // Wrap the function so it works with the goroutine limiting code.
    var f = t.Func()
    var dtFunc = func (id int64) {
        // Run the given function.
        f(id)

        // Decrement the process counter.
        gq.pLock.Lock()
        gq.processing--
        var procWaiting = gq.waitingToRun
        if procWaiting {
            gq.waitingToRun = false
        }
        gq.pLock.Unlock()

        // Start any waiting process.
        if procWaiting {
            gq.nextWake<-true
        }
    }
    var dt = StdTask{dtFunc}

    // Lock the queue and enqueue a new task.
    gq.qLock.Lock()
    gq.idcount++
    var id = gq.idcount
    gq.queue.Enqueue(dispatchTaskWrapper{id, dt})
    var loopWaiting = gq.waitingOnQ
    if loopWaiting {
        gq.waitingOnQ = false
    }
    gq.qLock.Unlock()

    // Restart the Start() loop if it was deemed necessary.
    if loopWaiting {
        gq.restart<-true
    }

    return id
}

//  Stop the queue after gq.Start() has been called. Any goroutines which
//  have not already been dequeued will not be executed until gq.Start()
//  is called again.
func (gq *Dispatch) Stop() {
    // Lock out Start() and queue ops for the entire call.
    gq.startLock.Lock()
    defer gq.startLock.Unlock()
    gq.qLock.Lock()
    defer gq.qLock.Unlock()

    if !gq.started {
        return
    }

    // Clear channel flags and close channels, stoping further processing.
    gq.started = false
    gq.waitingToRun = false
    gq.waitingOnQ = false
    close(gq.restart)
    close(gq.kill)
    close(gq.nextWake)
}

//  Start the next task in the queue. It's assumed that the queue is non-
//  empty. Furthermore, there should only be one goroutine in this method
//  (for this object) at a time. Both conditions are enforced in
//  gq.Start(), which calls gq.next() exclusively.
func (gq *Dispatch) next() {
    for true {
        // Attempt to start processing the file.
        gq.pLock.Lock()
        if gq.processing >= gq.MaxGo {
            gq.waitingToRun = true
            gq.pLock.Unlock()
            var cont, ok =<-gq.nextWake
            if !ok {
                gq.nextWake = make(chan bool)
                return
            }
            if !cont {
                return
            }
            continue
        }
        // Keep the books and reset wait time before unlocking.
        gq.waitingToRun = false
        gq.processing++
        gq.pLock.Unlock()

        // Get an element from the queue.
        gq.qLock.Lock()
        var wrapper = gq.queue.Dequeue().(RegisteredTask)<commit_msg>Fix broken pieces of dispatch after creating the queues package.<commit_after>// Copyright 2011, <NAME>. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
/*
 *  Filename:    godirs.go
 *  Author:      <NAME> <<EMAIL>>
 *  Created:     Tue Jul  5 22:13:49 PDT 2011
 *  Description: 
 *  Usage:       godirs [options] ARGUMENT ...
 */

//  Package dispatch provides goroutine dispatch and concurrency limiting.
//  It provides an object Dispatch which is a queueing system for concurrent
//  functions. It implements a dynamic limit on the number of routines that
//  it runs simultaneously. It also implements an interface Queue, allowing
//  for alternate queue implementations.
//
//  See github.com/bmatsuo/dispatch/examples for usage examples.
package dispatch
import (
    "sync"
    "github.com/bmatsuo/dispatch/queues"
)

//  A Dispatch is an automated function dispatch queue with a limited
//  number of concurrent gorountines.
type Dispatch struct {
    // The maximum number of goroutines can be changed while the queue is
    // processing.
    MaxGo      int

    // Handle waiting when the limit of concurrent goroutines has been reached.
    waitingToRun bool
    nextWake     chan bool

    // Handle waiting when function queue is empty.
    waitingOnQ   bool
    restart      chan bool

    // Manage the Start()'ing of a Dispatch, avoiding race conditions.
    startLock    *sync.Mutex
    started      bool

    // Handle goroutine-safe queue operations.
    qLock        *sync.Mutex
    queue        queues.Queue

    // Handle goroutine-safe limiting and identifier operations.
    pLock        *sync.Mutex
    processing   int         // Number of QueueTasks running
    idcount      int64       // pid counter

    // Handle stopping of the Start() method.
    kill         chan bool
}

//  Create a new queue object with a specified limit on concurrency.
func New(maxroutines int) *Dispatch {
    return NewCustom(maxroutines, queues.NewFIFO())
}
func NewCustom(maxroutines int, queue queues.Queue) *Dispatch {
    var rl = new(Dispatch)
    rl.startLock = new(sync.Mutex)
    rl.qLock     = new(sync.Mutex)
    rl.pLock     = new(sync.Mutex)
    rl.restart   = make(chan bool)
    rl.kill      = make(chan bool)
    rl.nextWake  = make(chan bool)
    rl.queue     = queue
    rl.MaxGo     = maxroutines
    rl.idcount   = 0
    return rl
}

//  Goroutines called from a Dispatch are given an int identifier unique
//  to that routine.
type StdTask struct {
    F func(id int64)
}
func (dt StdTask) Func() func(id int64) {
    return dt.F
}
type dispatchTaskWrapper struct {
    id int64
    t  queues.Task
}
func (dtw dispatchTaskWrapper) Func() func(id int64) {
    return dtw.t.Func()
}
func (dtw dispatchTaskWrapper) Id() int64 {
    return dtw.id
}
func (dtw dispatchTaskWrapper) Task() queues.Task {
    return dtw.t
}

//  Enqueue a task for execution as a goroutine.
func (gq *Dispatch) Enqueue(t queues.Task) int64 {
    // Wrap the function so it works with the goroutine limiting code.
    var f = t.Func()
    var dtFunc = func (id int64) {
        // Run the given function.
        f(id)

        // Decrement the process counter.
        gq.pLock.Lock()
        gq.processing--
        var procWaiting = gq.waitingToRun
        if procWaiting {
            gq.waitingToRun = false
        }
        gq.pLock.Unlock()

        // Start any waiting process.
        if procWaiting {
            gq.nextWake<-true
        }
    }
    var dt = StdTask{dtFunc}

    // Lock the queue and enqueue a new task.
    gq.qLock.Lock()
    gq.idcount++
    var id = gq.idcount
    gq.queue.Enqueue(dispatchTaskWrapper{id, dt})
    var loopWaiting = gq.waitingOnQ
    if loopWaiting {
        gq.waitingOnQ = false
    }
    gq.qLock.Unlock()

    // Restart the Start() loop if it was deemed necessary.
    if loopWaiting {
        gq.restart<-true
    }

    return id
}

//  Stop the queue after gq.Start() has been called. Any goroutines which
//  have not already been dequeued will not be executed until gq.Start()
//  is called again.
func (gq *Dispatch) Stop() {
    // Lock out Start() and queue ops for the entire call.
    gq.startLock.Lock()
    defer gq.startLock.Unlock()
    gq.qLock.Lock()
    defer gq.qLock.Unlock()

    if !gq.started {
        return
    }

    // Clear channel flags and close channels, stoping further processing.
    gq.started = false
    gq.waitingToRun = false
    gq.waitingOnQ = false
    close(gq.restart)
    close(gq.kill)
    close(gq.nextWake)
}

//  Start the next task in the queue. It's assumed that the queue is non-
//  empty. Furthermore, there should only be one goroutine in this method
//  (for this object) at a time. Both conditions are enforced in
//  gq.Start(), which calls gq.next() exclusively.
func (gq *Dispatch) next() {
    for true {
        // Attempt to start processing the file.
        gq.pLock.Lock()
        if gq.processing >= gq.MaxGo {
            gq.waitingToRun = true
            gq.pLock.Unlock()
            var cont, ok =<-gq.nextWake
            if !ok {
                gq.nextWake = make(chan bool)
                return
            }
            if !cont {
                return
            }
            continue
        }
        // Keep the books and reset wait time before unlocking.
        gq.waitingToRun = false
        gq.processing++
        gq.pLock.Unlock()

        // Get an element from the queue.
        gq.qLock.Lock()
        var wrapper = gq.queue.Dequeue().(queues.RegisteredTask)
{'fraction_non_alphanumeric': 0.08746618575293057, 'fraction_numerical': 0.005770964833183048, 'mean_word_length': 2.1189538807649044, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4170105', 'n_tokens_mistral': 3516, 'n_tokens_neox': 3367, 'n_words': 1162}