FGFS-4.1/simgear
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Emesary IReceiver no longer shared ptr.

Browse Source 
The use of SGShared for IReceiver derived conflicted with other uses of SGShared / SGReferenced - and after consideration transmitters should probably remain static for the lifetime of the application.

Any dynamic transmitters will need careful consideration if working in a threaded environment to ensure that a transmitter doesn't go out of scope before any notifications that it has issued have finished being processed by other threads.
This commit is contained in:
Richard Harrison
2021-04-16 21:13:13 +02:00
parent 41d06ee092
commit 2690fadca1
9 changed files with 454 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
simgear/emesary/INotification.hxx
View File

@@ -21,14 +21,20 @@
* Copyright (C)2019 Richard Harrison Licenced under GPL2 or later.
*
*---------------------------------------------------------------------------*/
#include <simgear/structure/SGSharedPtr.hxx>
namespace simgear
{
namespace Emesary
{
/// Interface (base class) for all notifications.
class INotification
class INotification: public SGReferenced
{
public:
virtual ~INotification()
{
}
// text representation of notification type. must be unique across all notifications
virtual const char *GetType() = 0;
@@ -49,6 +55,7 @@ namespace simgear
/// is returned as the status.
virtual bool IsComplete() { return true; }
};
typedef SGSharedPtr<INotification> INotificationPtr;
}
}
#endif

12
simgear/emesary/IReceiver.hxx
View File

@@ -19,33 +19,31 @@
* Copyright (C)2019 Richard Harrison Licenced under GPL2 or later.
*
*---------------------------------------------------------------------------*/
#include <simgear/structure/SGReferenced.hxx>
namespace simgear
{
namespace Emesary
{
/// Interface (base class) for a recipeint.
class IReceiver : public SGReferenced
class IReceiver
{
public:
virtual ~IReceiver() = default;
/// Receive notification - must be implemented
virtual ReceiptStatus Receive(INotification& message) = 0;
virtual ReceiptStatus Receive(INotificationPtr message) = 0;
/// Called when registered at a transmitter
virtual void OnRegisteredAtTransmitter(class Transmitter* p)
virtual void OnRegisteredAtTransmitter(class Transmitter *p)
{
}
/// Called when de-registered at a transmitter
virtual void OnDeRegisteredAtTransmitter(class Transmitter* p)
virtual void OnDeRegisteredAtTransmitter(class Transmitter *p)
{
}
};
typedef IReceiver* IReceiverPtr;
}
}

11
simgear/emesary/ITransmitter.hxx
View File

@@ -21,12 +21,11 @@
*---------------------------------------------------------------------------*/
#include <cstddef>
#include <simgear/structure/SGSharedPtr.hxx>
namespace simgear
{
namespace Emesary
{
typedef SGSharedPtr<IReceiver> IReceiverPtr;
/// Interface (base clasee) for a transmitter.
/// Transmits Message derived objects. Each instance of this class provides a
/// event/databus to which any number of receivers can attach to.
@@ -34,9 +33,9 @@ namespace simgear
{
public:
// Registers a recipient to receive message from this transmitter
virtual void Register(IReceiverPtr R) = 0;
virtual void Register(IReceiverPtr R) = 0;
// Removes a recipient from from this transmitter
virtual void DeRegister(IReceiverPtr R) = 0;
virtual void DeRegister(IReceiverPtr R) = 0;
//Notify all registered recipients. Stop when receipt status of abort or finished are received.
@@ -45,10 +44,10 @@ namespace simgear
// - Fail > message not handled. A status of Abort from a recipient will result in our status
// being fail as Abort means that the message was not and cannot be handled, and
// allows for usages such as access controls.
virtual ReceiptStatus NotifyAll(INotification& M) = 0;
virtual ReceiptStatus NotifyAll(INotificationPtr M) = 0;
/// number of recipients
virtual size_t Count() const = 0;
virtual size_t Count() const = 0;
};
}
}

