Spliterator implementation

Question

I'm trying to post a little tutorial on the new Spliterator class. There are many tutorials these days on using stream starting from a standard Java collection, but I want to explore the creation of a stream using data coming from the web (so no stream fully backed by a collection).

I decided to implement a infinite URL stream coming from a starting address. What I did is scraping the initial page and providing client code with URLs I found into that page. Then I repeat the scraping with new URLs. URLs that are returned to client are not repeated.

UrlScraper

package com.stackexchange.codereview.webscrapingwithstream;

import java.io.IOException;
import java.util.*;
import java.util.function.Consumer;
import java.util.stream.*;

import org.jsoup.Jsoup;
import org.jsoup.nodes.*;
import org.jsoup.select.Elements;

public class UrlScraper {

    private String url;
    private Set<String> index = Collections.synchronizedSet(new TreeSet<>());
    private List<String> startingReferences = new ArrayList<>();

    public UrlScraper(String url) {
        this.url = url;
    }

    public Stream<String> stream() {
        startingReferences.add(url);
        index.add(url);
        return StreamSupport.stream(new UrlSpliterator(startingReferences, index), false);
    }

    public Stream<String> parallelStream() {
        startingReferences.add(url);
        index.add(url);
        return StreamSupport.stream(new UrlSpliterator(startingReferences, index), true);
    }

    private static class UrlSpliterator implements Spliterator<String> {

        private static final int THRESHOLD = 10;
        // variable list of urls in the spliterator
        List<String> refs;
        // distinct set of already known urls 
        Set<String> index;
        // # urls returned by the stream
        int examined = 0;
        // position of last scraped url
        int scraped = -1;

        UrlSpliterator(List<String> refs, Set<String> index) {
            this.refs = refs;
            this.index = index;

        }

        @Override
        // the stream will contain distinct urls
        public int characteristics() {
            return DISTINCT | NONNULL | IMMUTABLE;
        }

        @Override
        // stream will be infinite as web
        public long estimateSize() {
            return Long.MAX_VALUE;
        }

        @Override
        public boolean tryAdvance(Consumer<? super String> consumer) {
            if (examined < refs.size()) {
                consumer.accept(refs.get(examined));

                // when under threshold, a new url is scraped 
                // to fill the list     
                if (refs.size() - examined < THRESHOLD) {
                    scraped++;
                    analyze(refs.get(scraped));
                }

                examined++;
                return true;
            } else {
                return false;
            }
        }

        @Override
        public Spliterator<String> trySplit() {
            int n = refs.size() - examined;
            // when I have more than one url, I split the scraping
            // into two parts. New Spliterator will receive a new
            // arraylist that contains half of the urls to be
            // returned by the stream. Index will be passed as is
            // since we do not want duplicates in urls
            if (n > 1) {
                int splitPoint = examined + n / 2;

                ArrayList<String> al = new ArrayList<>();
                al.addAll(splitPoint, refs);
                for (int i = refs.size() - 1; i >= splitPoint; i--)
                    refs.remove(i);
                return new UrlSpliterator(al, index);

            }
            return null;

        }

        // this is the core of scraping, I use Jsoup to do the
        // dirty part of the job
        private void analyze(String aUrl) {
            Document doc;
            Elements links = null;
            try {

                doc = Jsoup.connect(aUrl).get();

                links = doc.select("a[href]");
                for (Element link : links) {
                    String newUrl = (String) link.attr("abs:href");
                    if (!index.contains(newUrl)) {
                        refs.add(newUrl);
                        index.add(newUrl);
                    }
                }
            } catch (IOException e) {
                ; // if a link is broken, not a problem. It can happens
                  // but I do not want to stop scraping
            }

        }

    }
}

Client code

UrlScraper urlScraper = new UrlScraper("http://www.wikipedia.org");
urlScraper.parallelStream()
         .forEach(System.out::println);

My review questions:

