Component: TupleRouterM

The TupleRouter is the core of the TinyDB system -- it receives queries from the network, creates local state for them (converts them from Queries to ParsedQueries), and then collects results from local sensors and neighboring nodes and feeds them through local queries.

Queries consist of selections and aggregates. Results from queries without aggregates are simply forwarded to the root of the tree to be handled by the query processor.

Queries with aggregates are processed according to the TAG approach: each node collects partial aggregates from its children, combines those aggregates with its own sensor readings, and forwards a partial aggregate on to its parents.

There are three main execution paths within TUPLE_ROUTER; one for accepting new queries, one for accepting results from neighboring nodes, and one for generating local results and deliver data to parent nodes.

QUERY ARRIVAL

------------

1) New queries arrive in a TUPLE_ROUTER_QUERY_MESSAGE. Each query is assumed to be identified by a globally unique ID. Query messages contain a part of a query: either a single field (attribute) to retrieve, a single selection predicate to apply, or a single aggregation predicate to apply. All the QUERY_MESSAGEs describing a single query must arrive before the router will begin routing tuples for that query.

2) Once all the QUERY_MESSAGESs have arrived, the router calls parseQuery() to generate a compact representation of the query in which field names have been replaced with field ids that can be used as offsets into the sensors local catalog (SCHEMA).

3) Given a parsedQuery, the tuple router allocates space at the end of the query to hold a single, "in-flight" tuple for that query -- this tuple will be filled in with the appropriate data fields as the query executes.

4) TupleRouter then calls setSampleRate() to start (or restart) the mote's 32khz clock to fire at the appropriate data-delivery rate for all of the queries currently in the system. If there is only one query, it will fire once per "epoch" -- if there are multiple queries, it will fire at the GCD of the delivery intervals of all the queries.

TUPLE DELIVERY

--------------

1) Whenever a clock event occurs (TUPLE_ROUTER_TIMER_EVENT), the router must perform four actions:

a) Deliver tuples which were completed on the previous clock event (deliverTuplesTask). If the query contains an aggregate, deliver the aggregate data from the aggregate operator; if not, deliver the tuple filled out during the last iteration. Reset the counters that indicate when these queries should be fired again.

b) Decrement the counters for all queries. Any queries who's counters reach 0 need to have data delivered. Reset the expression specific state for these queries (this is specific to the expressions in the queries -- MAX aggregates, for instances, will want to reset the current maximum aggregate to some large negative number.)

c) Fetch data fields for each query firing this epoch. Loop through all fields of all queries, fetch them (using the SCHEMA interface), and fill in the appropriate values in the tuples on the appropriate queries.

d) Route filled in tuples to query operators. First route to selections, then the aggregate (if it exists). If any selection rejects a tuple, stop routing it.

NEIGHBOR RESULT ARRIVAL

-----------------------

When a result arrives from a neighbor (TUPLE_ROUTER_RESULT_MESSAGE), it needs to be integrated into the aggregate values being computed locally. If the result corresponds to an aggregate query, that result is forwarded into the AGG_OPERATOR component, otherwise it is simply forwarded up the routing tree towards the root.

Author:

Sam Madden (madden@cs.berkeley.edu)

Required Interfaces

• Network
• AttrUse
• TupleIntf
• QueryIntf
• ParsedQueryIntf
• Operator AggOperator
• Operator SelOperator
• QueryResultIntf
• MemAlloc
• Leds
• Timer
• Random
• Interrupt
• StdControl ChildControl
• StdControl TimerControl
• StdControl RadioControl
• DBBuffer
• CommandUse
• TinyDBError addResults(QueryResult *qr, ParsedQuery *q, Expr *e)

• QueryProcessor
• StdControl
• RadioQueue
• result_t setFixedComm(bool fixed) "fixed communication" means that we transmit during a time slot proportional to our mote id, rather than learning about the neighborhood size from the root and choosing a random time slot based on that size

• void signalError(TinyDBError err)
• void statusMessage(CharPtr m)

• typedef enum TupleRouterM.__nesc_unnamed4330 { TupleRouterM.STATE_ALLOC_PARSED_QUERY = 0, TupleRouterM.STATE_ALLOC_IN_FLIGHT_QUERY, TupleRouterM.STATE_RESIZE_QUERY, TupleRouterM.STATE_NOT_ALLOCING, TupleRouterM.STATE_ALLOC_QUERY_RESULT } AllocState
  number of clocks events that dictate size of interval -- every 3 seconds
