@return void

@throws IOException
@throws InterruptedException
@Description: 初始化
@author liaoqiqi
@date 2013-6-14
19496
private void initInternal(String hosts, String defaultPrefixString, boolean debug)
            throws IOException, InterruptedException {

        curHost = hosts;
        curDefaultPrefixString = defaultPrefixString;

        store = new ResilientActiveKeyValueStore(debug);
        store.connect(hosts);

        LOGGER.info("zoo prefix: " + defaultPrefixString);

        // 新建父目录
        makeDir(defaultPrefixString, ZooUtils.getIp());
    }
282912
private void handleReadBuffer() throws IOException {
        if (readBuffer.remaining() == 0) {
            readBuffer.clear();
            actualFile.read(actualFileOffset, readBuffer);
            actualFileOffset += readBuffer.limit();
            readBuffer.position(0);
        }
    }
1193009
public void put(Object o)
        throws InterruptedException
    {
        synchronized(lock)
        {
            while (size==maxSize)
                lock.wait();

            elements[tail]=o;
            if(++tail==maxSize)
                tail=0;
            size++;
            lock.notify();
        }
    }
1197749
private void requestAllOutputGatesToClose() throws IOException, InterruptedException {
		for (int i = 0; i < this.outputGates.size(); i++) {
			this.outputGates.get(i).requestClose();
		}
	}
1279233
public void lock() throws IOException, InterruptedException {
        if (channel == null) {
            initializeChannel();
        }
        lockSem.acquire();
        lock = channel.lock();
    }
996338
private boolean startReading() throws IOException, InterruptedException {
        //If the reader is not ready, gives it some milliseconds
        while (!this.reader.ready()) {
            Thread.sleep(this.waitPause);
            if (this.closed) {
                return false;
            }
        }
        return true;
    }
912388
public void stop() throws IOException, InterruptedException {
    for (int i = services.size() - 1; i >= 0; i--) {
      Service service = services.get(i);
      logger.info("Stopping Minicluster Service: "
          + service.getClass().getName());
      service.stop();
    }
    logger.info("All Minicluster Services stopped.");
  }
1163579
public function reset()
    {
        if (!$this->markSupported()) {
            throw new IOException(get_class($this) . " does not support mark() and reset() methods.");
        }
        // goes back to last mark, by default this would be 0 (i.e. rewind file).
        fseek($this->stream, SEEK_SET, $this->mark);
        $this->mark = 0;
    }
706138
public void stop() {
		isRunning = false;
		thread.interrupt();
		try {
			closeFileReader();
		} catch (IOException ioe) {
			System.out.println(new LogEntry(Level.CRITICAL, "I/O exception while closing file reader: " + ioe.getMessage(), ioe));
		}
	}
919411
public void waitAbortable() throws IOException, InterruptedException {
        try {
            isWaiting.set(true);
            terminal.waitAbortable(updater);
        } finally {
            close();
            updateLines();
            terminal.finish();
        }
    }
1094811
@Override
  public void close() throws IOException {
    if (!closed) {
      objectOutputStream.close();
      fileOutputStream.close();
      closed = true;
    }
  }
749294
@Override
	public final void cleanup(Context context) throws IOException, InterruptedException {
		cleanup(this.context, collector);
		collector.close();
		super.cleanup(context);
	}
304878
public void run() {
        synchronized (this) {
            started = true;
        }
        finished = false;
        finish = false;

        try {
            while (true) {
                waitForInput(is);

                if (finish || Thread.interrupted()) {
                    break;
                }

                int availableBytes = is.available();
                byte[] buf = new byte[availableBytes];
                int actualBytesRead = is.read(buf, 0, availableBytes);

                if (actualBytesRead <= 0 || finish || Thread.interrupted()) {
                    break;
                }
                os.write(buf, 0, actualBytesRead);
                if (autoflush) {
                    os.flush();
                }
            }
            os.flush();
        } catch (InterruptedException ie) {
            // likely PumpStreamHandler trying to stop us
        } catch (Exception e) {
            synchronized (this) {
                exception = e;
            }
        } finally {
            if (closeWhenExhausted) {
                FileUtils.close(os);
            }
            finished = true;
            synchronized (this) {
                notifyAll();
            }
        }
    }
