More information: http://www.cs.berkeley.edu/~mdw/proj/seda/

seda.sandStorm.internal
Class SinkProxy

java.lang.Object
  |
  +--seda.sandStorm.internal.SinkProxy
All Implemented Interfaces:
ProfilableIF, SinkIF

public class SinkProxy
extends java.lang.Object
implements SinkIF, ProfilableIF

Used as a proxy to observe and measure communication behavior between stages. By handing out a SinkProxy instead of a FiniteQueue, it is possible to gather statistics on event communication between stages. This is used by StageGraph to construct a graph of the communication patterns between stages.

Author:
Matt Welsh

Field Summary
 int enqueueCount
          Maintains a running sum of the number of elements enqueued onto this sink.
 int enqueueSuccessCount
          Maintains a running sum of the number of elements successfully enqueued onto this sink (that is, not rejected by the enqueue predicate).
 SinkIF thesink
           
 long timer
          Used to maintain a timer for statistics gathering.
 
Constructor Summary
SinkProxy(SinkIF sink, ManagerIF mgr, StageWrapperIF toStage)
          Create a SinkProxy for the given sink.
 
Method Summary
 void enqueue_abort(java.lang.Object key)
          Abort a previously prepared provisional enqueue operation (from the enqueue_prepare() method).
 void enqueue_commit(java.lang.Object key)
          Commit a previously prepared provisional enqueue operation (from the enqueue_prepare() method).
 boolean enqueue_lossy(QueueElementIF enqueueMe)
          Enqueues the given element onto the queue.
 void enqueue_many(QueueElementIF[] enqueueMe)
          Given an array of elements, atomically enqueues all of the elements in the array.
 java.lang.Object enqueue_prepare(QueueElementIF[] enqueueMe)
          Support for transactional enqueue.
 void enqueue(QueueElementIF enqueueMe)
          Enqueues the given element onto the queue.
 EnqueuePredicateIF getEnqueuePredicate()
          Return the enqueue predicate for this sink.
 int profileSize()
          Return the profile size of the queue.
 void setEnqueuePredicate(EnqueuePredicateIF pred)
          Set the enqueue predicate for this sink.
 int size()
          Return the size of the queue.
 java.lang.String toString()
           
 
Methods inherited from class java.lang.Object
, clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
 

Field Detail

thesink

public SinkIF thesink

enqueueCount

public int enqueueCount
Maintains a running sum of the number of elements enqueued onto this sink.

enqueueSuccessCount

public int enqueueSuccessCount
Maintains a running sum of the number of elements successfully enqueued onto this sink (that is, not rejected by the enqueue predicate).

timer

public long timer
Used to maintain a timer for statistics gathering.
Constructor Detail

SinkProxy

public SinkProxy(SinkIF sink,
                 ManagerIF mgr,
                 StageWrapperIF toStage)
Create a SinkProxy for the given sink.
Parameters:
sink - The sink to create a proxy for.
mgr - The associated manager.
toStage - The stage which this sink pushes events to.
Method Detail

size

public int size()
Return the size of the queue.
Specified by:
size in interface SinkIF

enqueue

public void enqueue(QueueElementIF enqueueMe)
             throws SinkException
Description copied from interface: SinkIF
Enqueues the given element onto the queue.
Specified by:
enqueue in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
element - The QueueElementIF to enqueue
Throws:
SinkFullException - Indicates that the sink is temporarily full.
SinkClosedException - Indicates that the sink is no longer being serviced.

enqueue_lossy

public boolean enqueue_lossy(QueueElementIF enqueueMe)
Description copied from interface: SinkIF
Enqueues the given element onto the queue. This is lossy in that this method drops the element if the element could not be enqueued, rather than throwing a SinkFullException or SinkClosedException. This is meant as a convenience interface for "low priority" enqueue events which can be safely dropped.
Specified by:
enqueue_lossy in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
element - The QueueElementIF to enqueue
Returns:
true if the element was enqueued, false otherwise.

enqueue_many

public void enqueue_many(QueueElementIF[] enqueueMe)
                  throws SinkException