• typedef struct TupleRouterM.__nesc_unnamed4331 { void **next; ParsedQuery q; } *QueryListPtr
  Linked list to track queries currently being processed
• typedef struct TupleRouterM.__nesc_unnamed4331 { void **next; ParsedQuery q; } **QueryListHandle
• typedef struct TupleRouterM.__nesc_unnamed4331 { void **next; ParsedQuery q; } QueryListEl
• typedef void (*MemoryCallback)(Handle *memory)
  Completion routine for memory allocation complete
• typedef struct TupleRouterM.__nesc_unnamed4332 { DbMsgHdr hdr; char qid; } QueryRequestMessage
• typedef struct TupleRouterM.__nesc_unnamed4334 { short start; short end; short size; TOS_Msg msgs[TupleRouterM.MSG_Q_LEN]; } MsgQ
• typedef struct TupleRouterM.__nesc_unnamed4335 { bool isAttr; bool isNull; union TupleRouterM.__nesc_unnamed4336 { AttrDescPtr attr; uint8_t tupleIdx; } u; } TupleFieldDesc
  A data structure for tracking the next tuple field to fill needed since some fields come from base sensors (attrs), and some come from nested queries
• char mDbgMsg[20]
• TOS_Msg mMsg
• TupleRouterM.MsgQ mMsgq
• uint16_t mPendingMask
• uint8_t mCycleToSend
• TupleRouterM.QueryListHandle mQs
• TupleRouterM.QueryListHandle mTail
• Query **mCurQuery
• Handle mTmpHandle
• TupleRouterM.MemoryCallback mAllocCallback
• uint8_t mFetchingFieldId
• char mCurExpr
• Tuple *mCurTuple
• TupleRouterM.QueryListHandle mCurRouteQuery
• QueryResult mResult
• QueryResult mEnqResult
• short mOutputCount
• short mFetchTries
• TupleRouterM.AllocState mAllocState
• short mOldRate
• TupleRouterM.QueryListHandle mCurSendingQuery
• char mCurSendingField
• char mCurSendingExpr
• TOS_Msg mQmsg
• bool mTriedAllocWaiting
• bool mTriedQueryRequest
• unsigned char mXmitSlots
• unsigned char mNumSenders
• short mTicksThisInterval
• short mMsgsThisInterval
• bool mFixedComm
• bool mSendQueryNextClock
• bool mSending
• char *mResultBuf
• ParsedQuery *mLastQuery
• void continueQuery(Handle *memory)
  Continuation after query is successfully alloc'ed

  Parameters:

  memory - The newly allocated handle (must to be non-null)

• bool addQueryField(TOS_MsgPtr msg)
  Given a query message, add the corresponding field or expression to a partially completed query

  Returns:	true iff we already knew about this field
  Parameters:	msg - The query message

• bool allocPendingQuery(TupleRouterM.MemoryCallback callback, Query *q)
  Allocates space for a query, given a query with numExprs and numFields filled in
• bool allocQuery(TupleRouterM.MemoryCallback callback, Query *q)
  Allocate space for a parsed query After request compltes, add the result to qs linked list, and then call callback.
• void parsedCallback(Handle *memory)
  Continuation after parsed query is successfully alloc'ed NOTE: after we setup query, need to resize for tuple at end of query...
• bool parseQuery(Query *q, ParsedQuery *pq)
  Given a query, parse it into pq
• bool queryComplete(Query q)
  Return TRUE if we've heard about all the fields for the specified query
• bool reallocQueryForTuple(TupleRouterM.MemoryCallback callback, TupleRouterM.QueryListHandle qlh)
  Resize qlh to have space for tuple at the end
• void resizedCallback(Handle *memory)
  Continuation after the query is realloced to include space for a tuple

  Parameters:

  memory - Resized parsed query

• void setSampleRate(void)
  Adjust the rate that the main tuple router clock fires at based on EPOCH DURATION of all of the queries that are currently installed