1. Is this code understandable in its intents?
2. Am I going towards multithreading problems?
3. Is my code honoring the spliterator attributes declared in characteristics?

Answer 1

I see a few, actually, a number of problems in this code. I am afraid this will be something of a scathing review, in part because you mention this is intended to be for a tutorial....

I see a number of call them 'critical' issues. Then also a number of lesser issues. The critical ones first:

Use Case

Why not use an Iterator? Since an Iterator would do this job, and is the traditional 'vehicle', you should indicate why a Spliterator is the right tool for the job. Note, Iterators are relatively well understood, and there is a clear wrapper-path from Iterator to Spliterator using Spliterators.spliteratorUnknownSize()

The obvious answer to "Why not use an Iterator?" is either:

• it's a tutorial
• for better control of parallel execution

In each of these cases, the code should be 'exemplar', and you should clarify exactly why this code is needed... but, if the answer includes the 'parallel execution' reason, then the concurrency had better be accurate.

forEachRemaining()

one of the performance advantages of Spliterator is the forEachRemaining() method, and you have chosen not to implement it. Why?

Queues are Queues

Your code really boils down to being a queue processing tool. The refs is a queue, you have content on it, and you process it from the beginning, and add stuff to the end. Occasionally you split the queue.

Use a Queue (Deque).

Specifically, you should probably use an LinkedList (which is a Deque).

