From a1b80ca79fb54e5b30867e72138b417fa6d079cf Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Thu, 18 Aug 2016 17:01:51 -0700 Subject: [PATCH 01/20] Added file lock and add extra comments --- README.md | 26 ++++++++++++++++++++++++++ phantom.go | 27 ++++++++++++++++++--------- phantom_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index cd9859e..1e6c924 100644 --- a/README.md +++ b/README.md @@ -28,3 +28,29 @@ func main() { // Output: 4 } ``` + +More Complex Usage +```go +import ( + "github.com/urturn/go-phantomjs" // exported package is phantomjs +) + +func main() { + p, err := phantomjs.Start() + if err != nil { + panic(err) + } + defer p.Exit() // Don't forget to kill phantomjs at some point. + var result interface{} + err = p.Run("function(done){ setTimeout(function() { done(3 + 3) ; }, 0);", &result) + if err != nil { + panic(err) + } + number, ok := result.(float64) + if !ok { + panic("Cannot convert result to float64") + } + fmt.Println(number) + // Output: 4 +} +``` diff --git a/phantom.go b/phantom.go index 0506146..5c21ad9 100644 --- a/phantom.go +++ b/phantom.go @@ -10,28 +10,36 @@ import ( "os" "os/exec" "strings" + "sync" ) +// Phantom a data structure that interacts with the wrapper file type Phantom struct { cmd *exec.Cmd in io.WriteCloser out io.ReadCloser errout io.ReadCloser + // wrapperFileName string } var nbInstance = 0 + var wrapperFileName = "" +var fileLock = new(sync.Mutex) /* +Start create phantomjs file Create a new `Phantomjs` instance and return it as a pointer. If an error occurs during command start, return it instead. */ func Start(args ...string) (*Phantom, error) { + fileLock.Lock() if nbInstance == 0 { wrapperFileName, _ = createWrapperFile() } - nbInstance += 1 + nbInstance++ + fileLock.Unlock() args = append(args, wrapperFileName) cmd := exec.Command("phantomjs", args...) @@ -81,19 +89,21 @@ func (p *Phantom) Exit() error { if err != nil { return err } - nbInstance -= 1 + fileLock.Lock() + nbInstance-- if nbInstance == 0 { os.Remove(wrapperFileName) } - + fileLock.Unlock() return nil } /* Run the javascript function passed as a string and wait for the result. -The result can be either in the return value of the function or the first argument passed -to the function first arguments. +The result can be either in the return value of the function or, +You can pass a function a closure which can take two arguments 1st is the successfull response +the 2nd is an err. See TestComplex in the phantom_test.go */ func (p *Phantom) Run(jsFunc string, res *interface{}) error { err := p.sendLine("RUN", jsFunc, "END") @@ -112,9 +122,8 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { resMsg <- parts[1] close(resMsg) return - } else { - fmt.Printf("LOG %s\n", line) } + fmt.Printf("LOG %s\n", line) } }() go func() { @@ -125,9 +134,8 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { errMsg <- errors.New(parts[1]) close(errMsg) return - } else { - fmt.Printf("LOG %s\n", line) } + fmt.Printf("LOG %s\n", line) } }() select { @@ -145,6 +153,7 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { } /* +Load will load more code Eval `jsCode` in the main context. */ func (p *Phantom) Load(jsCode string) error { diff --git a/phantom_test.go b/phantom_test.go index 6cf8452..fc4eb5b 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -1,7 +1,10 @@ package phantomjs import ( + "log" + "sync" "testing" + "time" ) func TestStartStop(t *testing.T) { @@ -81,6 +84,46 @@ func TestDoubleErrorSendDontCrash(t *testing.T) { p.Run("function(done) {done(null, 'manual'); done(null, 'should not panic');}", nil) } +func TestComplex(t *testing.T) { + var wg sync.WaitGroup + + for i := 0; i < 5; i++ { + wg.Add(1) + go func() { + defer wg.Done() + p, err := Start() + failOnError(err, t) + defer p.Exit() + var r interface{} + begin := time.Now() + + p.Run(`function(done){ + var a = 0; + var b = 1; + var c = 0; + for(var i=2; i<=25; i++) + { + c = b + a; + a = b; + b = c; + } + done(c, undefined); + }`, &r) + log.Println("Completed Run in", time.Since(begin)) + failOnError(err, t) + v, ok := r.(float64) + if !ok { + t.Errorf("Should be an int but is %v", r) + return + } + if v != 75025 { + t.Errorf("Should be %d but is %f", 75025, v) + } + }() + } + wg.Wait() +} + func assertFloatResult(jsFunc string, expected float64, p *Phantom, t *testing.T) { var r interface{} err := p.Run(jsFunc, &r) From 6e3f2e708458111330f973a79f63244e3646446d Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Fri, 19 Aug 2016 13:12:35 -0700 Subject: [PATCH 02/20] Added Forceful Shutdown and removed possible race conditions --- phantom.go | 109 +++++++++++++++++++++++++++++++++++++----------- phantom_test.go | 56 +++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 25 deletions(-) diff --git a/phantom.go b/phantom.go index 5c21ad9..97aefb4 100644 --- a/phantom.go +++ b/phantom.go @@ -15,11 +15,12 @@ import ( // Phantom a data structure that interacts with the wrapper file type Phantom struct { - cmd *exec.Cmd - in io.WriteCloser - out io.ReadCloser - errout io.ReadCloser - // wrapperFileName string + cmd *exec.Cmd + in io.WriteCloser + out io.ReadCloser + errout io.ReadCloser + scannerErrorLock *sync.Mutex + scannerLock *sync.Mutex } var nbInstance = 0 @@ -59,10 +60,12 @@ func Start(args ...string) (*Phantom, error) { } p := Phantom{ - cmd: cmd, - in: inPipe, - out: outPipe, - errout: errPipe, + cmd: cmd, + in: inPipe, + out: outPipe, + errout: errPipe, + scannerErrorLock: new(sync.Mutex), + scannerLock: new(sync.Mutex), } err = cmd.Start() @@ -80,12 +83,16 @@ and wait for the command to end. Return an error if one occured during exit command or if the program output a error value */ func (p *Phantom) Exit() error { + err := p.Load("phantom.exit()") if err != nil { return err } - + p.scannerErrorLock.Lock() + p.scannerLock.Lock() err = p.cmd.Wait() + p.scannerErrorLock.Unlock() + p.scannerLock.Unlock() if err != nil { return err } @@ -98,6 +105,23 @@ func (p *Phantom) Exit() error { return nil } +/* +ForceShutdown will forcefully kill phantomjs. +This will completly terminate the proccess compared to Exit which will safely Exit +*/ +func (p *Phantom) ForceShutdown() error { + if err := p.cmd.Process.Kill(); err != nil { + return err + } + fileLock.Lock() + nbInstance-- + if nbInstance == 0 { + os.Remove(wrapperFileName) + } + fileLock.Unlock() + return nil +} + /* Run the javascript function passed as a string and wait for the result. @@ -115,27 +139,35 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { resMsg := make(chan string) errMsg := make(chan error) go func() { - for scannerOut.Scan() { - line := scannerOut.Text() - parts := strings.SplitN(line, " ", 2) - if strings.HasPrefix(line, "RES") { - resMsg <- parts[1] - close(resMsg) + for { + value, scanError := readScanner(p.scannerLock, scannerOut) + if scanError != nil { + errMsg <- scanError return } - fmt.Printf("LOG %s\n", line) + + if value == "" { + // Nothing to see here + return + } + + resMsg <- value } }() go func() { - for scannerErrorOut.Scan() { - line := scannerErrorOut.Text() - parts := strings.SplitN(line, " ", 2) - if strings.HasPrefix(line, "RES") { - errMsg <- errors.New(parts[1]) - close(errMsg) + for { + value, scanError := readScanner(p.scannerErrorLock, scannerErrorOut) + if scanError != nil { + errMsg <- scanError + return + } + + if value == "" { + // Nothing to see here return } - fmt.Printf("LOG %s\n", line) + + errMsg <- errors.New(value) } }() select { @@ -179,11 +211,38 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } +func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { + scannerLock.Lock() + read := scanner.Scan() + scannerLock.Unlock() + if !read { + return "", errors.New("this instance of phantomjs is no longer running") + } + + if scanner.Err() != nil { + return "", scanner.Err() + } + + line := scanner.Text() + parts := strings.SplitN(line, " ", 2) + + if strings.HasPrefix(line, "RES") { + return parts[1], nil + } else if line != "" { + fmt.Printf("LOG %s\n", line) + return "", nil + } else if line != " " { + return "", errors.New("Error reading response, just got a space") + } + fmt.Printf("LOG |%s|\n", line) + return "", errors.New("Error reading response") +} + func (p *Phantom) sendLine(lines ...string) error { for _, l := range lines { _, err := io.WriteString(p.in, l+"\n") if err != nil { - return errors.New("Cannot Send: `" + l + "`") + return errors.New("Cannot Send: `" + l + "`" + "phantomjs instance might be dead") } } return nil diff --git a/phantom_test.go b/phantom_test.go index fc4eb5b..642bdf4 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -2,6 +2,7 @@ package phantomjs import ( "log" + "strings" "sync" "testing" "time" @@ -84,6 +85,22 @@ func TestDoubleErrorSendDontCrash(t *testing.T) { p.Run("function(done) {done(null, 'manual'); done(null, 'should not panic');}", nil) } +func TestThrow(t *testing.T) { + p, err := Start() + if err != nil { + panic(err) + } + defer p.Exit() // Don't forget to kill phantomjs at some point. + var result interface{} + err = p.Run("function() { throw 'Ooops' }", &result) + if err == nil { + t.Fatal("Expected an Error") + } + if !strings.Contains("\"Ooops\"", err.Error()) { + t.Fatal(err) + } +} + func TestComplex(t *testing.T) { var wg sync.WaitGroup @@ -124,6 +141,45 @@ func TestComplex(t *testing.T) { wg.Wait() } +func TestForceShutdown(t *testing.T) { + p, err := Start() + failOnError(err, t) + + for i := 0; i < 5; i++ { + var r interface{} + + err = p.Run(`function(done){ + var a = 0; + var b = 1; + var c = 0; + for(var i=2; i<=25; i++) + { + c = b + a; + a = b; + b = c; + } + done(c, undefined); + }`, &r) + if i == 2 { + p.ForceShutdown() + } + if err == nil { + v, ok := r.(float64) + if !ok { + t.Errorf("Should be an int but is %v", r) + return + } + if v != 75025 { + t.Errorf("Should be %d but is %f", 75025, v) + } + } else { + if !strings.Contains(err.Error(), "no longer running") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { + t.Fatal(err) + } + } + } +} + func assertFloatResult(jsFunc string, expected float64, p *Phantom, t *testing.T) { var r interface{} err := p.Run(jsFunc, &r) From a00e074ec763fb9054851530d232bafc9931c11d Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 11:41:50 -0700 Subject: [PATCH 03/20] updated tests --- phantom_test.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/phantom_test.go b/phantom_test.go index 642bdf4..846962c 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -145,9 +145,13 @@ func TestForceShutdown(t *testing.T) { p, err := Start() failOnError(err, t) + count := 0 for i := 0; i < 5; i++ { var r interface{} + if i == 2 { + p.ForceShutdown() + } err = p.Run(`function(done){ var a = 0; var b = 1; @@ -160,9 +164,7 @@ func TestForceShutdown(t *testing.T) { } done(c, undefined); }`, &r) - if i == 2 { - p.ForceShutdown() - } + if err == nil { v, ok := r.(float64) if !ok { @@ -171,6 +173,8 @@ func TestForceShutdown(t *testing.T) { } if v != 75025 { t.Errorf("Should be %d but is %f", 75025, v) + } else { + count++ } } else { if !strings.Contains(err.Error(), "no longer running") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { @@ -178,6 +182,9 @@ func TestForceShutdown(t *testing.T) { } } } + if count != 2 { + t.Fatalf("Didn' reaach distination %d", count) + } } func assertFloatResult(jsFunc string, expected float64, p *Phantom, t *testing.T) { From ca18fda5e0a3d8a7b40eac52787c78ce1643db41 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 14:08:18 -0700 Subject: [PATCH 04/20] Fixed errors when logging multiple times --- phantom.go | 85 ++++++++++++++++++++++++------------------------- phantom_test.go | 31 +++++++++++++++++- 2 files changed, 72 insertions(+), 44 deletions(-) diff --git a/phantom.go b/phantom.go index 97aefb4..b561d6f 100644 --- a/phantom.go +++ b/phantom.go @@ -139,36 +139,33 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { resMsg := make(chan string) errMsg := make(chan error) go func() { - for { - value, scanError := readScanner(p.scannerLock, scannerOut) - if scanError != nil { - errMsg <- scanError - return - } + value, scanError := readScanner(p.scannerLock, scannerOut) - if value == "" { - // Nothing to see here - return - } + if scanError != nil { + errMsg <- scanError + return + } - resMsg <- value + if value == "" { + // Nothing to see here + return } + + resMsg <- value }() go func() { - for { - value, scanError := readScanner(p.scannerErrorLock, scannerErrorOut) - if scanError != nil { - errMsg <- scanError - return - } - - if value == "" { - // Nothing to see here - return - } + value, scanError := readScanner(p.scannerErrorLock, scannerErrorOut) + if scanError != nil { + errMsg <- scanError + return + } - errMsg <- errors.New(value) + if value == "" { + // Nothing to see here + return } + + errMsg <- errors.New(value) }() select { case text := <-resMsg: @@ -212,30 +209,32 @@ func createWrapperFile() (fileName string, err error) { } func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { - scannerLock.Lock() - read := scanner.Scan() - scannerLock.Unlock() - if !read { - return "", errors.New("this instance of phantomjs is no longer running") - } + read := true + for read { + scannerLock.Lock() + read := scanner.Scan() + scannerLock.Unlock() + if !read { + return "", errors.New("phantomjs instance is no longer running") + } - if scanner.Err() != nil { - return "", scanner.Err() - } + if scanner.Err() != nil { + return "", scanner.Err() + } - line := scanner.Text() - parts := strings.SplitN(line, " ", 2) + line := scanner.Text() + parts := strings.SplitN(line, " ", 2) - if strings.HasPrefix(line, "RES") { - return parts[1], nil - } else if line != "" { - fmt.Printf("LOG %s\n", line) - return "", nil - } else if line != " " { - return "", errors.New("Error reading response, just got a space") + if strings.HasPrefix(line, "RES") { + return parts[1], nil + } else if line != "" { + fmt.Printf("LOG %s\n", line) + // return "", nil + } else if line != " " { + // return "", errors.New("Error reading response, just got a space") + } } - fmt.Printf("LOG |%s|\n", line) - return "", errors.New("Error reading response") + return "", errors.New("No Response") } func (p *Phantom) sendLine(lines ...string) error { diff --git a/phantom_test.go b/phantom_test.go index 846962c..e1528f9 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -183,7 +183,36 @@ func TestForceShutdown(t *testing.T) { } } if count != 2 { - t.Fatalf("Didn' reaach distination %d", count) + t.Fatalf("Didn' reach distination %d", count) + } +} + +func TestMultipleLogs(t *testing.T) { + p, err := Start() + failOnError(err, t) + var r interface{} + + err = p.Run(`function(done){ + var a = 0; + var b = 1; + var c = 0; + for(var i=2; i<=25; i++) + { + c = b + a; + a = b; + b = c; + console.log(c) + } + done(c, undefined); + }`, &r) + failOnError(err, t) + v, ok := r.(float64) + if !ok { + t.Errorf("Should be an int but is %v", r) + return + } + if v != 75025 { + t.Errorf("Should be %d but is %f", 75025, v) } } From bc29c980bc992629f26378db60c4b751d5133a2a Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 14:17:48 -0700 Subject: [PATCH 05/20] forgot to defer exit --- phantom_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/phantom_test.go b/phantom_test.go index e1528f9..268b338 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -144,6 +144,7 @@ func TestComplex(t *testing.T) { func TestForceShutdown(t *testing.T) { p, err := Start() failOnError(err, t) + defer p.Exit() count := 0 for i := 0; i < 5; i++ { @@ -189,6 +190,7 @@ func TestForceShutdown(t *testing.T) { func TestMultipleLogs(t *testing.T) { p, err := Start() + defer p.Exit() failOnError(err, t) var r interface{} From 64b08337622533b8650c4612e88a0b56075f208e Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 15:49:18 -0700 Subject: [PATCH 06/20] Added Custom Scanner --- phantom.go | 7 +- phantomScanner.go | 410 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 413 insertions(+), 4 deletions(-) create mode 100644 phantomScanner.go diff --git a/phantom.go b/phantom.go index b561d6f..ee71b3b 100644 --- a/phantom.go +++ b/phantom.go @@ -1,7 +1,6 @@ package phantomjs import ( - "bufio" "encoding/json" "errors" "fmt" @@ -134,8 +133,8 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { if err != nil { return err } - scannerOut := bufio.NewScanner(p.out) - scannerErrorOut := bufio.NewScanner(p.errout) + scannerOut := NewPhantomScanner(p.out) + scannerErrorOut := NewPhantomScanner(p.errout) resMsg := make(chan string) errMsg := make(chan error) go func() { @@ -208,7 +207,7 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { +func readScanner(scannerLock *sync.Mutex, scanner *PhantomScanner) (string, error) { read := true for read { scannerLock.Lock() diff --git a/phantomScanner.go b/phantomScanner.go new file mode 100644 index 0000000..954d793 --- /dev/null +++ b/phantomScanner.go @@ -0,0 +1,410 @@ +// Copyright 2013 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 phantomjs + +import ( + "bytes" + "errors" + "io" + "unicode/utf8" +) + +// PhantomScanner provides a convenient interface for reading data such as +// a file of newline-delimited lines of text. Successive calls to +// the Scan method will step through the 'tokens' of a file, skipping +// the bytes between the tokens. The specification of a token is +// defined by a split function of type SplitFunc; the default split +// function breaks the input into lines with line termination stripped. Split +// functions are defined in this package for scanning a file into +// lines, bytes, UTF-8-encoded runes, and space-delimited words. The +// client may instead provide a custom split function. +// +// Scanning stops unrecoverably at EOF, the first I/O error, or a token too +// large to fit in the buffer. When a scan stops, the reader may have +// advanced arbitrarily far past the last token. Programs that need more +// control over error handling or large tokens, or must run sequential scans +// on a reader, should use bufio.Reader instead. +// +type PhantomScanner struct { + r io.Reader // The reader provided by the client. + split SplitFunc // The function to split the tokens. + maxTokenSize int // Maximum size of a token; modified by tests. + token []byte // Last token returned by split. + buf []byte // Buffer used as argument to split. + start int // First non-processed byte in buf. + end int // End of data in buf. + err error // Sticky error. + empties int // Count of successive empty tokens. + scanCalled bool // Scan has been called; buffer is in use. + done bool // Scan has finished. +} + +// SplitFunc is the signature of the split function used to tokenize the +// input. The arguments are an initial substring of the remaining unprocessed +// data and a flag, atEOF, that reports whether the Reader has no more data +// to give. The return values are the number of bytes to advance the input +// and the next token to return to the user, plus an error, if any. If the +// data does not yet hold a complete token, for instance if it has no newline +// while scanning lines, SplitFunc can return (0, nil, nil) to signal the +// PhantomScanner to read more data into the slice and try again with a longer slice +// starting at the same point in the input. +// +// If the returned error is non-nil, scanning stops and the error +// is returned to the client. +// +// The function is never called with an empty data slice unless atEOF +// is true. If atEOF is true, however, data may be non-empty and, +// as always, holds unprocessed text. +type SplitFunc func(data []byte, atEOF bool) (advance int, token []byte, err error) + +// Errors returned by PhantomScanner. +var ( + ErrTooLong = errors.New("bufio.PhantomScanner: token too long") + ErrNegativeAdvance = errors.New("bufio.PhantomScanner: SplitFunc returns negative advance count") + ErrAdvanceTooFar = errors.New("bufio.PhantomScanner: SplitFunc returns advance count beyond input") +) + +const ( + // MaxScanTokenSize is the maximum size used to buffer a token + // unless the user provides an explicit buffer with Scan.Buffer. + // The actual maximum token size may be smaller as the buffer + // may need to include, for instance, a newline. + MaxScanTokenSize = 2048 * 1024 + + startBufSize = 4096 // Size of initial allocation for buffer. + maxConsecutiveEmptyReads = 1024 +) + +// NewPhantomScanner returns a new PhantomScanner to read from r. +// The split function defaults to ScanLines. +func NewPhantomScanner(r io.Reader) *PhantomScanner { + return &PhantomScanner{ + r: r, + split: ScanLines, + maxTokenSize: MaxScanTokenSize, + } +} + +// Err returns the first non-EOF error that was encountered by the PhantomScanner. +func (s *PhantomScanner) Err() error { + if s.err == io.EOF { + return nil + } + return s.err +} + +// Bytes returns the most recent token generated by a call to Scan. +// The underlying array may point to data that will be overwritten +// by a subsequent call to Scan. It does no allocation. +func (s *PhantomScanner) Bytes() []byte { + return s.token +} + +// Text returns the most recent token generated by a call to Scan +// as a newly allocated string holding its bytes. +func (s *PhantomScanner) Text() string { + return string(s.token) +} + +// ErrFinalToken is a special sentinel error value. It is intended to be +// returned by a Split function to indicate that the token being delivered +// with the error is the last token and scanning should stop after this one. +// After ErrFinalToken is received by Scan, scanning stops with no error. +// The value is useful to stop processing early or when it is necessary to +// deliver a final empty token. One could achieve the same behavior +// with a custom error value but providing one here is tidier. +// See the emptyFinalToken example for a use of this value. +var ErrFinalToken = errors.New("final token") + +// Scan advances the PhantomScanner to the next token, which will then be +// available through the Bytes or Text method. It returns false when the +// scan stops, either by reaching the end of the input or an error. +// After Scan returns false, the Err method will return any error that +// occurred during scanning, except that if it was io.EOF, Err +// will return nil. +// Scan panics if the split function returns 100 empty tokens without +// advancing the input. This is a common error mode for PhantomScanners. +func (s *PhantomScanner) Scan() bool { + if s.done { + return false + } + s.scanCalled = true + // Loop until we have a token. + for { + // See if we can get a token with what we already have. + // If we've run out of data but have an error, give the split function + // a chance to recover any remaining, possibly empty token. + if s.end > s.start || s.err != nil { + advance, token, err := s.split(s.buf[s.start:s.end], s.err != nil) + if err != nil { + if err == ErrFinalToken { + s.token = token + s.done = true + return true + } + s.setErr(err) + return false + } + if !s.advance(advance) { + return false + } + s.token = token + if token != nil { + if s.err == nil || advance > 0 { + s.empties = 0 + } else { + // Returning tokens not advancing input at EOF. + s.empties++ + if s.empties > 100 { + panic("bufio.Scan: 100 empty tokens without progressing") + } + } + return true + } + } + // We cannot generate a token with what we are holding. + // If we've already hit EOF or an I/O error, we are done. + if s.err != nil { + // Shut it down. + s.start = 0 + s.end = 0 + return false + } + // Must read more data. + // First, shift data to beginning of buffer if there's lots of empty space + // or space is needed. + if s.start > 0 && (s.end == len(s.buf) || s.start > len(s.buf)/2) { + copy(s.buf, s.buf[s.start:s.end]) + s.end -= s.start + s.start = 0 + } + // Is the buffer full? If so, resize. + if s.end == len(s.buf) { + // Guarantee no overflow in the multiplication below. + const maxInt = int(^uint(0) >> 1) + if len(s.buf) >= s.maxTokenSize || len(s.buf) > maxInt/2 { + s.setErr(ErrTooLong) + return false + } + newSize := len(s.buf) * 2 + if newSize == 0 { + newSize = startBufSize + } + if newSize > s.maxTokenSize { + newSize = s.maxTokenSize + } + newBuf := make([]byte, newSize) + copy(newBuf, s.buf[s.start:s.end]) + s.buf = newBuf + s.end -= s.start + s.start = 0 + continue + } + // Finally we can read some input. Make sure we don't get stuck with + // a misbehaving Reader. Officially we don't need to do this, but let's + // be extra careful: PhantomScanner is for safe, simple jobs. + for loop := 0; ; { + n, err := s.r.Read(s.buf[s.end:len(s.buf)]) + s.end += n + if err != nil { + s.setErr(err) + break + } + if n > 0 { + s.empties = 0 + break + } + loop++ + if loop > maxConsecutiveEmptyReads { + s.setErr(io.ErrNoProgress) + break + } + } + } +} + +// advance consumes n bytes of the buffer. It reports whether the advance was legal. +func (s *PhantomScanner) advance(n int) bool { + if n < 0 { + s.setErr(ErrNegativeAdvance) + return false + } + if n > s.end-s.start { + s.setErr(ErrAdvanceTooFar) + return false + } + s.start += n + return true +} + +// setErr records the first error encountered. +func (s *PhantomScanner) setErr(err error) { + if s.err == nil || s.err == io.EOF { + s.err = err + } +} + +// Buffer sets the initial buffer to use when scanning and the maximum +// size of buffer that may be allocated during scanning. The maximum +// token size is the larger of max and cap(buf). If max <= cap(buf), +// Scan will use this buffer only and do no allocation. +// +// By default, Scan uses an internal buffer and sets the +// maximum token size to MaxScanTokenSize. +// +// Buffer panics if it is called after scanning has started. +func (s *PhantomScanner) Buffer(buf []byte, max int) { + if s.scanCalled { + panic("Buffer called after Scan") + } + s.buf = buf[0:cap(buf)] + s.maxTokenSize = max +} + +// Split sets the split function for the PhantomScanner. +// The default split function is ScanLines. +// +// Split panics if it is called after scanning has started. +func (s *PhantomScanner) Split(split SplitFunc) { + if s.scanCalled { + panic("Split called after Scan") + } + s.split = split +} + +// Split functions + +// ScanBytes is a split function for a PhantomScanner that returns each byte as a token. +func ScanBytes(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + return 1, data[0:1], nil +} + +var errorRune = []byte(string(utf8.RuneError)) + +// ScanRunes is a split function for a PhantomScanner that returns each +// UTF-8-encoded rune as a token. The sequence of runes returned is +// equivalent to that from a range loop over the input as a string, which +// means that erroneous UTF-8 encodings translate to U+FFFD = "\xef\xbf\xbd". +// Because of the Scan interface, this makes it impossible for the client to +// distinguish correctly encoded replacement runes from encoding errors. +func ScanRunes(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + + // Fast path 1: ASCII. + if data[0] < utf8.RuneSelf { + return 1, data[0:1], nil + } + + // Fast path 2: Correct UTF-8 decode without error. + _, width := utf8.DecodeRune(data) + if width > 1 { + // It's a valid encoding. Width cannot be one for a correctly encoded + // non-ASCII rune. + return width, data[0:width], nil + } + + // We know it's an error: we have width==1 and implicitly r==utf8.RuneError. + // Is the error because there wasn't a full rune to be decoded? + // FullRune distinguishes correctly between erroneous and incomplete encodings. + if !atEOF && !utf8.FullRune(data) { + // Incomplete; get more bytes. + return 0, nil, nil + } + + // We have a real UTF-8 encoding error. Return a properly encoded error rune + // but advance only one byte. This matches the behavior of a range loop over + // an incorrectly encoded string. + return 1, errorRune, nil +} + +// dropCR drops a terminal \r from the data. +func dropCR(data []byte) []byte { + if len(data) > 0 && data[len(data)-1] == '\r' { + return data[0 : len(data)-1] + } + return data +} + +// ScanLines is a split function for a PhantomScanner that returns each line of +// text, stripped of any trailing end-of-line marker. The returned line may +// be empty. The end-of-line marker is one optional carriage return followed +// by one mandatory newline. In regular expression notation, it is `\r?\n`. +// The last non-empty line of input will be returned even if it has no +// newline. +func ScanLines(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + if i := bytes.IndexByte(data, '\n'); i >= 0 { + // We have a full newline-terminated line. + return i + 1, dropCR(data[0:i]), nil + } + // If we're at EOF, we have a final, non-terminated line. Return it. + if atEOF { + return len(data), dropCR(data), nil + } + // Request more data. + return 0, nil, nil +} + +// isSpace reports whether the character is a Unicode white space character. +// We avoid dependency on the unicode package, but check validity of the implementation +// in the tests. +func isSpace(r rune) bool { + if r <= '\u00FF' { + // Obvious ASCII ones: \t through \r plus space. Plus two Latin-1 oddballs. + switch r { + case ' ', '\t', '\n', '\v', '\f', '\r': + return true + case '\u0085', '\u00A0': + return true + } + return false + } + // High-valued ones. + if '\u2000' <= r && r <= '\u200a' { + return true + } + switch r { + case '\u1680', '\u2028', '\u2029', '\u202f', '\u205f', '\u3000': + return true + } + return false +} + +// ScanWords is a split function for a PhantomScanner that returns each +// space-separated word of text, with surrounding spaces deleted. It will +// never return an empty string. The definition of space is set by +// unicode.IsSpace. +func ScanWords(data []byte, atEOF bool) (advance int, token []byte, err error) { + // Skip leading spaces. + start := 0 + for width := 0; start < len(data); start += width { + var r rune + r, width = utf8.DecodeRune(data[start:]) + if !isSpace(r) { + break + } + } + // Scan until space, marking end of word. + for width, i := 0, start; i < len(data); i += width { + var r rune + r, width = utf8.DecodeRune(data[i:]) + if isSpace(r) { + return i + width, data[start:i], nil + } + } + // If we're at EOF, we have a final, non-empty, non-terminated word. Return it. + if atEOF && len(data) > start { + return len(data), data[start:], nil + } + // Request more data. + return start, nil, nil +} From 1bee947b5ed917e0d30cedaad5a775c95936d4c5 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 16:48:21 -0700 Subject: [PATCH 07/20] Better way of dealing with max token size --- phantom.go | 30 +++- phantomScanner.go | 410 ---------------------------------------------- 2 files changed, 23 insertions(+), 417 deletions(-) delete mode 100644 phantomScanner.go diff --git a/phantom.go b/phantom.go index ee71b3b..c186749 100644 --- a/phantom.go +++ b/phantom.go @@ -1,6 +1,7 @@ package phantomjs import ( + "bufio" "encoding/json" "errors" "fmt" @@ -27,6 +28,8 @@ var nbInstance = 0 var wrapperFileName = "" var fileLock = new(sync.Mutex) +var maxScannerTokenSize = 2048 + /* Start create phantomjs file Create a new `Phantomjs` instance and return it as a pointer. @@ -121,6 +124,18 @@ func (p *Phantom) ForceShutdown() error { return nil } +/* +SetMaxTokenSize will set the max Scanner Token Size +If your script will return a large input use this. +Specify the number of KB +Default value is 2048KB +*/ +func (p *Phantom) SetMaxTokenSize(tokenSize int) { + if tokenSize > 0 { + maxScannerTokenSize = tokenSize + } +} + /* Run the javascript function passed as a string and wait for the result. @@ -133,8 +148,9 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { if err != nil { return err } - scannerOut := NewPhantomScanner(p.out) - scannerErrorOut := NewPhantomScanner(p.errout) + + scannerOut := bufio.NewScanner(bufio.NewReaderSize(p.out, maxScannerTokenSize*1024)) + scannerErrorOut := bufio.NewScanner(bufio.NewReaderSize(p.errout, maxScannerTokenSize*1024)) resMsg := make(chan string) errMsg := make(chan error) go func() { @@ -207,19 +223,19 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readScanner(scannerLock *sync.Mutex, scanner *PhantomScanner) (string, error) { +func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { read := true for read { scannerLock.Lock() read := scanner.Scan() scannerLock.Unlock() - if !read { - return "", errors.New("phantomjs instance is no longer running") - } if scanner.Err() != nil { return "", scanner.Err() } + if !read { + break + } line := scanner.Text() parts := strings.SplitN(line, " ", 2) @@ -233,7 +249,7 @@ func readScanner(scannerLock *sync.Mutex, scanner *PhantomScanner) (string, erro // return "", errors.New("Error reading response, just got a space") } } - return "", errors.New("No Response") + return "", errors.New("PhantomJS Error, Instance is no longer running, try increasing max token size") } func (p *Phantom) sendLine(lines ...string) error { diff --git a/phantomScanner.go b/phantomScanner.go deleted file mode 100644 index 954d793..0000000 --- a/phantomScanner.go +++ /dev/null @@ -1,410 +0,0 @@ -// Copyright 2013 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 phantomjs - -import ( - "bytes" - "errors" - "io" - "unicode/utf8" -) - -// PhantomScanner provides a convenient interface for reading data such as -// a file of newline-delimited lines of text. Successive calls to -// the Scan method will step through the 'tokens' of a file, skipping -// the bytes between the tokens. The specification of a token is -// defined by a split function of type SplitFunc; the default split -// function breaks the input into lines with line termination stripped. Split -// functions are defined in this package for scanning a file into -// lines, bytes, UTF-8-encoded runes, and space-delimited words. The -// client may instead provide a custom split function. -// -// Scanning stops unrecoverably at EOF, the first I/O error, or a token too -// large to fit in the buffer. When a scan stops, the reader may have -// advanced arbitrarily far past the last token. Programs that need more -// control over error handling or large tokens, or must run sequential scans -// on a reader, should use bufio.Reader instead. -// -type PhantomScanner struct { - r io.Reader // The reader provided by the client. - split SplitFunc // The function to split the tokens. - maxTokenSize int // Maximum size of a token; modified by tests. - token []byte // Last token returned by split. - buf []byte // Buffer used as argument to split. - start int // First non-processed byte in buf. - end int // End of data in buf. - err error // Sticky error. - empties int // Count of successive empty tokens. - scanCalled bool // Scan has been called; buffer is in use. - done bool // Scan has finished. -} - -// SplitFunc is the signature of the split function used to tokenize the -// input. The arguments are an initial substring of the remaining unprocessed -// data and a flag, atEOF, that reports whether the Reader has no more data -// to give. The return values are the number of bytes to advance the input -// and the next token to return to the user, plus an error, if any. If the -// data does not yet hold a complete token, for instance if it has no newline -// while scanning lines, SplitFunc can return (0, nil, nil) to signal the -// PhantomScanner to read more data into the slice and try again with a longer slice -// starting at the same point in the input. -// -// If the returned error is non-nil, scanning stops and the error -// is returned to the client. -// -// The function is never called with an empty data slice unless atEOF -// is true. If atEOF is true, however, data may be non-empty and, -// as always, holds unprocessed text. -type SplitFunc func(data []byte, atEOF bool) (advance int, token []byte, err error) - -// Errors returned by PhantomScanner. -var ( - ErrTooLong = errors.New("bufio.PhantomScanner: token too long") - ErrNegativeAdvance = errors.New("bufio.PhantomScanner: SplitFunc returns negative advance count") - ErrAdvanceTooFar = errors.New("bufio.PhantomScanner: SplitFunc returns advance count beyond input") -) - -const ( - // MaxScanTokenSize is the maximum size used to buffer a token - // unless the user provides an explicit buffer with Scan.Buffer. - // The actual maximum token size may be smaller as the buffer - // may need to include, for instance, a newline. - MaxScanTokenSize = 2048 * 1024 - - startBufSize = 4096 // Size of initial allocation for buffer. - maxConsecutiveEmptyReads = 1024 -) - -// NewPhantomScanner returns a new PhantomScanner to read from r. -// The split function defaults to ScanLines. -func NewPhantomScanner(r io.Reader) *PhantomScanner { - return &PhantomScanner{ - r: r, - split: ScanLines, - maxTokenSize: MaxScanTokenSize, - } -} - -// Err returns the first non-EOF error that was encountered by the PhantomScanner. -func (s *PhantomScanner) Err() error { - if s.err == io.EOF { - return nil - } - return s.err -} - -// Bytes returns the most recent token generated by a call to Scan. -// The underlying array may point to data that will be overwritten -// by a subsequent call to Scan. It does no allocation. -func (s *PhantomScanner) Bytes() []byte { - return s.token -} - -// Text returns the most recent token generated by a call to Scan -// as a newly allocated string holding its bytes. -func (s *PhantomScanner) Text() string { - return string(s.token) -} - -// ErrFinalToken is a special sentinel error value. It is intended to be -// returned by a Split function to indicate that the token being delivered -// with the error is the last token and scanning should stop after this one. -// After ErrFinalToken is received by Scan, scanning stops with no error. -// The value is useful to stop processing early or when it is necessary to -// deliver a final empty token. One could achieve the same behavior -// with a custom error value but providing one here is tidier. -// See the emptyFinalToken example for a use of this value. -var ErrFinalToken = errors.New("final token") - -// Scan advances the PhantomScanner to the next token, which will then be -// available through the Bytes or Text method. It returns false when the -// scan stops, either by reaching the end of the input or an error. -// After Scan returns false, the Err method will return any error that -// occurred during scanning, except that if it was io.EOF, Err -// will return nil. -// Scan panics if the split function returns 100 empty tokens without -// advancing the input. This is a common error mode for PhantomScanners. -func (s *PhantomScanner) Scan() bool { - if s.done { - return false - } - s.scanCalled = true - // Loop until we have a token. - for { - // See if we can get a token with what we already have. - // If we've run out of data but have an error, give the split function - // a chance to recover any remaining, possibly empty token. - if s.end > s.start || s.err != nil { - advance, token, err := s.split(s.buf[s.start:s.end], s.err != nil) - if err != nil { - if err == ErrFinalToken { - s.token = token - s.done = true - return true - } - s.setErr(err) - return false - } - if !s.advance(advance) { - return false - } - s.token = token - if token != nil { - if s.err == nil || advance > 0 { - s.empties = 0 - } else { - // Returning tokens not advancing input at EOF. - s.empties++ - if s.empties > 100 { - panic("bufio.Scan: 100 empty tokens without progressing") - } - } - return true - } - } - // We cannot generate a token with what we are holding. - // If we've already hit EOF or an I/O error, we are done. - if s.err != nil { - // Shut it down. - s.start = 0 - s.end = 0 - return false - } - // Must read more data. - // First, shift data to beginning of buffer if there's lots of empty space - // or space is needed. - if s.start > 0 && (s.end == len(s.buf) || s.start > len(s.buf)/2) { - copy(s.buf, s.buf[s.start:s.end]) - s.end -= s.start - s.start = 0 - } - // Is the buffer full? If so, resize. - if s.end == len(s.buf) { - // Guarantee no overflow in the multiplication below. - const maxInt = int(^uint(0) >> 1) - if len(s.buf) >= s.maxTokenSize || len(s.buf) > maxInt/2 { - s.setErr(ErrTooLong) - return false - } - newSize := len(s.buf) * 2 - if newSize == 0 { - newSize = startBufSize - } - if newSize > s.maxTokenSize { - newSize = s.maxTokenSize - } - newBuf := make([]byte, newSize) - copy(newBuf, s.buf[s.start:s.end]) - s.buf = newBuf - s.end -= s.start - s.start = 0 - continue - } - // Finally we can read some input. Make sure we don't get stuck with - // a misbehaving Reader. Officially we don't need to do this, but let's - // be extra careful: PhantomScanner is for safe, simple jobs. - for loop := 0; ; { - n, err := s.r.Read(s.buf[s.end:len(s.buf)]) - s.end += n - if err != nil { - s.setErr(err) - break - } - if n > 0 { - s.empties = 0 - break - } - loop++ - if loop > maxConsecutiveEmptyReads { - s.setErr(io.ErrNoProgress) - break - } - } - } -} - -// advance consumes n bytes of the buffer. It reports whether the advance was legal. -func (s *PhantomScanner) advance(n int) bool { - if n < 0 { - s.setErr(ErrNegativeAdvance) - return false - } - if n > s.end-s.start { - s.setErr(ErrAdvanceTooFar) - return false - } - s.start += n - return true -} - -// setErr records the first error encountered. -func (s *PhantomScanner) setErr(err error) { - if s.err == nil || s.err == io.EOF { - s.err = err - } -} - -// Buffer sets the initial buffer to use when scanning and the maximum -// size of buffer that may be allocated during scanning. The maximum -// token size is the larger of max and cap(buf). If max <= cap(buf), -// Scan will use this buffer only and do no allocation. -// -// By default, Scan uses an internal buffer and sets the -// maximum token size to MaxScanTokenSize. -// -// Buffer panics if it is called after scanning has started. -func (s *PhantomScanner) Buffer(buf []byte, max int) { - if s.scanCalled { - panic("Buffer called after Scan") - } - s.buf = buf[0:cap(buf)] - s.maxTokenSize = max -} - -// Split sets the split function for the PhantomScanner. -// The default split function is ScanLines. -// -// Split panics if it is called after scanning has started. -func (s *PhantomScanner) Split(split SplitFunc) { - if s.scanCalled { - panic("Split called after Scan") - } - s.split = split -} - -// Split functions - -// ScanBytes is a split function for a PhantomScanner that returns each byte as a token. -func ScanBytes(data []byte, atEOF bool) (advance int, token []byte, err error) { - if atEOF && len(data) == 0 { - return 0, nil, nil - } - return 1, data[0:1], nil -} - -var errorRune = []byte(string(utf8.RuneError)) - -// ScanRunes is a split function for a PhantomScanner that returns each -// UTF-8-encoded rune as a token. The sequence of runes returned is -// equivalent to that from a range loop over the input as a string, which -// means that erroneous UTF-8 encodings translate to U+FFFD = "\xef\xbf\xbd". -// Because of the Scan interface, this makes it impossible for the client to -// distinguish correctly encoded replacement runes from encoding errors. -func ScanRunes(data []byte, atEOF bool) (advance int, token []byte, err error) { - if atEOF && len(data) == 0 { - return 0, nil, nil - } - - // Fast path 1: ASCII. - if data[0] < utf8.RuneSelf { - return 1, data[0:1], nil - } - - // Fast path 2: Correct UTF-8 decode without error. - _, width := utf8.DecodeRune(data) - if width > 1 { - // It's a valid encoding. Width cannot be one for a correctly encoded - // non-ASCII rune. - return width, data[0:width], nil - } - - // We know it's an error: we have width==1 and implicitly r==utf8.RuneError. - // Is the error because there wasn't a full rune to be decoded? - // FullRune distinguishes correctly between erroneous and incomplete encodings. - if !atEOF && !utf8.FullRune(data) { - // Incomplete; get more bytes. - return 0, nil, nil - } - - // We have a real UTF-8 encoding error. Return a properly encoded error rune - // but advance only one byte. This matches the behavior of a range loop over - // an incorrectly encoded string. - return 1, errorRune, nil -} - -// dropCR drops a terminal \r from the data. -func dropCR(data []byte) []byte { - if len(data) > 0 && data[len(data)-1] == '\r' { - return data[0 : len(data)-1] - } - return data -} - -// ScanLines is a split function for a PhantomScanner that returns each line of -// text, stripped of any trailing end-of-line marker. The returned line may -// be empty. The end-of-line marker is one optional carriage return followed -// by one mandatory newline. In regular expression notation, it is `\r?\n`. -// The last non-empty line of input will be returned even if it has no -// newline. -func ScanLines(data []byte, atEOF bool) (advance int, token []byte, err error) { - if atEOF && len(data) == 0 { - return 0, nil, nil - } - if i := bytes.IndexByte(data, '\n'); i >= 0 { - // We have a full newline-terminated line. - return i + 1, dropCR(data[0:i]), nil - } - // If we're at EOF, we have a final, non-terminated line. Return it. - if atEOF { - return len(data), dropCR(data), nil - } - // Request more data. - return 0, nil, nil -} - -// isSpace reports whether the character is a Unicode white space character. -// We avoid dependency on the unicode package, but check validity of the implementation -// in the tests. -func isSpace(r rune) bool { - if r <= '\u00FF' { - // Obvious ASCII ones: \t through \r plus space. Plus two Latin-1 oddballs. - switch r { - case ' ', '\t', '\n', '\v', '\f', '\r': - return true - case '\u0085', '\u00A0': - return true - } - return false - } - // High-valued ones. - if '\u2000' <= r && r <= '\u200a' { - return true - } - switch r { - case '\u1680', '\u2028', '\u2029', '\u202f', '\u205f', '\u3000': - return true - } - return false -} - -// ScanWords is a split function for a PhantomScanner that returns each -// space-separated word of text, with surrounding spaces deleted. It will -// never return an empty string. The definition of space is set by -// unicode.IsSpace. -func ScanWords(data []byte, atEOF bool) (advance int, token []byte, err error) { - // Skip leading spaces. - start := 0 - for width := 0; start < len(data); start += width { - var r rune - r, width = utf8.DecodeRune(data[start:]) - if !isSpace(r) { - break - } - } - // Scan until space, marking end of word. - for width, i := 0, start; i < len(data); i += width { - var r rune - r, width = utf8.DecodeRune(data[i:]) - if isSpace(r) { - return i + width, data[start:i], nil - } - } - // If we're at EOF, we have a final, non-empty, non-terminated word. Return it. - if atEOF && len(data) > start { - return len(data), data[start:], nil - } - // Request more data. - return start, nil, nil -} From 8ca63f0c970b586f77ca57c336610baebcfb251b Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Tue, 13 Sep 2016 16:55:45 -0700 Subject: [PATCH 08/20] added custom buffer --- phantom.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/phantom.go b/phantom.go index c186749..5331913 100644 --- a/phantom.go +++ b/phantom.go @@ -224,6 +224,8 @@ func createWrapperFile() (fileName string, err error) { } func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { + buf := make([]byte, maxScannerTokenSize) + scanner.Buffer(buf, maxScannerTokenSize) read := true for read { scannerLock.Lock() From 3675dc887266f84d7f3098f7d6a6d76a89dea440 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Wed, 14 Sep 2016 08:41:27 -0700 Subject: [PATCH 09/20] Moved scanner to reader --- phantom.go | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/phantom.go b/phantom.go index 5331913..6aca52f 100644 --- a/phantom.go +++ b/phantom.go @@ -149,8 +149,8 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { return err } - scannerOut := bufio.NewScanner(bufio.NewReaderSize(p.out, maxScannerTokenSize*1024)) - scannerErrorOut := bufio.NewScanner(bufio.NewReaderSize(p.errout, maxScannerTokenSize*1024)) + scannerOut := bufio.NewReaderSize(p.out, maxScannerTokenSize*1024) + scannerErrorOut := bufio.NewReaderSize(p.errout, maxScannerTokenSize*1024) resMsg := make(chan string) errMsg := make(chan error) go func() { @@ -223,23 +223,24 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readScanner(scannerLock *sync.Mutex, scanner *bufio.Scanner) (string, error) { - buf := make([]byte, maxScannerTokenSize) - scanner.Buffer(buf, maxScannerTokenSize) - read := true - for read { +func readScanner(scannerLock *sync.Mutex, reader *bufio.Reader) (string, error) { + for { scannerLock.Lock() - read := scanner.Scan() + data, _, err := reader.ReadLine() scannerLock.Unlock() - if scanner.Err() != nil { - return "", scanner.Err() + if err != nil { + eof := "EOF" + if strings.Compare(eof, err.Error()) == 0 { + return "", errors.New("phantomjs instance is no longer running") + } + return "", err } - if !read { + if len(data) == 0 { break } - line := scanner.Text() + line := string(data) parts := strings.SplitN(line, " ", 2) if strings.HasPrefix(line, "RES") { @@ -258,7 +259,7 @@ func (p *Phantom) sendLine(lines ...string) error { for _, l := range lines { _, err := io.WriteString(p.in, l+"\n") if err != nil { - return errors.New("Cannot Send: `" + l + "`" + "phantomjs instance might be dead") + return errors.New("Cannot Send: `" + l + "` " + "phantomjs instance might be dead") } } return nil From e9bc4bc3c4dbf89c30190ff61a58aa2613ec5dd5 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Wed, 14 Sep 2016 10:41:59 -0700 Subject: [PATCH 10/20] Fixed Errors caused by bad file descriptor --- phantom.go | 76 ++++++++++++++++++++++++++++-------------------------- 1 file changed, 40 insertions(+), 36 deletions(-) diff --git a/phantom.go b/phantom.go index 6aca52f..9f2c9ba 100644 --- a/phantom.go +++ b/phantom.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "io/ioutil" + "log" "os" "os/exec" "strings" @@ -15,12 +16,12 @@ import ( // Phantom a data structure that interacts with the wrapper file type Phantom struct { - cmd *exec.Cmd - in io.WriteCloser - out io.ReadCloser - errout io.ReadCloser - scannerErrorLock *sync.Mutex - scannerLock *sync.Mutex + cmd *exec.Cmd + in io.WriteCloser + out io.ReadCloser + errout io.ReadCloser + readerErrorLock *sync.Mutex + readerLock *sync.Mutex } var nbInstance = 0 @@ -28,7 +29,7 @@ var nbInstance = 0 var wrapperFileName = "" var fileLock = new(sync.Mutex) -var maxScannerTokenSize = 2048 +var readerBufferSize = 2048 /* Start create phantomjs file @@ -62,12 +63,12 @@ func Start(args ...string) (*Phantom, error) { } p := Phantom{ - cmd: cmd, - in: inPipe, - out: outPipe, - errout: errPipe, - scannerErrorLock: new(sync.Mutex), - scannerLock: new(sync.Mutex), + cmd: cmd, + in: inPipe, + out: outPipe, + errout: errPipe, + readerErrorLock: new(sync.Mutex), + readerLock: new(sync.Mutex), } err = cmd.Start() @@ -90,12 +91,13 @@ func (p *Phantom) Exit() error { if err != nil { return err } - p.scannerErrorLock.Lock() - p.scannerLock.Lock() + p.readerErrorLock.Lock() + p.readerLock.Lock() err = p.cmd.Wait() - p.scannerErrorLock.Unlock() - p.scannerLock.Unlock() + p.readerErrorLock.Unlock() + p.readerLock.Unlock() if err != nil { + log.Println(err) return err } fileLock.Lock() @@ -125,14 +127,14 @@ func (p *Phantom) ForceShutdown() error { } /* -SetMaxTokenSize will set the max Scanner Token Size +SetMaxBufferSize will set the max Buffer Size If your script will return a large input use this. Specify the number of KB Default value is 2048KB */ -func (p *Phantom) SetMaxTokenSize(tokenSize int) { - if tokenSize > 0 { - maxScannerTokenSize = tokenSize +func (p *Phantom) SetMaxBufferSize(bufferSize int) { + if bufferSize > 0 { + readerBufferSize = bufferSize } } @@ -149,12 +151,12 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { return err } - scannerOut := bufio.NewReaderSize(p.out, maxScannerTokenSize*1024) - scannerErrorOut := bufio.NewReaderSize(p.errout, maxScannerTokenSize*1024) + readerOut := bufio.NewReaderSize(p.out, readerBufferSize*1024) + readerErrorOut := bufio.NewReaderSize(p.errout, readerBufferSize*1024) resMsg := make(chan string) errMsg := make(chan error) go func() { - value, scanError := readScanner(p.scannerLock, scannerOut) + value, scanError := readreader(p.readerLock, readerOut) if scanError != nil { errMsg <- scanError @@ -169,7 +171,7 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { resMsg <- value }() go func() { - value, scanError := readScanner(p.scannerErrorLock, scannerErrorOut) + value, scanError := readreader(p.readerErrorLock, readerErrorOut) if scanError != nil { errMsg <- scanError return @@ -223,21 +225,23 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readScanner(scannerLock *sync.Mutex, reader *bufio.Reader) (string, error) { +func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { for { - scannerLock.Lock() + readerLock.Lock() data, _, err := reader.ReadLine() - scannerLock.Unlock() + readerLock.Unlock() if err != nil { - eof := "EOF" - if strings.Compare(eof, err.Error()) == 0 { - return "", errors.New("phantomjs instance is no longer running") + // Nothing else to read + if strings.Compare("EOF", err.Error()) == 0 && len(data) == 0 { + break + } + // Nothing to read and waiting to exit + if len(data) == 0 && strings.Contains(err.Error(), "bad file descriptor") { + break + } else if strings.Compare("EOF", err.Error()) != 0 && len(data) == 0 { + return "", err } - return "", err - } - if len(data) == 0 { - break } line := string(data) @@ -252,7 +256,7 @@ func readScanner(scannerLock *sync.Mutex, reader *bufio.Reader) (string, error) // return "", errors.New("Error reading response, just got a space") } } - return "", errors.New("PhantomJS Error, Instance is no longer running, try increasing max token size") + return "", errors.New("PhantomJS Error, Instance is no longer running, try increasing max buffer size") } func (p *Phantom) sendLine(lines ...string) error { From 78ba575845d0146bbeba6c39eea939cde7238f29 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Fri, 16 Sep 2016 16:44:25 -0700 Subject: [PATCH 11/20] updated shutdown method --- phantom.go | 12 ++++++++---- phantom_test.go | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/phantom.go b/phantom.go index 9f2c9ba..5423c95 100644 --- a/phantom.go +++ b/phantom.go @@ -114,16 +114,20 @@ ForceShutdown will forcefully kill phantomjs. This will completly terminate the proccess compared to Exit which will safely Exit */ func (p *Phantom) ForceShutdown() error { - if err := p.cmd.Process.Kill(); err != nil { - return err - } + p.readerErrorLock.Lock() + p.readerLock.Lock() + defer p.readerErrorLock.Unlock() + defer p.readerLock.Unlock() + + err := p.cmd.Process.Kill() + fileLock.Lock() nbInstance-- if nbInstance == 0 { os.Remove(wrapperFileName) } fileLock.Unlock() - return nil + return err } /* diff --git a/phantom_test.go b/phantom_test.go index 268b338..5914c7c 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -188,6 +188,44 @@ func TestForceShutdown(t *testing.T) { } } +func TestRestartInstance(t *testing.T) { + p, err := Start() + defer p.Exit() + failOnError(err, t) + + err = p.ForceShutdown() + failOnError(err, t) + + p, err = Start() + defer p.Exit() + failOnError(err, t) + + var r interface{} + + err = p.Run(`function(done){ + var a = 0; + var b = 1; + var c = 0; + for(var i=2; i<=25; i++) + { + c = b + a; + a = b; + b = c; + } + done(c, undefined); + }`, &r) + + failOnError(err, t) + v, ok := r.(float64) + if !ok { + t.Errorf("Should be an int but is %v", r) + return + } + if v != 75025 { + t.Errorf("Should be %d but is %f", 75025, v) + } +} + func TestMultipleLogs(t *testing.T) { p, err := Start() defer p.Exit() From 77d24120b327d60f87414a7a42441d3b6d021f59 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Fri, 16 Sep 2016 16:53:10 -0700 Subject: [PATCH 12/20] removed locks on shutdown --- phantom.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/phantom.go b/phantom.go index 5423c95..28ba2da 100644 --- a/phantom.go +++ b/phantom.go @@ -114,10 +114,6 @@ ForceShutdown will forcefully kill phantomjs. This will completly terminate the proccess compared to Exit which will safely Exit */ func (p *Phantom) ForceShutdown() error { - p.readerErrorLock.Lock() - p.readerLock.Lock() - defer p.readerErrorLock.Unlock() - defer p.readerLock.Unlock() err := p.cmd.Process.Kill() From 7789e5b81372a0855662870866cfc6ddfb655471 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Mon, 19 Sep 2016 10:06:19 -0700 Subject: [PATCH 13/20] better error error messages --- phantom.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/phantom.go b/phantom.go index 28ba2da..22ef361 100644 --- a/phantom.go +++ b/phantom.go @@ -194,6 +194,9 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { } return nil case err := <-errMsg: + if strings.Compare(err.Error(), "EOF") == 0 { + return errors.New("PhantomJS is no longer running") + } return err } } @@ -256,7 +259,7 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { // return "", errors.New("Error reading response, just got a space") } } - return "", errors.New("PhantomJS Error, Instance is no longer running, try increasing max buffer size") + return "", errors.New("EOF") } func (p *Phantom) sendLine(lines ...string) error { From 7c310ef67293ec8eb50964f5657de332d4fb6af5 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Mon, 19 Sep 2016 15:21:03 -0700 Subject: [PATCH 14/20] Added error message --- phantom.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/phantom.go b/phantom.go index 22ef361..d5b3e8e 100644 --- a/phantom.go +++ b/phantom.go @@ -191,6 +191,9 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { if err != nil { return err } + if len([]byte(text)) == 0 { + return errors.New("No Response Data") + } } return nil case err := <-errMsg: From 0b2bc970ea687ad66c9577e18e369c6543b7a223 Mon Sep 17 00:00:00 2001 From: kcajmagic Date: Mon, 19 Sep 2016 15:48:03 -0700 Subject: [PATCH 15/20] fixed reader with EOF --- phantom.go | 49 ++++++++++++++++++++++++++++++++----------------- 1 file changed, 32 insertions(+), 17 deletions(-) diff --git a/phantom.go b/phantom.go index d5b3e8e..3a412de 100644 --- a/phantom.go +++ b/phantom.go @@ -16,12 +16,13 @@ import ( // Phantom a data structure that interacts with the wrapper file type Phantom struct { - cmd *exec.Cmd - in io.WriteCloser - out io.ReadCloser - errout io.ReadCloser - readerErrorLock *sync.Mutex - readerLock *sync.Mutex + cmd *exec.Cmd + in io.WriteCloser + out io.ReadCloser + errout io.ReadCloser + readerErrorLock *sync.Mutex + readerLock *sync.Mutex + nothingReadCount int64 } var nbInstance = 0 @@ -30,6 +31,7 @@ var wrapperFileName = "" var fileLock = new(sync.Mutex) var readerBufferSize = 2048 +var maxReadTimes = 100 /* Start create phantomjs file @@ -63,12 +65,13 @@ func Start(args ...string) (*Phantom, error) { } p := Phantom{ - cmd: cmd, - in: inPipe, - out: outPipe, - errout: errPipe, - readerErrorLock: new(sync.Mutex), - readerLock: new(sync.Mutex), + cmd: cmd, + in: inPipe, + out: outPipe, + errout: errPipe, + readerErrorLock: new(sync.Mutex), + readerLock: new(sync.Mutex), + nothingReadCount: 0, } err = cmd.Start() @@ -138,6 +141,16 @@ func (p *Phantom) SetMaxBufferSize(bufferSize int) { } } +/* +SetMaxReadTimes will set the max read times +If the +*/ +func (p *Phantom) SetMaxReadTimes(bufferSize int) { + if bufferSize > 0 { + readerBufferSize = bufferSize + } +} + /* Run the javascript function passed as a string and wait for the result. @@ -191,9 +204,6 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { if err != nil { return err } - if len([]byte(text)) == 0 { - return errors.New("No Response Data") - } } return nil case err := <-errMsg: @@ -232,6 +242,7 @@ func createWrapperFile() (fileName string, err error) { } func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { + var count = 0 for { readerLock.Lock() data, _, err := reader.ReadLine() @@ -240,11 +251,15 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { if err != nil { // Nothing else to read if strings.Compare("EOF", err.Error()) == 0 && len(data) == 0 { - break + // like the scanner if 100 empty reads panic + count++ + if count >= 100 { + break + } } // Nothing to read and waiting to exit if len(data) == 0 && strings.Contains(err.Error(), "bad file descriptor") { - break + return "", errors.New("Wrapper Error: Bad File Descriptor") } else if strings.Compare("EOF", err.Error()) != 0 && len(data) == 0 { return "", err } From 8dc823216ddb196cddb979c0aee2d449c06ca5fc Mon Sep 17 00:00:00 2001 From: Michael Ernst Date: Sun, 15 Jan 2017 19:55:23 +0100 Subject: [PATCH 16/20] make command customizable --- phantom.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/phantom.go b/phantom.go index 3a412de..33a793c 100644 --- a/phantom.go +++ b/phantom.go @@ -39,7 +39,15 @@ Create a new `Phantomjs` instance and return it as a pointer. If an error occurs during command start, return it instead. */ -func Start(args ...string) (*Phantom, error) { + +var cmd = "phantomjs" + +// SetCommand lets you specify the binary for phantomjs +func SetCommand(cmd string) { + cmd = cmd +} + +func Start(command string, args ...string) (*Phantom, error) { fileLock.Lock() if nbInstance == 0 { wrapperFileName, _ = createWrapperFile() @@ -47,7 +55,7 @@ func Start(args ...string) (*Phantom, error) { nbInstance++ fileLock.Unlock() args = append(args, wrapperFileName) - cmd := exec.Command("phantomjs", args...) + cmd := exec.Command(cmd, args...) inPipe, err := cmd.StdinPipe() if err != nil { From 85853d09a270ab36f958d7620efd21a66b4776a7 Mon Sep 17 00:00:00 2001 From: Michael Ernst Date: Sun, 15 Jan 2017 20:39:40 +0100 Subject: [PATCH 17/20] i am with stupid. i should not have commited that --- phantom.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/phantom.go b/phantom.go index 33a793c..25fed76 100644 --- a/phantom.go +++ b/phantom.go @@ -47,7 +47,7 @@ func SetCommand(cmd string) { cmd = cmd } -func Start(command string, args ...string) (*Phantom, error) { +func Start(args ...string) (*Phantom, error) { fileLock.Lock() if nbInstance == 0 { wrapperFileName, _ = createWrapperFile() From 8064236eb035a74eee5ed3e7d057523633174f5b Mon Sep 17 00:00:00 2001 From: Jack Murdock Date: Wed, 8 Feb 2017 13:06:32 -0800 Subject: [PATCH 18/20] fixed threads leaking --- phantom.go | 82 ++++++++++++++++++++++++++++++++++--------------- phantom_test.go | 75 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 25 deletions(-) diff --git a/phantom.go b/phantom.go index 3a412de..f5ef912 100644 --- a/phantom.go +++ b/phantom.go @@ -10,6 +10,7 @@ import ( "log" "os" "os/exec" + "runtime" "strings" "sync" ) @@ -25,6 +26,12 @@ type Phantom struct { nothingReadCount int64 } +// return Value is used to bundle up the value returned from the reader +type returnValue struct { + val string + err error +} + var nbInstance = 0 var wrapperFileName = "" @@ -166,37 +173,51 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { readerOut := bufio.NewReaderSize(p.out, readerBufferSize*1024) readerErrorOut := bufio.NewReaderSize(p.errout, readerBufferSize*1024) - resMsg := make(chan string) - errMsg := make(chan error) + resMsg := make(chan string, 1) + errMsg := make(chan error, 1) + quit := make(chan bool, 1) + + // var wg sync.WaitGroup + // wg.Add(1) go func() { - value, scanError := readreader(p.readerLock, readerOut) + select { + case value := <-readreader(p.readerLock, readerOut): - if scanError != nil { - errMsg <- scanError - return - } + if value.err != nil { + errMsg <- value.err + return + } + + if value.val == "" { + // Nothing to see here + return + } - if value == "" { - // Nothing to see here + resMsg <- value.val + case <-quit: return } - - resMsg <- value }() go func() { - value, scanError := readreader(p.readerErrorLock, readerErrorOut) - if scanError != nil { - errMsg <- scanError - return - } + select { + case value := <-readreader(p.readerErrorLock, readerErrorOut): + if value.err != nil { + errMsg <- value.err + return + } + + if value.val == "" { + // Nothing to see here + return + } - if value == "" { - // Nothing to see here + errMsg <- errors.New(value.val) + case <-quit: return } - - errMsg <- errors.New(value) }() + + // Wait for Threads to complete select { case text := <-resMsg: if res != nil { @@ -205,11 +226,17 @@ func (p *Phantom) Run(jsFunc string, res *interface{}) error { return err } } + log.Printf("ret val res GT %d", runtime.NumGoroutine()) + + // wg.Wait() return nil case err := <-errMsg: if strings.Compare(err.Error(), "EOF") == 0 { return errors.New("PhantomJS is no longer running") } + log.Printf("ret val err GT %d", runtime.NumGoroutine()) + + // wg.Wait() return err } } @@ -241,8 +268,9 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { +func readreader(readerLock *sync.Mutex, reader *bufio.Reader) chan returnValue { var count = 0 + retVal := make(chan returnValue, 1) for { readerLock.Lock() data, _, err := reader.ReadLine() @@ -259,9 +287,11 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { } // Nothing to read and waiting to exit if len(data) == 0 && strings.Contains(err.Error(), "bad file descriptor") { - return "", errors.New("Wrapper Error: Bad File Descriptor") + retVal <- returnValue{"", errors.New("Wrapper Error: Bad File Descriptor")} + return retVal } else if strings.Compare("EOF", err.Error()) != 0 && len(data) == 0 { - return "", err + retVal <- returnValue{"", err} + return retVal } } @@ -269,7 +299,8 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { parts := strings.SplitN(line, " ", 2) if strings.HasPrefix(line, "RES") { - return parts[1], nil + retVal <- returnValue{parts[1], nil} + return retVal } else if line != "" { fmt.Printf("LOG %s\n", line) // return "", nil @@ -277,7 +308,8 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { // return "", errors.New("Error reading response, just got a space") } } - return "", errors.New("EOF") + retVal <- returnValue{"", errors.New("EOF")} + return retVal } func (p *Phantom) sendLine(lines ...string) error { diff --git a/phantom_test.go b/phantom_test.go index 5914c7c..796719e 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -2,6 +2,7 @@ package phantomjs import ( "log" + "runtime" "strings" "sync" "testing" @@ -226,6 +227,80 @@ func TestRestartInstance(t *testing.T) { } } +func TestMutlipleThreads(t *testing.T) { + + var wg sync.WaitGroup + var stats runtime.MemStats + for i := 0; i < 1; i++ { + wg.Add(1) + + runtime.ReadMemStats(&stats) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Starting GO ROUTINE") + go func() { + var insideStats runtime.MemStats + runtime.ReadMemStats(&insideStats) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Start of Processor") + // Staring new Thread + p, err := Start() + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Started Processor") + + failOnError(err, t) + var r interface{} + for j := 0; j < 5; j++ { + runtime.ReadMemStats(&insideStats) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Starting RUN && BEGIN OF LOOP") + err = p.Run(`function(done){ + var a = 0; + var b = 1; + var c = 0; + var page = new WebPage(); + page.open('http://charles.lescampeurs.org/'); + page.onLoadFinished = function(status) { + for(var i=2; i<=25; i++) + { + c = b + a; + a = b; + b = c; + } + done(c, undefined); + } + }`, &r) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending RUN") + + if err == nil { + v, ok := r.(float64) + if !ok { + t.Errorf("Should be an int but is %v", r) + p.Exit() + defer wg.Done() + return + } + if v != 75025 { + t.Errorf("Should be %d but is %f", 75025, v) + } + } else { + if !strings.Contains(err.Error(), "no longer running") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { + t.Fatal(err) + } + } + runtime.ReadMemStats(&insideStats) + + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending Loop \n\n\n ") + + } + p.Exit() + defer wg.Done() + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Done with Processor") + //Thread Done + }() + runtime.ReadMemStats(&stats) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Started GO ROUTINE") + } + wg.Wait() + runtime.ReadMemStats(&stats) + log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Done") +} + func TestMultipleLogs(t *testing.T) { p, err := Start() defer p.Exit() From 88d35f294999d7963e5ff0dce80300cbce9816ed Mon Sep 17 00:00:00 2001 From: Jack Murdock Date: Thu, 9 Feb 2017 10:16:04 -0800 Subject: [PATCH 19/20] fixed threads leaking --- phantom.go | 318 +++++++++++++++++++++++++++++------------------- phantom_test.go | 37 +++--- 2 files changed, 211 insertions(+), 144 deletions(-) diff --git a/phantom.go b/phantom.go index f5ef912..5525964 100644 --- a/phantom.go +++ b/phantom.go @@ -10,19 +10,25 @@ import ( "log" "os" "os/exec" - "runtime" "strings" "sync" ) // Phantom a data structure that interacts with the wrapper file type Phantom struct { - cmd *exec.Cmd - in io.WriteCloser - out io.ReadCloser - errout io.ReadCloser - readerErrorLock *sync.Mutex - readerLock *sync.Mutex + cmd *exec.Cmd + in io.WriteCloser + out io.ReadCloser + errout io.ReadCloser + readerErrorLock *sync.Mutex + readerLock *sync.Mutex + readerOut *bufio.Reader + readerErr *bufio.Reader + lineOut chan string + lineErr chan string + quit chan bool + stopReading bool + nothingReadCount int64 } @@ -32,6 +38,12 @@ type returnValue struct { err error } +// return Value is used to bundle up the value returned from the reader +type exitValue struct { + val []byte + err error +} + var nbInstance = 0 var wrapperFileName = "" @@ -78,45 +90,164 @@ func Start(args ...string) (*Phantom, error) { errout: errPipe, readerErrorLock: new(sync.Mutex), readerLock: new(sync.Mutex), + lineOut: make(chan string, 100), + lineErr: make(chan string, 100), + quit: make(chan bool, 1), + stopReading: false, nothingReadCount: 0, } - err = cmd.Start() + p.readerOut = bufio.NewReaderSize(p.out, readerBufferSize*1024) + p.readerErr = bufio.NewReaderSize(p.errout, readerBufferSize*1024) + + go p.readFromSTDOut() + go p.readFromSTDERR() + err = cmd.Start() if err != nil { return nil, err } + // time.Sleep(time.Millisecond) return &p, nil } +func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { + var count = 0 + for { + readerLock.Lock() + data, _, err := reader.ReadLine() + readerLock.Unlock() + + if err != nil { + // Nothing else to read + if err == io.EOF && len(data) == 0 { + // like the scanner if 100 empty reads panic + count++ + if count >= 100 { + break + } + } + // Nothing to read and waiting to exit + if len(data) == 0 && strings.Contains(err.Error(), "bad file descriptor") { + return "", errors.New("Wrapper Error: Bad File Descriptor") + } else if err == io.EOF && len(data) == 0 { + return "", err + } + } + + line := string(data) + parts := strings.SplitN(line, " ", 2) + + if strings.HasPrefix(line, "RES") { + return parts[1], nil + } else if line != "" { + fmt.Printf("LOG %s\n", line) + // return "", nil + } else if line != " " { + // return "", errors.New("Error reading response, just got a space") + } + } + return "", errors.New("EOF") +} + +func (p *Phantom) readFromSTDOut() { + for !p.stopReading { + line, err := readreader(p.readerLock, p.readerOut) + if err == io.EOF && line == "" { + // Done Reading Data + return + } else if err != nil && !p.stopReading { + // p.Exit() + log.Printf("An error occurred while reading stdout: %+v", err) + return + } else if p.stopReading && line == "" { + return + } else if line == "" { + continue + } + + select { + case p.lineOut <- line: + default: + log.Println("no listener attached to stdout " + line) + } + } +} + +func (p *Phantom) readFromSTDERR() { + for !p.stopReading { + line, err := readreader(p.readerErrorLock, p.readerErr) + if err == io.EOF && line == "" { + // Done Reading Data + return + } else if err != nil && !p.stopReading { + // p.Exit() + log.Printf("an error occurred while reading stderr: %+v", err) + return + } else if p.stopReading && line == "" { + return + } else if line == "" { + continue + } + + select { + case p.lineErr <- line: + default: + log.Println("no listener attached to sderr " + line) + } + } +} + +func drainBool(commch chan bool) { + for { + select { + case <-commch: + default: + return + } + } +} + /* Exit Phantomjs by sending the "phantomjs.exit()" command and wait for the command to end. -Return an error if one occured during exit command or if the program output a error value +Return an error if one occurred during exit command or if the program output a error value */ func (p *Phantom) Exit() error { err := p.Load("phantom.exit()") + stoping := false + if !p.stopReading { + drainBool(p.quit) + p.quit <- true + stoping = true + } + p.stopReading = true if err != nil { return err } + p.readerErrorLock.Lock() p.readerLock.Lock() err = p.cmd.Wait() p.readerErrorLock.Unlock() p.readerLock.Unlock() if err != nil { - log.Println(err) - return err + if !strings.Contains(err.Error(), "signal: killed") { + return errors.New("Failed to kill instance :" + err.Error()) + } + err = nil } - fileLock.Lock() - nbInstance-- - if nbInstance == 0 { - os.Remove(wrapperFileName) + if !stoping { + fileLock.Lock() + nbInstance-- + if nbInstance == 0 { + err = os.Remove(wrapperFileName) + } + fileLock.Unlock() } - fileLock.Unlock() - return nil + return err } /* @@ -124,15 +255,30 @@ ForceShutdown will forcefully kill phantomjs. This will completly terminate the proccess compared to Exit which will safely Exit */ func (p *Phantom) ForceShutdown() error { + stoping := false + if !p.stopReading { + drainBool(p.quit) + p.quit <- true + stoping = true + } + p.stopReading = true err := p.cmd.Process.Kill() + if err != nil { + if !strings.Contains(err.Error(), "signal: killed") { + return errors.New("Failed to kill instance :" + err.Error()) + } + err = nil + } - fileLock.Lock() - nbInstance-- - if nbInstance == 0 { - os.Remove(wrapperFileName) + if !stoping { + fileLock.Lock() + nbInstance-- + if nbInstance == 0 { + err = os.Remove(wrapperFileName) + } + fileLock.Unlock() } - fileLock.Unlock() return err } @@ -158,6 +304,16 @@ func (p *Phantom) SetMaxReadTimes(bufferSize int) { } } +func drainchan(commch chan string) { + for { + select { + case <-commch: + default: + return + } + } +} + /* Run the javascript function passed as a string and wait for the result. @@ -166,79 +322,33 @@ You can pass a function a closure which can take two arguments 1st is the succes the 2nd is an err. See TestComplex in the phantom_test.go */ func (p *Phantom) Run(jsFunc string, res *interface{}) error { + if p.stopReading { + return errors.New("PhantomJS Instance is dead") + } + // flushing channel incase it read some left over data + drainchan(p.lineOut) + drainchan(p.lineErr) + // end flush err := p.sendLine("RUN", jsFunc, "END") if err != nil { return err } - - readerOut := bufio.NewReaderSize(p.out, readerBufferSize*1024) - readerErrorOut := bufio.NewReaderSize(p.errout, readerBufferSize*1024) - resMsg := make(chan string, 1) - errMsg := make(chan error, 1) - quit := make(chan bool, 1) - - // var wg sync.WaitGroup - // wg.Add(1) - go func() { - select { - case value := <-readreader(p.readerLock, readerOut): - - if value.err != nil { - errMsg <- value.err - return - } - - if value.val == "" { - // Nothing to see here - return - } - - resMsg <- value.val - case <-quit: - return - } - }() - go func() { - select { - case value := <-readreader(p.readerErrorLock, readerErrorOut): - if value.err != nil { - errMsg <- value.err - return - } - - if value.val == "" { - // Nothing to see here - return - } - - errMsg <- errors.New(value.val) - case <-quit: - return - } - }() - - // Wait for Threads to complete select { - case text := <-resMsg: + case text := <-p.lineOut: if res != nil { + err = json.Unmarshal([]byte(text), res) if err != nil { return err } } - log.Printf("ret val res GT %d", runtime.NumGoroutine()) - - // wg.Wait() return nil - case err := <-errMsg: - if strings.Compare(err.Error(), "EOF") == 0 { - return errors.New("PhantomJS is no longer running") - } - log.Printf("ret val err GT %d", runtime.NumGoroutine()) - - // wg.Wait() - return err + case errLine := <-p.lineErr: + return errors.New(errLine) + case <-p.quit: + return errors.New("PhantomJS Instance Killed") } + } /* @@ -268,50 +378,6 @@ func createWrapperFile() (fileName string, err error) { return wrapper.Name(), nil } -func readreader(readerLock *sync.Mutex, reader *bufio.Reader) chan returnValue { - var count = 0 - retVal := make(chan returnValue, 1) - for { - readerLock.Lock() - data, _, err := reader.ReadLine() - readerLock.Unlock() - - if err != nil { - // Nothing else to read - if strings.Compare("EOF", err.Error()) == 0 && len(data) == 0 { - // like the scanner if 100 empty reads panic - count++ - if count >= 100 { - break - } - } - // Nothing to read and waiting to exit - if len(data) == 0 && strings.Contains(err.Error(), "bad file descriptor") { - retVal <- returnValue{"", errors.New("Wrapper Error: Bad File Descriptor")} - return retVal - } else if strings.Compare("EOF", err.Error()) != 0 && len(data) == 0 { - retVal <- returnValue{"", err} - return retVal - } - } - - line := string(data) - parts := strings.SplitN(line, " ", 2) - - if strings.HasPrefix(line, "RES") { - retVal <- returnValue{parts[1], nil} - return retVal - } else if line != "" { - fmt.Printf("LOG %s\n", line) - // return "", nil - } else if line != " " { - // return "", errors.New("Error reading response, just got a space") - } - } - retVal <- returnValue{"", errors.New("EOF")} - return retVal -} - func (p *Phantom) sendLine(lines ...string) error { for _, l := range lines { _, err := io.WriteString(p.in, l+"\n") diff --git a/phantom_test.go b/phantom_test.go index 796719e..768b493 100644 --- a/phantom_test.go +++ b/phantom_test.go @@ -1,7 +1,6 @@ package phantomjs import ( - "log" "runtime" "strings" "sync" @@ -127,7 +126,7 @@ func TestComplex(t *testing.T) { } done(c, undefined); }`, &r) - log.Println("Completed Run in", time.Since(begin)) + t.Logf("Completed Run in %s", time.Since(begin)) failOnError(err, t) v, ok := r.(float64) if !ok { @@ -145,14 +144,14 @@ func TestComplex(t *testing.T) { func TestForceShutdown(t *testing.T) { p, err := Start() failOnError(err, t) - defer p.Exit() count := 0 for i := 0; i < 5; i++ { var r interface{} if i == 2 { - p.ForceShutdown() + err = p.ForceShutdown() + failOnError(err, t) } err = p.Run(`function(done){ var a = 0; @@ -166,7 +165,6 @@ func TestForceShutdown(t *testing.T) { } done(c, undefined); }`, &r) - if err == nil { v, ok := r.(float64) if !ok { @@ -179,14 +177,16 @@ func TestForceShutdown(t *testing.T) { count++ } } else { - if !strings.Contains(err.Error(), "no longer running") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { + if !strings.Contains(err.Error(), "PhantomJS Instance is dead") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { t.Fatal(err) } } } if count != 2 { - t.Fatalf("Didn' reach distination %d", count) + t.Fatalf("Didn't reach destination %d", count) } + err = p.Exit() + failOnError(err, t) } func TestRestartInstance(t *testing.T) { @@ -231,24 +231,25 @@ func TestMutlipleThreads(t *testing.T) { var wg sync.WaitGroup var stats runtime.MemStats - for i := 0; i < 1; i++ { + for i := 0; i < 3; i++ { wg.Add(1) runtime.ReadMemStats(&stats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Starting GO ROUTINE") + + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Starting GO ROUTINE") go func() { var insideStats runtime.MemStats runtime.ReadMemStats(&insideStats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Start of Processor") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Start of Processor") // Staring new Thread p, err := Start() - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Started Processor") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Started Processor") failOnError(err, t) var r interface{} for j := 0; j < 5; j++ { runtime.ReadMemStats(&insideStats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Starting RUN && BEGIN OF LOOP") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Starting RUN && BEGIN OF LOOP") err = p.Run(`function(done){ var a = 0; var b = 1; @@ -265,7 +266,7 @@ func TestMutlipleThreads(t *testing.T) { done(c, undefined); } }`, &r) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending RUN") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending RUN") if err == nil { v, ok := r.(float64) @@ -279,26 +280,26 @@ func TestMutlipleThreads(t *testing.T) { t.Errorf("Should be %d but is %f", 75025, v) } } else { - if !strings.Contains(err.Error(), "no longer running") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { + if !strings.Contains(err.Error(), "PhantomJS Instance is dead") && !strings.Contains(err.Error(), "phantomjs instance might be dead") { t.Fatal(err) } } runtime.ReadMemStats(&insideStats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending Loop \n\n\n ") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Ending Loop \n ") } p.Exit() defer wg.Done() - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Done with Processor") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (insideStats.Sys / 1000000), insideStats.Mallocs, runtime.NumGoroutine(), "Done with Processor") //Thread Done }() runtime.ReadMemStats(&stats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Started GO ROUTINE") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Started GO ROUTINE") } wg.Wait() runtime.ReadMemStats(&stats) - log.Printf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Done") + t.Logf("MEM: %d, MALC: %d, GOTHREAD: %d, %s", (stats.Sys / 1000000), stats.Mallocs, runtime.NumGoroutine(), "Done") } func TestMultipleLogs(t *testing.T) { From 372ff0a49fc76d5f501cab4888934ba223721736 Mon Sep 17 00:00:00 2001 From: Jack Murdock Date: Thu, 9 Feb 2017 12:51:58 -0800 Subject: [PATCH 20/20] fixed terminating of all phantomjs threads --- phantom.go | 103 +++++++++++++++++++++++++++++++++-------------------- 1 file changed, 64 insertions(+), 39 deletions(-) diff --git a/phantom.go b/phantom.go index 5525964..55d364e 100644 --- a/phantom.go +++ b/phantom.go @@ -4,7 +4,6 @@ import ( "bufio" "encoding/json" "errors" - "fmt" "io" "io/ioutil" "log" @@ -12,6 +11,7 @@ import ( "os/exec" "strings" "sync" + "time" ) // Phantom a data structure that interacts with the wrapper file @@ -28,6 +28,7 @@ type Phantom struct { lineErr chan string quit chan bool stopReading bool + once *sync.Once nothingReadCount int64 } @@ -94,6 +95,7 @@ func Start(args ...string) (*Phantom, error) { lineErr: make(chan string, 100), quit: make(chan bool, 1), stopReading: false, + once: new(sync.Once), nothingReadCount: 0, } p.readerOut = bufio.NewReaderSize(p.out, readerBufferSize*1024) @@ -141,7 +143,7 @@ func readreader(readerLock *sync.Mutex, reader *bufio.Reader) (string, error) { if strings.HasPrefix(line, "RES") { return parts[1], nil } else if line != "" { - fmt.Printf("LOG %s\n", line) + log.Printf("JS-LOG %s\n", line) // return "", nil } else if line != " " { // return "", errors.New("Error reading response, just got a space") @@ -215,38 +217,28 @@ and wait for the command to end. Return an error if one occurred during exit command or if the program output a error value */ func (p *Phantom) Exit() error { - - err := p.Load("phantom.exit()") - stoping := false - if !p.stopReading { - drainBool(p.quit) + var err error + p.once.Do(func() { + err = p.Load("phantom.exit()") + if err != nil { + return + } p.quit <- true - stoping = true - } - p.stopReading = true - if err != nil { - return err - } - p.readerErrorLock.Lock() - p.readerLock.Lock() - err = p.cmd.Wait() - p.readerErrorLock.Unlock() - p.readerLock.Unlock() - if err != nil { - if !strings.Contains(err.Error(), "signal: killed") { - return errors.New("Failed to kill instance :" + err.Error()) - } - err = nil - } - if !stoping { + p.readerErrorLock.Lock() + p.readerLock.Lock() + err = p.cmd.Wait() + p.readerLock.Unlock() + p.readerErrorLock.Unlock() + fileLock.Lock() nbInstance-- if nbInstance == 0 { err = os.Remove(wrapperFileName) } fileLock.Unlock() - } + }) + return err } @@ -255,23 +247,32 @@ ForceShutdown will forcefully kill phantomjs. This will completly terminate the proccess compared to Exit which will safely Exit */ func (p *Phantom) ForceShutdown() error { - stoping := false - if !p.stopReading { - drainBool(p.quit) + var err error + p.once.Do(func() { + err = p.Load("phantom.exit()") + if err != nil { + return + } p.quit <- true - stoping = true - } - p.stopReading = true - err := p.cmd.Process.Kill() - if err != nil { - if !strings.Contains(err.Error(), "signal: killed") { - return errors.New("Failed to kill instance :" + err.Error()) + p.readerErrorLock.Lock() + p.readerLock.Lock() + err = stopExec(p.cmd) + p.readerLock.Unlock() + p.readerErrorLock.Unlock() + + fileLock.Lock() + nbInstance-- + if nbInstance == 0 { + err = os.Remove(wrapperFileName) } - err = nil - } + fileLock.Unlock() + }) + + if !p.cmd.ProcessState.Exited() { + err = p.cmd.Process.Kill() + err = p.cmd.Process.Release() - if !stoping { fileLock.Lock() nbInstance-- if nbInstance == 0 { @@ -279,9 +280,33 @@ func (p *Phantom) ForceShutdown() error { } fileLock.Unlock() } + return err } +func stopExec(cmd *exec.Cmd) error { + done := make(chan error, 1) + go func() { + done <- cmd.Wait() + }() + select { + case <-time.After(3 * time.Second): + if err := cmd.Process.Kill(); err != nil { + errOther := cmd.Process.Release() + if errOther != nil { + log.Printf("Error not Handled %s", errOther) + } + return err + } + return nil + case err := <-done: + if err != nil { + return err + } + return nil + } +} + /* SetMaxBufferSize will set the max Buffer Size If your script will return a large input use this.