351
simgear/emesary/Transmitter.hxx
View File

@@ -28,231 +28,232 @@
#include <vector>
#include <atomic>
#include <mutex>
#include <cstddef>
#include "ITransmitter.hxx"
namespace simgear
{
namespace Emesary
{
// Implementation of a ITransmitter
class Transmitter : public ITransmitter
{
protected:
namespace Emesary
{
// Implementation of a ITransmitter
class Transmitter : public ITransmitter
{
protected:
typedef std::vector<IReceiverPtr> RecipientList;
RecipientList recipient_list;
RecipientList new_recipient_list;
RecipientList deleted_recipient_list;
RecipientList recipient_list;
RecipientList new_recipient_list;
RecipientList deleted_recipient_list;
std::mutex _lock;
std::atomic<size_t> receiveDepth;
std::atomic<size_t> sentMessageCount;
std::atomic<size_t> recipientCount;
std::atomic<size_t> pendingDeletions;
std::atomic<size_t> pendingAdditions;
std::mutex _lock;
std::atomic<size_t> receiveDepth;
std::atomic<size_t> sentMessageCount;
std::atomic<size_t> recipientCount;
std::atomic<size_t> pendingDeletions;
std::atomic<size_t> pendingAdditions;
public:
Transmitter() :
receiveDepth(0),
sentMessageCount(0),
recipientCount(0),
pendingDeletions(0),
pendingAdditions(0)
{
}
virtual ~Transmitter()
{
}
// Registers an object to receive messsages from this transmitter.
// This object is added to the top of the list of objects to be notified. This is deliberate as
// the sequence of registration and message receipt can influence the way messages are processing
// when ReceiptStatus of Abort or Finished are encountered. So it was a deliberate decision that the
// most recently registered recipients should process the messages/events first.
virtual void Register(IReceiverPtr r)
{
std::lock_guard<std::mutex> scopeLock(_lock);
RecipientList::iterator deleted_location = std::find(deleted_recipient_list.begin(), deleted_recipient_list.end(), r);
if (deleted_location != deleted_recipient_list.end())
deleted_recipient_list.erase(deleted_location);
RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r);
if (location == recipient_list.end())
public:
Transmitter() :
receiveDepth(0),
sentMessageCount(0),
recipientCount(0),
pendingDeletions(0),
pendingAdditions(0)
{
RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r);
if (location == new_recipient_list.end()) {
new_recipient_list.insert(new_recipient_list.begin(), r);
pendingAdditions++;
}
virtual ~Transmitter()
{
}
// Registers an object to receive messsages from this transmitter.
// This object is added to the top of the list of objects to be notified. This is deliberate as
// the sequence of registration and message receipt can influence the way messages are processing
// when ReceiptStatus of Abort or Finished are encountered. So it was a deliberate decision that the
// most recently registered recipients should process the messages/events first.
virtual void Register(IReceiverPtr r)
{
std::lock_guard<std::mutex> scopeLock(_lock);
RecipientList::iterator deleted_location = std::find(deleted_recipient_list.begin(), deleted_recipient_list.end(), r);
if (deleted_location != deleted_recipient_list.end())
deleted_recipient_list.erase(deleted_location);
RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r);
if (location == recipient_list.end())
{
RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r);
if (location == new_recipient_list.end()) {
new_recipient_list.insert(new_recipient_list.begin(), r);
pendingAdditions++;
}
}
}
}
// Removes an object from receving message from this transmitter
virtual void DeRegister(IReceiverPtr r)
{
std::lock_guard<std::mutex> scopeLock(_lock);
// Removes an object from receving message from this transmitter
virtual void DeRegister(IReceiverPtr r)
{
std::lock_guard<std::mutex> scopeLock(_lock);
if (new_recipient_list.size())
{
RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r);
if (new_recipient_list.size())
{
RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r);
if (location != new_recipient_list.end())
new_recipient_list.erase(location);
if (location != new_recipient_list.end())
new_recipient_list.erase(location);
}
deleted_recipient_list.push_back(r);
pendingDeletions++;
}
deleted_recipient_list.push_back(r);
pendingDeletions++;
}
// this will purge the recipients that are marked as deleted
// it will only do this when the receive depth is zero - i.e. this
// notification is being sent out from outside a recipient notify.(because it is
// fine for a recipient to retransmit another notification as a result of receiving a notification)
//
// also we can quickly check to see if we have any pending deletions before doing anything.
void AddRemoveIFAppropriate()
{
std::lock_guard<std::mutex> scopeLock(_lock);
// this will purge the recipients that are marked as deleted
// it will only do this when the receive depth is zero - i.e. this
// notification is being sent out from outside a recipient notify.(because it is
// fine for a recipient to retransmit another notification as a result of receiving a notification)
//
// also we can quickly check to see if we have any pending deletions before doing anything.
void AddRemoveIFAppropriate()
{
std::lock_guard<std::mutex> scopeLock(_lock);
/// handle pending deletions first.
if (pendingDeletions > 0) {
/// handle pending deletions first.
if (pendingDeletions > 0) {
///
/// remove deleted recipients from the main list.
std::for_each(deleted_recipient_list.begin(), deleted_recipient_list.end(),
[this](IReceiverPtr r) {
///
/// remove deleted recipients from the main list.
std::for_each(deleted_recipient_list.begin(), deleted_recipient_list.end(),
[this](IReceiverPtr r) {
RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r);
if (location != recipient_list.end()) {
r->OnDeRegisteredAtTransmitter(this);
recipient_list.erase(location);
}
}
});
recipientCount -= pendingDeletions;
deleted_recipient_list.erase(deleted_recipient_list.begin(), deleted_recipient_list.end());
pendingDeletions = 0; // can do this because we are guarded
}
recipientCount -= pendingDeletions;
deleted_recipient_list.erase(deleted_recipient_list.begin(), deleted_recipient_list.end());
pendingDeletions = 0; // can do this because we are guarded
}
if (pendingAdditions) {
/// firstly remove items from the new list that are already in the list
std::for_each(recipient_list.begin(), recipient_list.end(),
[this](IReceiverPtr r) {
if (pendingAdditions) {
/// firstly remove items from the new list that are already in the list
std::for_each(recipient_list.begin(), recipient_list.end(),
[this](IReceiverPtr r) {
RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r);
if (location != new_recipient_list.end()) {
new_recipient_list.erase(location);
pendingAdditions--;
}
}
});
std::for_each(new_recipient_list.begin(), new_recipient_list.end(),
[this](IReceiverPtr r) {
std::for_each(new_recipient_list.begin(), new_recipient_list.end(),
[this](IReceiverPtr r) {
r->OnRegisteredAtTransmitter(this);
});
recipient_list.insert(recipient_list.begin(),
std::make_move_iterator(new_recipient_list.begin()),
std::make_move_iterator(new_recipient_list.end()));
recipient_list.insert(recipient_list.begin(),
std::make_move_iterator(new_recipient_list.begin()),
std::make_move_iterator(new_recipient_list.end()));
new_recipient_list.erase(new_recipient_list.begin(), new_recipient_list.end());
recipientCount += pendingAdditions;
pendingAdditions = 0;
new_recipient_list.erase(new_recipient_list.begin(), new_recipient_list.end());
recipientCount += pendingAdditions;
pendingAdditions = 0;
}
}
}
// Notify all registered recipients. Stop when receipt status of abort or finished are received.
// The receipt status from this method will be
// - OK > message handled
// - Fail > message not handled. A status of Abort from a recipient will result in our status
// being fail as Abort means that the message was not and cannot be handled, and
// allows for usages such as access controls.
virtual ReceiptStatus NotifyAll(INotification& M)
{
ReceiptStatus return_status = ReceiptStatus::NotProcessed;
// Notify all registered recipients. Stop when receipt status of abort or finished are received.
// The receipt status from this method will be
// - OK > message handled
// - Fail > message not handled. A status of Abort from a recipient will result in our status
// being fail as Abort means that the message was not and cannot be handled, and
// allows for usages such as access controls.
virtual ReceiptStatus NotifyAll(INotificationPtr M)
{
ReceiptStatus return_status = ReceiptStatus::NotProcessed;
auto v = receiveDepth.fetch_add(1, std::memory_order_relaxed);
if (v == 0)
AddRemoveIFAppropriate();
auto v = receiveDepth.fetch_add(1, std::memory_order_relaxed);
if (v == 0)
AddRemoveIFAppropriate();
sentMessageCount++;
sentMessageCount++;
bool finished = false;
bool finished = false;
size_t idx = 0;
do {
size_t idx = 0;
do {
if (idx < recipient_list.size()) {
IReceiverPtr R = recipient_list[idx++];
if (idx < recipient_list.size()) {
IReceiverPtr R = recipient_list[idx++];
if (R != nullptr)
{
Emesary::ReceiptStatus rstat = R->Receive(M);
switch (rstat)
{
case ReceiptStatus::Fail:
return_status = ReceiptStatus::Fail;
break;
if (R != nullptr)
{
Emesary::ReceiptStatus rstat = R->Receive(M);
switch (rstat)
{
case ReceiptStatus::Fail:
return_status = ReceiptStatus::Fail;
break;
case ReceiptStatus::Pending:
return_status = ReceiptStatus::Pending;
break;
case ReceiptStatus::Pending:
return_status = ReceiptStatus::Pending;
break;
case ReceiptStatus::PendingFinished:
return_status = rstat;
finished = true;
break;
case ReceiptStatus::PendingFinished:
return_status = rstat;
finished = true;
break;
case ReceiptStatus::NotProcessed:
break;
case ReceiptStatus::NotProcessed:
break;
case ReceiptStatus::OK:
if (return_status == ReceiptStatus::NotProcessed)
return_status = rstat;
break;
case ReceiptStatus::OK:
if (return_status == ReceiptStatus::NotProcessed)
return_status = rstat;
break;
case ReceiptStatus::Abort:
finished = true;
return_status = ReceiptStatus::Abort;
break;
case ReceiptStatus::Abort:
finished = true;
return_status = ReceiptStatus::Abort;
break;
case ReceiptStatus::Finished:
finished = true;
return_status = ReceiptStatus::OK;;
break;
}
}
case ReceiptStatus::Finished:
finished = true;
return_status = ReceiptStatus::OK;;
break;
}
}
}
else
break;
} while (!finished);
}
else
break;
} while (!finished);
receiveDepth--;
return return_status;
}
receiveDepth--;
return return_status;
}
// number of sent messages.
size_t SentMessageCount() const
{
return sentMessageCount;
}
// number of sent messages.
int SentMessageCount() const
{
return sentMessageCount;
}
// number of currently registered recipients
// better to avoid using the size members on the list directly as
// using the atomics is lock free and threadsafe.
virtual size_t Count() const
{
return (recipientCount + pendingAdditions) - pendingDeletions;
}
// ascertain if a receipt status can be interpreted as failure.
static bool Failed(ReceiptStatus receiptStatus)
{
//
// failed is either Fail or Abort.
// NotProcessed isn't a failure because it hasn't been processed.
return receiptStatus == ReceiptStatus::Fail
|| receiptStatus == ReceiptStatus::Abort;
}
};
}
// number of currently registered recipients
// better to avoid using the size members on the list directly as
// using the atomics is lock free and threadsafe.
virtual size_t Count() const
{
return (recipientCount + pendingAdditions) - pendingDeletions;
}
// ascertain if a receipt status can be interpreted as failure.
static bool Failed(ReceiptStatus receiptStatus)
{
//
// failed is either Fail or Abort.
// NotProcessed isn't a failure because it hasn't been processed.
return receiptStatus == ReceiptStatus::Fail
|| receiptStatus == ReceiptStatus::Abort;
}
};
}
}
#endif