If you use a queue, there is no need for the examined variable, and code like the trySplit becomes (here, refs is an LinkedList`):

    @Override
    public Spliterator<String> trySplit() {
        int mid = refs.size() / 2;
        if (mid < 1) {
            return null;
        }

        List<String> toclear = refs.subList(mid, refs.size());
        LinkedList<String> newrefs = new LinkedList<>(toclear);
        toclear.clear();
        return new UrlSpliterator(newrefs, index);
    }

Threshold Logic

There is no apparent reason to have the weird logic in the tryAdvance() for the 'scrape()` call.

Something simple like:

        if (examined < refs.size()) {
            String url = refs.get(examined++);
            analyze(url);
            consumer.accept(url);
            return true;
        }
        return false;

Concurrency bugs

1. The custom wrapper class introduces a significant concurrency bug. The UrlScraper contains the control information for the entire spliteration. The use case you present is:

   UrlScraper urlScraper = new UrlScraper("http://www.wikipedia.org");
   urlScraper.parallelStream()
            .forEach(System.out::println);
   

   This code can be very easily broken by calling stream() and then parallelStream().....

    UrlScraper urlScraper = new UrlScraper("http://www.wikipedia.org");
    urlScraper.stream()
             .forEach(System.out::println);
    urlScraper.parallelStream()
             .forEach(System.out::println);
   

   I would expect the second stream to produce the same results as the first, but the second stream will produce..... from what I can tell, the same content, plus an extra copy of the start URL. It will also re-scrape the last THRESHOLD number of pages (but assuming the pages did not change, it will ignore the scraped URLS).

   Odder things will happen if you call parallelStream multiple times, from multiple different threads...

2. You have a race condition in your analyze code. The following is broken:

               for (Element link : links) {
                   String newUrl = (String) link.attr("abs:href");
                   if (!index.contains(newUrl)) {
                       refs.add(newUrl);
                       index.add(newUrl);
                   }
               }
   

   You are not supposed to add a newUrl to the refs multiple times. But, it is possible that two threads, each processing a page, both pages with a link to the same URL, will hit this code at the same time.

   Both threads may cal !index.contains(newURL) at 'the same time', and both threads will add the newUrl to the refs (even though the refs are different lists in each thread, the index is not. Only one of the threads will successfully add the newUrl to the index though.

   You can use this to your advantage with the Set.add() return value:

                for (Element link : links) {
                    String newUrl = (String) link.attr("abs:href");
                    if (index.add(newUrl)) {
                        refs.add(newUrl);
                    }
                }
   

Code Style

You should declare variables where they are used. try ... catch blocks sometimes mess things up, but in this case, they do not. The following code:

        Document doc;
        Elements links = null;
        try {

            doc = Jsoup.connect(aUrl).get();

            links = doc.select("a[href]");

should be:

        try {
            Document doc = Jsoup.connect(aUrl).get();
            Elements links = doc.select("a[href]");

When you have 'early-return' methods (which are a good thing...), you should have conditionals without else-blocks.... The code:

   @Override
    public boolean tryAdvance(Consumer<? super String> consumer) {
        if (examined < refs.size()) {
            ......
            return true;
        } else {
            return false;
        }
    }

Would be better written as:

   @Override
    public boolean tryAdvance(Consumer<? super String> consumer) {
        if (examined >= refs.size()) {
            return false;
        }
        .......
        return true;
    }

One-Liner conditional statements should still have braces {...}. This code:

            for (int i = refs.size() - 1; i >= splitPoint; i--)
                refs.remove(i);

should be:

            for (int i = refs.size() - 1; i >= splitPoint; i--) {
                refs.remove(i);
            }

Spliterator Characteristics:

Question 3, does the spliterator honour the characteristics specified in the Spliterator interface. These are the available characteristics:

• CONCURRENT - The source data can be modified safely from other threads without lock controls
• DISTINCT - No two values are the same
• IMMUTABLE - It is not possible to change the source
• NONNULL - No values will be null
• ORDERED - there is a sequence to the results
• SIZED - the size is finite, and also estimateSize() is accurate
• SORTED - (requires ORDERED as well) The sequence is pre-defined on some characteristic
• SUBSIZED - (requires SIZED as well) All trySplit() Spliterators will be SIZED & SUBSIZED

Going through that list, your code defines DISTINCT | NONNULL | IMMUTABLE.

I agree with DISTINCT, and NONNULL. Those are 'easy'.

IMMUTABLE - Is it possible to change the source? Can a thread change a page that is being crawled? Yes, I don't think it can, so I do not agree with IMMUTABLE.

CONCURRENT - is it safe to modify the source while Spliterating? Yes, I think it is. The Documentation for CONCURRENT includes:

Characteristic value signifying that the element source may be safely concurrently modified (allowing additions, replacements, and/or removals) by multiple threads without external synchronization. If so, the Spliterator is expected to have a documented policy concerning the impact of modifications during traversal.

Your code can accommodate that easily. Your Characteristics should include CONCURRENT.

ORDERED, SORTED, SIZED, and SUBSIZED do not apply to your data.

CONCURRENT and IMMUTABLE... these are the complicated ones. They are mutually exclusive. If you cannot modify the source data, then you cannot modify it in a concurrent way. If it is concurrent, then it is mutable.

Bottom line, your characteristics should be:

CONCURRENT | NONNULL | DISTINCT

Answer 2

From my reading of the Spliterator documentation, your code looks like it should handle parallel usage. Each list of URLs is owned by exactly one spliterator while they all share the thread-safe set of scraped URLs. If you're concerned about performance, you may want to switch to ConcurrentHashMap, but synchronizedSet is sufficient for example code.

I believe you can add two more characteristics:

• CONCURRENT: Someone, somewhere, might change a page on the web while you're iterating. And when that happens, your code will reflect that change if it hasn't already scraped the page. You should note that this class doesn't detect structural changes to pages you've already scraped in the documentation.

• ORDERED: Since you follow links from page to page and never return to the same page twice, you can claim that the URLs are returned in a parent-then-child order. Granted, that order is a bit hand-wavy, but it's an order nonetheless. :)

---

The only real problem I see is that the refs lists will grow unconstrained. Create a new list as a copy of the unacquired URLs.

// when under threshold, a new url is scraped 
// to fill the list     
if (refs.size() - examined < THRESHOLD) {
    refs = new ArrayList<>(refs.subList(examined, refs.size());
    scraped = scraped - examined + 1;
    analyze(refs.get(scraped));
}