Description copied from interface: SinkIF
Given an array of elements, atomically enqueues all of the elements in the array. This guarantees that no other thread can interleave its own elements with those being inserted from this array. The implementation must enqueue all of the elements or none of them; if a SinkFullException or SinkClosedException is thrown, none of the elements will have been enqueued. This implies that the enqueue predicate (if any) must accept all elements in the array for the enqueue to proceed.
Specified by:
enqueue_many in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
elements - The element array to enqueue
Throws:
SinkFullException - Indicates that the sink is temporarily full.
SinkClosedException - Indicates that the sink is no longer being serviced.

profileSize

public int profileSize()
Return the profile size of the queue.
Specified by:
profileSize in interface ProfilableIF

enqueue_prepare

public java.lang.Object enqueue_prepare(QueueElementIF[] enqueueMe)
                                 throws SinkException
Description copied from interface: SinkIF
Support for transactional enqueue.

This method allows a client to provisionally enqueue a number of elements onto the queue, and then later commit the enqueue (with a enqueue_commit() call), or abort (with a enqueue_abort() call). This mechanism can be used to perform "split-phase" enqueues, where a client first enqueues a set of elements on the queue and then performs some work to "fill in" those elements before performing a commit. This can also be used to perform multi-queue transactional enqueue operations, with an "all-or-nothing" strategy for enqueueing events on multiple queues.

This method would generally be used in the following manner:

   Object key = sink.enqueue_prepare(someElements);
   if (can_commit) {
     sink.enqueue_commit(key);
   } else {
     sink.enqueue_abort(key);
   }
 

Note that this method does not protect against "dangling prepares" -- that is, a prepare without an associated commit or abort operation. This method should be used with care. In particular, be sure that all code paths (such as exceptions) after a prepare include either a commit or an abort.

Like enqueue_many, enqueue_prepare is an "all or none" operation: the enqueue predicate must accept all elements for enqueue, or none of them will be enqueued.

Specified by:
enqueue_prepare in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
elements - The element array to provisionally enqueue
Returns:
A "transaction key" that may be used to commit or abort the provisional enqueue
Throws:
SinkFullException - Indicates that the sink is temporarily full and that the requested elements could not be provisionally enqueued.
SinkClosedException - Indicates that the sink is no longer being serviced.
See Also:
enqueue_commit, enqueue_abort

enqueue_commit

public void enqueue_commit(java.lang.Object key)
Description copied from interface: SinkIF
Commit a previously prepared provisional enqueue operation (from the enqueue_prepare() method). Causes the provisionally enqueued elements to appear on the queue for future dequeue operations. Note that once a enqueue_prepare() has returned an enqueue key, the queue cannot reject the entries.
Specified by:
enqueue_commit in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
key - The enqueue key returned by a previous call to enqueue_prepare().
Throws:
java.lang.IllegalArgumentException - Thrown if an unknown enqueue key is provided.

enqueue_abort

public void enqueue_abort(java.lang.Object key)
Description copied from interface: SinkIF
Abort a previously prepared provisional enqueue operation (from the enqueue_prepare() method). Causes the queue to discard the provisionally enqueued elements.
Specified by:
enqueue_abort in interface SinkIF
Following copied from interface: seda.sandStorm.api.SinkIF
Parameters:
key - The enqueue key returned by a previous call to enqueue_prepare().
Throws:
java.lang.IllegalArgumentException - Thrown if an unknown enqueue key is provided.

setEnqueuePredicate

public void setEnqueuePredicate(EnqueuePredicateIF pred)
Description copied from interface: SinkIF
Set the enqueue predicate for this sink. This mechanism allows user to define a method that will 'screen' QueueElementIF's during the enqueue procedure to either accept or reject them. The enqueue predicate runs in the context of the caller of enqueue(), which means it must be simple and fast. This can be used to implement many interesting queue-thresholding policies, such as simple count threshold, credit-based mechanisms, and more.
Specified by:
setEnqueuePredicate in interface SinkIF

getEnqueuePredicate

public EnqueuePredicateIF getEnqueuePredicate()
Description copied from interface: SinkIF
Return the enqueue predicate for this sink.
Specified by:
getEnqueuePredicate in interface SinkIF

toString

public java.lang.String toString()
Overrides:
toString in class java.lang.Object

More information: http://www.cs.berkeley.edu/~mdw/proj/seda/