Mirrored_Repos/LookingGlass
1
0
Fork 0
mirror of https://github.com/gnif/LookingGlass.git synced 2026-08-22 15:11:31 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity
Files
4477ac0de16fd30ed1bc743dbf3d398b63f86cbd
LookingGlass/client/tests/lgmp_transport_test.c
Geoffrey McRae 4768396180 [client] transport: track video component availability
2026-08-13 04:28:09 +10:00

1038 lines
39 KiB
C
Raw Blame History

/**
* Looking Glass
* Copyright © 2017-2026 The Looking Glass Authors
* https://looking-glass.io
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the Free
* Software Foundation; either version 2 of the License, or (at your option)
* any later version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
* more details.
*
* You should have received a copy of the GNU General Public License along
* with this program; if not, write to the Free Software Foundation, Inc., 59
* Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
#include "interface/transport.h"
#include "common/KVMFR.h"
#include "common/KVMFRRecovery.h"
#include "common/LGMPConfig.h"
#include "common/debug.h"
#include "common/option.h"
#include <lgmp/host.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <unistd.h>
#define TEST_SHM_SIZE (2U * 1024U * 1024U)
#define TEST_TIMEOUT 20U
#define WAIT_TIMEOUT (TEST_TIMEOUT + 100U)
#define POLL_INTERVAL 50000U
extern const LG_TransportOps LGT_LGMP;
extern void lgmp_testSetVideoStatusDispatchGate(
LG_Transport * target, atomic_bool * waiting, atomic_bool * hold);
extern LG_RecoveryError lgmp_testRecoveryError(uint32_t error);
extern bool lgmp_testRecoveryProbeCancellation(
LG_TransportCancelledFn cancelled, void * opaque);
static bool cancelled(void * opaque)
{
return *(const bool *)opaque;
}
struct CancelAfter
{
unsigned int calls;
unsigned int limit;
};
static bool cancelAfter(void * opaque)
{
struct CancelAfter * state = opaque;
return ++state->calls >= state->limit;
}
#define CHECK(x) \
do \
{ \
if (!(x)) \
{ \
fprintf(stderr, "check failed at %s:%d: %s\n", __FILE__, __LINE__, #x); \
goto cleanup; \
} \
} \
while (0)
static bool waitForQueuesEmpty(PLGMPHost host, PLGMPHostQueue frameQueue,
PLGMPHostQueue pointerQueue)
{
for (unsigned i = 0; i < WAIT_TIMEOUT; ++i)
{
if (lgmpHostProcess(host) != LGMP_OK)
return false;
if (lgmpHostQueuePending(frameQueue) == 0 &&
lgmpHostQueuePending(pointerQueue) == 0)
return true;
usleep(1000);
}
return false;
}
struct VideoStatusTrace;
struct WaitState
{
LG_Transport * transport;
const LG_FrameOps * ops;
LG_TransportStatus status;
atomic_bool done;
bool frame;
};
struct UnregisterState
{
LG_Transport * transport;
const LG_FrameOps * ops;
atomic_bool started;
atomic_bool done;
};
struct VideoStatusTrace
{
atomic_uint count;
atomic_uint_fast64_t frameEpoch;
atomic_uint_fast64_t pointerEpoch;
atomic_int frameReason;
atomic_int pointerReason;
atomic_bool frameAvailable;
atomic_bool pointerAvailable;
atomic_bool active;
atomic_bool overlap;
atomic_bool block;
atomic_bool entered;
atomic_bool release;
atomic_bool * waitBeforeReplace;
atomic_bool unregisterDone;
atomic_bool replaceDone;
LG_Transport * unregisterTransport;
LG_Transport ** destroyTransport;
const LG_FrameOps * unregisterOps;
struct VideoStatusTrace * registerTrace;
struct VideoStatusTrace * replaceTrace;
unsigned int registerAtCount;
LG_Transport * transport;
const LG_FrameOps * ops;
};
static void videoStatusChanged(void * opaque,
const LG_VideoStatus * status)
{
struct VideoStatusTrace * trace = opaque;
if (atomic_exchange_explicit(&trace->active, true,
memory_order_acq_rel))
atomic_store_explicit(&trace->overlap, true, memory_order_release);
atomic_store_explicit(&trace->frameEpoch,
status->frame.epoch, memory_order_relaxed);
atomic_store_explicit(&trace->pointerEpoch,
status->pointer.epoch, memory_order_relaxed);
atomic_store_explicit(&trace->frameReason,
status->frame.reason, memory_order_relaxed);
atomic_store_explicit(&trace->pointerReason,
status->pointer.reason, memory_order_relaxed);
atomic_store_explicit(&trace->frameAvailable,
status->frame.available, memory_order_relaxed);
atomic_store_explicit(&trace->pointerAvailable,
status->pointer.available, memory_order_relaxed);
const unsigned int count = atomic_fetch_add_explicit(
&trace->count, 1, memory_order_release) + 1;
if (trace->unregisterTransport)
{
trace->unregisterOps->setStatusListener(
trace->unregisterTransport, NULL, NULL);
atomic_store_explicit(
&trace->unregisterDone, true, memory_order_release);
}
if (trace->destroyTransport)
{
LGT_LGMP.destroy(trace->destroyTransport);
atomic_store_explicit(
&trace->unregisterDone, true, memory_order_release);
}
if (trace->registerTrace &&
(!trace->registerAtCount || count >= trace->registerAtCount))
trace->registerTrace->ops->setStatusListener(
trace->registerTrace->transport, videoStatusChanged,
trace->registerTrace);
if (atomic_load_explicit(&trace->block, memory_order_acquire))
{
atomic_store_explicit(&trace->entered, true, memory_order_release);
while (!atomic_load_explicit(&trace->release, memory_order_acquire))
usleep(1000);
}
if (trace->replaceTrace)
{
if (trace->waitBeforeReplace)
while (!atomic_load_explicit(
trace->waitBeforeReplace, memory_order_acquire))
usleep(1000);
trace->replaceTrace->ops->setStatusListener(
trace->replaceTrace->transport, NULL, NULL);
trace->replaceTrace->ops->setStatusListener(
trace->replaceTrace->transport, videoStatusChanged,
trace->replaceTrace);
atomic_store_explicit(
&trace->replaceDone, true, memory_order_release);
}
atomic_store_explicit(&trace->active, false, memory_order_release);
}
static bool waitStatusCount(
const struct VideoStatusTrace * trace, unsigned int count)
{
for (unsigned i = 0; i < WAIT_TIMEOUT; ++i)
{
if (atomic_load_explicit(&trace->count, memory_order_acquire) >= count)
return true;
usleep(1000);
}
return false;
}
static void * disconnectVideo(void * opaque)
{
struct VideoStatusTrace * trace = opaque;
LGT_LGMP.disconnect(trace->transport);
return NULL;
}
static void * disconnectVideoSignaled(void * opaque)
{
struct VideoStatusTrace * trace = opaque;
LGT_LGMP.disconnect(trace->transport);
return NULL;
}
static void * registerVideoStatus(void * opaque)
{
struct VideoStatusTrace * trace = opaque;
trace->ops->setStatusListener(
trace->transport, videoStatusChanged, trace);
return NULL;
}
static void * unregisterVideoStatus(void * opaque)
{
struct UnregisterState * state = opaque;
atomic_store_explicit(&state->started, true, memory_order_release);
state->ops->setStatusListener(state->transport, NULL, NULL);
atomic_store_explicit(&state->done, true, memory_order_release);
return NULL;
}
static void * waitForVideo(void * opaque)
{
struct WaitState * state = opaque;
if (state->frame)
{
LG_TransportFrame frame;
state->status = state->ops->nextFrame(
state->transport, false, &frame);
}
else
{
LG_TransportPointer pointer;
state->status = state->ops->nextPointer(state->transport, &pointer);
}
atomic_store_explicit(&state->done, true, memory_order_release);
return NULL;
}
static bool checkWaitCancellation(LG_Transport * transport,
const LG_FrameOps * ops, PLGMPHostQueue frameQueue,
PLGMPHostQueue pointerQueue)
{
struct WaitState frame = {
.transport = transport,
.ops = ops,
.status = LG_TRANSPORT_ERROR,
.frame = true,
};
struct WaitState pointer = {
.transport = transport,
.ops = ops,
.status = LG_TRANSPORT_ERROR,
};
pthread_t frameThread;
pthread_t pointerThread;
if (pthread_create(&frameThread, NULL, waitForVideo, &frame) != 0)
return false;
if (pthread_create(&pointerThread, NULL, waitForVideo, &pointer) != 0)
{
ops->cancelFrameWait(transport);
pthread_join(frameThread, NULL);
return false;
}
bool subscribed = false;
for (unsigned i = 0; i < WAIT_TIMEOUT; ++i)
{
if (lgmpHostQueueHasSubs(frameQueue) &&
lgmpHostQueueHasSubs(pointerQueue))
{
subscribed = true;
break;
}
usleep(1000);
}
ops->cancelFrameWait(transport);
ops->cancelPointerWait(transport);
const int frameJoin = pthread_join(frameThread, NULL);
const int pointerJoin = pthread_join(pointerThread, NULL);
return subscribed && frameJoin == 0 && pointerJoin == 0 &&
frame.status == LG_TRANSPORT_TIMEOUT &&
pointer.status == LG_TRANSPORT_TIMEOUT;
}
int main(void)
{
int result = 1;
int fd = -1;
char path[] = "/tmp/lgmp-transport-test-XXXXXX";
bool pathExists = false;
void * hostMemory = MAP_FAILED;
PLGMPHost host = NULL;
PLGMPHostQueue frameQueue = NULL;
PLGMPHostQueue pointerQueue = NULL;
PLGMPMemory frameMemory = NULL;
PLGMPMemory malformedFrameMemory = NULL;
PLGMPMemory pointerMemory = NULL;
PLGMPMemory malformedPointerMemory = NULL;
LG_Transport * transport = NULL;
LG_Transport * secondTransport = NULL;
LG_Transport * thirdTransport = NULL;
LG_Transport * nestedTransport = NULL;
LG_Transport * admittedTransport = NULL;
const LG_FrameOps * frameOps = NULL;
struct VideoStatusTrace videoTrace;
struct VideoStatusTrace secondVideoTrace;
memset(&videoTrace, 0, sizeof(videoTrace));
memset(&secondVideoTrace, 0, sizeof(secondVideoTrace));
debug_init();
CHECK(lgmp_testRecoveryError(KVMFR_R_ERR_BUSY) ==
LG_RECOVERY_ERR_BUSY);
CHECK(lgmp_testRecoveryError(KVMFR_R_ERR_CAPACITY) ==
LG_RECOVERY_ERR_CAPACITY);
struct CancelAfter cancelDuringProbe = { .limit = 3 };
CHECK(!lgmp_testRecoveryProbeCancellation(
cancelAfter, &cancelDuringProbe));
CHECK(cancelDuringProbe.calls >= cancelDuringProbe.limit);
fd = mkstemp(path);
CHECK(fd >= 0);
pathExists = true;
CHECK(ftruncate(fd, TEST_SHM_SIZE) == 0);
hostMemory = mmap(NULL, TEST_SHM_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED,
fd, 0);
CHECK(hostMemory != MAP_FAILED);
KVMFR session = { 0 };
memcpy(session.magic, KVMFR_MAGIC, sizeof(session.magic));
session.version = KVMFR_VERSION;
memcpy(session.hostver, "transport-test", sizeof("transport-test"));
CHECK(lgmpHostInit(hostMemory, TEST_SHM_SIZE, &host, sizeof(session),
(uint8_t *)&session) == LGMP_OK);
const struct LGMPQueueConfig frameConfig = {
.queueID = LGMP_Q_FRAME,
.numMessages = LGMP_Q_FRAME_LEN,
.subTimeout = TEST_TIMEOUT,
};
const struct LGMPQueueConfig pointerConfig = {
.queueID = LGMP_Q_POINTER,
.numMessages = LGMP_Q_POINTER_LEN,
.subTimeout = TEST_TIMEOUT,
};
const uint32_t frameSize =
sizeof(KVMFRFrame) + sizeof(FrameBuffer) + sizeof(uint32_t);
CHECK(lgmpHostMemAlloc(host, frameSize, &frameMemory) == LGMP_OK);
KVMFRFrame * wireFrame = lgmpHostMemPtr(frameMemory);
wireFrame->formatVer = 1;
wireFrame->frameSerial = 1;
wireFrame->type = FRAME_TYPE_BGRA;
wireFrame->screenWidth = 1;
wireFrame->screenHeight = 1;
wireFrame->dataWidth = 1;
wireFrame->dataHeight = 1;
wireFrame->frameWidth = 1;
wireFrame->frameHeight = 1;
wireFrame->rotation = FRAME_ROT_0;
wireFrame->stride = 1;
wireFrame->pitch = sizeof(uint32_t);
wireFrame->offset = sizeof(*wireFrame);
wireFrame->sdrWhiteLevel = KVMFR_SDR_WHITE_LEVEL_DEFAULT;
wireFrame->captureTime = 100;
wireFrame->postProcessTime = 200;
wireFrame->copyTime = 300;
wireFrame->readyTime = 400;
wireFrame->timingSerial = wireFrame->frameSerial;
wireFrame->timingValid = 1;
FrameBuffer * framebuffer = (FrameBuffer *)((uint8_t *)wireFrame +
wireFrame->offset);
atomic_store(&framebuffer->wp, sizeof(uint32_t));
CHECK(lgmpHostMemAlloc(host, sizeof(KVMFRFrame) - 1,
&malformedFrameMemory) == LGMP_OK);
CHECK(lgmpHostMemAlloc(host, sizeof(KVMFRCursor), &pointerMemory) ==
LGMP_OK);
CHECK(lgmpHostMemAlloc(host, sizeof(KVMFRCursor),
&malformedPointerMemory) == LGMP_OK);
KVMFRCursor * malformedPointer = lgmpHostMemPtr(malformedPointerMemory);
malformedPointer->height = 2;
malformedPointer->pitch = UINT32_MAX;
KVMFRCursor * wirePointer = lgmpHostMemPtr(pointerMemory);
wirePointer->x = 1;
wirePointer->y = 1;
LGT_LGMP.setup();
option_set_string("lgmp", "shmDevice", path);
option_set_bool("lgmp", "allowDMA", false);
option_set_int("lgmp", "framePollInterval", POLL_INTERVAL);
option_set_int("lgmp", "cursorPollInterval", POLL_INTERVAL);
CHECK(LGT_LGMP.create(&transport));
const LG_VideoOps * videoOps = LGT_LGMP.getVideoOps(transport);
CHECK(videoOps);
CHECK(videoOps->type == LG_VIDEO_TYPE_FRAME);
CHECK(videoOps->frame);
frameOps = videoOps->frame;
CHECK(frameOps->cancelFrameWait);
CHECK(frameOps->cancelPointerWait);
CHECK(frameOps->setStatusListener);
CHECK(LGT_LGMP.create(&secondTransport));
const LG_VideoOps * secondVideoOps =
LGT_LGMP.getVideoOps(secondTransport);
CHECK(secondVideoOps && secondVideoOps->type == LG_VIDEO_TYPE_FRAME);
CHECK(secondVideoOps->frame && secondVideoOps->frame->setStatusListener);
CHECK(LGT_LGMP.create(&thirdTransport));
const LG_VideoOps * thirdVideoOps =
LGT_LGMP.getVideoOps(thirdTransport);
CHECK(thirdVideoOps && thirdVideoOps->type == LG_VIDEO_TYPE_FRAME);
CHECK(thirdVideoOps->frame && thirdVideoOps->frame->setStatusListener);
CHECK(LGT_LGMP.create(&nestedTransport));
const LG_VideoOps * nestedVideoOps =
LGT_LGMP.getVideoOps(nestedTransport);
CHECK(nestedVideoOps && nestedVideoOps->type == LG_VIDEO_TYPE_FRAME);
CHECK(nestedVideoOps->frame && nestedVideoOps->frame->setStatusListener);
CHECK(LGT_LGMP.create(&admittedTransport));
const LG_VideoOps * admittedVideoOps =
LGT_LGMP.getVideoOps(admittedTransport);
CHECK(admittedVideoOps && admittedVideoOps->type == LG_VIDEO_TYPE_FRAME);
CHECK(admittedVideoOps->frame &&
admittedVideoOps->frame->setStatusListener);
usleep(300000);
CHECK(lgmpHostProcess(host) == LGMP_OK);
LG_TransportSession transportSession;
bool cancelConnect = true;
CHECK(LGT_LGMP.connectCancellable(transport, &transportSession,
cancelled, &cancelConnect) == LG_TRANSPORT_DISCONNECTED);
CHECK(!LGT_LGMP.sessionValid(transport));
CHECK(LGT_LGMP.connectCancellable(
transport, &transportSession, NULL, NULL) == LG_TRANSPORT_OK);
LG_TransportSession secondSession;
CHECK(LGT_LGMP.connectCancellable(
secondTransport, &secondSession, NULL, NULL) == LG_TRANSPORT_OK);
LG_TransportSession thirdSession;
CHECK(LGT_LGMP.connectCancellable(
thirdTransport, &thirdSession, NULL, NULL) == LG_TRANSPORT_OK);
LG_TransportSession nestedSession;
CHECK(LGT_LGMP.connectCancellable(
nestedTransport, &nestedSession, NULL, NULL) == LG_TRANSPORT_OK);
LG_TransportSession admittedSession;
CHECK(LGT_LGMP.connectCancellable(
admittedTransport, &admittedSession, NULL, NULL) == LG_TRANSPORT_OK);
videoTrace.transport = transport;
videoTrace.ops = frameOps;
frameOps->setStatusListener(
transport, videoStatusChanged, &videoTrace);
CHECK(atomic_load_explicit(&videoTrace.count, memory_order_acquire) == 1);
CHECK(!atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
CHECK(!atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.frameReason, memory_order_acquire) ==
LG_TRANSPORT_UNAVAILABLE);
CHECK(atomic_load_explicit(
&videoTrace.pointerReason, memory_order_acquire) ==
LG_TRANSPORT_UNAVAILABLE);
CHECK(lgmpHostQueueNew(host, frameConfig, &frameQueue) == LGMP_OK);
CHECK(lgmpHostQueueNew(host, pointerConfig, &pointerQueue) == LGMP_OK);
LG_TransportFrame frame;
LG_TransportPointer pointer;
CHECK(checkWaitCancellation(
transport, frameOps, frameQueue, pointerQueue));
CHECK(waitStatusCount(&videoTrace, 3));
CHECK(atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.frameReason, memory_order_acquire) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.pointerReason, memory_order_acquire) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire) != 0);
CHECK(atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire) != 0);
CHECK(!atomic_load_explicit(&videoTrace.overlap, memory_order_acquire));
CHECK(lgmpHostQueueHasSubs(frameQueue));
CHECK(lgmpHostQueueHasSubs(pointerQueue));
CHECK(lgmpHostQueueNewSubs(frameQueue) == 1);
CHECK(lgmpHostQueueNewSubs(pointerQueue) == 1);
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
CHECK(frame.epoch == atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire));
LG_TransportFrameTiming timing;
frameOps->getFrameTiming(transport, &frame, &timing);
CHECK(timing.captureTime == wireFrame->captureTime);
CHECK(timing.postProcessTime == wireFrame->postProcessTime);
CHECK(timing.copyTime == wireFrame->copyTime);
CHECK(timing.readyTime == wireFrame->readyTime);
CHECK(timing.providerValid);
frameOps->releaseFrame(transport, &frame);
wireFrame->frameSerial = 2;
wireFrame->timingSerial = wireFrame->frameSerial;
wireFrame->timingValid = 0;
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
/* Work which began before producer timing became coherent is not reported
* as an independent provider stage. */
wireFrame->timingValid = 1;
frameOps->getFrameTiming(transport, &frame, &timing);
CHECK(timing.valid);
CHECK(!timing.providerValid);
CHECK(timing.receiveTime == 0);
CHECK(timing.prepareTime == 0);
frameOps->releaseFrame(transport, &frame);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_OK);
CHECK(pointer.epoch == atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire));
frameOps->releasePointer(transport, &pointer);
const uint64_t frameEpoch = atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire);
const uint64_t pointerEpoch = atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire);
CHECK(lgmpHostQueuePost(frameQueue, 0, malformedFrameMemory) == LGMP_OK);
frameOps->cancelFrameWait(transport);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_ERROR);
CHECK(!atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire) != frameEpoch);
CHECK(atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire) == pointerEpoch);
CHECK(atomic_load_explicit(
&videoTrace.frameReason, memory_order_acquire) ==
LG_TRANSPORT_ERROR);
CHECK(lgmpHostQueuePost(frameQueue, 0, malformedFrameMemory) == LGMP_OK);
struct WaitState frameErrorWait =
{
.transport = transport,
.ops = frameOps,
.status = LG_TRANSPORT_OK,
.frame = true,
};
pthread_t frameErrorThread;
CHECK(pthread_create(&frameErrorThread, NULL,
waitForVideo, &frameErrorWait) == 0);
CHECK(waitForQueuesEmpty(host, frameQueue, pointerQueue));
usleep(POLL_INTERVAL / 10U);
CHECK(!atomic_load_explicit(
&frameErrorWait.done, memory_order_acquire));
frameOps->cancelFrameWait(transport);
CHECK(pthread_join(frameErrorThread, NULL) == 0);
CHECK(atomic_load_explicit(&frameErrorWait.done, memory_order_acquire));
CHECK(frameErrorWait.status == LG_TRANSPORT_ERROR);
wireFrame->frameSerial = 3;
wireFrame->timingSerial = wireFrame->frameSerial;
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
frameOps->releaseFrame(transport, &frame);
const uint64_t frameEpochAfterRecovery = atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire);
const uint64_t pointerEpochBeforeLoss = atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_SHAPE,
malformedPointerMemory) == LGMP_OK);
frameOps->cancelPointerWait(transport);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_ERROR);
CHECK(atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
CHECK(!atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire) ==
frameEpochAfterRecovery);
CHECK(atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire) !=
pointerEpochBeforeLoss);
CHECK(atomic_load_explicit(
&videoTrace.pointerReason, memory_order_acquire) ==
LG_TRANSPORT_ERROR);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_SHAPE,
malformedPointerMemory) == LGMP_OK);
struct WaitState pointerErrorWait =
{
.transport = transport,
.ops = frameOps,
.status = LG_TRANSPORT_OK,
};
pthread_t pointerErrorThread;
CHECK(pthread_create(&pointerErrorThread, NULL,
waitForVideo, &pointerErrorWait) == 0);
CHECK(waitForQueuesEmpty(host, frameQueue, pointerQueue));
usleep(POLL_INTERVAL / 10U);
CHECK(!atomic_load_explicit(
&pointerErrorWait.done, memory_order_acquire));
frameOps->cancelPointerWait(transport);
CHECK(pthread_join(pointerErrorThread, NULL) == 0);
CHECK(atomic_load_explicit(&pointerErrorWait.done, memory_order_acquire));
CHECK(pointerErrorWait.status == LG_TRANSPORT_ERROR);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
frameOps->releasePointer(transport, &pointer);
/*
* Stop with messages pending. Unsubscribing before the host timeout is the
* lifecycle behavior that was lost when LGMP moved behind the transport API.
*/
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
CHECK(frameOps->stopFrame);
CHECK(frameOps->stopPointer);
frameOps->stopFrame(transport);
frameOps->stopPointer(transport);
CHECK(!lgmpHostQueueHasSubs(frameQueue));
CHECK(!lgmpHostQueueHasSubs(pointerQueue));
CHECK(waitForQueuesEmpty(host, frameQueue, pointerQueue));
CHECK(lgmpHostQueuePending(frameQueue) == 0);
CHECK(lgmpHostQueuePending(pointerQueue) == 0);
frameOps->cancelFrameWait(transport);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_TIMEOUT);
frameOps->cancelPointerWait(transport);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT);
CHECK(lgmpHostQueueNewSubs(frameQueue) == 1);
CHECK(lgmpHostQueueNewSubs(pointerQueue) == 1);
/* The host resends the same serial after a new subscription. */
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
frameOps->releaseFrame(transport, &frame);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_OK);
frameOps->releasePointer(transport, &pointer);
/*
* Also recover when the host has already marked the cached handles bad.
* LGMP leaves a timed-out handle non-NULL when unsubscribe fails.
*/
wireFrame->frameSerial = 4;
wireFrame->timingSerial = wireFrame->frameSerial;
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
CHECK(waitForQueuesEmpty(host, frameQueue, pointerQueue));
CHECK(lgmpHostQueueHasSubs(frameQueue));
CHECK(lgmpHostQueueHasSubs(pointerQueue));
frameOps->stopFrame(transport);
frameOps->stopPointer(transport);
frameOps->cancelFrameWait(transport);
const uint64_t oldFrameEpoch = atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire);
const uint64_t oldPointerEpoch = atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_TIMEOUT);
CHECK(atomic_load_explicit(
&videoTrace.frameEpoch, memory_order_acquire) != oldFrameEpoch);
CHECK(atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire) == oldPointerEpoch);
frameOps->cancelPointerWait(transport);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT);
CHECK(atomic_load_explicit(
&videoTrace.pointerEpoch, memory_order_acquire) != oldPointerEpoch);
CHECK(lgmpHostQueueNewSubs(frameQueue) == 1);
CHECK(lgmpHostQueueNewSubs(pointerQueue) == 1);
/* Resetting the transport serial makes the host's resend visible. */
CHECK(lgmpHostQueuePost(frameQueue, 0, frameMemory) == LGMP_OK);
CHECK(lgmpHostQueuePost(pointerQueue, CURSOR_FLAG_POSITION,
pointerMemory) == LGMP_OK);
const bool pointerBeforeFrameRecovery = atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire);
CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.frameAvailable, memory_order_acquire));
CHECK(atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire) ==
pointerBeforeFrameRecovery);
frameOps->releaseFrame(transport, &frame);
CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_OK);
CHECK(atomic_load_explicit(
&videoTrace.pointerAvailable, memory_order_acquire));
frameOps->releasePointer(transport, &pointer);
atomic_store_explicit(&videoTrace.block, true, memory_order_release);
pthread_t disconnectThread;
pthread_t crossUnregisterThread;
CHECK(pthread_create(&disconnectThread, NULL,
disconnectVideo, &videoTrace) == 0);
for (unsigned i = 0; i < WAIT_TIMEOUT && !atomic_load_explicit(
&videoTrace.entered, memory_order_acquire); ++i)
usleep(1000);
CHECK(atomic_load_explicit(&videoTrace.entered, memory_order_acquire));
secondVideoTrace.transport = secondTransport;
secondVideoTrace.ops = secondVideoOps->frame;
secondVideoTrace.unregisterTransport = transport;
secondVideoTrace.unregisterOps = frameOps;
CHECK(pthread_create(&crossUnregisterThread, NULL,
registerVideoStatus, &secondVideoTrace) == 0);
usleep(POLL_INTERVAL);
CHECK(atomic_load_explicit(
&secondVideoTrace.count, memory_order_acquire) == 0);
CHECK(!atomic_load_explicit(&secondVideoTrace.unregisterDone,
memory_order_acquire));
atomic_store_explicit(&videoTrace.release, true, memory_order_release);
CHECK(pthread_join(disconnectThread, NULL) == 0);
CHECK(pthread_join(crossUnregisterThread, NULL) == 0);
CHECK(waitStatusCount(&secondVideoTrace, 1));
CHECK(atomic_load_explicit(&secondVideoTrace.unregisterDone,
memory_order_acquire));
CHECK(!atomic_load_explicit(&videoTrace.overlap, memory_order_acquire));
CHECK(!atomic_load_explicit(
&secondVideoTrace.overlap, memory_order_acquire));
frameOps->setStatusListener(transport, NULL, NULL);
secondVideoOps->frame->setStatusListener(secondTransport, NULL, NULL);
/* A callback can unregister another instance while a callback for that
* instance is waiting to dispatch. The queued callback revalidates its
* listener and is skipped. */
struct VideoStatusTrace queuedTrace;
struct VideoStatusTrace nestedTrace;
struct VideoStatusTrace replacementTrace;
memset(&queuedTrace, 0, sizeof(queuedTrace));
memset(&nestedTrace, 0, sizeof(nestedTrace));
memset(&replacementTrace, 0, sizeof(replacementTrace));
queuedTrace.transport = transport;
queuedTrace.ops = frameOps;
nestedTrace.transport = secondTransport;
nestedTrace.ops = secondVideoOps->frame;
nestedTrace.unregisterTransport = transport;
nestedTrace.unregisterOps = frameOps;
replacementTrace.transport = secondTransport;
replacementTrace.ops = secondVideoOps->frame;
nestedTrace.registerTrace = &replacementTrace;
frameOps->setStatusListener(
transport, videoStatusChanged, &queuedTrace);
const unsigned int queuedCount = atomic_load_explicit(
&queuedTrace.count, memory_order_acquire);
secondVideoOps->frame->setStatusListener(
secondTransport, videoStatusChanged, &nestedTrace);
CHECK(atomic_load_explicit(
&nestedTrace.unregisterDone, memory_order_acquire));
CHECK(atomic_load_explicit(
&replacementTrace.count, memory_order_acquire) == 1);
LGT_LGMP.disconnect(transport);
CHECK(atomic_load_explicit(
&queuedTrace.count, memory_order_acquire) == queuedCount);
/* Re-registering the identical callback pair creates a new listener
* lifetime. A publication queued for the prior lifetime must not be
* delivered after that replacement. */
struct VideoStatusTrace gateTrace;
struct VideoStatusTrace abaTrace;
memset(&gateTrace, 0, sizeof(gateTrace));
memset(&abaTrace, 0, sizeof(abaTrace));
gateTrace.transport = transport;
gateTrace.ops = frameOps;
abaTrace.transport = secondTransport;
abaTrace.ops = secondVideoOps->frame;
frameOps->setStatusListener(transport, NULL, NULL);
secondVideoOps->frame->setStatusListener(
secondTransport, videoStatusChanged, &abaTrace);
const unsigned int abaCount = atomic_load_explicit(
&abaTrace.count, memory_order_acquire);
gateTrace.block = true;
atomic_bool abaWaiting = false;
gateTrace.replaceTrace = &abaTrace;
gateTrace.waitBeforeReplace = &abaWaiting;
pthread_t gateThread;
CHECK(pthread_create(&gateThread, NULL,
registerVideoStatus, &gateTrace) == 0);
for (unsigned i = 0; i < WAIT_TIMEOUT && !atomic_load_explicit(
&gateTrace.entered, memory_order_acquire); ++i)
usleep(1000);
CHECK(atomic_load_explicit(&gateTrace.entered, memory_order_acquire));
atomic_bool abaHold = true;
lgmp_testSetVideoStatusDispatchGate(
secondTransport, &abaWaiting, &abaHold);
pthread_t abaPublishThread;
CHECK(pthread_create(&abaPublishThread, NULL,
disconnectVideoSignaled, &abaTrace) == 0);
for (unsigned i = 0; i < WAIT_TIMEOUT && !atomic_load_explicit(
&abaWaiting, memory_order_acquire); ++i)
usleep(1000);
CHECK(atomic_load_explicit(&abaWaiting, memory_order_acquire));
atomic_store_explicit(&gateTrace.release, true, memory_order_release);
usleep(POLL_INTERVAL);
CHECK(atomic_load_explicit(
&gateTrace.replaceDone, memory_order_acquire));
/* External unregister waits for the stale accepted dispatch to drain. */
struct UnregisterState unregisterState =
{
.transport = secondTransport,
.ops = secondVideoOps->frame,
};
pthread_t unregisterThread;
CHECK(pthread_create(&unregisterThread, NULL,
unregisterVideoStatus, &unregisterState) == 0);
for (unsigned i = 0; i < WAIT_TIMEOUT && !atomic_load_explicit(
&unregisterState.started, memory_order_acquire); ++i)
usleep(1000);
CHECK(atomic_load_explicit(
&unregisterState.started, memory_order_acquire));
usleep(POLL_INTERVAL);
CHECK(!atomic_load_explicit(
&unregisterState.done, memory_order_acquire));
atomic_store_explicit(&abaHold, false, memory_order_release);
CHECK(pthread_join(gateThread, NULL) == 0);
CHECK(pthread_join(abaPublishThread, NULL) == 0);
CHECK(pthread_join(unregisterThread, NULL) == 0);
CHECK(atomic_load_explicit(&gateTrace.replaceDone, memory_order_acquire));
CHECK(atomic_load_explicit(&unregisterState.done, memory_order_acquire));
lgmp_testSetVideoStatusDispatchGate(NULL, NULL, NULL);
CHECK(atomic_load_explicit(
&abaTrace.count, memory_order_acquire) == abaCount + 1);
/* Symmetric cross-instance unregisters are safe when both registrations
* start together. Both callbacks run sequentially and complete their
* cross-instance unregister. */
struct VideoStatusTrace symmetricA;
struct VideoStatusTrace symmetricB;
memset(&symmetricA, 0, sizeof(symmetricA));
memset(&symmetricB, 0, sizeof(symmetricB));
symmetricA.transport = transport;
symmetricA.ops = frameOps;
symmetricA.unregisterTransport = secondTransport;
symmetricA.unregisterOps = secondVideoOps->frame;
symmetricB.transport = secondTransport;
symmetricB.ops = secondVideoOps->frame;
symmetricB.unregisterTransport = transport;
symmetricB.unregisterOps = frameOps;
pthread_t symmetricThreadA;
pthread_t symmetricThreadB;
CHECK(pthread_create(&symmetricThreadA, NULL,
registerVideoStatus, &symmetricA) == 0);
CHECK(pthread_create(&symmetricThreadB, NULL,
registerVideoStatus, &symmetricB) == 0);
CHECK(pthread_join(symmetricThreadA, NULL) == 0);
CHECK(pthread_join(symmetricThreadB, NULL) == 0);
CHECK(atomic_load_explicit(
&symmetricA.count, memory_order_acquire) == 1);
CHECK(atomic_load_explicit(
&symmetricB.count, memory_order_acquire) == 1);
CHECK(atomic_load_explicit(
&symmetricA.unregisterDone, memory_order_acquire));
CHECK(atomic_load_explicit(
&symmetricB.unregisterDone, memory_order_acquire));
/* A callback may destroy the instance on which an admitted callback is
* active. Physical teardown is deferred until that callback retires. */
frameOps->setStatusListener(transport, NULL, NULL);
secondVideoOps->frame->setStatusListener(secondTransport, NULL, NULL);
admittedVideoOps->frame->setStatusListener(admittedTransport, NULL, NULL);
struct VideoStatusTrace admittedDestroy;
struct VideoStatusTrace admittedOuter;
memset(&admittedDestroy, 0, sizeof(admittedDestroy));
memset(&admittedOuter, 0, sizeof(admittedOuter));
admittedDestroy.transport = secondTransport;
admittedDestroy.ops = secondVideoOps->frame;
admittedDestroy.destroyTransport = &admittedTransport;
admittedOuter.transport = admittedTransport;
admittedOuter.ops = admittedVideoOps->frame;
admittedOuter.registerTrace = &admittedDestroy;
admittedOuter.registerAtCount = 2;
admittedVideoOps->frame->setStatusListener(
admittedTransport, videoStatusChanged, &admittedOuter);
CHECK(atomic_load_explicit(
&admittedOuter.count, memory_order_acquire) == 1);
LGT_LGMP.disconnect(admittedTransport);
CHECK(!admittedTransport);
CHECK(atomic_load_explicit(
&admittedOuter.count, memory_order_acquire) == 2);
CHECK(atomic_load_explicit(
&admittedDestroy.count, memory_order_acquire) == 1);
/* A synchronous callback stays pinned while it drops serialization to
* drain a different instance, including when a third callback destroys it.
*/
thirdVideoOps->frame->setStatusListener(thirdTransport, NULL, NULL);
nestedVideoOps->frame->setStatusListener(nestedTransport, NULL, NULL);
struct VideoStatusTrace drainTrace;
struct VideoStatusTrace pinnedTrace;
struct VideoStatusTrace thirdTrace;
memset(&drainTrace, 0, sizeof(drainTrace));
memset(&pinnedTrace, 0, sizeof(pinnedTrace));
memset(&thirdTrace, 0, sizeof(thirdTrace));
drainTrace.transport = thirdTransport;
drainTrace.ops = thirdVideoOps->frame;
thirdVideoOps->frame->setStatusListener(
thirdTransport, videoStatusChanged, &drainTrace);
atomic_bool drainWaiting = false;
atomic_bool drainHold = true;
lgmp_testSetVideoStatusDispatchGate(
thirdTransport, &drainWaiting, &drainHold);
pthread_t drainThread;
CHECK(pthread_create(&drainThread, NULL,
disconnectVideoSignaled, &drainTrace) == 0);
for (unsigned i = 0; i < WAIT_TIMEOUT && !atomic_load_explicit(
&drainWaiting, memory_order_acquire); ++i)
usleep(1000);
CHECK(atomic_load_explicit(&drainWaiting, memory_order_acquire));
pinnedTrace.transport = transport;
pinnedTrace.ops = frameOps;
pinnedTrace.unregisterTransport = thirdTransport;
pinnedTrace.unregisterOps = thirdVideoOps->frame;
pthread_t pinnedThread;
CHECK(pthread_create(&pinnedThread, NULL,
registerVideoStatus, &pinnedTrace) == 0);
CHECK(waitStatusCount(&pinnedTrace, 1));
thirdTrace.transport = nestedTransport;
thirdTrace.ops = nestedVideoOps->frame;
nestedVideoOps->frame->setStatusListener(
nestedTransport, videoStatusChanged, &thirdTrace);
thirdTrace.destroyTransport = &transport;
pthread_t thirdThread;
CHECK(pthread_create(&thirdThread, NULL,
disconnectVideoSignaled, &thirdTrace) == 0);
CHECK(waitStatusCount(&thirdTrace, 2));
atomic_store_explicit(&drainHold, false, memory_order_release);
CHECK(pthread_join(drainThread, NULL) == 0);
CHECK(pthread_join(pinnedThread, NULL) == 0);
CHECK(pthread_join(thirdThread, NULL) == 0);
lgmp_testSetVideoStatusDispatchGate(NULL, NULL, NULL);
CHECK(!transport);
CHECK(atomic_load_explicit(
&pinnedTrace.unregisterDone, memory_order_acquire));
CHECK(atomic_load_explicit(
&thirdTrace.unregisterDone, memory_order_acquire));
/* The same lifetime rule applies when nested synchronous callbacks destroy
* the instance belonging to the enclosing callback. */
secondVideoOps->frame->setStatusListener(secondTransport, NULL, NULL);
struct VideoStatusTrace syncDestroy;
struct VideoStatusTrace syncOuter;
memset(&syncDestroy, 0, sizeof(syncDestroy));
memset(&syncOuter, 0, sizeof(syncOuter));
syncDestroy.transport = secondTransport;
syncDestroy.ops = secondVideoOps->frame;
syncDestroy.destroyTransport = &thirdTransport;
syncOuter.transport = thirdTransport;
syncOuter.ops = thirdVideoOps->frame;
syncOuter.registerTrace = &syncDestroy;
thirdVideoOps->frame->setStatusListener(
thirdTransport, videoStatusChanged, &syncOuter);
CHECK(!thirdTransport);
CHECK(atomic_load_explicit(
&syncOuter.count, memory_order_acquire) == 1);
CHECK(atomic_load_explicit(
&syncDestroy.count, memory_order_acquire) == 1);
result = 0;
cleanup:
if (admittedTransport)
LGT_LGMP.destroy(&admittedTransport);
if (nestedTransport)
LGT_LGMP.destroy(&nestedTransport);
if (thirdTransport)
LGT_LGMP.destroy(&thirdTransport);
if (secondTransport)
LGT_LGMP.destroy(&secondTransport);
if (transport)
LGT_LGMP.destroy(&transport);
option_free();
lgmpHostMemFree(&pointerMemory);
lgmpHostMemFree(&malformedPointerMemory);
lgmpHostMemFree(&malformedFrameMemory);
lgmpHostMemFree(&frameMemory);
lgmpHostFree(&host);
if (hostMemory != MAP_FAILED)
munmap(hostMemory, TEST_SHM_SIZE);
if (fd >= 0)
close(fd);
if (pathExists)
unlink(path);
return result;
}
Reference in New Issue View Git Blame Copy Permalink