• short gcd(short a, short b)
  Find the GCD of two non-negative integers

  Parameters:	a - The first integer b - The secnd integer
  Returns:	the GCD of a and b

• bool fetchNextAttr(void)
  Fetch the next needed attribute, and
• TupleRouterM.TupleFieldDesc getNextQueryField(ParsedQuery **q)
  Scan queries, looking for fields that haven't been defined yet
• TupleRouterM.QueryListHandle nextQueryToRoute(TupleRouterM.QueryListHandle curQuery)
  
• bool routeToQuery(ParsedQuery *q, Tuple *t)
  Route the specified tuple to the first operator of the specified query.
• Expr *nextExpr(ParsedQuery *q)
  Uses mCurExpr to track the current expression in q that is being applied.
• bool getQuery(uint8_t qid, ParsedQuery **q)
  
• void startFetchingTuples(void)
• void resetTupleState(ParsedQuery *q)
• void fillInAttrVal(char *resultBuf, SchemaErrorNo errorNo)
  Set the value of field mFetchingFieldId (which is the id of an attribute in the schema) in all queries that need the field to the data contained in resultBuf

  Parameters:

  resultBuf - The data to be placed in the field errorNo - An error (if any) returned by the schema in response to the getAttr command that generated resultBuf

• void aggregateResult(ParsedQuery *q, QueryResult *qr, char exprId)
  Apply all aggregate operators to this result.
• void computeOutputRate(void)
  Determine how many communication slots there are in the shortest duration query -- this will determine how long we can afford to maximally wait before sending a result.
• TinyDBError dequeueMessage(TOS_Msg *msg)
• void sendWaitingMessages(void)
  Called from the main timer loop -- task that drains the internal message queue.
• TinyDBError removeQuery(uint8_t qid, BoolPtr success)
  Remove a query from the tuple router
• TinyDBError forwardQuery(TOS_MsgPtr msg)
  Forward out a query message, setting errors as appropriate if the radio is already busy.
• void finishedBufferSetup(void)
  Called when the buffer has been allocated
• void keepRouting(void)
  Continue filling in tuple attributes and (if done with that) routing completed tuples to operators
• task void deliverTuplesTask(void)
  Walk through queries, finding ones that have gone off (timer reached 0), and where the tuples are complete.
• task void routeTask(void)
  routeTask does the heavy lifting of routing tuples to queries.
• task void sendQuery(void)
  Send mCurSendingQuery to a neighbor (we assume the query is in mCurSendingQuery because the sending must be done in multiple phases)
• task void fillInTask(void)
  Used to make attribute setting split phase even when its not ...
• task void mainTask(void)

• uint16_t max(uint16_t a, uint16_t b)
• void SET_READING_QUERY(void)
• void UNSET_READING_QUERY(void)
• bool IS_READING_QUERY(void)
• void SET_PARSING_QUERY(void)
• void UNSET_PARSING_QUERY(void)
• bool IS_PARSING_QUERY(void)
• bool IS_SPACE_ALLOCED(void)
• void UNSET_SPACE_ALLOCED(void)
• void SET_SPACE_ALLOCED(void)
• bool IS_FETCHING_ATTRIBUTE(void)
• void UNSET_FETCHING_ATTRIBUTE(void)
• void SET_FETCHING_ATTRIBUTE(void)
• bool IS_ROUTING_TUPLES(void)
• void UNSET_ROUTING_TUPLES(void)
• void SET_ROUTING_TUPLES(void)
• bool IS_DELIVERING_TUPLES(void)
• void UNSET_DELIVERING_TUPLES(void)
• void SET_DELIVERING_TUPLES(void)
• bool IS_SENDING_MESSAGE(void)
• void UNSET_SENDING_MESSAGE(void)
• void SET_SENDING_MESSAGE(void)
• bool IS_AGGREGATING_RESULT(void)
• void UNSET_AGGREGATING_RESULT(void)
• void SET_AGGREGATING_RESULT(void)
• bool IS_SENDING_QUERY(void)
• void UNSET_SENDING_QUERY(void)
• void SET_SENDING_QUERY(void)
• bool IS_IN_QUERY_MSG(void)
• void UNSET_IS_IN_QUERY_MSG(void)
• void SET_IS_IN_QUERY_MSG(void)
• command result_t StdControl.init (void) Intialize the tuple router

