/[svn]/linuxsampler/trunk/src/engines/gig/DiskThread.cpp
ViewVC logotype

Contents of /linuxsampler/trunk/src/engines/gig/DiskThread.cpp

Parent Directory Parent Directory | Revision Log Revision Log


Revision 840 - (show annotations) (download)
Sun Feb 26 13:00:08 2006 UTC (12 years, 1 month ago) by persson
File size: 15886 byte(s)
* fixed some concurrency problems between the lscp thread and the
  audio/midi threads for these commands: load engine, set channel
  audio output device, set channel midi input device/port/channel and
  remove channel. Thanks to Vincent Bongiorno for bug reports and
  testing.
* removed an autotools warning
* fixed an iterator bug
* minor code clean-ups

1 /***************************************************************************
2 * *
3 * LinuxSampler - modular, streaming capable sampler *
4 * *
5 * Copyright (C) 2003, 2004 by Benno Senoner and Christian Schoenebeck *
6 * Copyright (C) 2005 Christian Schoenebeck *
7 * *
8 * This program is free software; you can redistribute it and/or modify *
9 * it under the terms of the GNU General Public License as published by *
10 * the Free Software Foundation; either version 2 of the License, or *
11 * (at your option) any later version. *
12 * *
13 * This program is distributed in the hope that it will be useful, *
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of *
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
16 * GNU General Public License for more details. *
17 * *
18 * You should have received a copy of the GNU General Public License *
19 * along with this program; if not, write to the Free Software *
20 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, *
21 * MA 02111-1307 USA *
22 ***************************************************************************/
23
24 #include <sstream>
25
26 #include "DiskThread.h"
27
28 namespace LinuxSampler { namespace gig {
29
30 // *********** DiskThread **************
31 // *
32
33
34 // just a placeholder to mark a cell in the pickup array as 'reserved'
35 Stream* DiskThread::SLOT_RESERVED = (Stream*) &SLOT_RESERVED;
36
37
38 // #########################################################################
39 // # Foreign Thread Section
40 // # (following code intended to be interface for audio thread)
41
42
43 /**
44 * Suspend disk thread, kill all active streams, clear all queues and the
45 * pickup array and reset all streams. Call this method to bring everything
46 * in the disk thread to day one. If the disk thread was running, it will be
47 * respawned right after everything was reset.
48 */
49 void DiskThread::Reset() {
50 bool running = this->IsRunning();
51 if (running) this->StopThread();
52 for (int i = 0; i < CONFIG_MAX_STREAMS; i++) {
53 pStreams[i]->Kill();
54 }
55 for (int i = 1; i <= CONFIG_MAX_STREAMS; i++) {
56 pCreatedStreams[i] = NULL;
57 }
58 GhostQueue->init();
59 CreationQueue->init();
60 DeletionQueue->init();
61 ActiveStreamCount = 0;
62 ActiveStreamCountMax = 0;
63 if (running) this->StartThread(); // start thread only if it was running before
64 }
65
66 String DiskThread::GetBufferFillBytes() {
67 bool activestreams = false;
68 std::stringstream ss;
69 for (uint i = 0; i < this->Streams; i++) {
70 if (pStreams[i]->GetState() == Stream::state_unused) continue;
71 uint bufferfill = pStreams[i]->GetReadSpace() * sizeof(sample_t);
72 uint streamid = (uint) pStreams[i]->GetHandle();
73 if (!streamid) continue;
74
75 if (activestreams) ss << ",[" << streamid << ']' << bufferfill;
76 else {
77 ss << '[' << streamid << ']' << bufferfill;
78 activestreams = true;
79 }
80 }
81 return ss.str();
82 }
83
84 String DiskThread::GetBufferFillPercentage() {
85 bool activestreams = false;
86 std::stringstream ss;
87 for (uint i = 0; i < this->Streams; i++) {
88 if (pStreams[i]->GetState() == Stream::state_unused) continue;
89 uint bufferfill = (uint) ((float) pStreams[i]->GetReadSpace() / (float) CONFIG_STREAM_BUFFER_SIZE * 100);
90 uint streamid = (uint) pStreams[i]->GetHandle();
91 if (!streamid) continue;
92
93 if (activestreams) ss << ",[" << streamid << ']' << bufferfill << '%';
94 else {
95 ss << '[' << streamid << ']' << bufferfill;
96 activestreams = true;
97 }
98 }
99 return ss.str();
100 }
101
102 /**
103 * Returns -1 if command queue or pickup pool is full, 0 on success (will be
104 * called by audio thread within the voice class).
105 */
106 int DiskThread::OrderNewStream(Stream::reference_t* pStreamRef, ::gig::Sample* pSample, unsigned long SampleOffset, bool DoLoop) {
107 dmsg(4,("Disk Thread: new stream ordered\n"));
108 if (CreationQueue->write_space() < 1) {
109 dmsg(1,("DiskThread: Order queue full!\n"));
110 return -1;
111 }
112
113 const Stream::OrderID_t newOrder = CreateOrderID();
114 if (!newOrder) {
115 dmsg(1,("DiskThread: there was no free slot\n"));
116 return -1; // there was no free slot
117 }
118
119 pStreamRef->State = Stream::state_active;
120 pStreamRef->OrderID = newOrder;
121 pStreamRef->hStream = CreateHandle();
122 pStreamRef->pStream = NULL; // a stream has to be activated by the disk thread first
123
124 create_command_t cmd;
125 cmd.OrderID = pStreamRef->OrderID;
126 cmd.hStream = pStreamRef->hStream;
127 cmd.pStreamRef = pStreamRef;
128 cmd.pSample = pSample;
129 cmd.SampleOffset = SampleOffset;
130 cmd.DoLoop = DoLoop;
131
132 CreationQueue->push(&cmd);
133 return 0;
134 }
135
136 /**
137 * Returns -1 if command queue is full, 0 on success (will be called by audio
138 * thread within the voice class).
139 */
140 int DiskThread::OrderDeletionOfStream(Stream::reference_t* pStreamRef) {
141 dmsg(4,("Disk Thread: stream deletion ordered\n"));
142 if (DeletionQueue->write_space() < 1) {
143 dmsg(1,("DiskThread: Deletion queue full!\n"));
144 return -1;
145 }
146
147 delete_command_t cmd;
148 cmd.pStream = pStreamRef->pStream;
149 cmd.hStream = pStreamRef->hStream;
150 cmd.OrderID = pStreamRef->OrderID;
151
152 DeletionQueue->push(&cmd);
153 return 0;
154 }
155
156 /**
157 * Returns the pointer to a disk stream if the ordered disk stream
158 * represented by the \a StreamOrderID was already activated by the disk
159 * thread, returns NULL otherwise. If the call was successful, thus if it
160 * returned a valid stream pointer, the caller has to the store that pointer
161 * by himself, because it's not possible to call this method again with the
162 * same used order ID; this method is just intended for picking up an ordered
163 * disk stream. This method will usually be called by the voice class (within
164 * the audio thread).
165 *
166 * @param StreamOrderID - ID previously returned by OrderNewStream()
167 * @returns pointer to created stream object, NULL otherwise
168 */
169 Stream* DiskThread::AskForCreatedStream(Stream::OrderID_t StreamOrderID) {
170 dmsg(4,("Disk Thread: been asked if stream already created, OrderID=%x ", StreamOrderID));
171 Stream* pStream = pCreatedStreams[StreamOrderID];
172 if (pStream && pStream != SLOT_RESERVED) {
173 dmsg(4,("(yes created)\n"));
174 pCreatedStreams[StreamOrderID] = NULL; // free the slot for a new order
175 return pStream;
176 }
177 dmsg(4,("(no not yet created)\n"));
178 return NULL;
179 }
180
181
182
183 // #########################################################################
184 // # Disk Thread Only Section
185 // # (following code should only be executed by the disk thread)
186
187
188 DiskThread::DiskThread(uint BufferWrapElements) : Thread(true, false, 1, -2) {
189 DecompressionBuffer = ::gig::Sample::CreateDecompressionBuffer(CONFIG_STREAM_MAX_REFILL_SIZE);
190 CreationQueue = new RingBuffer<create_command_t>(1024);
191 DeletionQueue = new RingBuffer<delete_command_t>(1024);
192 GhostQueue = new RingBuffer<Stream::Handle>(CONFIG_MAX_STREAMS);
193 Streams = CONFIG_MAX_STREAMS;
194 RefillStreamsPerRun = CONFIG_REFILL_STREAMS_PER_RUN;
195 for (int i = 0; i < CONFIG_MAX_STREAMS; i++) {
196 pStreams[i] = new Stream(&DecompressionBuffer, CONFIG_STREAM_BUFFER_SIZE, BufferWrapElements); // 131072 sample words
197 }
198 for (int i = 1; i <= CONFIG_MAX_STREAMS; i++) {
199 pCreatedStreams[i] = NULL;
200 }
201 ActiveStreamCountMax = 0;
202 }
203
204 DiskThread::~DiskThread() {
205 for (int i = 0; i < CONFIG_MAX_STREAMS; i++) {
206 if (pStreams[i]) delete pStreams[i];
207 }
208 if (CreationQueue) delete CreationQueue;
209 if (DeletionQueue) delete DeletionQueue;
210 if (GhostQueue) delete GhostQueue;
211 ::gig::Sample::DestroyDecompressionBuffer(DecompressionBuffer);
212 }
213
214 int DiskThread::Main() {
215 dmsg(3,("Disk thread running\n"));
216 while (true) {
217 pthread_testcancel(); // mandatory for OSX
218 IsIdle = true; // will be set to false if a stream got filled
219
220 // if there are ghost streams, delete them
221 for (int i = 0; i < GhostQueue->read_space(); i++) { //FIXME: unefficient
222 Stream::Handle hGhostStream;
223 GhostQueue->pop(&hGhostStream);
224 bool found = false;
225 for (int i = 0; i < this->Streams; i++) {
226 if (pStreams[i]->GetHandle() == hGhostStream) {
227 pStreams[i]->Kill();
228 found = true;
229 break;
230 }
231 }
232 if (!found) GhostQueue->push(&hGhostStream); // put ghost stream handle back to the queue
233 }
234
235 // if there are creation commands, create new streams
236 while (Stream::UnusedStreams > 0 && CreationQueue->read_space() > 0) {
237 create_command_t command;
238 CreationQueue->pop(&command);
239 CreateStream(command);
240 }
241
242 // if there are deletion commands, delete those streams
243 while (Stream::UnusedStreams < Stream::TotalStreams && DeletionQueue->read_space() > 0) {
244 delete_command_t command;
245 DeletionQueue->pop(&command);
246 DeleteStream(command);
247 }
248
249 RefillStreams(); // refill the most empty streams
250
251 // if nothing was done during this iteration (eg no streambuffer
252 // filled with data) then sleep for 30ms
253 if (IsIdle) usleep(30000);
254
255 int streamsInUsage = 0;
256 for (int i = Streams - 1; i >= 0; i--) {
257 if (pStreams[i]->GetState() != Stream::state_unused) streamsInUsage++;
258 }
259 ActiveStreamCount = streamsInUsage;
260 if (streamsInUsage > ActiveStreamCountMax) ActiveStreamCountMax = streamsInUsage;
261 }
262
263 return EXIT_FAILURE;
264 }
265
266 void DiskThread::CreateStream(create_command_t& Command) {
267 // search for unused stream
268 Stream* newstream = NULL;
269 for (int i = Streams - 1; i >= 0; i--) {
270 if (pStreams[i]->GetState() == Stream::state_unused) {
271 newstream = pStreams[i];
272 break;
273 }
274 }
275 if (!newstream) {
276 std::cerr << "No unused stream found (OrderID:" << Command.OrderID << ") - report if this happens, this is a bug!\n" << std::flush;
277 return;
278 }
279 newstream->Launch(Command.hStream, Command.pStreamRef, Command.pSample, Command.SampleOffset, Command.DoLoop);
280 dmsg(4,("new Stream launched by disk thread (OrderID:%d,StreamHandle:%d)\n", Command.OrderID, Command.hStream));
281 if (pCreatedStreams[Command.OrderID] != SLOT_RESERVED) {
282 std::cerr << "DiskThread: Slot " << Command.OrderID << " already occupied! Please report this!\n" << std::flush;
283 newstream->Kill();
284 return;
285 }
286 pCreatedStreams[Command.OrderID] = newstream;
287 }
288
289 void DiskThread::DeleteStream(delete_command_t& Command) {
290 if (Command.pStream) Command.pStream->Kill();
291 else { // the stream wasn't created by disk thread or picked up by audio thread yet
292
293 // if stream was created but not picked up yet
294 Stream* pStream = pCreatedStreams[Command.OrderID];
295 if (pStream && pStream != SLOT_RESERVED) {
296 pStream->Kill();
297 pCreatedStreams[Command.OrderID] = NULL; // free slot for new order
298 return;
299 }
300
301 // the stream was not created yet
302 if (GhostQueue->write_space() > 0) {
303 GhostQueue->push(&Command.hStream);
304 }
305 else dmsg(1,("DiskThread: GhostQueue full!\n"));
306 }
307 }
308
309 void DiskThread::RefillStreams() {
310 // sort the streams by most empty stream
311 qsort(pStreams, Streams, sizeof(Stream*), CompareStreamWriteSpace);
312
313 // refill the most empty streams
314 for (uint i = 0; i < RefillStreamsPerRun; i++) {
315 if (pStreams[i]->GetState() == Stream::state_active) {
316
317 //float filledpercentage = (float) pStreams[i]->GetReadSpace() / 131072.0 * 100.0;
318 //dmsg(("\nbuffer fill: %.1f%\n", filledpercentage));
319
320 int writespace = pStreams[i]->GetWriteSpaceToEnd();
321 if (writespace == 0) break;
322
323 int capped_writespace = writespace;
324 // if there is too much buffer space available then cut the read/write
325 // size to CONFIG_STREAM_MAX_REFILL_SIZE which is by default 65536 samples = 256KBytes
326 if (writespace > CONFIG_STREAM_MAX_REFILL_SIZE) capped_writespace = CONFIG_STREAM_MAX_REFILL_SIZE;
327
328 // adjust the amount to read in order to ensure that the buffer wraps correctly
329 int read_amount = pStreams[i]->AdjustWriteSpaceToAvoidBoundary(writespace, capped_writespace);
330 // if we wasn't able to refill one of the stream buffers by more than
331 // CONFIG_STREAM_MIN_REFILL_SIZE we'll send the disk thread to sleep later
332 if (pStreams[i]->ReadAhead(read_amount) > CONFIG_STREAM_MIN_REFILL_SIZE) this->IsIdle = false;
333 }
334 }
335 }
336
337 /// Handle Generator
338 Stream::Handle DiskThread::CreateHandle() {
339 static uint32_t counter = 0;
340 if (counter == 0xffffffff) counter = 1; // we use '0' as 'invalid handle' only, so we skip 0
341 else counter++;
342 return counter;
343 }
344
345 /// order ID Generator
346 Stream::OrderID_t DiskThread::CreateOrderID() {
347 static Stream::OrderID_t counter(0);
348 for (int i = 0; i < CONFIG_MAX_STREAMS; i++) {
349 if (counter == CONFIG_MAX_STREAMS) counter = 1; // we use '0' as 'invalid order' only, so we skip 0
350 else counter++;
351 if (!pCreatedStreams[counter]) {
352 pCreatedStreams[counter] = SLOT_RESERVED; // mark this slot as reserved
353 return counter; // found empty slot
354 }
355 }
356 return 0; // no free slot
357 }
358
359
360
361 // *********** C functions **************
362 // *
363
364 /**
365 * This is the comparison function the qsort algo uses to determine if a value is
366 * bigger than another one or special in our case; if the writespace of a stream
367 * is bigger than another one.
368 */
369 int CompareStreamWriteSpace(const void* A, const void* B) {
370 Stream* a = *(Stream**) A;
371 Stream* b = *(Stream**) B;
372 return b->GetWriteSpace() - a->GetWriteSpace();
373 }
374
375 }} // namespace LinuxSampler::gig

  ViewVC Help
Powered by ViewVC