2
simgear/emesary/notifications.hxx
View File

@@ -29,7 +29,7 @@ namespace simgear
class MainLoopNotification : public simgear::Emesary::INotification
{
public:
enum Type { Started, Stopped, Begin, End };
enum class Type { Started, Stopped, Begin, End };
MainLoopNotification(Type v) : _type(v) {}
virtual Type GetValue() { return _type; }

277
simgear/emesary/test_emesary.cxx
View File

@@ -5,11 +5,12 @@
#include <simgear_config.h>
#include <simgear/compiler.h>
#include <list>
#include <iostream>
#include <simgear/threads/SGThread.hxx>
#include <simgear/emesary/Emesary.hxx>
#include <list>
#include <simgear/misc/test_macros.hxx>
using std::cout;
using std::cerr;
@@ -17,7 +18,18 @@ using std::endl;
std::atomic<int> nthread {0};
std::atomic<int> noperations {0};
const int MaxIterations = 9999;
const int MaxIterationsBase = 9999999;
const int MaxIterationsThreaded = 90;
const int MaxIterationsBaseMultipleRecipients = 9999999;
const int num_threads = 220;
void summary(const SGTimeStamp& timeStamp, simgear::Emesary::Transmitter* transmitter, const char* id) {
printf("[%s]: invocations %zu\n", id, transmitter->SentMessageCount());
double elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f;
printf("[%s]: -> elapsed %d\n", id, timeStamp.elapsedMSec());
printf("[%s]: took %lf seconds which is %lf/sec\n", id, elapsed_seconds, transmitter->SentMessageCount() / elapsed_seconds);
}
class TestThreadNotification : public simgear::Emesary::INotification
{
@@ -29,51 +41,76 @@ public:
virtual const char* GetType () { return baseValue; }
};
class TestThreadRecipient : public simgear::Emesary::IReceiver
class TestThreadBaseRecipient : public simgear::Emesary::IReceiver
{
public:
TestThreadRecipient() : receiveCount(0)
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n)
{
return simgear::Emesary::ReceiptStatus::NotProcessed;
}
};
class TestThreadRecipient : public simgear::Emesary::IReceiver
{
simgear::Emesary::ITransmitter* transmitter;
public:
TestThreadRecipient(simgear::Emesary::ITransmitter* _transmitter, bool addDuringReceive)
: transmitter(_transmitter), addDuringReceive(addDuringReceive), receiveCount(0), ourType("TestThread")
{
r1 = new TestThreadBaseRecipient();
}
simgear::Emesary::IReceiver *r1;
std::string ourType;
bool addDuringReceive;
std::atomic<int> receiveCount;
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotification &n)
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n)
{
if (n.GetType() == (const char*)this)
if (ourType == n->GetType())
{
// SGSharedPtr<TestThreadBaseRecipient> r1 = new TestThreadBaseRecipient();
// Unused: TestThreadNotification *tn = dynamic_cast<TestThreadNotification *>(&n);
receiveCount++;
TestThreadNotification onwardNotification("AL");
simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(onwardNotification);
SGSharedPtr<TestThreadNotification> onwardNotification(new TestThreadNotification("AL"));
transmitter->NotifyAll(onwardNotification);
if (addDuringReceive) {
transmitter->Register(r1);
transmitter->NotifyAll(onwardNotification);
transmitter->DeRegister(r1);
}
return simgear::Emesary::ReceiptStatus::OK;
}
return simgear::Emesary::ReceiptStatus::OK;
}
};
class EmesaryTestThread : public SGThread
{
public:
EmesaryTestThread(simgear::Emesary::ITransmitter* transmitter, bool _addDuringReceive) : addDuringReceive(_addDuringReceive), transmitter(transmitter) {
}
protected:
simgear::Emesary::ITransmitter* transmitter;
bool addDuringReceive;
virtual void run() {
int threadId = nthread.fetch_add(1);
//System.Threading.Interlocked.Increment(ref nthread);
//var rng = new Random();
SGSharedPtr<TestThreadRecipient> r = new TestThreadRecipient;
TestThreadRecipient *r = new TestThreadRecipient(transmitter, addDuringReceive);
char temp[100];
sprintf(temp, "Notif %d", threadId);
printf("starting thread %s\n", temp);
TestThreadNotification tn((const char*)&r);
for (int i = 0; i < MaxIterations; i++)
SGSharedPtr<TestThreadNotification> tn(new TestThreadNotification("TestThread"));
for (int i = 0; i < MaxIterationsThreaded; i++)
{
simgear::Emesary::IReceiverPtr ir(r);
simgear::Emesary::GlobalTransmitter::instance()->Register(ir);
simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(tn);
simgear::Emesary::GlobalTransmitter::instance()->DeRegister(ir);
transmitter->Register(r);
transmitter->NotifyAll(tn);
transmitter->DeRegister(r);
//System.Threading.Thread.Sleep(rng.Next(MaxSleep));
noperations++;
}
@@ -86,14 +123,13 @@ class EmesaryTest
{
public:
void Emesary_MultiThreadTransmitterTest()
void Emesary_MultiThreadTransmitterTest(simgear::Emesary::ITransmitter *transmitter, bool addDuringReceive)
{
int num_threads = 12;
std::list<EmesaryTestThread*> threads;
for (int i = 0; i < num_threads; i++)
{
EmesaryTestThread *thread = new EmesaryTestThread();
EmesaryTestThread *thread = new EmesaryTestThread(transmitter, addDuringReceive);
threads.push_back(thread);
thread->start();
}
@@ -106,28 +142,211 @@ public:
void testEmesaryThreaded()
{
SGSharedPtr<TestThreadRecipient> r = new TestThreadRecipient;
simgear::Emesary::IReceiverPtr ir(r);
TestThreadNotification tn((const char*)r.ptr());
simgear::Emesary::GlobalTransmitter::instance()->Register(ir);
for (int i = 0; i < MaxIterations*MaxIterations; i++)
printf("Testing multithreaded operations\n");
simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance();
TestThreadRecipient *r = new TestThreadRecipient(globalTransmitter, false);
SGSharedPtr < TestThreadNotification> tn (new TestThreadNotification("TestThread"));
globalTransmitter->Register(r);
SGTimeStamp timeStamp;
timeStamp.stamp();
printf(" -- simple receive\n");
for (int i = 0; i < MaxIterationsThreaded; i++)
{
simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(tn);
globalTransmitter->NotifyAll(tn);
//System.Threading.Thread.Sleep(rng.Next(MaxSleep));
noperations++;
}
simgear::Emesary::GlobalTransmitter::instance()->DeRegister(ir);
printf("invocations %d\n", simgear::Emesary::GlobalTransmitter::instance()->SentMessageCount());
globalTransmitter->DeRegister(r);
printf("invocations %zu\n", globalTransmitter->SentMessageCount());
double elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f;
printf(" -> elapsed %d\n", timeStamp.elapsedMSec());
printf("took %lf seconds which is %lf/sec\n", elapsed_seconds , globalTransmitter->SentMessageCount() / elapsed_seconds);
EmesaryTest t;
t.Emesary_MultiThreadTransmitterTest();
t.Emesary_MultiThreadTransmitterTest(globalTransmitter, false);
elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f;
printf(" -> elapsed %d\n", timeStamp.elapsedMSec());
printf("took %lf seconds which is %lf/sec\n", elapsed_seconds, globalTransmitter->SentMessageCount() / elapsed_seconds);
}
void testEmesaryThreadedAddDuringReceive()
{
SGTimeStamp timeStamp;
timeStamp.stamp();
simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance();
EmesaryTest t;
t.Emesary_MultiThreadTransmitterTest(globalTransmitter, true);
summary(timeStamp, globalTransmitter, "ThreadedAddReceive");
}
//////////////////////
/// basic tests
class TestBaseNotification : public simgear::Emesary::INotification
{
public:
TestBaseNotification() : index(0) {}
int index;
virtual const char* GetType() { return "Test"; }
};
static size_t TestDerefNotification_destructor_count = 0;
class TestDerefNotification : public simgear::Emesary::INotification
{
public:
TestDerefNotification() {}
~TestDerefNotification() override {
TestDerefNotification_destructor_count ++;
}
virtual const char* GetType() { return "TestDerefNotification"; }
};
class TestBaseRecipient : public simgear::Emesary::IReceiver
{
public:
TestBaseRecipient() : receiveCount(0)
{
}
std::atomic<int> receiveCount;
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n)
{
if (n->GetType() == "Test")
{
auto tbn = dynamic_pointer_cast<TestBaseNotification>(n);
SG_CHECK_EQUAL(receiveCount, tbn->index);
receiveCount++;
return simgear::Emesary::ReceiptStatus::OK;
}
return simgear::Emesary::ReceiptStatus::OK;
}
};
void testEmesaryBase()
{
printf("Testing base functions\n");
simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance();
SGTimeStamp timeStamp;
timeStamp.stamp();
TestBaseRecipient *r = new TestBaseRecipient();
globalTransmitter->Register(r);
SG_CHECK_EQUAL(globalTransmitter->Count(), 1);
for (int i = 0; i < MaxIterationsBase; i++)
{
SGSharedPtr<TestBaseNotification> tn (new TestBaseNotification());
tn->index = i;
globalTransmitter->NotifyAll(tn);
//System.Threading.Thread.Sleep(rng.Next(MaxSleep));
noperations++;
}
globalTransmitter->DeRegister(r);
{
SGSharedPtr<TestBaseNotification> tn(new TestBaseNotification());
SG_CHECK_EQUAL(globalTransmitter->Count(), 0);
globalTransmitter->NotifyAll(tn);
SG_CHECK_EQUAL(globalTransmitter->Count(), 0);
}
summary(timeStamp, globalTransmitter, "base");
}
void testEmesaryMultipleRecipients()
{
printf("Testing multiple recipients\n");
simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance();
SGTimeStamp timeStamp;
timeStamp.stamp();
// create and register test recipients
std::vector<TestBaseRecipient*> recips;
for (int i = 0; i < 5; i++)
{
TestBaseRecipient *newRecipient = new TestBaseRecipient;
recips.push_back(newRecipient);
globalTransmitter->Register(newRecipient);
}
// check that TestThreadNotification are ignored
int rcount = globalTransmitter->SentMessageCount();
SG_CHECK_EQUAL(globalTransmitter->Count(), recips.size());
{
// send a bunch of notifications.
for (int i = 0; i < MaxIterationsBaseMultipleRecipients; i++)
{
SGSharedPtr<TestThreadNotification> ttn(new TestThreadNotification("TestThread"));
globalTransmitter->NotifyAll(ttn);
//System.Threading.Thread.Sleep(rng.Next(MaxSleep));
}
std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) {
SG_CHECK_EQUAL(0, r->receiveCount);
});
}
SG_CHECK_NE(rcount, globalTransmitter->SentMessageCount());
rcount = globalTransmitter->SentMessageCount();
{
SGSharedPtr < TestBaseNotification>tbn(new TestBaseNotification());
for (int i = 0; i < MaxIterationsBaseMultipleRecipients; i++)
{
tbn->index = i;
globalTransmitter->NotifyAll(tbn);
noperations++;
}
std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) {
SG_CHECK_EQUAL(r->receiveCount, MaxIterationsBaseMultipleRecipients);
});
}
// now degregister all
std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) {
globalTransmitter->DeRegister(r);
});
SG_CHECK_EQUAL(globalTransmitter->Count(), 0);
{
SGSharedPtr < TestThreadNotification >ttn(new TestThreadNotification("TestThread"));
globalTransmitter->NotifyAll(ttn);
}
SG_CHECK_EQUAL(globalTransmitter->Count(), 0);
summary(timeStamp, globalTransmitter, "base");
}
int main(int ac, char ** av)
{
{
SGSharedPtr<TestDerefNotification> tn(new TestDerefNotification());
}
SG_CHECK_EQUAL(TestDerefNotification_destructor_count, 1);
// should still be deleted when the inner pointer goes out of scope.
auto tn1 = new TestDerefNotification();
{
SGSharedPtr<TestDerefNotification> tn(tn1);
}
SG_CHECK_EQUAL(TestDerefNotification_destructor_count, 2);
testEmesaryBase();
testEmesaryMultipleRecipients();
testEmesaryThreaded();
testEmesaryThreadedAddDuringReceive();
std::cout << "all tests passed" << std::endl;
return 0;
}

