protected void parseDocumentsMultiThreaded(
        final FixedDurationTemporalRandomIndexing fdTri, 
        final Iterator<TemporalDocument> docIter,
        final TimeSpan timeSpan, int numThreads)
            throws IOException, InterruptedException {

        Collection<Thread> processingThreads = new LinkedList<Thread>();

        final AtomicInteger count = new AtomicInteger(0);

        final AtomicLong curSSpaceStartTime = new AtomicLong();
        final Object calendarLock = new Object();
        final DateFormat df = new SimpleDateFormat("yyyy_MM_ww_dd_hh");

        final AtomicLong lastWriteTime = new AtomicLong();
        
        // barrier for setting up the initial time stamp based on the first
        // document processed
        final AtomicBoolean startBarrier = new AtomicBoolean(false);

        // Before a Thread blocks waiting for s-space serialization, it enqueues
        // the time for its next document (outside the time-span).  These times
        // are used to select the start time for the next s-sspace.
        final Queue<Long> futureStartTimes = new ConcurrentLinkedQueue<Long>();

        // final variables necessary due to the anonymous inner class
        final boolean writeSemanticPartitions = savePartitions;
        final boolean writeSemanticShifts = printInterestingTokenShifts;
        final boolean writeInterestingWordNeighbors = 
            interestingWordNeighbors > 0;
        final boolean writeShiftRankings = printShiftRankings;

        /**
         * A runnable that serializes the current semantic space to disk and
         * annotates it with the time at which the space started.
         */
        Runnable serializeTimeSpan = new Runnable() {
                public void run() {
                    Calendar c = Calendar.getInstance();
                    c.setTimeInMillis(curSSpaceStartTime.get());
                    String dateString = df.format(c.getTime());
                                    
                    try {

                        // Save the s-space only when requried to.  This
                        // operation can be very slow due to I/O requirements,
                        // and is not mandatory when computing the shifts
                        if (writeSemanticPartitions) {
                            LOGGER.info("writing semantic partition starting " +
                                        "at: " + dateString);
                            // save the current contets of the semantic space
                            printSpace(fdTri, "-" + dateString);
                        }

                        // Add the semantics from the current semantic partition
                        // for each of the interesting words
                        updateTemporalSemantics(curSSpaceStartTime.get(),
                                                fdTri);
                        if (writeSemanticShifts) 
                            printSemanticShifts(dateString);

                        if (writeShiftRankings) 
                            printShiftRankings(dateString, 
                                               curSSpaceStartTime.get(),
                                               timeSpan);

                        // NOTE: since the FD-TRI implementaiton resets
                        // its semantics after every 
                        if (interestingWordNeighbors > 0) 
                            printWordNeighbors(dateString, fdTri);

                    } catch (IOException ioe) {
                        // rethrow
                        throw new IOError(ioe);
                    }

                    // Pick the earlier start time available as the new starting
                    // time for the s-space
                    assert futureStartTimes.size() > 0;
                    Long ssStart = new TreeSet<Long>(futureStartTimes).first();
                    futureStartTimes.clear();

                    // last update the date with the new time
                    curSSpaceStartTime.set(ssStart);
                }
            };
            
        // barrier for document processing threads.  When their next document is
        // outside of the time range, the immediately increase the release on
        // this semaphore and lock on the an object while the serialization
        // thread writes out the current time span's .sspace
        final CyclicBarrier exceededTimeSpanBarrier =
            new CyclicBarrier(numThreads, serializeTimeSpan);
                    
        
        for (int i = 0; i < numThreads; ++i) {

            Thread processingThread = new Thread() {
                    
                    public void run() {
                        // repeatedly try to process documents while some still
                        // remain
                        while (docIter.hasNext()) {
                            
                            TemporalDocument doc = docIter.next();
                            int docNumber = count.incrementAndGet();
                            long docTime = doc.timeStamp();

                            // special case for first document
                            if (docNumber == 1) {
                                curSSpaceStartTime.set(docTime);
                                startBarrier.set(true);
                            }
                            
                            // Spin until the Thread with the first document
                            // sets the initial starting document time.  Note
                            // that we spin here instead of block, because this
                            // is expected that another thread will immediately
                            // set this and so it will be a quick no-op
                            while (startBarrier.get() == false)
                                ;

                            // Check whether the time for this document would
                            // exceed the maximum time span for any TRI
                            // partition.  Loop to ensure that if this thread
                            // does loop and another thread has an earlier time
                            // that would cause this thread's time span to
                            // exceeds the other thread's time period, then this
                            // thread will block and loop again.
                            while (!timeSpan.insideRange(
                                   curSSpaceStartTime.get(), docTime)) {
                                try {
                                    // notify the barrier that this Thread is
                                    // now processing a document in the next
                                    // time span and so the serialization thread
                                    // should write the .sspace to disk.  In
                                    // addition, enqueue the time for this
                                    // document so the serialization thread can
                                    // reset the correct s-sspace start time
                                    futureStartTimes.offer(docTime);
                                    exceededTimeSpanBarrier.await();
                                } catch (InterruptedException ex) {
                                    return;
                                } catch (BrokenBarrierException ex) {
                                    return;
                                }
                            }

                            try {
                                fdTri.processDocument(doc.reader());
                            } catch (IOException ioe) {
                                // rethrow
                                throw new IOError(ioe);
                            }
                            LOGGER.fine("parsed document #" + docNumber);
                        }
                    }
                };
            processingThreads.add(processingThread);
        }

        long threadStart = System.currentTimeMillis();
        
        // start all the threads processing
        for (Thread t : processingThreads)
            t.start();

        verbose("Beginning processing using %d threads", numThreads);

        // wait until all the documents have been parsed
        for (Thread t : processingThreads)
            t.join();

        verbose("parsed %d document in %.3f total seconds)%n",
                count.get(),
                ((System.currentTimeMillis() - threadStart) / 1000d));
    }
