← Back to branch summary

~ubuntu-branches/ubuntu/hardy/avidemux/hardy

« back to all changes in this revision

Viewing changes to avidemux/ADM_toolkit/ADM_packetQueue.cpp

  • Committer: Bazaar Package Importer
  • Author(s): Daniel T Chen
  • Date: 2006-12-15 17:13:20 UTC
  • mfrom: (1.1.6 upstream)
  • Revision ID: james.westby@ubuntu.com-20061215171320-w79pvpehxx2fr217
Tags: 1:2.3.0-0.0ubuntu1
* Merge from debian-multimedia.org, remaining Ubuntu change:
  - desktop file,
  - no support for ccache and make -j.
* Closes Ubuntu: #69614.

Show diffs side-by-side

added added

removed removed

Lines of Context:
avidemux/ADM_toolkit/ADM_packetQueue.cpp
 
1
/***************************************************************************
 
2
              
 
3
    copyright            : (C) 2006 by mean
 
4
    email                : fixounet@free.fr
 
5
 ***************************************************************************/
 
6
 
 
7
/***************************************************************************
 
8
 *                                                                         *
 
9
 *   This program is free software; you can redistribute it and/or modify  *
 
10
 *   it under the terms of the GNU General Public License as published by  *
 
11
 *   the Free Software Foundation; either version 2 of the License, or     *
 
12
 *   (at your option) any later version.                                   *
 
13
 *                                                                         *
 
14
 ***************************************************************************/
 
15
#include "config.h"
 
16
 
 
17
#include <stdio.h>
 
18
#include <stdlib.h>
 
19
#include <string.h>
 
20
#include <ADM_assert.h>
 
21
#include "ADM_library/default.h"
 
22
#include "ADM_toolkit/ADM_threads.h"
 
23
 
 
24
#include "ADM_packetQueue.h"
 
25
/*!
 
26
  Constructor for packetQueue
 
27
  \param name   : name of the packetQueue, useful for debugging
 
28
  \param nbSlot : How many packets can the packetQueue hold
 
29
  \param buffSize : BufferSize in bytes
 
30
 */
 
31
 
 
32
PacketQueue::PacketQueue(const char *name,uint32_t nbSlot,uint32_t buffSize)
 
33
{
 
34
  
 
35
  _nbSlots=nbSlot;
 
36
  _bufferSize=buffSize;
 
37
  _name=ADM_strdup(name);
 
38
  _buffer=new uint8_t[_bufferSize];
 
39
  _slots=new Slots[_nbSlots];
 
40
  _mutex=new admMutex(_name);
 
41
  _pusherCond=new admCond(_mutex);
 
42
  _poperCond=new admCond(_mutex);
 
43
  _slotQueue=_slotHead=0;
 
44
  _bufferQueue=_bufferHead=0;
 
45
  _eof=0;
 
46
  _aborted=0;
 
47
  printf("PacketQueue %s created\n",_name);
 
48
}
 
49
/*!
 
50
  Destructor
 
51
 */
 
52
 
 
53
PacketQueue::~PacketQueue()
 
54
{
 
55
  if(_pusherCond) delete _pusherCond;
 
56
  _pusherCond=NULL;
 
57
  if(_poperCond) delete _poperCond;
 
58
  _poperCond=NULL;
 
59
  if(_mutex) delete _mutex;
 
60
  _mutex=NULL;
 
61
  if(_name) ADM_dealloc(_name);
 
62
  _name=NULL;
 
63
  if(_buffer) delete [] _buffer;
 
64
  _buffer=NULL;
 
65
  if(_slots) delete [] _slots;
 
66
}
 
67
/*!
 
68
  Signal the pusher has finished, wake the poper.
 
69
  Warning : Must be called with mutex held
 
70
 */
 
71
 
 
72
uint8_t   PacketQueue::IsEmpty(void)
 
73
{
 
74
  uint8_t r=0;
 
75
  
 
76
  if(_slotHead==_slotQueue)
 
77
  { 
 
78
      r=1;
 
79
  }
 
80
  
 
81
  return r;
 
82
}
 
83
/*!
 
84
  Signal the pusher has finished, wake the poper.
 
85
 
 
86
 */
 
87
 
 
88
uint8_t  PacketQueue::Finished(void)
 
89
{
 
90
  uint8_t r=0;
 
91
  _mutex->lock();
 
92
  _eof=1;
 
93
  _poperCond->abort();
 
94
  _mutex->unlock();
 
95
  return 1;
 
96
  
 
97
}
 
98
/*!
 
99
  Put a packet in the buffer
 
100
  Wait on _pusherCond if no slot available or buffer full
 
101
  \param ptr  : Ptr where to take the packet from
 
102
  \param size : packetsize
 
103
 
 
104
 */
 
105
uint8_t   PacketQueue::Push(uint8_t *ptr, uint32_t size,uint32_t sample)
 
