Introduced hybrid event model

VonRosenchild · Aug 1, 2012 · e6d7fd6 · e6d7fd6
1 parent 9d3c5a2
commit e6d7fd6
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 2 deletions.
diff --git a/...n/java/eu/stratosphere/nephele/taskmanager/bytebuffered/AbstractOutputChannelContext.java b/...n/java/eu/stratosphere/nephele/taskmanager/bytebuffered/AbstractOutputChannelContext.java
@@ -18,6 +18,7 @@
 import java.util.Iterator;
 
 import eu.stratosphere.nephele.event.task.AbstractEvent;
+import eu.stratosphere.nephele.event.task.AbstractTaskEvent;
 import eu.stratosphere.nephele.taskmanager.transferenvelope.TransferEnvelope;
 
 public abstract class AbstractOutputChannelContext implements OutputChannelContext {
@@ -44,10 +45,40 @@ public void queueTransferEnvelope(final TransferEnvelope transferEnvelope) {
 
  final Iterator<AbstractEvent> it = transferEnvelope.getEventList().iterator();
  while (it.hasNext()) {
- this.forwardingChain.offerEvent(it.next());
+
+ final AbstractEvent event = it.next();
+ if (event instanceof AbstractTaskEvent) {
+ processEventAsynchronously(event);
+ } else {
+ processEventSynchronously(event);
+ }
  }
  }
 
+ /**
+ * Processes an event received from the framework in a synchronous fashion, i.e. the event processing is done by the
+ * thread the event is destined for (usually the task thread).
+ * 
+ * @param event
+ * the event to be processed
+ */
+ protected void processEventSynchronously(final AbstractEvent event) {
+
+ this.forwardingChain.offerEvent(event);
+ }
+
+ /**
+ * Processes an event received from the framework in an asynchronous fashion, i.e. the event processing is done by
+ * the thread which delivers the event.
+ * 
+ * @param event
+ * the event to be processed
+ */
+ protected void processEventAsynchronously(final AbstractEvent event) {
+
+ // The default implementation does nothing
+ }
+
  /**
  * {@inheritDoc}
  */

diff --git a/...src/main/java/eu/stratosphere/nephele/taskmanager/runtime/RuntimeOutputChannelBroker.java b/...src/main/java/eu/stratosphere/nephele/taskmanager/runtime/RuntimeOutputChannelBroker.java
@@ -122,7 +122,7 @@ public void processEvent(final AbstractEvent event) {
  } else if (event instanceof ReceiverNotFoundEvent) {
  this.lastSequenceNumberWithReceiverNotFound = ((ReceiverNotFoundEvent) event).getSequenceNumber();
  } else if (event instanceof AbstractTaskEvent) {
- this.byteBufferedOutputChannel.processEvent(event);
+ throw new IllegalStateException("Received synchronous task event " + event);
  }
 
  getNext().processEvent(event);

diff --git a/...rc/main/java/eu/stratosphere/nephele/taskmanager/runtime/RuntimeOutputChannelContext.java b/...rc/main/java/eu/stratosphere/nephele/taskmanager/runtime/RuntimeOutputChannelContext.java
@@ -15,6 +15,7 @@
 
 package eu.stratosphere.nephele.taskmanager.runtime;
 
+import eu.stratosphere.nephele.event.task.AbstractEvent;
 import eu.stratosphere.nephele.io.channels.ChannelID;
 import eu.stratosphere.nephele.io.channels.ChannelType;
 import eu.stratosphere.nephele.io.channels.bytebuffered.AbstractByteBufferedOutputChannel;
@@ -77,4 +78,13 @@ public ChannelType getType() {
 
  return this.byteBufferedOutputChannel.getType();
  }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void processEventAsynchronously(final AbstractEvent event) {
+
+ this.byteBufferedOutputChannel.processEvent(event);
+ }
 }