Author: shawkins Date: 2012-01-30 13:18:50 -0500 (Mon, 30 Jan 2012) New Revision: 3833 Modified: branches/7.7.x/engine/src/main/java/org/teiid/query/processor/relational/XMLTableNode.java Log: TEIID-1915 correction to thread coordination in XMLTableNode Modified: branches/7.7.x/engine/src/main/java/org/teiid/query/processor/relational/XMLTableNode.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/query/processor/relational/XMLTableNode.java 2012-01-27 18:10:06 UTC (rev 3832) +++ branches/7.7.x/engine/src/main/java/org/teiid/query/processor/relational/XMLTableNode.java 2012-01-30 18:18:50 UTC (rev 3833) @@ -88,8 +88,15 @@ private Item item; private TupleBuffer buffer; - private boolean batchAvailable = false; - private TeiidRuntimeException asynchException; + + private enum State { + BUILDING, + AVAILABLE, + DONE + }; + + private State state = State.BUILDING; + private volatile TeiidRuntimeException asynchException; private int outputRow = 1; private boolean usingOutput; @@ -121,7 +128,7 @@ outputRow = 1; usingOutput = false; this.buffer = null; - this.batchAvailable = false; + this.state = State.BUILDING; this.asynchException = null; } @@ -149,7 +156,7 @@ evaluate(false); if (this.table.getXQueryExpression().isStreaming()) { - while (!batchAvailable) { + while (state == State.BUILDING) { try { this.wait(); } catch (InterruptedException e) { @@ -159,7 +166,9 @@ unwrapException(asynchException); TupleBatch batch = this.buffer.getBatch(outputRow); outputRow = batch.getEndRow() + 1; - batchAvailable = hasNextBatch(); + if (state != State.DONE && !batch.getTerminationFlag()) { + state = hasNextBatch()?State.AVAILABLE:State.BUILDING; + } return batch; } @@ -195,16 +204,20 @@ if (this.table.getXQueryExpression().isStreaming()) { if (this.buffer == null) { this.buffer = this.getBufferManager().createTupleBuffer(getOutputElements(), getConnectionID(), TupleSourceType.PROCESSOR); + if (!useFinalBuffer) { + this.buffer.setForwardOnly(true); + } } Runnable r = new Runnable() { @Override public void run() { try { - if (!useFinalBuffer) { - buffer.setForwardOnly(true); + XQueryEvaluator.evaluateXQuery(table.getXQueryExpression(), contextItem, parameters, XMLTableNode.this, getContext()); + synchronized (XMLTableNode.this) { + if (buffer != null) { + buffer.close(); + } } - XQueryEvaluator.evaluateXQuery(table.getXQueryExpression(), contextItem, parameters, XMLTableNode.this, getContext()); - buffer.close(); } catch (TeiidException e) { asynchException = new TeiidRuntimeException(e); } catch (TeiidRuntimeException e) { @@ -214,8 +227,8 @@ asynchException = new TeiidRuntimeException(e); } } finally { - batchAvailable = true; synchronized (XMLTableNode.this) { + state = State.DONE; XMLTableNode.this.notifyAll(); } } @@ -323,12 +336,13 @@ if (isClosed()) { throw EARLY_TERMINATION; } + assert this.state != State.DONE; this.item = row; rowCount++; try { this.buffer.addTuple(processRow()); - if (hasNextBatch()) { - this.batchAvailable = true; + if (state == State.BUILDING && hasNextBatch()) { + this.state = State.AVAILABLE; this.notifyAll(); } } catch (TeiidException e) {