Home | History | Annotate | Download | only in src 
      1 /* ------------------------------------------------------------------
      2  * Copyright (C) 1998-2009 PacketVideo
      3  *
      4  * Licensed under the Apache License, Version 2.0 (the "License");
      5  * you may not use this file except in compliance with the License.
      6  * You may obtain a copy of the License at
      7  *
      8  *      http://www.apache.org/licenses/LICENSE-2.0
      9  *
     10  * Unless required by applicable law or agreed to in writing, software
     11  * distributed under the License is distributed on an "AS IS" BASIS,
     12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
     13  * express or implied.
     14  * See the License for the specific language governing permissions
     15  * and limitations under the License.
     16  * -------------------------------------------------------------------
     17  */
     18 #ifndef THREADSAFE_QUEUE_H_INCLUDED
     19 #define THREADSAFE_QUEUE_H_INCLUDED
     20 
     21 #ifndef OSCL_BASE_H_INCLUDED
     22 #include "oscl_base.h"
     23 #endif
     24 
     25 #ifndef OSCL_BASE_ALLOC_H_INCLUDED
     26 #include "oscl_base_alloc.h"
     27 #endif
     28 
     29 #ifndef OSCL_SCHEDULER_AO_H_INCLUDED
     30 #include "oscl_scheduler_ao.h"
     31 #endif
     32 
     33 #ifndef OSCL_SEMAPHORE_H_INCLUDED
     34 #include "oscl_semaphore.h"
     35 #endif
     36 
     37 #ifndef OSCL_MUTEX_H_INCLUDED
     38 #include "oscl_mutex.h"
     39 #endif
     40 
     41 #ifndef OSCL_VECTOR_H_INCLUDED
     42 #include "oscl_vector.h"
     43 #endif
     44 
     45 class ThreadSafeQueue;
     46 class ThreadSafeQueueObserver
     47 {
     48     public:
     49         /*
     50         ** Observer receives this callback when data has been added to the queue.
     51         ** There may be multiple data available when this is called.  There will
     52         ** not necessarily be one notice per item, so the observer should process
     53         ** the entire queue when it receives this notice.
     54         **
     55         ** param (in) aQueue: originating queue, in case multiple queues share an observer.
     56         */
     57         OSCL_IMPORT_REF virtual void ThreadSafeQueueDataAvailable(ThreadSafeQueue* aQueue) = 0;
     58 };
     59 
     60 /*
     61 ** A thread-safe queue.
     62 ** This queue resides in a thread that has a scheduler.
     63 ** Any thread(s) can add data.
     64 ** The queue will call ThreadSafeQueueDataAvailable from the owner thread
     65 ** when data has been added to the queue.
     66 */
     67 
     68 typedef uint32 ThreadSafeQueueId;
     69 
     70 class ThreadSafeQueue: public OsclActiveObject
     71 {
     72     public:
     73 
     74         OSCL_IMPORT_REF ThreadSafeQueue();
     75         OSCL_IMPORT_REF virtual ~ThreadSafeQueue();
     76 
     77         /*
     78         ** Configure
     79         ** @param (in) aObs: observer
     80         ** @param (in) aReserve: queue reserve
     81         ** @param (in) aId: initial value for the assigned data IDs.
     82         */
     83         OSCL_IMPORT_REF void Configure(ThreadSafeQueueObserver* aObs, uint32 aReserve = 1, uint32 aId = 0);
     84 
     85         /*
     86         ** Add data to the queue from any thread.
     87         ** @param (in) aData: item to queue
     88         ** @param (in) aId: optional command ID.  If none is input, then one will be
     89         **   generated from the internal counter.
     90         ** @return: a unique ID for the data.
     91         */
     92         OSCL_IMPORT_REF ThreadSafeQueueId AddToQueue(OsclAny *aData, ThreadSafeQueueId* aId = NULL);
     93 
     94         /*
     95         ** DeQueue data from any thread.  Data is returned in FIFO order.
     96         ** @param (out) aId: the ID that was returned by AddToQueue.
     97         ** @param (out) aData: the queued item.
     98         ** @return: number of items de-queued (either 0 or 1)
     99         */
    100         OSCL_IMPORT_REF uint32 DeQueue(ThreadSafeQueueId& aId, OsclAny*& aData);
    101 
    102         /*
    103         ** Checks whether calling context is the same as the queue thread context.
    104         ** @return true if same thread context.
    105         */
    106         OSCL_IMPORT_REF bool IsInThread();
    107 
    108     protected:
    109 
    110         class ThreadSafeQueueElement
    111         {
    112             public:
    113                 ThreadSafeQueueElement(ThreadSafeQueueId aId, OsclAny* aData)
    114                         : iId(aId)
    115                         , iData(aData)
    116                 {}
    117                 ThreadSafeQueueElement(const ThreadSafeQueueElement& aElem)
    118                         : iId(aElem.iId)
    119                         , iData(aElem.iData)
    120                 {}
    121                 ThreadSafeQueueId iId;
    122                 OsclAny *iData;
    123         };
    124 
    125         void Run();
    126 
    127         ThreadSafeQueueObserver *iObserver;
    128 
    129         OsclSemaphore iQueueReadySem;//this sem is signaled when it's OK to signal the AO.
    130         OsclNoYieldMutex iQueueMut; //data lock
    131 
    132         Oscl_Vector<ThreadSafeQueueElement, _OsclBasicAllocator> iQueue;
    133 
    134         uint32 iCounter;//data ID counter
    135 
    136         TOsclThreadId iThreadId;
    137 
    138         void Lock();
    139         void Unlock();
    140 };
    141 
    142 #endif
    143