7
simgear/nasal/cppbind/NasalEmesaryInterface.hxx
View File

@@ -71,10 +71,9 @@ namespace nasal
}
std::atomic<int> receiveCount;
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotification &n)
virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n)
{
simgear::Notifications::MainLoopNotification *mln = dynamic_cast<simgear::Notifications::MainLoopNotification *>(&n);
auto mln = dynamic_pointer_cast<simgear::Notifications::MainLoopNotification>(n);
if (mln) {
switch (mln->GetValue()) {
@@ -101,8 +100,8 @@ namespace nasal
}
return simgear::Emesary::ReceiptStatus::OK;
}
auto gccn = dynamic_pointer_cast<simgear::Notifications::NasalGarbageCollectionConfigurationNotification>(n);
auto *gccn = dynamic_cast<simgear::Notifications::NasalGarbageCollectionConfigurationNotification *>(&n);
if (gccn) {
CanWait = gccn->GetCanWait();
Active = gccn->GetActive();

10
simgear/nasal/cppbind/detail/to_nasal_helper.cxx
View File

@@ -30,15 +30,7 @@ namespace nasal
// create single instance of the main loop recipient for Nasal - this will self register at the
// global transmitter - and that's all that is needed to link up the background GC to the main
// loop in FG that will send out the MainLoop notifications.
//class NasalMainLoopRecipientSingleton : public simgear::Singleton<NasalMainLoopRecipient>
//{
//public:
// NasalMainLoopRecipientSingleton()
// {
// }
// virtual ~NasalMainLoopRecipientSingleton() {}
//};
SGSharedPtr<NasalMainLoopRecipient> mrl;
NasalMainLoopRecipient mrl;
//----------------------------------------------------------------------------
naRef to_nasal_helper(naContext c, const std::string& str)

8
simgear/structure/SGSharedPtr.hxx
View File

@@ -44,7 +44,7 @@ class SGWeakPtr;
/// ordinary pointers or SGWeakPtr's.
/// There is a very good description of OpenSceneGraphs ref_ptr which is
/// pretty much the same than this one at
/// http://dburns.dhs.org/OSG/Articles/RefPointers/RefPointers.html
/// https://web.archive.org/web/20061012014801/http://dburns.dhs.org/OSG/Articles/RefPointers/RefPointers.html
template<typename T>
class SGSharedPtr {
@@ -158,6 +158,12 @@ SGSharedPtr<T> static_pointer_cast(SGSharedPtr<U> const & r)
return SGSharedPtr<T>( static_cast<T*>(r.get()) );
}
template<class T, class U>
SGSharedPtr<T> dynamic_pointer_cast(SGSharedPtr<U> const& r)
{
return SGSharedPtr<T>(dynamic_cast<T*>(r.get()));
}
/**
* Compare two SGSharedPtr<T> objects for equality.
*