1354829
private void waitForInputChannelsToBeClosed() throws IOException, InterruptedException {
		// Wait for disconnection of all output gates
		while (true) {

			// Make sure, we leave this method with an InterruptedException when the task has been canceled
			if (this.executionObserver.isCanceled()) {
				throw new InterruptedException();
			}

			boolean allClosed = true;
			for (int i = 0; i < getNumberOfInputGates(); i++) {
				final InputGate<? extends IOReadableWritable> eig = this.inputGates.get(i);
				if (!eig.isClosed()) {
					allClosed = false;
				}
			}

			if (allClosed) {
				break;
			}
			else {
				Thread.sleep(SLEEPINTERVAL);
			}
		}
	}
804772
private void detectFileChange() throws IOException {
		checkClosed();
		assert Thread.holdsLock(filePosLock);
		while(!file.exists()) {
			try {
				System.err.println("File not found, waiting: " + file.getPath());
				Thread.sleep(pollInterval);
			} catch(InterruptedException e) {
				InterruptedIOException newExc = new InterruptedIOException(e.getMessage());
				newExc.initCause(e);
				throw newExc;
			}
			checkClosed();
		}
		long fileLen = file.length();
		if(fileLen<filePos) filePos = 0;
	}
492430
public void close() throws IOException {
    // multiple close should return with no errors
    // readThread will close underlying buffer
    readThread.close();
    try {
      readThread.join();
    } catch (InterruptedException e) {
      throw new IOException(e);
    }
  }