• command result_t StdControl.start (void)
• command result_t StdControl.stop (void)
• event TOS_MsgPtr Network.querySub (TOS_MsgPtr msg) Message indicating the arrival of (part of) a query

• TinyDBError forwardQuery (TOS_MsgPtr msg) Forward out a query message, setting errors as appropriate if the radio is already busy.

• void continueQuery(Handle *memory) Continuation after query is successfully alloc'ed

  Parameters:

  memory - The newly allocated handle (must to be non-null)

• bool addQueryField(TOS_MsgPtr msg) Given a query message, add the corresponding field or expression to a partially completed query

  Returns:	true iff we already knew about this field
  Parameters:	msg - The query message

• void finishedBufferSetup(void) Called when the buffer has been allocated

• void parsedCallback (Handle *memory) Continuation after parsed query is successfully alloc'ed NOTE: after we setup query, need to resize for tuple at end of query...

• void resizedCallback(Handle *memory) Continuation after the query is realloced to include space for a tuple

  Parameters:

  memory - Resized parsed query

• TinyDBError removeQuery (uint8_t qid, BoolPtr success) Remove a query from the tuple router

• task void sendQuery(void) Send mCurSendingQuery to a neighbor (we assume the query is in mCurSendingQuery because the sending must be done in multiple phases)

• event TOS_MsgPtr Network.queryRequestSub (TOS_MsgPtr msg) Message indicating neighbor requested a query from us If we know about the query, and aren't otherwise occupied with sending / routing, send the query back

  Parameters:

  The - QueryResultMessage from the neighbor

• event TOS_MsgPtr Network.snoopedSub (TOS_MsgPtr msg, uint8_t amId, bool isParent) A message not directly addressed to us that we overhead Use this for time synchronization with our parent, and to snoop on queries.

• event TinyDBError SelOperator.processedTuple (Tuple *t, ParsedQuery *q, Expr *e, bool passed) Continue processing a tuple after a selection operator Basically, if the tuple passed the selection, we continue routing it to additional operators.

• event TinyDBError SelOperator.processedResult (QueryResult *qr, ParsedQuery *q, Expr *e)
• event TinyDBError AggOperator.processedTuple (Tuple *t, ParsedQuery *q, Expr *e, bool passed) Continue processing a tuple after an aggregation operator has been applied

  Parameters:

  t - The tuple passed into the operator q - The query that the tuple belongs to e - The expression that processed the tuple passed - (Should be true for aggregates)

• event TinyDBError AggOperator.processedResult (QueryResult *qr, ParsedQuery *q, Expr *e) Called every time we route a query result through an aggregate operator.

• event TOS_MsgPtr Network.dataSub (TOS_MsgPtr msg) Received a result from a neighbor -- need to either:

  1) process it, if is an aggregate result

  or

  2) forward it, if it is a non-aggregate result

  Parameters:

  msg - The message that was received

• void aggregateResult (ParsedQuery *q, QueryResult *qr, char exprId) Apply all aggregate operators to this result.

• void setSampleRate(void) Adjust the rate that the main tuple router clock fires at based on EPOCH DURATION of all of the queries that are currently installed

• void computeOutputRate(void) Determine how many communication slots there are in the shortest duration query -- this will determine how long we can afford to maximally wait before sending a result.

• short gcd(short a, short b) Find the GCD of two non-negative integers

  Parameters:	a - The first integer b - The secnd integer
  Returns:	the GCD of a and b

• event result_t Timer.fired (void) Clock fired event --
  Works as follows:
  1) Output tuples from previous epochs
  2) Deterimine what queries fire in this epoch
  3) Collect samples from those queries
  4) Fill in the tuples in those queries
  5) Apply operators to those tuples
  

  While this is happening, results may arrive from other sensors nodes representing results from the last epoch.

• task void mainTask(void)
• task void deliverTuplesTask (void) Walk through queries, finding ones that have gone off (timer reached 0), and where the tuples are complete.

• void sendWaitingMessages (void) Called from the main timer loop -- task that drains the internal message queue.

• event result_t Network.outputDone (TOS_MsgPtr msg, uint8_t amId) Event that's signalled when a send is complete

