<commit_before>/*******************************************************************************
 *
 *    Copyright (C) 2015-2016
 *  
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License. 
 *    
 *******************************************************************************/
package de.fernunihagen.dna.scalephant.network.client.future;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.scalephant.network.client.ScalephantClient;
import de.fernunihagen.dna.scalephant.storage.entity.BoundingBox;
import de.fernunihagen.dna.scalephant.storage.entity.Tuple;
import de.fernunihagen.dna.scalephant.storage.queryprocessor.CloseableIterator;

public class TupleListFuture extends OperationFutureImpl<List<Tuple>> implements Iterable<Tuple> {

	private static final class ThreadedTupleListFutureIterator implements CloseableIterator<Tuple> {
		/**
		 * The size of the transfer queue
		 */
		protected final static int QUEUE_SIZE = 25;
		
		/**
		 * The transfer queue
		 */
		protected final BlockingQueue<Tuple> tupleQueue = new LinkedBlockingQueue<Tuple>(QUEUE_SIZE);
		
		/**
		 * The amount of seen terminals, the iterator is exaustet, 
		 * when all producers are terminated and the queue is empty
		 */
		protected int seenTerminals = 0;
		
		/**
		 * The terminal (or poison) element
		 */
		protected final static Tuple QUEUE_TERMINAL = new Tuple("", BoundingBox.EMPTY_BOX, "".getBytes());
		
		/**
		 * The next tuple for the next operation
		 */
		protected Tuple nextTuple = null;
		
		/**
		 * The number of results, that needs to be queired
		 */
		protected final int futuresToQuery;
		
		/**
		 * The instance to iterate 
		 * 
		 * @param tupleListFuture
		 */
		protected final TupleListFuture tupleListFuture;
		
		/**
		 * The executor pool
		 */
		protected final ExecutorService executor = Executors.newCachedThreadPool();

		
		/**
		 * The Logger
		 */
		private final static Logger logger = LoggerFactory.getLogger(ThreadedTupleListFutureIterator.class);

		
		public ThreadedTupleListFutureIterator(final TupleListFuture tupleListFuture) {
			this.tupleListFuture = tupleListFuture;
			this.futuresToQuery = tupleListFuture.getNumberOfResultObjets();
			
			for(int i = 0; i < tupleListFuture.getNumberOfResultObjets(); i++) {
				setupProducer(i);
			}
		}
		
		/**
		 * Setup the worker that fetches the data from the futures
		 */
		public void setupProducer(final int resultId) {
			
			logger.trace("Start producer for {}", resultId);
			
			final Runnable producer = new Runnable() {
				
				@Override
				public void run() {

					try {
						final List<Tuple> tupleList = tupleListFuture.get(resultId);
						for(final Tuple tuple : tupleList) {
							tupleQueue.put(tuple);
						}
												
						// TODO: Handle paging
						
					} catch (InterruptedException | ExecutionException e) {
						logger.warn("Got exception while writing data to queue", e);
					} finally {
						addTerminalNE();
						logger.trace("Producer {} is done", resultId);
					}
				}

				/**
				 * Add the terminal to the queue
				 */
				protected void addTerminalNE() {
					try {
						tupleQueue.put(QUEUE_TERMINAL);
					} catch (InterruptedException e) {
						// Got the interrupted exception while addint the 
						// terminal, ignoring
					}
				}
			};
			
			executor.submit(producer);
		}

		@Override
		public boolean hasNext() {
			
			if(nextTuple != null) {
				logger.warn("Last tuple was not requested, did you call next before?");
				nextTuple = null;
			}
			
			while(nextTuple == null) {
				
				// All worker are done
				if(seenTerminals == futuresToQuery) {
					return (nextTuple != null);
				}
				
				try {
					// Wait until element is available
					nextTuple = tupleQueue.take(); 
				} catch (InterruptedException e) {
					// Ignore exception
					return false;
				}
										
				if(nextTuple == QUEUE_TERMINAL) {
					seenTerminals++;
					nextTuple = null;
				}
			}
			
			return true;
		}

		@Override
		public Tuple next() {
			if(nextTuple == null) {
				throw new IllegalStateException("Tuple is null, did you called hasNext before?");
			}
			
			final Tuple resultTuple = nextTuple;
			nextTuple = null;
			return resultTuple;
		}

		@Override
		public void close() throws Exception {
			logger.trace("Close called on interator");
			executor.shutdown();
		}
	}

	/**
	 * Is the result complete or only a page?
	 */
	protected final Map<Integer, Boolean> resultComplete = new HashMap<>();
	
	/**<commit_msg>Added the missing paging code<commit_after>/*******************************************************************************
 *
 *    Copyright (C) 2015-2016
 *  
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License. 
 *    
 *******************************************************************************/
package de.fernunihagen.dna.scalephant.network.client.future;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.scalephant.network.client.ScalephantClient;
import de.fernunihagen.dna.scalephant.storage.entity.BoundingBox;
import de.fernunihagen.dna.scalephant.storage.entity.Tuple;
import de.fernunihagen.dna.scalephant.storage.queryprocessor.CloseableIterator;

public class TupleListFuture extends OperationFutureImpl<List<Tuple>> implements Iterable<Tuple> {

	private static final class ThreadedTupleListFutureIterator implements CloseableIterator<Tuple> {
		/**
		 * The size of the transfer queue
		 */
		protected final static int QUEUE_SIZE = 25;
		
		/**
		 * The transfer queue
		 */
		protected final BlockingQueue<Tuple> tupleQueue = new LinkedBlockingQueue<Tuple>(QUEUE_SIZE);
		