106
{
 
107
  uint8_t r=0;
 
108
  uint32_t slot;
 
109
  uint32_t available;
 
110
  _mutex->lock();
 
111
  // First try to allocate a slot
 
112
  while(((_nbSlots+_slotHead-_slotQueue)%_nbSlots)==1)
 
113
  {
 
114
    _pusherCond->wait();
 
115
    _mutex->lock(); 
 
116
  }
 
117
  // Ok we have a slot,
 
118
  // Now lets's see if we have enough data in the buffer (we are still under lock)
 
119
  while(!_eof)
 
120
  {
 
121
      available=availableSpace(); 
 
122
      if(available>=size)
 
123
      {
 
124
        
 
125
         slot=_slotHead;
 
126
        _slotHead++;
 
127
        _slotHead%=_nbSlots;
 
128
        _slots[slot].size=size;
 
129
        _slots[slot].sample=sample;
 
130
        _slots[slot].startAt=_bufferHead;
 
131
        
 
132
       // printf("Pushing slot %d at %u\n",slot,_bufferHead,_slots[slot].startAt);
 
133
        if(_bufferSize>=(_bufferHead+size))
 
134
        {
 
135
          memcpy(&_buffer[ _bufferHead],ptr,size);
 
136
          _bufferHead+=size;
 
137
        }
 
138
        else  // Split
 
139
        {
 
140
          uint32_t part1=_bufferSize-_bufferHead;
 
141
          memcpy(&_buffer[ _bufferHead],ptr,part1);
 
142
          memcpy(&_buffer[ 0],ptr+part1,size-part1);
 
143
          _bufferHead=size-part1;
 
144
        }
 
145
        // Look if someone was waiting ...
 
146
        if(_poperCond->iswaiting())
 
147
        {
 
148
            _poperCond->wakeup();
 
149
        }
 
150
        _mutex->unlock();
 
151
        return 1;
 
152
      }
 
153
      _pusherCond->wait();
 
154
      _mutex->lock();
 
155
  }
 
156
  _mutex->unlock();
 
157
  printf("[PKTQ] %s is eof\n",_name);
 
158
  return 0;
 
159
}
 
160
/*!
 
161
  Read a packet from the queue
 
162
  Wait on the _poperCond if empty
 
163
  If aborted returns immediatly
 
164
  \param ptr : Ptr where to put the packet
 
165
  \param size : returns packetsize
 
166
 
 
167
*/
 
168
uint8_t   PacketQueue::Pop(uint8_t *ptr, uint32_t *size,uint32_t *sample)
 
169
{
 
170
  uint8_t r=0;
 
171
  uint32_t slot;
 
172
  uint32_t available,sz,position;
 
173
  _mutex->lock();
 
174
  // is there something ?
 
175
  while(IsEmpty() && !_eof)
 
176
  {
 
177
    _poperCond->wait();
 
178
    _mutex->lock();
 
179
  }
 
180
  if(IsEmpty() && _eof)
 
181
  {
 
182
    *size=0;
 
183
    _mutex->unlock();
 
184
    return 0;
 
185
  }
 
186
  //
 
187
  //printf("Pop : Head %u Tail %u\n",_slotHead,_slotQueue);
 
188
  // ok, which slot ?
 
189
  slot=_slotQueue;
 
190
  _slotQueue++;
 
191
  _slotQueue%=_nbSlots;
 
192
  sz=*size=_slots[slot].size;
 
193
  *sample=_slots[slot].sample;
 
194
  //printf("Poping slot %d at %u\n",slot,_bufferQueue,_slots[slot].startAt);
 
195
  if(!_bufferQueue==_slots[slot].startAt)
 
196
  {
 
197
    printf("Buffer Q:%u\n",_bufferQueue);
 
198
    printf("Slot :%u\n",_slots[slot].startAt);
 
199
    ADM_assert(0);
 
200
  }
 
201
  
 
202
  if(_bufferSize>=(_bufferQueue+sz))
 
203
  {
 
204
    memcpy(ptr,&(_buffer[_bufferQueue]),sz);
 
205
    _bufferQueue+=sz;
 
206
    _bufferQueue%=_bufferSize;
 
207
  }
 
208
  else  // Split
 
209
  {
 
210
    uint32_t part1=_bufferSize-_bufferQueue;
 
211
    memcpy(ptr,&_buffer[ _bufferQueue],part1);
 
212
    memcpy(ptr+part1,&_buffer[ 0],sz-part1);
 
213
    _bufferQueue=sz-part1;
 
214
  }
 
215
  // In case we made some space
 
216
  if(_pusherCond->iswaiting())
 
217
  {
 
218
    uint32_t third=(2*_bufferSize)/3;
 
219
    uint32_t available=availableSpace(); 
 
220
    // Only wakeup if we go over 1/3 of the buffer
 
221
    ADM_assert(available>=sz);
 
222
    if(available > third)
 
223
        _pusherCond->wakeup();
 
224
  }
 
225
  _mutex->unlock();
 
226
  return 1;
 
227
}
 
228
/*!
 
229
  Returns available space in bytes in the buffer
 
230
  Warning, this method must be called with mutex
 
231
    hold!.
 
232
 
 
233
*/
 
234
uint32_t PacketQueue::availableSpace(void)
 
235
{
 
236
  
 
237
  uint32_t space=_bufferSize+_bufferHead-_bufferQueue;
 
238
  space=space%_bufferSize;
 
239
  space=_bufferSize-space;
 
240
  
 
241
  return space;
 
242
}
 
243
uint8_t PacketQueue::Abort(void)
 
244
{
 
245
  
 
246
  Finished();
 
247
  
 
248
  _mutex->lock();
 
249
  _aborted=1;
 
250
  _slotHead=_slotQueue=0;
 
251
  _bufferQueue=_bufferHead=0;
 
252
  _mutex->unlock();
 
253
  _pusherCond->abort();
 
254
  
 
255
  return 1;
 
256
}
 
257
uint8_t   PacketQueue::isEof(void)
 
258
{
 
259
uint8_t r=0;
 
260
_mutex->lock();
 
261
    if((_eof||_aborted) && _slotHead==_slotQueue) r=1;
 
262
_mutex->unlock();
 
263
return r;
 
264
}
 
265
//EOF
 
266