From e80ef1b1b92498d238e17424391259240057971d Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 17 Feb 2015 09:24:26 -0800 Subject: [PATCH 1/4] flow: Created branch 'feature/use_crawler_workers'. From 3f85a0e7c03dd78c2f5c737a329322bc6f189a01 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 17 Feb 2015 09:24:52 -0800 Subject: [PATCH 2/4] switch to using a limited number of crawler worker functions --- src/docrawler/docrawler.go | 49 +++++++++++++++++++++++++++++--------- 1 file changed, 38 insertions(+), 11 deletions(-) diff --git a/src/docrawler/docrawler.go b/src/docrawler/docrawler.go index abbc01a..c2f094d 100644 --- a/src/docrawler/docrawler.go +++ b/src/docrawler/docrawler.go @@ -7,7 +7,7 @@ import ( "time" ) -// docrawl begins crawling the site at "homeurl" +// doCrawl begins crawling the site at "homeurl" func doCrawl(homeurl string) itemSlice { // set of what we have already crawled, our results crawled := make(itemMap) @@ -17,8 +17,14 @@ func doCrawl(homeurl string) itemSlice { // already crawled http://a.com/index.html, or vice versa crawledStripped := make(itemMap) - // a channel to receive our results on - rchan := make(chan *httpItem) + // channels to send work to and receive our results on + rxchan := make(chan *httpItem) + txchan := make(chan *httpItem) + + // spin up our crawler workers + for i := 0; i < 10; i++ { + go crawlListner(txchan, rxchan) + } // start a ticker which we'll use to output status and check for completion ticker := time.Tick(250 * time.Millisecond) @@ -35,12 +41,12 @@ func doCrawl(homeurl string) itemSlice { // start the home page crawl crawled[homeitem.url.String()] = homeitem crawledStripped[stripURL(homeitem.url)] = homeitem - go crawlItem(homeitem, rchan) + txchan <- homeitem // wait for results for { select { - case r := <-rchan: // new results? + case r := <-rxchan: // new results? // add result to our results map crawled[r.url.String()] = r @@ -77,7 +83,9 @@ func doCrawl(homeurl string) itemSlice { crawlingCount++ crawled[c.url.String()] = c crawledStripped[stripURL(c.url)] = c - go crawlItem(c, rchan) + go func(newItem *httpItem) { + txchan <- newItem + }(c) } case <-ticker: // our regular ticker. for status output and checking for completion. @@ -86,6 +94,10 @@ func doCrawl(homeurl string) itemSlice { // see if we're finished if crawlingCount == 0 { + // close our channels, signalling any workers to exit + close(txchan) + close(rxchan) + // finished! convert results map to a slice and return it rslice := itemSlice{} for _, v := range crawled { @@ -97,21 +109,36 @@ func doCrawl(homeurl string) itemSlice { } } +// crawlListner is a for+select wrapper around crawlItem that listens +// for new jobs and sends them off to crawlItem +func crawlListner(txchan <-chan *httpItem, rxchan chan<- *httpItem) { + for { + select { + case newJob, ok := <-txchan: + if !ok { + // channel closed, finish + return + } + crawlItem(newJob, rxchan) + } + } +} + // crawlItem crawls a single httpItem, fetching the header, hte page, parsing it, // and filling out its structure as much as possible -func crawlItem(item *httpItem, rchan chan<- *httpItem) { +func crawlItem(item *httpItem, rxchan chan<- *httpItem) { // make sure this item is the same domain (i.e. URL "host part") as its referrer if item.refurl != nil && item.url.Host != item.refurl.Host { // skip URLs associated with other Hosts item.linkType = tRemote - rchan <- item + rxchan <- item return } // fetch page text, err := fetchPage(item) if err != nil { - rchan <- item + rxchan <- item return } @@ -119,7 +146,7 @@ func crawlItem(item *httpItem, rchan chan<- *httpItem) { title, links := parseLinks(text) // TODO remove error from parseLinks, not needed if err != nil { item.linkType = tBroken - rchan <- item + rxchan <- item return } item.title = title @@ -134,7 +161,7 @@ func crawlItem(item *httpItem, rchan chan<- *httpItem) { } // send back our item struct now that it's all filled out - rchan <- item + rxchan <- item } func main() { From 13b84ee452c3d3aa753dd8bcc81749f7628e81bc Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 17 Feb 2015 09:28:24 -0800 Subject: [PATCH 3/4] completion of crawler worker mode. tests still passing! --- src/docrawler/docrawler.go | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/src/docrawler/docrawler.go b/src/docrawler/docrawler.go index c2f094d..26a84f8 100644 --- a/src/docrawler/docrawler.go +++ b/src/docrawler/docrawler.go @@ -119,36 +119,30 @@ func crawlListner(txchan <-chan *httpItem, rxchan chan<- *httpItem) { // channel closed, finish return } - crawlItem(newJob, rxchan) + // perform the crawl + rxchan <- crawlItem(newJob) } } } // crawlItem crawls a single httpItem, fetching the header, hte page, parsing it, // and filling out its structure as much as possible -func crawlItem(item *httpItem, rxchan chan<- *httpItem) { +func crawlItem(item *httpItem) *httpItem { // make sure this item is the same domain (i.e. URL "host part") as its referrer if item.refurl != nil && item.url.Host != item.refurl.Host { // skip URLs associated with other Hosts item.linkType = tRemote - rxchan <- item - return + return item } // fetch page text, err := fetchPage(item) if err != nil { - rxchan <- item - return + return item } // parse links - title, links := parseLinks(text) // TODO remove error from parseLinks, not needed - if err != nil { - item.linkType = tBroken - rxchan <- item - return - } + title, links := parseLinks(text) item.title = title // walk links and add them as children to the current item @@ -161,7 +155,7 @@ func crawlItem(item *httpItem, rxchan chan<- *httpItem) { } // send back our item struct now that it's all filled out - rxchan <- item + return item } func main() { From b502cfd5b1a1c30bd62eb553349bd3e25a15d446 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 17 Feb 2015 09:28:34 -0800 Subject: [PATCH 4/4] flow: Closed 'use_crawler_workers'.