a7b6545028
I fail to build TIFF code: it requires access to libtiff internals but it unavailable with libtiff prom ports. Probably pre build form thirdparty/tiff-4.0.3 should fix this, but I don't check this. Tiff code disabled for work but available in options. I fail with SSE2 too, there is some mess with class initialization: TRaster32P rout32 = rout; have no idea how to fix it. If some one want to play with it, then try to build with CFLAGS+= -DUSE_SSE2 -DDWORD=uint32_t -D_aligned_malloc=aligned_alloc -D_aligned_free=free on non windows system. FarmServer::queryHwInfo() - now uses TSystem::***() to collect system info, to avoid code duplication. Add pthread as required lib.
736 lines
22 KiB
C++
736 lines
22 KiB
C++
|
|
|
|
// Qt includes
|
|
#include <QCoreApplication>
|
|
#include <QThread>
|
|
#include <QTime>
|
|
#include <QSharedMemory>
|
|
#include <QProcess>
|
|
#include <QMutex>
|
|
#include <QSemaphore>
|
|
#include <QAtomicInt>
|
|
#include <QEventLoop>
|
|
#include <QTimer>
|
|
|
|
// System-specific includes
|
|
#if defined(_WIN32)
|
|
#include <windows.h>
|
|
#elif defined(MACOSX)
|
|
#include <sys/sysctl.h>
|
|
#include <unistd.h>
|
|
#elif defined(LINUX) || defined(FREEBSD)
|
|
#include <sys/time.h>
|
|
#include <unistd.h>
|
|
#endif
|
|
|
|
#include "tipc.h"
|
|
|
|
/*
|
|
PLATFORM-SPECIFIC REMINDERS:
|
|
|
|
There are few remarks to be aware when maintaining this code.
|
|
Please, be careful that:
|
|
|
|
- It seems that, on Windows, QLocalSocket::waitForBytesWritten does not return
|
|
success unless the data is actually read on the other end. On Unix this is not
|
|
the case, presumably because the data is written to a buffer which can be read
|
|
by
|
|
the other process at some later point.
|
|
|
|
Thus, *BE SURE* that every data written is received on the other side.
|
|
|
|
- On MACOSX, the default shared memory settings can be quite restrictive.
|
|
On a standard machine, the maximum size of a shared segment is 4 MB, exactly
|
|
the same as the TOTAL size of shared memory available.
|
|
|
|
Whereas tipc respects the former parameter, there must be a way to circumvent
|
|
the
|
|
latter in order to make the allocation of multiple shared segments of the
|
|
maximum
|
|
size.
|
|
*/
|
|
|
|
//********************************************************
|
|
// Diagnostics Stuff
|
|
//********************************************************
|
|
|
|
//#define TIPC_DEBUG
|
|
|
|
#ifdef TIPC_DEBUG
|
|
#define tipc_debug(expr) expr
|
|
#else
|
|
#define tipc_debug(expr)
|
|
#endif
|
|
|
|
#ifdef TIPC_DEBUG
|
|
#include <QTime>
|
|
#endif
|
|
|
|
//********************************************************
|
|
// Local namespace Stuff
|
|
//********************************************************
|
|
|
|
namespace {
|
|
int shm_max = -1;
|
|
int shm_all = -1;
|
|
int shm_seg = -1;
|
|
int shm_mni = -1;
|
|
}
|
|
|
|
//********************************************************
|
|
// tipc Stream Implementation
|
|
//********************************************************
|
|
|
|
int tipc::Stream::readSize() {
|
|
if (m_socket->bytesAvailable() < sizeof(TINT32)) return -1;
|
|
|
|
TINT32 msgSize = -1;
|
|
m_socket->peek((char *)&msgSize, sizeof(TINT32));
|
|
|
|
return msgSize;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
bool tipc::Stream::messageReady() {
|
|
TINT32 msgSize;
|
|
return (msgSize = readSize()) >= 0 && m_socket->bytesAvailable() >= msgSize;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
bool tipc::Stream::readData(char *data, qint64 dataSize, int msecs) {
|
|
tipc_debug(qDebug("tipc::Stream::readData entry"));
|
|
qint64 r, dataRead = 0;
|
|
char *currData = data;
|
|
|
|
while (dataRead < dataSize) {
|
|
if ((m_socket->bytesAvailable() == 0) &&
|
|
!m_socket->waitForReadyRead(msecs)) {
|
|
tipc_debug(
|
|
qDebug("tipc::Stream::readData exit (unexpected loss of data)"));
|
|
return false;
|
|
}
|
|
|
|
// Read the supplied data
|
|
currData += r = m_socket->read(currData, dataSize - dataRead);
|
|
dataRead += r;
|
|
}
|
|
|
|
tipc_debug(qDebug("tipc::Stream::readData exit"));
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
bool tipc::Stream::readDataNB(char *data, qint64 dataSize, int msecs,
|
|
QEventLoop::ProcessEventsFlag flag) {
|
|
tipc_debug(qDebug("tipc::Stream::readDataNB entry"));
|
|
qint64 r, dataRead = 0;
|
|
char *currData = data;
|
|
|
|
QEventLoop loop;
|
|
QObject::connect(m_socket, SIGNAL(readyRead()), &loop, SLOT(quit()));
|
|
QObject::connect(m_socket, SIGNAL(error(QLocalSocket::LocalSocketError)),
|
|
&loop, SLOT(quit()));
|
|
|
|
if (msecs >= 0) QTimer::singleShot(msecs, &loop, SLOT(quit()));
|
|
|
|
while (dataRead < dataSize) {
|
|
if (m_socket->bytesAvailable() == 0) {
|
|
loop.exec(flag);
|
|
if (m_socket->bytesAvailable() == 0) {
|
|
tipc_debug(
|
|
qDebug("tipc::Stream::readDataNB exit (unexpected loss of data)"));
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// Read the supplied data
|
|
currData += r = m_socket->read(currData, dataSize - dataRead);
|
|
dataRead += r;
|
|
}
|
|
|
|
tipc_debug(qDebug("tipc::Stream::readDataNB exit"));
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Reads the message and returns its header.
|
|
This function reads a complete message from the socket, waiting
|
|
until it is completely available. The function accepts
|
|
an inactivity timeout which can be supplied to drop the operation
|
|
after msecs milliseconds no data has been received.
|
|
*/
|
|
bool tipc::Stream::readMessage(Message &msg, int msecs) {
|
|
TINT32 msgSize = 0;
|
|
if (!readData((char *)&msgSize, sizeof(TINT32), msecs)) return false;
|
|
|
|
msg.ba().resize(msgSize);
|
|
if (!readData(msg.ba().data(), msgSize, msecs)) return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
The non-blocking equivalent to readMessage(), this function
|
|
performs event processing in a local event loop until all
|
|
message data has been received.
|
|
*/
|
|
bool tipc::Stream::readMessageNB(Message &msg, int msecs,
|
|
QEventLoop::ProcessEventsFlag flag) {
|
|
TINT32 msgSize = 0;
|
|
if (!readDataNB((char *)&msgSize, sizeof(TINT32), msecs, flag)) return false;
|
|
|
|
msg.ba().resize(msgSize);
|
|
if (!readDataNB(msg.ba().data(), msgSize, msecs, flag)) return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Flushes all data written to the stream.
|
|
This function waits until all data written on the stream
|
|
has been successfully delivered in output.
|
|
Returns true if the operation was successful, false if
|
|
it timed out or an error occurred.
|
|
*/
|
|
bool tipc::Stream::flush(int msecs) {
|
|
tipc_debug(qDebug("tipc:flush entry"));
|
|
|
|
while (m_socket->bytesToWrite() > 0) {
|
|
tipc_debug(qDebug() << "bytes to write:" << m_socket->bytesToWrite());
|
|
bool ok = m_socket->flush();
|
|
tipc_debug(qDebug() << "flush success:" << ok
|
|
<< "bytes to write:" << m_socket->bytesToWrite());
|
|
if (m_socket->bytesToWrite() > 0 && !m_socket->waitForBytesWritten(msecs))
|
|
return false;
|
|
}
|
|
|
|
tipc_debug(qDebug() << "tipc:flush exit - bytes to write:"
|
|
<< m_socket->bytesToWrite());
|
|
return (m_socket->bytesToWrite() == 0);
|
|
}
|
|
|
|
//********************************************************
|
|
// tipc Stream Operators
|
|
//********************************************************
|
|
|
|
//! \warning This operation assumes that all the message is available for read.
|
|
//! Use tipc::stream::readMessage if this cannot be ensured.
|
|
tipc::Stream &operator>>(tipc::Stream &stream, tipc::Message &msg) {
|
|
QLocalSocket *socket = stream.socket();
|
|
msg.clear();
|
|
|
|
TINT32 msgSize;
|
|
socket->read((char *)&msgSize, sizeof(TINT32));
|
|
msg.ba().resize(msgSize);
|
|
socket->read(msg.ba().data(), msgSize);
|
|
return stream;
|
|
}
|
|
|
|
tipc::Stream &operator<<(tipc::Stream &stream, tipc::Message &msg) {
|
|
QLocalSocket *socket = stream.socket();
|
|
|
|
TINT32 size = msg.ba().size();
|
|
socket->write((char *)&size, sizeof(TINT32));
|
|
socket->write(msg.ba().data(), size);
|
|
|
|
return stream;
|
|
}
|
|
|
|
//********************************************************
|
|
// tipc Utilities
|
|
//********************************************************
|
|
|
|
/*!
|
|
Appends the invoking process' pid to the passed srvName.
|
|
*/
|
|
QString tipc::applicationSpecificServerName(QString srvName) {
|
|
return srvName + QString::number(QCoreApplication::applicationPid());
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
bool tipc::startBackgroundProcess(QString cmdline) {
|
|
#ifdef _WIN32
|
|
QProcess *proc = new QProcess;
|
|
proc->start(cmdline);
|
|
if (proc->state() == QProcess::NotRunning) {
|
|
delete proc;
|
|
return false;
|
|
}
|
|
|
|
QObject::connect(proc, SIGNAL(finished(int, QProcess::ExitStatus)), proc,
|
|
SLOT(deleteLater()));
|
|
QObject::connect(proc, SIGNAL(error(QProcess::ProcessError)), proc,
|
|
SLOT(deleteLater()));
|
|
return true;
|
|
#else
|
|
return QProcess::startDetached(cmdline);
|
|
;
|
|
#endif
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Invokes the passed command line to run a slave server.
|
|
A slave server is hereby intended as a 'child' server process which
|
|
automatically destroys itself in case the calling application
|
|
crashes.
|
|
This process \b MUST support one server, running in the \b MAIN \b THREAD,
|
|
whose name is <srvName>_main.
|
|
This function waits until the main server is up and ready to
|
|
listen for incoming connections - no timeout accepted.
|
|
|
|
\warning Please, observe that a correct slave server name should be
|
|
ensured to be unique to the system.
|
|
*/
|
|
bool tipc::startSlaveServer(QString srvName, QString cmdline) {
|
|
if (!tipc::startBackgroundProcess(cmdline)) return false;
|
|
|
|
QString mainSrvName(srvName + "_main");
|
|
|
|
// Establish a dummy socket connection to provide a mean for the process
|
|
// to tell whether the calling process exited unexpectedly.
|
|
QLocalSocket *dummySock = new QLocalSocket;
|
|
dummySock->connectToServer(mainSrvName);
|
|
|
|
// Wait up to msecs until the socket is connecting. Wait a small amount of
|
|
// time
|
|
// until the server is up and listening to connection (there is no other way
|
|
// to tell).
|
|
while (dummySock->state() == QLocalSocket::UnconnectedState) {
|
|
#ifdef _WIN32
|
|
Sleep(10);
|
|
#else
|
|
usleep(10 << 10); // 10.24 msecs
|
|
#endif
|
|
|
|
dummySock->connectToServer(mainSrvName);
|
|
}
|
|
|
|
dummySock->waitForConnected(-1);
|
|
|
|
tipc::Stream stream(dummySock);
|
|
tipc::Message msg;
|
|
|
|
// Supply the 'quit if this socket connection fails' command
|
|
// This command ensure termination of the child process in case of some errors
|
|
// or ending of the program
|
|
stream << (msg << QString("$quit_on_error"));
|
|
if (tipc::readMessage(stream, msg, 3000) == QString()) {
|
|
std::cout << "tipc::startSlaveServer - tipc::readMessage TIMEOUT"
|
|
<< std::endl;
|
|
return false;
|
|
}
|
|
|
|
// The server should die if dummyDock is destroyed. This should happen when
|
|
// the *MAIN* thread
|
|
// in *this process* exits. So, if this is not the main thread, we must move
|
|
// the socket there.
|
|
if (QCoreApplication::instance() &&
|
|
QThread::currentThread() != QCoreApplication::instance()->thread())
|
|
dummySock->moveToThread(QCoreApplication::instance()->thread());
|
|
|
|
// If a connection error takes place, release the dummy socket.
|
|
// Please, observe that this QObject::connect is invoked *AFTER* the
|
|
// connection trials above...
|
|
QObject::connect(dummySock, SIGNAL(error(QLocalSocket::LocalSocketError)),
|
|
dummySock, SLOT(deleteLater()));
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Connects the passed socket to the server with name <srvName> + <threadName>.
|
|
Awaits for the connection up to msecs milliseconds before returning false.
|
|
If no server was found, a new slave server is started by invoking
|
|
the supplied command line and connection is re-attempted.
|
|
Returns true on success, false otherwise.
|
|
|
|
\warning Please, observe that a correct slave server name should be
|
|
ensured to be unique to the parent process.
|
|
*/
|
|
bool tipc::startSlaveConnection(QLocalSocket *socket, QString srvName,
|
|
int msecs, QString cmdline,
|
|
QString threadName) {
|
|
QTime time;
|
|
time.start();
|
|
|
|
if (msecs == -1) msecs = (std::numeric_limits<int>::max)();
|
|
|
|
QString fullSrvName(srvName + threadName);
|
|
socket->connectToServer(fullSrvName);
|
|
|
|
// If the socket is not connecting, the server lookup table returned that the
|
|
// no server with
|
|
// the passed name exists. This means that a server must be created.
|
|
if (socket->state() == QLocalSocket::UnconnectedState && !cmdline.isEmpty()) {
|
|
// Completely serialize the server start
|
|
static QMutex mutex;
|
|
QMutexLocker locker(&mutex);
|
|
|
|
// Retry connection - this is required due to the mutex
|
|
socket->connectToServer(fullSrvName);
|
|
if (socket->state() != QLocalSocket::UnconnectedState) goto connecting;
|
|
|
|
// Invoke the supplied command line to start the server
|
|
if (!tipc::startSlaveServer(srvName, cmdline)) return false;
|
|
|
|
// Reconnect to the server
|
|
socket->connectToServer(fullSrvName);
|
|
if (socket->state() == QLocalSocket::UnconnectedState) return false;
|
|
}
|
|
|
|
connecting:
|
|
|
|
// Now, the server is connecting or already connected. Wait until the socket
|
|
// is connected.
|
|
socket->waitForConnected(msecs - time.elapsed());
|
|
if (socket->state() != QLocalSocket::ConnectedState) return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Waits and reads the next message from stream.
|
|
This function is mainly a useful macro that encapsulates
|
|
the following steps in one call:
|
|
|
|
\li Flush the write buffer (output messages)
|
|
\li Wait until an input message is completely readable
|
|
\li Read the message from stream
|
|
\li Read the first string from the message and return it
|
|
|
|
This function returns an empty QString if the message could not be
|
|
entirely retrieved from the stream.
|
|
*/
|
|
QString tipc::readMessage(Stream &stream, Message &msg, int msecs) {
|
|
msg.clear();
|
|
stream.flush();
|
|
if (!stream.readMessage(msg, msecs)) return QString();
|
|
|
|
QString res;
|
|
msg >> res;
|
|
return res;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
The non-blocking equivalent to tipc::readMessage.
|
|
*/
|
|
QString tipc::readMessageNB(Stream &stream, Message &msg, int msecs,
|
|
QEventLoop::ProcessEventsFlag flag) {
|
|
msg.clear();
|
|
if (!stream.readMessageNB(msg, msecs, flag)) return QString();
|
|
|
|
QString res;
|
|
msg >> res;
|
|
return res;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Returns an inter-process unique id string; the returned
|
|
id should be used to create QSharedMemory objects.
|
|
*/
|
|
QString tipc::uniqueId() {
|
|
static QAtomicInt count;
|
|
count.ref();
|
|
return QString::number(QCoreApplication::applicationPid()) + "_" +
|
|
QString::number((int)count);
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
//! Returns the maximum size of a shared memory segment allowed by the system
|
|
int tipc::shm_maxSegmentSize() {
|
|
if (shm_max < 0) {
|
|
#ifdef MACOSX
|
|
// Retrieve it by invoking sysctl
|
|
size_t valSize = sizeof(TINT64);
|
|
TINT64 val;
|
|
sysctlbyname("kern.sysv.shmmax", &val, &valSize, NULL, 0);
|
|
shm_max = std::min(val, (TINT64)(std::numeric_limits<int>::max)());
|
|
#else
|
|
// Windows case: no such limit
|
|
// Observe that QSharedMemory accepts only an int size - so the num_lim is
|
|
// against int.
|
|
shm_max = (std::numeric_limits<int>::max)();
|
|
#endif
|
|
}
|
|
|
|
return shm_max;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
//! Returns the maximum number of shared segments allowed by the system
|
|
int tipc::shm_maxSegmentCount() {
|
|
if (shm_seg < 0) {
|
|
#ifdef MACOSX
|
|
size_t valSize = sizeof(TINT64);
|
|
TINT64 val;
|
|
sysctlbyname("kern.sysv.shmseg", &val, &valSize, NULL, 0);
|
|
shm_seg = std::min(val, (TINT64)(std::numeric_limits<int>::max)());
|
|
#else
|
|
// Windows case: no such limit - again, using limit against max due to Qt
|
|
shm_seg = (std::numeric_limits<int>::max)();
|
|
#endif
|
|
}
|
|
|
|
return shm_seg;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
int tipc::shm_maxSharedPages() {
|
|
if (shm_all < 0) {
|
|
#ifdef MACOSX
|
|
size_t valSize = sizeof(TINT64);
|
|
TINT64 val;
|
|
sysctlbyname("kern.sysv.shmall", &val, &valSize, NULL, 0);
|
|
shm_all = std::min(val, (TINT64)(std::numeric_limits<int>::max)());
|
|
#else
|
|
shm_all = (std::numeric_limits<int>::max)();
|
|
#endif
|
|
}
|
|
|
|
return shm_all;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
int tipc::shm_maxSharedCount() {
|
|
if (shm_mni < 0) {
|
|
#ifdef MACOSX
|
|
size_t valSize = sizeof(TINT64);
|
|
TINT64 val;
|
|
sysctlbyname("kern.sysv.shmmni", &val, &valSize, NULL, 0);
|
|
shm_mni = std::min(val, (TINT64)(std::numeric_limits<int>::max)());
|
|
#else
|
|
shm_mni = (std::numeric_limits<int>::max)();
|
|
#endif
|
|
}
|
|
|
|
return shm_mni;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Attempts to set the shared memory parameters to the system.
|
|
This is only working on MAC's SystemV shm, it's a no-op on Win.
|
|
This function will fail anyway if the process is not owned by an
|
|
admin.
|
|
*/
|
|
void tipc::shm_set(int shmmax, int shmseg, int shmall, int shmmni) {
|
|
tipc_debug(qDebug("shmmax: %i, shmseg: %i, shmall: %i, shmmni: %i", shmmax,
|
|
shmseg, shmall, shmmni));
|
|
#ifdef MACOSX
|
|
TINT64 val;
|
|
int err;
|
|
if (shmmax > 0) {
|
|
val = shmmax;
|
|
err = sysctlbyname("kern.sysv.shmmax", NULL, NULL, &val, sizeof(TINT64));
|
|
if (!err) shm_max = shmmax;
|
|
}
|
|
if (shmseg > 0) {
|
|
val = shmseg;
|
|
err = sysctlbyname("kern.sysv.shmseg", NULL, NULL, &val, sizeof(TINT64));
|
|
if (!err) shm_seg = shmseg;
|
|
}
|
|
if (shmall > 0) {
|
|
val = shmall;
|
|
err = sysctlbyname("kern.sysv.shmall", NULL, NULL, &val, sizeof(TINT64));
|
|
if (!err) shm_all = shmall;
|
|
}
|
|
if (shmmni > 0) {
|
|
val = shmmni;
|
|
err = sysctlbyname("kern.sysv.shmmni", NULL, NULL, &val, sizeof(TINT64));
|
|
if (!err) shm_mni = shmmni;
|
|
}
|
|
#endif
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Creates a shared memory segment for passed QSharedMemory.
|
|
|
|
This function attempts creation of a shared memory segment
|
|
in the form of Qt's QSharedMemory, with the following \b UNIX-specific
|
|
distinctions:
|
|
|
|
<LI> If the segment size is beyond that supported by the system,
|
|
the function can be set to either fail or return a segment with
|
|
the maximum supported size. <\LI>
|
|
|
|
<LI> Unlike QSharedMemory::create, this function attempts to
|
|
reclaim an already existing memory id before creating a new one. <\LI>
|
|
*/
|
|
int tipc::create(QSharedMemory &shmem, int size, bool strictSize) {
|
|
bool ok, retried = false;
|
|
|
|
if (!strictSize) size = std::min(size, (int)shm_maxSegmentSize());
|
|
|
|
tipc_debug(qDebug() << "shMem create: size =" << size);
|
|
|
|
retry:
|
|
|
|
ok = shmem.create(size);
|
|
if (!ok) {
|
|
tipc_debug(qDebug() << "Error: Shared Segment could not be created: #"
|
|
<< shmem.errorString());
|
|
|
|
// Unix-specific error recovery follows. See Qt's docs about it.
|
|
|
|
// Try to recover error #AlreadyExists - supposedly, the server crashed in a
|
|
// previous instance.
|
|
// As shared memory segments that happen to go this way are owned by the
|
|
// server process with 1
|
|
// reference count, detaching it now may solve the issue.
|
|
if (shmem.error() == QSharedMemory::AlreadyExists && !retried) {
|
|
retried = true; // We're trying this only once... for now it works.
|
|
shmem.attach();
|
|
shmem.detach();
|
|
goto retry;
|
|
}
|
|
|
|
return -1;
|
|
}
|
|
|
|
return size;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Writes data through a shared memory segment medium.
|
|
*/
|
|
bool tipc::writeShMemBuffer(Stream &stream, Message &msg, int bufSize,
|
|
ShMemWriter *dataWriter) {
|
|
tipc_debug(QTime time; time.start());
|
|
tipc_debug(qDebug("tipc::writeShMemBuffer entry"));
|
|
|
|
static QSemaphore sem(tipc::shm_maxSegmentCount());
|
|
sem.acquire(1);
|
|
|
|
{
|
|
// Create a shared memory segment, possibly of passed size
|
|
QSharedMemory shmem(tipc::uniqueId());
|
|
bool ok = (tipc::create(shmem, bufSize) > 0);
|
|
if (!ok) goto err;
|
|
|
|
// Communicate the shared memory id and bufSize to the reader
|
|
msg << QString("shm") << shmem.key() << bufSize;
|
|
|
|
// Fill in data until all the buffer has been sent
|
|
int chunkData, remainingData = bufSize;
|
|
while (remainingData > 0) {
|
|
// Write to the shared memory segment
|
|
tipc_debug(QTime xchTime; xchTime.start());
|
|
shmem.lock();
|
|
remainingData -= chunkData = dataWriter->write(
|
|
(char *)shmem.data(), std::min(shmem.size(), remainingData));
|
|
shmem.unlock();
|
|
tipc_debug(qDebug() << "exchange time:" << xchTime.elapsed());
|
|
|
|
stream << (msg << QString("chk") << chunkData);
|
|
|
|
if (tipc::readMessage(stream, msg) != "ok") goto err;
|
|
|
|
msg.clear();
|
|
}
|
|
}
|
|
|
|
sem.release(1);
|
|
tipc_debug(qDebug("tipc::writeShMemBuffer exit"));
|
|
tipc_debug(qDebug() << "tipc::writeShMemBuffer time:" << time.elapsed());
|
|
return true;
|
|
|
|
err:
|
|
|
|
tipc_debug(qDebug("tipc::writeShMemBuffer exit (error)"));
|
|
|
|
msg.clear();
|
|
sem.release(1);
|
|
return false;
|
|
}
|
|
|
|
//-------------------------------------------------------------
|
|
|
|
/*!
|
|
Reads data through a shared memory segment medium.
|
|
*/
|
|
bool tipc::readShMemBuffer(Stream &stream, Message &msg,
|
|
ShMemReader *dataReader) {
|
|
tipc_debug(QTime time; time.start(););
|
|
tipc_debug(qDebug("tipc::readShMemBuffer entry"));
|
|
|
|
// Read the id from stream
|
|
QString res(tipc::readMessage(stream, msg));
|
|
if (res != "shm") {
|
|
tipc_debug(qDebug("tipc::readShMemBuffer exit (res != \"shm\")"));
|
|
return false;
|
|
}
|
|
|
|
// Read message and reply
|
|
QString id, chkStr;
|
|
int bufSize;
|
|
msg >> id >> bufSize >> chkStr;
|
|
|
|
// Data is ready to be read - attach to the shared memory segment.
|
|
QSharedMemory shmem(id);
|
|
shmem.attach();
|
|
if (!shmem.isAttached()) {
|
|
tipc_debug(qDebug("tipc::readShMemBuffer exit (shmem not attached)"));
|
|
return false;
|
|
}
|
|
|
|
// Start reading from it
|
|
int chunkData, remainingData = bufSize;
|
|
while (true) {
|
|
msg >> chunkData;
|
|
|
|
tipc_debug(QTime xchTime; xchTime.start());
|
|
shmem.lock();
|
|
remainingData -= dataReader->read((const char *)shmem.data(), chunkData);
|
|
shmem.unlock();
|
|
tipc_debug(qDebug() << "exchange time:" << xchTime.elapsed());
|
|
|
|
// Data was read. Inform the writer
|
|
stream << (msg << clr << QString("ok"));
|
|
stream.flush();
|
|
|
|
if (remainingData <= 0) break;
|
|
|
|
// Wait for more chunks
|
|
if (tipc::readMessage(stream, msg) != "chk") {
|
|
tipc_debug(
|
|
qDebug("tipc::readShMemBuffer exit (unexpected chunk absence)"));
|
|
return false;
|
|
}
|
|
}
|
|
|
|
shmem.detach();
|
|
tipc_debug(qDebug("tipc::readShMemBuffer exit"));
|
|
tipc_debug(qDebug() << "tipc::readShMemBuffer time:" << time.elapsed());
|
|
return true;
|
|
}
|