Commit 004016f2 authored by sletz's avatar sletz
Browse files

Rework the mutex/signal classes. Use them in JackMessageBuffer.

git-svn-id: http://subversion.jackaudio.org/jack/jack2/trunk/jackmp@3319 0c269be4-1314-0410-8aa9-9f06e86f4224
parent e583db25
......@@ -23,10 +23,14 @@ Michael Voigt
Jackdmp changes log
---------------------------
2009-02-16 Stephane Letz <letz@grame.fr>
* Rework the mutex/signal classes. Use them in JackMessageBuffer.
2009-02-11 Stephane Letz <letz@grame.fr>
* Merge Solaris branch back on trunk.
* Equality of input and output buffer size removed (for now) in JackOSSDriver.
* Equality of input and output buffer size removed (for now) in JackOSSDriver.
2009-02-10 Stephane Letz <letz@grame.fr>
......
......@@ -24,7 +24,6 @@ Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#include "jack.h"
#include "JackError.h"
#include "JackResampler.h"
#include "JackConstants.h"
#include "JackFilters.h"
#include <samplerate.h>
......
......@@ -86,7 +86,6 @@ int JackEngine::Close()
}
}
fSignal.Destroy();
return 0;
}
......@@ -135,7 +134,7 @@ void JackEngine::ProcessNext(jack_time_t cur_cycle_begin)
fLastSwitchUsecs = cur_cycle_begin;
if (fGraphManager->RunNextGraph()) // True if the graph actually switched to a new state
fChannel.Notify(ALL_CLIENTS, kGraphOrderCallback, 0);
fSignal.SignalAll(); // Signal for threads waiting for next cycle
fSignal.Signal(); // Signal for threads waiting for next cycle
}
void JackEngine::ProcessCurrent(jack_time_t cur_cycle_begin)
......@@ -148,7 +147,7 @@ void JackEngine::ProcessCurrent(jack_time_t cur_cycle_begin)
bool JackEngine::Process(jack_time_t cur_cycle_begin, jack_time_t prev_cycle_end)
{
bool res = true;
// Cycle begin
fEngineControl->CycleBegin(fClientTable, fGraphManager, cur_cycle_begin, prev_cycle_end);
......@@ -174,7 +173,6 @@ bool JackEngine::Process(jack_time_t cur_cycle_begin, jack_time_t prev_cycle_end
return res;
}
/*
Client that finish *after* the callback date are considered late even if their output buffers may have been
correctly mixed in the time window: callbackUsecs <==> Read <==> Write.
......@@ -501,7 +499,7 @@ int JackEngine::ClientExternalOpen(const char* name, int pid, int* ref, int* sha
goto error;
}
if (!fSignal.TimedWait(DRIVER_OPEN_TIMEOUT * 1000000)) {
if (!fSignal.LockedTimedWait(DRIVER_OPEN_TIMEOUT * 1000000)) {
// Failure if RT thread is not running (problem with the driver...)
jack_error("Driver is not running");
goto error;
......@@ -546,7 +544,7 @@ int JackEngine::ClientInternalOpen(const char* name, int* ref, JackEngineControl
goto error;
}
if (wait && !fSignal.TimedWait(DRIVER_OPEN_TIMEOUT * 1000000)) {
if (wait && !fSignal.LockedTimedWait(DRIVER_OPEN_TIMEOUT * 1000000)) {
// Failure if RT thread is not running (problem with the driver...)
jack_error("Driver is not running");
goto error;
......@@ -624,7 +622,7 @@ int JackEngine::ClientCloseAux(int refnum, JackClientInterface* client, bool wai
// Wait until next cycle to be sure client is not used anymore
if (wait) {
if (!fSignal.TimedWait(fEngineControl->fTimeOutUsecs * 2)) { // Must wait at least until a switch occurs in Process, even in case of graph end failure
if (!fSignal.LockedTimedWait(fEngineControl->fTimeOutUsecs * 2)) { // Must wait at least until a switch occurs in Process, even in case of graph end failure
jack_error("JackEngine::ClientCloseAux wait error ref = %ld", refnum);
}
}
......@@ -649,7 +647,7 @@ int JackEngine::ClientActivate(int refnum, bool state)
fGraphManager->Activate(refnum);
// Wait for graph state change to be effective
if (!fSignal.TimedWait(fEngineControl->fTimeOutUsecs * 10)) {
if (!fSignal.LockedTimedWait(fEngineControl->fTimeOutUsecs * 10)) {
jack_error("JackEngine::ClientActivate wait error ref = %ld name = %s", refnum, client->GetClientControl()->fName);
return -1;
} else {
......@@ -686,7 +684,7 @@ int JackEngine::ClientDeactivate(int refnum)
fLastSwitchUsecs = 0; // Force switch to occur next cycle, even when called with "dead" clients
// Wait for graph state change to be effective
if (!fSignal.TimedWait(fEngineControl->fTimeOutUsecs * 10)) {
if (!fSignal.LockedTimedWait(fEngineControl->fTimeOutUsecs * 10)) {
jack_error("JackEngine::ClientDeactivate wait error ref = %ld name = %s", refnum, client->GetClientControl()->fName);
return -1;
} else {
......
......@@ -23,7 +23,6 @@ Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#include "jack.h"
#include "JackAtomicState.h"
#include <math.h>
#include <stdio.h>
namespace Jack
{
......
......@@ -31,6 +31,7 @@ JackMessageBuffer* JackMessageBuffer::fInstance = NULL;
JackMessageBuffer::JackMessageBuffer()
:fThread(this),fInBuffer(0),fOutBuffer(0),fOverruns(0)
{
fRunning = true;
fThread.StartSync();
}
......@@ -41,7 +42,11 @@ JackMessageBuffer::~JackMessageBuffer()
} else {
jack_info("no message buffer overruns");
}
fThread.Kill();
fGuard.Lock();
fRunning = false;
fGuard.Signal();
fGuard.Unlock();
fThread.Stop();
Flush();
}
......@@ -55,12 +60,12 @@ void JackMessageBuffer::Flush()
void JackMessageBuffer::AddMessage(int level, const char *message)
{
if (fMutex.Trylock()) {
if (fGuard.Trylock()) {
fBuffers[fInBuffer].level = level;
strncpy(fBuffers[fInBuffer].message, message, MB_BUFFERSIZE);
fInBuffer = MB_NEXT(fInBuffer);
fSignal.SignalAll();
fMutex.Unlock();
fGuard.Signal();
fGuard.Unlock();
} else { /* lock collision */
INC_ATOMIC(&fOverruns);
}
......@@ -68,11 +73,13 @@ void JackMessageBuffer::AddMessage(int level, const char *message)
bool JackMessageBuffer::Execute()
{
fSignal.Wait();
fMutex.Lock();
Flush();
fMutex.Unlock();
return true;
while (fRunning) {
fGuard.Lock();
fGuard.Wait();
Flush();
fGuard.Unlock();
}
return false;
}
void JackMessageBuffer::Create()
......
......@@ -58,12 +58,12 @@ class JackMessageBuffer : public JackRunnableInterface
private:
JackMessage fBuffers[MB_BUFFERS];
JackMutex fMutex;
JackThread fThread;
JackProcessSync fSignal;
JackProcessSync fGuard;
volatile unsigned int fInBuffer;
volatile unsigned int fOutBuffer;
SInt32 fOverruns;
bool fRunning;
void Flush();
......
......@@ -22,7 +22,6 @@ Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#include <assert.h>
#include "JackAudioAdapterInterface.h"
#include "JackPlatformPlug.h"
#include "JackNetInterface.h"
namespace Jack
......
......@@ -17,14 +17,13 @@ Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
*/
#include "JackConstants.h"
#include "JackMidiPort.h"
#include "JackTools.h"
#include "JackPlatformPlug.h"
#include "types.h"
#include "transport.h"
#include "transport.h"
#ifndef WIN32
#include <netinet/in.h>
#include <netinet/in.h>
#endif
#include <cmath>
......
......@@ -34,8 +34,6 @@ void JackMachSemaphore::BuildName(const char* name, const char* server_name, cha
bool JackMachSemaphore::Signal()
{
kern_return_t res;
if (!fSemaphore) {
jack_error("JackMachSemaphore::Signal name = %s already desallocated!!", fName);
return false;
......@@ -44,6 +42,7 @@ bool JackMachSemaphore::Signal()
if (fFlush)
return true;
kern_return_t res;
if ((res = semaphore_signal(fSemaphore)) != KERN_SUCCESS) {
jack_error("JackMachSemaphore::Signal name = %s err = %s", fName, mach_error_string(res));
}
......@@ -52,8 +51,6 @@ bool JackMachSemaphore::Signal()
bool JackMachSemaphore::SignalAll()
{
kern_return_t res;
if (!fSemaphore) {
jack_error("JackMachSemaphore::SignalAll name = %s already desallocated!!", fName);
return false;
......@@ -61,6 +58,8 @@ bool JackMachSemaphore::SignalAll()
if (fFlush)
return true;
kern_return_t res;
// When signaled several times, do not accumulate signals...
if ((res = semaphore_signal_all(fSemaphore)) != KERN_SUCCESS) {
jack_error("JackMachSemaphore::SignalAll name = %s err = %s", fName, mach_error_string(res));
......@@ -70,13 +69,12 @@ bool JackMachSemaphore::SignalAll()
bool JackMachSemaphore::Wait()
{
kern_return_t res;
if (!fSemaphore) {
jack_error("JackMachSemaphore::Wait name = %s already desallocated!!", fName);
return false;
}
kern_return_t res;
if ((res = semaphore_wait(fSemaphore)) != KERN_SUCCESS) {
jack_error("JackMachSemaphore::Wait name = %s err = %s", fName, mach_error_string(res));
}
......@@ -85,15 +83,15 @@ bool JackMachSemaphore::Wait()
bool JackMachSemaphore::TimedWait(long usec)
{
kern_return_t res;
mach_timespec time;
time.tv_sec = usec / 1000000;
time.tv_nsec = (usec % 1000000) * 1000;
if (!fSemaphore) {
jack_error("JackMachSemaphore::TimedWait name = %s already desallocated!!", fName);
return false;
}
kern_return_t res;
mach_timespec time;
time.tv_sec = usec / 1000000;
time.tv_nsec = (usec % 1000000) * 1000;
if ((res = semaphore_timedwait(fSemaphore, time)) != KERN_SUCCESS) {
jack_error("JackMachSemaphore::TimedWait name = %s usec = %ld err = %s", fName, usec, mach_error_string(res));
......
......@@ -23,9 +23,6 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#include "JackCompilerDeps.h"
#include "JackSynchro.h"
#include <mach/mach.h>
#include <mach/mach_types.h>
#include <mach/message.h>
#include <mach/mach_error.h>
#include <servers/bootstrap.h>
#include <mach/semaphore.h>
......
......@@ -47,6 +47,8 @@ namespace Jack { typedef JackMachSemaphore JackSynchro; }
/* __JackPlatformProcessSync__ */
#include "JackProcessSync.h"
//#include "JackMachProcessSync.h"
//namespace Jack { typedef JackMachProcessSync JackProcessSync; }
/* Only on windows a special JackProcessSync is used. It is directly defined by including JackProcessSync.h here */
/* __JackPlatformServerChannel__ */
......
This diff is collapsed.
......@@ -23,7 +23,7 @@
#define __JackPosixMutex__
#include <pthread.h>
#include <stdio.h>
#include <assert.h>
#include "JackError.h"
......@@ -36,7 +36,7 @@ namespace Jack
class JackPosixMutex
{
private:
protected:
pthread_mutex_t fMutex;
......@@ -56,7 +56,8 @@ class JackPosixMutex
res = pthread_mutexattr_destroy(&mutex_attr);
assert(res == 0);
}
~JackPosixMutex()
virtual ~JackPosixMutex()
{
pthread_mutex_destroy(&fMutex);
}
......
......@@ -144,6 +144,7 @@ int JackPosixThread::StartImp(pthread_t* thread, int priority, int realtime, voi
return -1;
}
pthread_attr_destroy(&attributes);
return 0;
}
......
......@@ -23,6 +23,68 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
namespace Jack
{
void JackProcessSync::Signal()
{
int res = pthread_cond_signal(&fCond);
if (res != 0)
jack_error("pthread_cond_signal error err = %s", strerror(res));
}
void JackProcessSync::LockedSignal()
{
int res;
res = pthread_mutex_lock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_lock error err = %s", strerror(res));
res = pthread_cond_signal(&fCond);
if (res != 0)
jack_error("pthread_cond_signal error err = %s", strerror(res));
res = pthread_mutex_unlock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_unlock error err = %s", strerror(res));
}
void JackProcessSync::SignalAll()
{
int res = pthread_cond_broadcast(&fCond);
if (res != 0)
jack_error("pthread_cond_broadcast error err = %s", strerror(res));
}
void JackProcessSync::LockedSignalAll()
{
int res;
res = pthread_mutex_lock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_lock error err = %s", strerror(res));
res = pthread_cond_broadcast(&fCond);
if (res != 0)
jack_error("pthread_cond_broadcast error err = %s", strerror(res));
res = pthread_mutex_unlock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_unlock error err = %s", strerror(res));
}
void JackProcessSync::Wait()
{
int res;
if ((res = pthread_cond_wait(&fCond, &fMutex)) != 0)
jack_error("pthread_cond_wait error err = %s", strerror(errno));
}
void JackProcessSync::LockedWait()
{
int res;
res = pthread_mutex_lock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_lock error err = %s", strerror(res));
if ((res = pthread_cond_wait(&fCond, &fMutex)) != 0)
jack_error("pthread_cond_wait error err = %s", strerror(errno));
res = pthread_mutex_unlock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_unlock error err = %s", strerror(res));
}
bool JackProcessSync::TimedWait(long usec)
{
struct timeval T0, T1;
......@@ -30,7 +92,6 @@ bool JackProcessSync::TimedWait(long usec)
struct timeval now;
int res;
pthread_mutex_lock(&fLock);
jack_log("JackProcessSync::TimedWait time out = %ld", usec);
gettimeofday(&T0, 0);
......@@ -38,38 +99,49 @@ bool JackProcessSync::TimedWait(long usec)
unsigned int next_date_usec = now.tv_usec + usec;
time.tv_sec = now.tv_sec + (next_date_usec / 1000000);
time.tv_nsec = (next_date_usec % 1000000) * 1000;
res = pthread_cond_timedwait(&fCond, &fLock, &time);
res = pthread_cond_timedwait(&fCond, &fMutex, &time);
if (res != 0)
jack_error("pthread_cond_timedwait error usec = %ld err = %s", usec, strerror(res));
gettimeofday(&T1, 0);
pthread_mutex_unlock(&fLock);
jack_log("JackProcessSync::TimedWait finished delta = %5.1lf",
(1e6 * T1.tv_sec - 1e6 * T0.tv_sec + T1.tv_usec - T0.tv_usec));
return (res == 0);
}
void JackProcessSync::Wait()
bool JackProcessSync::LockedTimedWait(long usec)
{
struct timeval T0, T1;
timespec time;
struct timeval now;
int res;
pthread_mutex_lock(&fLock);
//jack_log("JackProcessSync::Wait...");
if ((res = pthread_cond_wait(&fCond, &fLock)) != 0)
jack_error("pthread_cond_wait error err = %s", strerror(errno));
pthread_mutex_unlock(&fLock);
//jack_log("JackProcessSync::Wait finished");
}
bool JackInterProcessSync::TimedWait(long usec)
{
struct timeval T0, T1;
//jack_log("JackInterProcessSync::TimedWait...");
res = pthread_mutex_lock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_lock error err = %s", usec, strerror(res));
jack_log("JackProcessSync::TimedWait time out = %ld", usec);
gettimeofday(&T0, 0);
bool res = fSynchro->TimedWait(usec);
gettimeofday(&now, 0);
unsigned int next_date_usec = now.tv_usec + usec;
time.tv_sec = now.tv_sec + (next_date_usec / 1000000);
time.tv_nsec = (next_date_usec % 1000000) * 1000;
res = pthread_cond_timedwait(&fCond, &fMutex, &time);
if (res != 0)
jack_error("pthread_cond_timedwait error usec = %ld err = %s", usec, strerror(res));
gettimeofday(&T1, 0);
//jack_log("JackInterProcessSync::TimedWait finished delta = %5.1lf", (1e6 * T1.tv_sec - 1e6 * T0.tv_sec + T1.tv_usec - T0.tv_usec));
return res;
pthread_mutex_unlock(&fMutex);
if (res != 0)
jack_error("pthread_mutex_unlock error err = %s", usec, strerror(res));
jack_log("JackProcessSync::TimedWait finished delta = %5.1lf",
(1e6 * T1.tv_sec - 1e6 * T0.tv_sec + T1.tv_usec - T0.tv_usec));
return (res == 0);
}
} // end of namespace
......@@ -21,7 +21,7 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#define __JackProcessSync__
#include "JackPlatformPlug.h"
#include <pthread.h>
#include "JackPosixMutex.h"
#include <sys/time.h>
#include <unistd.h>
......@@ -32,110 +32,39 @@ namespace Jack
\brief A synchronization primitive built using a condition variable.
*/
class JackProcessSync
class JackProcessSync : public JackPosixMutex
{
private:
pthread_mutex_t fLock; // Mutex
pthread_cond_t fCond; // Condition variable
public:
JackProcessSync()
JackProcessSync():JackPosixMutex()
{
pthread_mutex_init(&fLock, NULL);
pthread_cond_init(&fCond, NULL);
}
~JackProcessSync()
{
pthread_mutex_destroy(&fLock);
pthread_cond_destroy(&fCond);
}
bool Allocate(const char* name)
{
return true;
}
bool Connect(const char* name)
{
return true;
}
void Destroy()
{}
bool TimedWait(long usec);
void Wait();
bool LockedTimedWait(long usec);
void Signal()
{
pthread_mutex_lock(&fLock);
pthread_cond_signal(&fCond);
pthread_mutex_unlock(&fLock);
}
void SignalAll()
{
//pthread_mutex_lock(&fLock);
pthread_cond_broadcast(&fCond);
//pthread_mutex_unlock(&fLock);
}
};
/*!
\brief A synchronization primitive built using an inter-process synchronization object.
*/
class JackInterProcessSync
{
private:
JackSynchro* fSynchro;
public:
JackInterProcessSync(JackSynchro* synchro): fSynchro(synchro)
{}
~JackInterProcessSync()
{
delete fSynchro;
}
bool Allocate(const char* name)
{
return fSynchro->Allocate(name, "", 0);
}
void Destroy()
{
fSynchro->Destroy();
}
bool Connect(const char* name)
{
return fSynchro->Connect(name, "");
}
bool TimedWait(long usec);
void Wait()
{
fSynchro->Wait();
}
void SignalAll()
{
fSynchro->SignalAll();
}
void Wait();
void LockedWait();
void Signal();
void LockedSignal();
void SignalAll();
void LockedSignalAll();
};
} // end of namespace
#endif
......
......@@ -32,7 +32,7 @@ namespace Jack
class JackWinMutex
{
private:
protected:
HANDLE fMutex;
......@@ -43,7 +43,8 @@ class JackWinMutex
// In recursive mode by default
fMutex = (HANDLE)CreateMutex(0, FALSE, 0);
}
~JackWinMutex()
virtual ~JackWinMutex()
{
CloseHandle(fMutex);