• void startFetchingTuples(void)
• void resetTupleState(ParsedQuery *q)
• bool fetchNextAttr (void) Fetch the next needed attribute, and

• TupleRouterM.TupleFieldDesc getNextQueryField(ParsedQuery **pq) Scan queries, looking for fields that haven't been defined yet

• void keepRouting(void) Continue filling in tuple attributes and (if done with that) routing completed tuples to operators

• void fillInAttrVal(char *resultBuf, SchemaErrorNo errorNo) Set the value of field mFetchingFieldId (which is the id of an attribute in the schema) in all queries that need the field to the data contained in resultBuf

  Parameters:

  resultBuf - The data to be placed in the field errorNo - An error (if any) returned by the schema in response to the getAttr command that generated resultBuf

• task void fillInTask(void) Used to make attribute setting split phase even when its not ...

• event result_t AttrUse.getAttrDone (char *name, char *resultBuf, SchemaErrorNo errorNo) Completion event after some data was fetched Params should be filled out with the result of the command

  Parameters:

  name - The name of the attribute that was fetched resultBuf - The value of the attribute errorNo - Errors that occurred while fetching the result

• task void routeTask (void) routeTask does the heavy lifting of routing tuples to queries.

• TupleRouterM.QueryListHandle nextQueryToRoute (TupleRouterM.QueryListHandle curQuery)

• bool routeToQuery (ParsedQuery *q, Tuple *t) Route the specified tuple to the first operator of the specified query.

• Expr nextExpr (ParsedQuery *q) Uses mCurExpr to track the current expression in q that is being applied.

• bool getQuery (uint8_t qid, ParsedQuery **q)

• command ParsedQueryPtr QueryProcessor.getQueryCmd (uint8_t qid)
• command short QueryProcessor.msgToQueryRoot (TOS_MsgPtr msg) Given a processor message return the owner (origninating node) of the query, or -1 if the query is unknown or the message is a query processor message.

• bool parseQuery (Query *q, ParsedQuery *pq) Given a query, parse it into pq

• bool allocPendingQuery (TupleRouterM.MemoryCallback callback, Query *q) Allocates space for a query, given a query with numExprs and numFields filled in

• bool allocQuery (TupleRouterM.MemoryCallback callback, Query *q) Allocate space for a parsed query After request compltes, add the result to qs linked list, and then call callback.

• bool reallocQueryForTuple(TupleRouterM.MemoryCallback callback, TupleRouterM.QueryListHandle qlh) Resize qlh to have space for tuple at the end

• bool queryComplete(Query q) Return TRUE if we've heard about all the fields for the specified query

• command result_t setFixedComm(bool fixed) "fixed communication" means that we transmit during a time slot proportional to our mote id, rather than learning about the neighborhood size from the root and choosing a random time slot based on that size

• event result_t MemAlloc.allocComplete (Handle *handle, result_t complete)
• event result_t MemAlloc.reallocComplete (Handle handle, result_t complete)
• event result_t MemAlloc.compactComplete (void)
• command TinyDBError RadioQueue.enqueue (const char *msg, uint16_t len) Copy the specified bytes into the message queue.

• TinyDBError dequeueMessage(TOS_Msg *msg)
• command void signalError(TinyDBError err)
• command void statusMessage(CharPtr m)
• event result_t DBBuffer.resultReady (uint8_t bufferId)
• event result_t DBBuffer.getNext (uint8_t bufferId)
• event result_t DBBuffer.allocComplete (uint8_t bufferId, TinyDBError result)
• event result_t CommandUse.commandDone (char *commandName, char *resultBuf, SchemaErrorNo err)
• event result_t DBBuffer.getComplete (uint8_t bufferId, QueryResult *buf)
• event result_t DBBuffer.putComplete (uint8_t bufferId, QueryResult *buf, TinyDBError err)

forwardQuery

parsedCallback

removeQuery

Network.snoopedSub

SelOperator.processedTuple

AggOperator.processedResult

aggregateResult

Timer.fired

deliverTuplesTask

sendWaitingMessages

fetchNextAttr

routeTask

nextQueryToRoute

routeToQuery

nextExpr

getQuery

QueryProcessor.msgToQueryRoot

parseQuery

allocPendingQuery

allocQuery

RadioQueue.enqueue