import (
"bytes"
+ "context"
"flag"
"fmt"
"go/ast"
"strings"
"cmd/internal/diff"
+
+ "golang.org/x/sync/semaphore"
)
var (
)
var (
- fileSet = token.NewFileSet() // per process FileSet
- exitCode = 0
- rewrite func(*ast.File) *ast.File
+ rewrite func(*token.FileSet, *ast.File) *ast.File
parserMode parser.Mode
)
-func report(err error) {
- scanner.PrintError(os.Stderr, err)
- exitCode = 2
-}
-
func usage() {
fmt.Fprintf(os.Stderr, "usage: gofmt [flags] [path ...]\n")
flag.PrintDefaults()
func isGoFile(f fs.DirEntry) bool {
// ignore non-Go files
name := f.Name()
- return !f.IsDir() && !strings.HasPrefix(name, ".") && strings.HasSuffix(name, ".go")
+ return !strings.HasPrefix(name, ".") && strings.HasSuffix(name, ".go") && !f.IsDir()
+}
+
+// A sequencer performs concurrent tasks that may write output, but emits that
+// output in a deterministic order.
+type sequencer struct {
+ maxWeight int64
+ sem *semaphore.Weighted // weighted by input bytes (an approximate proxy for memory overhead)
+ prev <-chan *reporterState // 1-buffered
+}
+
+// newSequencer returns a sequencer that allows concurrent tasks up to maxWeight
+// and writes tasks' output to out and err.
+func newSequencer(maxWeight int64, out, err io.Writer) *sequencer {
+ sem := semaphore.NewWeighted(maxWeight)
+ prev := make(chan *reporterState, 1)
+ prev <- &reporterState{out: out, err: err}
+ return &sequencer{
+ maxWeight: maxWeight,
+ sem: sem,
+ prev: prev,
+ }
+}
+
+// exclusive is a weight that can be passed to a sequencer to cause
+// a task to be executed without any other concurrent tasks.
+const exclusive = -1
+
+// Add blocks until the sequencer has enough weight to spare, then adds f as a
+// task to be executed concurrently.
+//
+// If the weight is either negative or larger than the sequencer's maximum
+// weight, Add blocks until all other tasks have completed, then the task
+// executes exclusively (blocking all other calls to Add until it completes).
+//
+// f may run concurrently in a goroutine, but its output to the passed-in
+// reporter will be sequential relative to the other tasks in the sequencer.
+//
+// If f invokes a method on the reporter, execution of that method may block
+// until the previous task has finished. (To maximize concurrency, f should
+// avoid invoking the reporter until it has finished any parallelizable work.)
+//
+// If f returns a non-nil error, that error will be reported after f's output
+// (if any) and will cause a nonzero final exit code.
+func (s *sequencer) Add(weight int64, f func(*reporter) error) {
+ if weight < 0 || weight > s.maxWeight {
+ weight = s.maxWeight
+ }
+ if err := s.sem.Acquire(context.TODO(), weight); err != nil {
+ // Change the task from "execute f" to "report err".
+ weight = 0
+ f = func(*reporter) error { return err }
+ }
+
+ r := &reporter{prev: s.prev}
+ next := make(chan *reporterState, 1)
+ s.prev = next
+
+ // Start f in parallel: it can run until it invokes a method on r, at which
+ // point it will block until the previous task releases the output state.
+ go func() {
+ if err := f(r); err != nil {
+ r.Report(err)
+ }
+ next <- r.getState() // Release the next task.
+ s.sem.Release(weight)
+ }()
+}
+
+// AddReport prints an error to s after the output of any previously-added
+// tasks, causing the final exit code to be nonzero.
+func (s *sequencer) AddReport(err error) {
+ s.Add(0, func(*reporter) error { return err })
+}
+
+// GetExitCode waits for all previously-added tasks to complete, then returns an
+// exit code for the sequence suitable for passing to os.Exit.
+func (s *sequencer) GetExitCode() int {
+ c := make(chan int, 1)
+ s.Add(0, func(r *reporter) error {
+ c <- r.ExitCode()
+ return nil
+ })
+ return <-c
+}
+
+// A reporter reports output, warnings, and errors.
+type reporter struct {
+ prev <-chan *reporterState
+ state *reporterState
+}
+
+// reporterState carries the state of a reporter instance.
+//
+// Only one reporter at a time may have access to a reporterState.
+type reporterState struct {
+ out, err io.Writer
+ exitCode int
+}
+
+// getState blocks until any prior reporters are finished with the reporter
+// state, then returns the state for manipulation.
+func (r *reporter) getState() *reporterState {
+ if r.state == nil {
+ r.state = <-r.prev
+ }
+ return r.state
}
+// Warnf emits a warning message to the reporter's error stream,
+// without changing its exit code.
+func (r *reporter) Warnf(format string, args ...interface{}) {
+ fmt.Fprintf(r.getState().err, format, args...)
+}
+
+// Write emits a slice to the reporter's output stream.
+//
+// Any error is returned to the caller, and does not otherwise affect the
+// reporter's exit code.
+func (r *reporter) Write(p []byte) (int, error) {
+ return r.getState().out.Write(p)
+}
+
+// Report emits a non-nil error to the reporter's error stream,
+// changing its exit code to a nonzero value.
+func (r *reporter) Report(err error) {
+ if err == nil {
+ panic("Report with nil error")
+ }
+ st := r.getState()
+ scanner.PrintError(st.err, err)
+ st.exitCode = 2
+}
+
+func (r *reporter) ExitCode() int {
+ return r.getState().exitCode
+}
+
+// If info == nil, we are formatting stdin instead of a file.
// If in == nil, the source is the contents of the file with the given filename.
-func processFile(filename string, in io.Reader, out io.Writer, stdin bool) error {
- var perm fs.FileMode = 0644
+func processFile(filename string, info fs.FileInfo, in io.Reader, r *reporter) error {
if in == nil {
- f, err := os.Open(filename)
+ var err error
+ in, err = os.Open(filename)
if err != nil {
return err
}
- defer f.Close()
- fi, err := f.Stat()
+ }
+
+ // Compute the file's size and read its contents with minimal allocations.
+ //
+ // If the size is unknown (or bogus, or overflows an int), fall back to
+ // a size-independent ReadAll.
+ var src []byte
+ size := -1
+ if info != nil && info.Mode().IsRegular() && int64(int(info.Size())) == info.Size() {
+ size = int(info.Size())
+ }
+ if size+1 > 0 {
+ // If we have the FileInfo from filepath.WalkDir, use it to make
+ // a buffer of the right size and avoid ReadAll's reallocations.
+ //
+ // We try to read size+1 bytes so that we can detect modifications: if we
+ // read more than size bytes, then the file was modified concurrently.
+ // (If that happens, we could, say, append to src to finish the read, or
+ // proceed with a truncated buffer — but the fact that it changed at all
+ // indicates a possible race with someone editing the file, so we prefer to
+ // stop to avoid corrupting it.)
+ src = make([]byte, size+1)
+ n, err := io.ReadFull(in, src)
+ if err != nil && err != io.ErrUnexpectedEOF {
+ return err
+ }
+ if n < size {
+ return fmt.Errorf("error: size of %s changed during reading (from %d to %d bytes)", filename, size, n)
+ } else if n > size {
+ return fmt.Errorf("error: size of %s changed during reading (from %d to >=%d bytes)", filename, size, len(src))
+ }
+ src = src[:n]
+ } else {
+ // The file is not known to be regular, so we don't have a reliable size for it.
+ var err error
+ src, err = io.ReadAll(in)
if err != nil {
return err
}
- in = f
- perm = fi.Mode().Perm()
}
- src, err := io.ReadAll(in)
- if err != nil {
- return err
+ fileSet := token.NewFileSet()
+ fragmentOk := false
+ if info == nil {
+ // If we are formatting stdin, we accept a program fragment in lieu of a
+ // complete source file.
+ fragmentOk = true
}
-
- file, sourceAdj, indentAdj, err := parse(fileSet, filename, src, stdin)
+ file, sourceAdj, indentAdj, err := parse(fileSet, filename, src, fragmentOk)
if err != nil {
return err
}
if rewrite != nil {
if sourceAdj == nil {
- file = rewrite(file)
+ file = rewrite(fileSet, file)
} else {
- fmt.Fprintf(os.Stderr, "warning: rewrite ignored for incomplete programs\n")
+ r.Warnf("warning: rewrite ignored for incomplete programs\n")
}
}
if !bytes.Equal(src, res) {
// formatting has changed
if *list {
- fmt.Fprintln(out, filename)
+ fmt.Fprintln(r, filename)
}
if *write {
+ if info == nil {
+ panic("-w should not have been allowed with stdin")
+ }
// make a temporary backup before overwriting original
+ perm := info.Mode().Perm()
bakname, err := backupFile(filename+".", src, perm)
if err != nil {
return err
if err != nil {
return fmt.Errorf("computing diff: %s", err)
}
- fmt.Fprintf(out, "diff -u %s %s\n", filepath.ToSlash(filename+".orig"), filepath.ToSlash(filename))
- out.Write(data)
+ fmt.Fprintf(r, "diff -u %s %s\n", filepath.ToSlash(filename+".orig"), filepath.ToSlash(filename))
+ r.Write(data)
}
}
if !*list && !*write && !*doDiff {
- _, err = out.Write(res)
+ _, err = r.Write(res)
}
return err
}
-func visitFile(path string, f fs.DirEntry, err error) error {
- if err != nil || !isGoFile(f) {
- return err
- }
- if err := processFile(path, nil, os.Stdout, false); err != nil {
- report(err)
- }
- return nil
-}
-
func main() {
+ // Arbitrarily limit in-flight work to 2MiB times the number of threads.
+ //
+ // The actual overhead for the parse tree and output will depend on the
+ // specifics of the file, but this at least keeps the footprint of the process
+ // roughly proportional to GOMAXPROCS.
+ maxWeight := (2 << 20) * int64(runtime.GOMAXPROCS(0))
+ s := newSequencer(maxWeight, os.Stdout, os.Stderr)
+
// call gofmtMain in a separate function
// so that it can use defer and have them
// run before the exit.
- gofmtMain()
- os.Exit(exitCode)
+ gofmtMain(s)
+ os.Exit(s.GetExitCode())
}
-func gofmtMain() {
+func gofmtMain(s *sequencer) {
flag.Usage = usage
flag.Parse()
if *cpuprofile != "" {
f, err := os.Create(*cpuprofile)
if err != nil {
- fmt.Fprintf(os.Stderr, "creating cpu profile: %s\n", err)
- exitCode = 2
+ s.AddReport(fmt.Errorf("creating cpu profile: %s", err))
return
}
defer f.Close()
args := flag.Args()
if len(args) == 0 {
if *write {
- fmt.Fprintln(os.Stderr, "error: cannot use -w with standard input")
- exitCode = 2
+ s.AddReport(fmt.Errorf("error: cannot use -w with standard input"))
return
}
- if err := processFile("<standard input>", os.Stdin, os.Stdout, true); err != nil {
- report(err)
- }
+ s.Add(0, func(r *reporter) error {
+ return processFile("<standard input>", nil, os.Stdin, r)
+ })
return
}
for _, arg := range args {
switch info, err := os.Stat(arg); {
case err != nil:
- report(err)
+ s.AddReport(err)
case !info.IsDir():
// Non-directory arguments are always formatted.
- if err := processFile(arg, nil, os.Stdout, false); err != nil {
- report(err)
- }
+ arg := arg
+ s.Add(fileWeight(arg, info), func(r *reporter) error {
+ return processFile(arg, info, nil, r)
+ })
default:
// Directories are walked, ignoring non-Go files.
- if err := filepath.WalkDir(arg, visitFile); err != nil {
- report(err)
+ err := filepath.WalkDir(arg, func(path string, f fs.DirEntry, err error) error {
+ if err != nil || !isGoFile(f) {
+ return err
+ }
+ info, err := f.Info()
+ if err != nil {
+ s.AddReport(err)
+ return nil
+ }
+ s.Add(fileWeight(path, info), func(r *reporter) error {
+ return processFile(path, info, nil, r)
+ })
+ return nil
+ })
+ if err != nil {
+ s.AddReport(err)
}
}
}
}
+func fileWeight(path string, info fs.FileInfo) int64 {
+ if info == nil {
+ return exclusive
+ }
+ if info.Mode().Type() == fs.ModeSymlink {
+ var err error
+ info, err = os.Stat(path)
+ if err != nil {
+ return exclusive
+ }
+ }
+ if !info.Mode().IsRegular() {
+ // For non-regular files, FileInfo.Size is system-dependent and thus not a
+ // reliable indicator of weight.
+ return exclusive
+ }
+ return info.Size()
+}
+
func diffWithReplaceTempFile(b1, b2 []byte, filename string) ([]byte, error) {
data, err := diff.Diff("gofmt", b1, b2)
if len(data) > 0 {
--- /dev/null
+// Copyright 2017 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package semaphore provides a weighted semaphore implementation.
+package semaphore // import "golang.org/x/sync/semaphore"
+
+import (
+ "container/list"
+ "context"
+ "sync"
+)
+
+type waiter struct {
+ n int64
+ ready chan<- struct{} // Closed when semaphore acquired.
+}
+
+// NewWeighted creates a new weighted semaphore with the given
+// maximum combined weight for concurrent access.
+func NewWeighted(n int64) *Weighted {
+ w := &Weighted{size: n}
+ return w
+}
+
+// Weighted provides a way to bound concurrent access to a resource.
+// The callers can request access with a given weight.
+type Weighted struct {
+ size int64
+ cur int64
+ mu sync.Mutex
+ waiters list.List
+}
+
+// Acquire acquires the semaphore with a weight of n, blocking until resources
+// are available or ctx is done. On success, returns nil. On failure, returns
+// ctx.Err() and leaves the semaphore unchanged.
+//
+// If ctx is already done, Acquire may still succeed without blocking.
+func (s *Weighted) Acquire(ctx context.Context, n int64) error {
+ s.mu.Lock()
+ if s.size-s.cur >= n && s.waiters.Len() == 0 {
+ s.cur += n
+ s.mu.Unlock()
+ return nil
+ }
+
+ if n > s.size {
+ // Don't make other Acquire calls block on one that's doomed to fail.
+ s.mu.Unlock()
+ <-ctx.Done()
+ return ctx.Err()
+ }
+
+ ready := make(chan struct{})
+ w := waiter{n: n, ready: ready}
+ elem := s.waiters.PushBack(w)
+ s.mu.Unlock()
+
+ select {
+ case <-ctx.Done():
+ err := ctx.Err()
+ s.mu.Lock()
+ select {
+ case <-ready:
+ // Acquired the semaphore after we were canceled. Rather than trying to
+ // fix up the queue, just pretend we didn't notice the cancelation.
+ err = nil
+ default:
+ isFront := s.waiters.Front() == elem
+ s.waiters.Remove(elem)
+ // If we're at the front and there're extra tokens left, notify other waiters.
+ if isFront && s.size > s.cur {
+ s.notifyWaiters()
+ }
+ }
+ s.mu.Unlock()
+ return err
+
+ case <-ready:
+ return nil
+ }
+}
+
+// TryAcquire acquires the semaphore with a weight of n without blocking.
+// On success, returns true. On failure, returns false and leaves the semaphore unchanged.
+func (s *Weighted) TryAcquire(n int64) bool {
+ s.mu.Lock()
+ success := s.size-s.cur >= n && s.waiters.Len() == 0
+ if success {
+ s.cur += n
+ }
+ s.mu.Unlock()
+ return success
+}
+
+// Release releases the semaphore with a weight of n.
+func (s *Weighted) Release(n int64) {
+ s.mu.Lock()
+ s.cur -= n
+ if s.cur < 0 {
+ s.mu.Unlock()
+ panic("semaphore: released more than held")
+ }
+ s.notifyWaiters()
+ s.mu.Unlock()
+}
+
+func (s *Weighted) notifyWaiters() {
+ for {
+ next := s.waiters.Front()
+ if next == nil {
+ break // No more waiters blocked.
+ }
+
+ w := next.Value.(waiter)
+ if s.size-s.cur < w.n {
+ // Not enough tokens for the next waiter. We could keep going (to try to
+ // find a waiter with a smaller request), but under load that could cause
+ // starvation for large requests; instead, we leave all remaining waiters
+ // blocked.
+ //
+ // Consider a semaphore used as a read-write lock, with N tokens, N
+ // readers, and one writer. Each reader can Acquire(1) to obtain a read
+ // lock. The writer can Acquire(N) to obtain a write lock, excluding all
+ // of the readers. If we allow the readers to jump ahead in the queue,
+ // the writer will starve — there is always one token available for every
+ // reader.
+ break
+ }
+
+ s.cur += w.n
+ s.waiters.Remove(next)
+ close(w.ready)
+ }
+}