protected void parseDocumentsMultiThreaded ( final FixedDurationTemporalRandomIndexing fdTri , final Iterator < TemporalDocument > docIter , final TimeSpan timeSpan , int numThreads ) throws IOException , InterruptedException { Collection < Thread > processingThreads = new LinkedList < Thread > ( ) ; final AtomicInteger count = new AtomicInteger ( 0 ) ; final AtomicLong curSSpaceStartTime = new AtomicLong ( ) ; final Object calendarLock = new Object ( ) ; final DateFormat df = new SimpleDateFormat ( "yyyy_MM_ww_dd_hh" ) ; final AtomicLong lastWriteTime = new AtomicLong ( ) ; final AtomicBoolean startBarrier = new AtomicBoolean ( false ) ; final Queue < Long > futureStartTimes = new ConcurrentLinkedQueue < Long > ( ) ; final boolean writeSemanticPartitions = savePartitions ; final boolean writeSemanticShifts = printInterestingTokenShifts ; final boolean writeInterestingWordNeighbors = interestingWordNeighbors > 0 ; final boolean writeShiftRankings = printShiftRankings ; Runnable serializeTimeSpan = new Runnable ( ) { public void run ( ) { Calendar c = Calendar . getInstance ( ) ; c . setTimeInMillis ( curSSpaceStartTime . get ( ) ) ; String dateString = df . format ( c . getTime ( ) ) ; try { if ( writeSemanticPartitions ) { LOGGER . info ( "writing semantic partition starting " + "at: " + dateString ) ; printSpace ( fdTri , "-" + dateString ) ; } updateTemporalSemantics ( curSSpaceStartTime . get ( ) , fdTri ) ; if ( writeSemanticShifts ) printSemanticShifts ( dateString ) ; if ( writeShiftRankings ) printShiftRankings ( dateString , curSSpaceStartTime . get ( ) , timeSpan ) ; if ( interestingWordNeighbors > 0 ) printWordNeighbors ( dateString , fdTri ) ; } catch ( IOException ioe ) { throw new IOError ( ioe ) ; } assert futureStartTimes . size ( ) > 0 ; Long ssStart = new TreeSet < Long > ( futureStartTimes ) . first ( ) ; futureStartTimes . clear ( ) ; curSSpaceStartTime . set ( ssStart ) ; } } ; final CyclicBarrier exceededTimeSpanBarrier = new CyclicBarrier ( numThreads , serializeTimeSpan ) ; for ( int i = 0 ; i < numThreads ; ++ i ) { Thread processingThread = new Thread ( ) { public void run ( ) { while ( docIter . hasNext ( ) ) { TemporalDocument doc = docIter . next ( ) ; int docNumber = count . incrementAndGet ( ) ; long docTime = doc . timeStamp ( ) ; if ( docNumber == 1 ) { curSSpaceStartTime . set ( docTime ) ; startBarrier . set ( true ) ; } while ( startBarrier . get ( ) == false ) ; while ( ! timeSpan . insideRange ( curSSpaceStartTime . get ( ) , docTime ) ) { try { futureStartTimes . offer ( docTime ) ; exceededTimeSpanBarrier . await ( ) ; } catch ( InterruptedException ex ) { return ; } catch ( BrokenBarrierException ex ) { return ; } } try { fdTri . processDocument ( doc . reader ( ) ) ; } catch ( IOException ioe ) { throw new IOError ( ioe ) ; } LOGGER . fine ( "parsed document #" + docNumber ) ; } } } ; processingThreads . add ( processingThread ) ; } long threadStart = System . currentTimeMillis ( ) ; for ( Thread t : processingThreads ) t . start ( ) ; verbose ( "Beginning processing using %d threads" , numThreads ) ; for ( Thread t : processingThreads ) t . join ( ) ; verbose ( "parsed %d document in %.3f total seconds)%n" , count . get ( ) , ( ( System . currentTimeMillis ( ) - threadStart ) / 1000d ) ) ; }
Calls {@link TemporalSemanticSpace#processDocument(BufferedReader,long)
processDocument} once for every document in {@code docIter} using a the
specified number thread to call {@code processSpace} on the {@code
TemporalSemanticSpace} instance.

@param sspace the space to build
@param docIter an iterator over all the documents to process
@param numThreads the number of threads to use