68756
private void init() 
            throws IOException, InterruptedException {
        
        // get job config properties
        boolean advancedMode = 
            conf.get(INPUT_MODE, BASIC_MODE).equals(ADVANCED_MODE);
        redactionRuleCol = conf.getStrings(REDACTION_RULE_COLLECTION);
        
        // initialize the total length
        float recToFragRatio = conf.getFloat(RECORD_TO_FRAGMENT_RATIO, 
                getDefaultRatio());
        length = mlSplit.getLength() * recToFragRatio;
        
        // generate the query
        String queryLanguage = null;
        String queryText = null;
        long start = mlSplit.getStart() + 1 + count;
        long end = mlSplit.isLastSplit() ? 
                   Long.MAX_VALUE : start + mlSplit.getLength() - count - 1;
        if (!advancedMode) { 
            StringBuilder buf = new StringBuilder();      
            buf.append("xquery version \"1.0-ml\"; \n");
            if (redactionRuleCol != null) {
            	buf.append(
            		"import module namespace rdt = \"http://marklogic.com/xdmp/redaction\" at \"/MarkLogic/redaction.xqy\";\n");
            }
            String indent = conf.get(INDENTED, "FALSE");
            Indentation ind = Indentation.valueOf(indent);
            buf.append(
                    "declare namespace mlmr=\"http://marklogic.com/hadoop\";\n");
            buf.append(
                    "declare variable $mlmr:splitstart as xs:integer external;\n");
            buf.append(
                    "declare variable $mlmr:splitend as xs:integer external;\n");
           
            buf.append(ind.getStatement());
            buf.append("\n");
            String docExpr = conf.get(DOCUMENT_SELECTOR);
            String ctsQuery = conf.get(QUERY_FILTER);
            String subExpr = conf.get(SUBDOCUMENT_EXPRESSION);
            Collection<String> nsCol = docExpr != null || subExpr != null? 
                    conf.getStringCollection(PATH_NAMESPACE) : null;
            if (ctsQuery != null) {
                buildSearchQuery(docExpr, ctsQuery, nsCol, buf);
                queryText = buf.toString();
            } else if (docExpr != null || subExpr != null) {   
                buildDocExprQuery(docExpr, nsCol, subExpr, buf);
                queryText = buf.toString();
            } else {
                LexiconFunction function = null;
                Class<? extends LexiconFunction> lexiconClass = 
                        conf.getClass(INPUT_LEXICON_FUNCTION_CLASS, null,
                                LexiconFunction.class);
                if (lexiconClass != null) {
                    function = ReflectionUtils.newInstance(lexiconClass, 
                            conf);
                    if (nsCol == null) {
                        nsCol = conf.getStringCollection(PATH_NAMESPACE);
                    }
                    queryText = function.getInputQuery(nsCol, start, 
                            mlSplit.isLastSplit() ? 
                                    Long.MAX_VALUE : (long)length);
                }
            } 
            if (queryText == null) {
                buildDocExprQuery(docExpr, nsCol, subExpr, buf);
                queryText = buf.toString();
            }
        } else {
            queryText = conf.get(MarkLogicConstants.INPUT_QUERY);
            queryLanguage = conf.get(MarkLogicConstants.INPUT_QUERY_LANGUAGE);
            if (queryText == null) {
                throw new IllegalStateException(
                  "Input query is required in advanced mode but not defined.");
            }
        }
        
        if (LOG.isDebugEnabled()) {
            LOG.debug(queryText);
        }


        // set up a connection to the server
        while (retry < maxRetries) {
        try {
            if (retry == 1) {
                LOG.info("Retrying connect");
            }
            String curForestName = "";
            String curHostName = "";
            if (curForest == -1) {
                curForestName = mlSplit.getForestId().toString();
                curHostName = hostNames[0];
            } else {
                curForestName = replicas.get(curForest).getForest();
                curHostName = replicas.get(curForest).getHostName();
            }
            ContentSource cs = InternalUtilities.getInputContentSource(conf, 
                    curHostName);
            session = cs.newSession("#"+curForestName);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Connect to forest "
                    + curForestName + " on "
                    + session.getConnectionUri().getHost());
            }
            AdhocQuery query = session.newAdhocQuery(queryText);
            if (advancedMode) {
                boolean bindSplitRange = 
                        conf.getBoolean(BIND_SPLIT_RANGE, false);
                if (bindSplitRange) {
                    query.setNewIntegerVariable(MR_NAMESPACE, 
                            SPLIT_START_VARNAME, start);
                    query.setNewIntegerVariable(MR_NAMESPACE, 
                            SPLIT_END_VARNAME, end);
                } else {
                    query.setPosition(start);
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("split start position: " + start);
                    }        
                    query.setCount(mlSplit.isLastSplit() ?
                            Long.MAX_VALUE : mlSplit.getLength());
                }
            } else {
                query.setNewIntegerVariable(MR_NAMESPACE, 
                        SPLIT_START_VARNAME, start);
                query.setNewIntegerVariable(MR_NAMESPACE, 
                        SPLIT_END_VARNAME, end);
            }
            RequestOptions options = new RequestOptions();
            options.setCacheResult(false);
            if (queryLanguage != null) {
            	options.setQueryLanguage(queryLanguage);
            }
            String ts = conf.get(INPUT_QUERY_TIMESTAMP);
            if (ts != null) {
                options.setEffectivePointInTime(new BigInteger(ts));
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Query timestamp: " + ts);
                }
            }
            if (LOG.isDebugEnabled()) {
                LOG.debug("Input query: " + query.getQuery());
            }
            query.setOptions(options);       
            result = session.submitRequest(query);
        } catch (XccConfigException e) {
            LOG.error("XccConfigException:" + e);
            throw new IOException(e);
        } catch (QueryException e) {
            LOG.error("QueryException:" + e);
            throw new IOException(e);
        } catch (RequestException e) {
            LOG.error("RequestException:" + e.getMessage());
            if (curForest != -1) {
                if (++retry < maxRetries) {
                    // failover
                    try {
                        InternalUtilities.sleep(sleepTime);
                    } catch (Exception e2) {
                    }
                    sleepTime = Math.min(sleepTime * 2,maxSleepTime);

                    curForest = (curForest+1)%replicas.size();
                    continue;
                 }
                 LOG.info("Exceeded max retry");
            }
            LOG.info("Query: " + queryText);
            if (LOG.isDebugEnabled()) {
                LOG.debug("RequestException: " + e);
            }
            throw new IOException(e);
        }
        break;
        } // retry
    }