		/**
		 * The amount of seen terminals, the iterator is exaustet, 
		 * when all producers are terminated and the queue is empty
		 */
		protected int seenTerminals = 0;
		
		/**
		 * The terminal (or poison) element
		 */
		protected final static Tuple QUEUE_TERMINAL = new Tuple("", BoundingBox.EMPTY_BOX, "".getBytes());
		
		/**
		 * The next tuple for the next operation
		 */
		protected Tuple nextTuple = null;
		
		/**
		 * The number of results, that needs to be queired
		 */
		protected final int futuresToQuery;
		
		/**
		 * The instance to iterate 
		 * 
		 * @param tupleListFuture
		 */
		protected final TupleListFuture tupleListFuture;
		
		/**
		 * The executor pool
		 */
		protected final ExecutorService executor = Executors.newCachedThreadPool();

		
		/**
		 * The Logger
		 */
		private final static Logger logger = LoggerFactory.getLogger(ThreadedTupleListFutureIterator.class);

		
		public ThreadedTupleListFutureIterator(final TupleListFuture tupleListFuture) {
			this.tupleListFuture = tupleListFuture;
			this.futuresToQuery = tupleListFuture.getNumberOfResultObjets();
			
			for(int i = 0; i < tupleListFuture.getNumberOfResultObjets(); i++) {
				setupProducer(i);
			}
		}
		
		/**
		 * Setup the worker that fetches the data from the futures
		 */
		public void setupProducer(final int resultId) {
			
			logger.trace("Start producer for {}", resultId);
			
			final Runnable producer = new Runnable() {
				
				@Override
				public void run() {

					try {
						final List<Tuple> tupleList = tupleListFuture.get(resultId);
						
						addTupleListToQueue(tupleList);
												
						if(! tupleListFuture.isCompleteResult(resultId)) {
							handleAdditionalPages();
						}
						
					} catch (InterruptedException | ExecutionException e) {
						logger.warn("Got exception while writing data to queue", e);
					} finally {
						addTerminalNE();
						logger.trace("Producer {} is done", resultId);
					}
				}
				
				/**
				 * Request and add the additional pages to the queue
				 * @throws ExecutionException 
				 * @throws InterruptedException 
				 */
				protected void handleAdditionalPages() throws InterruptedException, ExecutionException {
					final ScalephantClient scalephantClient = tupleListFuture.getConnectionForResult(resultId);
					final short queryRequestId = tupleListFuture.getRequestId(resultId);
					
					if(scalephantClient == null) {
						logger.error("Unable to get connection for paging: {}", resultId);
						return;
					}
					
					TupleListFuture nextPage = null;
					do {
						 nextPage = scalephantClient.getNextPage(queryRequestId);
						 nextPage.waitForAll();
						 
						 if(nextPage.isFailed()) {
							 logger.error("Requesting next page failed! Query result is incomplete: {}", nextPage.getAllMessages());
							 return;
						 }
						 
						 // Query is send to one server, so the number of
						 // resupt objects should be 1
						 if(nextPage.getNumberOfResultObjets() != 1) {
							 logger.error("Got a non expected number of result objects {}", nextPage.getNumberOfResultObjets());
						 }
						 
						 addTupleListToQueue(nextPage.get(1));
						 
					} while(! nextPage.isCompleteResult(1));
				}

				/**
				 * Add the tuple list into the queue
				 * @param tupleList
				 * @throws InterruptedException
				 */
				protected void addTupleListToQueue(final List<Tuple> tupleList) throws InterruptedException {
					for(final Tuple tuple : tupleList) {
						tupleQueue.put(tuple);
					}
				}

				/**
				 * Add the terminal to the queue
				 */
				protected void addTerminalNE() {
					try {
						tupleQueue.put(QUEUE_TERMINAL);
					} catch (InterruptedException e) {
						// Got the interrupted exception while addint the 
						// terminal, ignoring
					}
				}
			};
			
			executor.submit(producer);
		}

		@Override
		public boolean hasNext() {
			
			if(nextTuple != null) {
				logger.warn("Last tuple was not requested, did you call next before?");
				nextTuple = null;
			}
			
			while(nextTuple == null) {
				
				// All worker are done
				if(seenTerminals == futuresToQuery) {
					return (nextTuple != null);
				}
				
				try {
					// Wait until element is available
					nextTuple = tupleQueue.take(); 
				} catch (InterruptedException e) {
					// Ignore exception
					return false;
				}
										
				if(nextTuple == QUEUE_TERMINAL) {
					seenTerminals++;
					nextTuple = null;
				}
			}
			
			return true;
		}

		@Override
		public Tuple next() {
			if(nextTuple == null) {
				throw new IllegalStateException("Tuple is null, did you called hasNext before?");
			}
			
			final Tuple resultTuple = nextTuple;
			nextTuple = null;
			return resultTuple;
		}

		@Override
		public void close() throws Exception {
			logger.trace("Close called on interator");
			executor.shutdown();
		}
		
		@Override
		protected void finalize() throws Throwable {
			close();
			super.finalize();
		}
	}

	/**
	 * Is the result complete or only a page?
	 */
	protected final Map<Integer, Boolean> resultComplete = new HashMap<>();
	
	/**
{'fraction_non_alphanumeric': 0.11740696278511405, 'fraction_numerical': 0.003201280512204882, 'mean_word_length': 3.047942986718497, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5221530', 'n_tokens_mistral': 4485, 'n_tokens_neox': 3396, 'n_words': 989}